competitor_product_analysis/spider/cpa_agg.py

# -*- coding: utf-8 -*-
# @Time : 2022/12/1 14:46
# @Author : XuJiakai
# @File : cpa_agg
# @Software: PyCharm
import json
import time, queue
from threading import Thread
from utils.datetime_utils import datetime_format
from log import get_log
from sdk.WinhcAllClient import get_all_client
from utils.datetime_utils import get_ds, get_now, datetime_format_transform
from utils import map_2_json_str, json_path
from utils.base_utils import tuple_max
from utils.mysql_utils import insert_many
from utils.xxl_queue import xxl_queue
import re
import sys
import argparse
from project_const import TOPIC_NAME, MONGODB_NAME

date_part = re.compile('\\d{4}年\\d{2}月\\d{2}日')
all_client = get_all_client()

col = all_client.get_mongo_collection(MONGODB_NAME)
del_col = all_client.get_mongo_collection(MONGODB_NAME + '_del')
log = get_log('cpa_agg')

holo_client = all_client.get_holo_client(db='winhc_biz')
HOLO_TABLE_NAME = 'public.ads_waa_dim_info'


def get_max_data(data: list, key: str, exclude_product_name: list = ['winhc']):
    max_data = None
    for i in data:
        tmp_v = json_path(i, key)
        if tmp_v is None:
            continue
        product_name = i['competitor_product_name']

        if product_name in exclude_product_name:
            continue
            pass

        if max_data is None:
            max_data = (tmp_v, product_name)
        else:
            max_data = tuple_max(max_data, (tmp_v, product_name))

    if max_data is None:
        return None, None
    return max_data


def get_all_data_by_item(data: list, key):
    result_data = {}
    for i in data:
        result_data[i['competitor_product_name']] = json_path(i, key)
    return result_data


def data_transform(data: list):
    log.info('input data: {}'.format(data))
    deleted_key = [i['_id'] for i in data][0]
    deleted_key = deleted_key[:deleted_key.rfind('_')]

    base_info = data[0]['base_info']
    ds = get_ds()
    key_set = set()
    winhc_data = None
    for i in data:
        key_set = key_set | set(i['summary'].keys())
        key_set = key_set | set(i['latest_date'].keys())
        if i['competitor_product_name'] == 'winhc':
            winhc_data = i
            pass
        pass

    if winhc_data is None:
        return

    li = []

    winhc_spider_date = winhc_data['spider_date']

    holo_keys = None

    for i in key_set:
        tmp_json = base_info.copy()

        summary_max, summary_max_p_name = get_max_data(data, "$.summary." + i)
        latest_date_max, latest_date_max_p_name = get_max_data(data, "$.latest_date." + i)

        winhc_dim_num = json_path(winhc_data, '$.summary.' + i)

        latest_date_max = datetime_format(latest_date_max)

        winhc_dim_date = json_path(winhc_data, '$.latest_date.' + i)
        if winhc_dim_date is not None and winhc_dim_date == '':
            winhc_dim_date = None

        winhc_dim_date = datetime_format(winhc_dim_date)

        if (latest_date_max is None or latest_date_max == '') and (
                summary_max is None or summary_max == 0) and winhc_dim_date is None and (
                winhc_dim_num is None or winhc_dim_num == 0):
            # print('这个维度为空...', i, )
            continue
            pass

        if winhc_spider_date is None:
            winhc_spider_date = get_now()

        other_data = {
            "id": tmp_json['company_id'] + "_" + ds + "_" + i,
            "dim_name": i,
            "dim_max_num": summary_max,
            "dim_max_num_business_name": summary_max_p_name,
            "winhc_dim_num": winhc_dim_num,
            "dim_max_date": latest_date_max,
            "dim_max_date_business_name": latest_date_max_p_name,
            "winhc_dim_date": winhc_dim_date,
            "other_info": json.dumps({"summary": get_all_data_by_item(data, '$.summary.' + i),
                                      'latest_date': get_all_data_by_item(data, '$.latest_date.' + i)}),
            "update_time": winhc_spider_date,
            "create_time": winhc_spider_date,
            "ds": ds,
        }
        tmp_json.update(other_data)
        li.append(tmp_json)
        if holo_keys is None:
            holo_keys = list(tmp_json.keys())
        pass

    log.info('output data: {}'.format(li))
    insert_many(li, holo_keys, HOLO_TABLE_NAME, holo_client)
    del_num = 0
    try:
        del_col.insert_many(data, ordered=False)
        del_num = col.delete_many({"_id": {"$regex": "^" + deleted_key}}).deleted_count
    except:
        pass
    log.info("deleted mongo _id: {} , deleted count: {}".format(deleted_key, del_num))

    return li


q = queue.Queue(5000)


class Work(Thread):
    def run(self):
        while True:
            data_transform(q.get())


today_ds = get_ds()

scan_ds = today_ds[:-2]


def overwrite_handle(key, obj_list):
    if obj_list is None or len(obj_list) == 0:
        return
    _id = obj_list[0]['_id']
    if not key.startswith(today_ds) and len(obj_list) == 1:
        deleted_count = col.delete_one({'_id': _id}).deleted_count
        log.info(f"delete id: {_id} , {deleted_count}")
    else:
        # log.info(f"skip :{_id}")
        pass

    pass


def main(max_round: int = 2, interval_of_sed: int = 300):
    thread_num = 10

    for i in range(thread_num):
        w = Work()
        w.setDaemon(True)
        w.start()
        pass

    round_num = 0
    while True:
        round_num += 1
        ds = get_ds()
        # ds = '20221205'
        log.info('{},第{}遍轮循...'.format(ds, round_num))
        xxl_q = xxl_queue(pop_threshold=2, overwrite_handle=overwrite_handle)
        # for i in col.find({"_id": {"$regex": "^" + ds}}).batch_size(200):
        for i in col.find({"_id": {"$regex": "^" + ds}}).batch_size(200):
            # for i in col.find().batch_size(200):
            _id = i['_id']
            key = _id[:_id.rfind('_')]
            result = xxl_q.append(key=key, obj=i)
            if result:
                q.put(result)
            pass
        if round_num >= max_round:
            # sys.exit(0)
            break

        try:
            log.info('{},第{}遍轮循结束.'.format(ds, round_num))
            time.sleep(interval_of_sed)
            pass
        except:
            pass
        pass

    pass


# tmp_data = {
#     "_id": "20221214_0000b94de6aa5fba1f4daa0f2c353815_winhc",
#     "base_info": {
#         "cate_first": "租赁和商务服务业",
#         "cate_second": "商务服务业",
#         "cate_third": "旅行社及相关服务",
#         "city": "衢州市",
#         "company_id": "0000b94de6aa5fba1f4daa0f2c353815",
#         "company_name": "龙游县文化旅游发展有限公司",
#         "company_org_type": "有限责任公司（非自然人投资或控股的法人独资）",
#         "county": "龙游县",
#         "credit_code": "91330825573984254D",
#         "org_number": "573984254",
#         "province": "浙江省",
#         "reg_number": "330825000024620"
#     },
#     "competitor_product_name": "winhc",
#     "latest_date": {
#         "严重违法": None,
#         "公示催告": None,
#         "历史变更": "2022-04-15 00:00:00",
#         "双随机抽查": None,
#         "司法拍卖": None,
#         "土地公示": "2022-06-14 00:00:00",
#         "基本信息": "2021-12-17 00:00:00",
#         "失信信息": None,
#         "开庭公告": "2022-10-24 09:00:00",
#         "抽查检查": None,
#         "招投标": None,
#         "欠税公告": None,
#         "法院公告": None,
#         "环保处罚": None,
#         "税收违法": None,
#         "立案信息": "2019-08-21 00:00:00",
#         "终本案件": None,
#         "经营异常": None,
#         "行政处罚": None,
#         "行政许可": "2022-08-17 00:00:00",
#         "被执行人": None,
#         "裁判文书": "2020-05-29 00:00:00",
#         "诉前调解": None,
#         "询价评估": None,
#         "购地信息": "2022-06-27 00:00:00",
#         "送达公告": None,
#         "限制消费": None
#     },
#     "spider_date": "2022-12-14 10:09:43",
#     "summary": {
#         "主要成员": 2,
#         "企业年报": 9,
#         "历史主要成员": 1,
#         "历史变更": 32,
#         "历史对外投资": 0,
#         "历史股东信息": 0,
#         "商标": 97,
#         "土地公示": 11,
#         "对外投资": 21,
#         "股东信息": 1,
#         "行政许可": 10,
#         "裁判文书": 5,
#         "购地信息": 12
#     }
# }
#
# tmp_data_2 = {
#     "_id": "20221215_0000b94de6aa5fba1f4daa0f2c353815_winhc",
#     "base_info": {
#         "cate_first": "租赁和商务服务业",
#         "cate_second": "商务服务业",
#         "cate_third": "旅行社及相关服务",
#         "city": "衢州市",
#         "company_id": "0000b94de6aa5fba1f4daa0f2c353815",
#         "company_name": "龙游县文化旅游发展有限公司",
#         "company_org_type": "有限责任公司（非自然人投资或控股的法人独资）",
#         "county": "龙游县",
#         "credit_code": "91330825573984254D",
#         "org_number": "573984254",
#         "province": "浙江省",
#         "reg_number": "330825000024620"
#     },
#     "competitor_product_name": "qcc",
#     "latest_date": {
#         "严重违法": None,
#         "公示催告": None,
#         "双随机抽查": None,
#         "司法拍卖": None,
#         "失信信息": None,
#         "抽查检查": None,
#         "招投标": None,
#         "欠税公告": None,
#         "法院公告": None,
#         "环保处罚": None,
#         "税收违法": None,
#         "终本案件": None,
#         "经营异常": None,
#         "行政处罚": None,
#         "被执行人": None,
#         "诉前调解": None,
#         "询价评估": None,
#         "送达公告": None,
#         "限制消费": None
#     },
#     "spider_date": "2022-12-14 10:09:43",
#     "summary": {
#         "主要成员": 0,
#         "企业年报": 0,
#         "历史主要成员": 0,
#         "历史变更": 0,
#         "历史对外投资": 0,
#         "历史股东信息": 0,
#         "商标": 0,
#         "土地公示": 0,
#         "对外投资": 0,
#         "股东信息": 0,
#         "行政许可": 0,
#         "裁判文书": 0,
#         "购地信息": None
#     }
# }

# def test():
#     ds = get_ds()
#     for i in col.find({"_id": {"$regex": "^" + ds}}).batch_size(200):
#         print(map_2_json_str(i))
#         break
#         pass
#
#     data_transform([tmp_data,tmp_data_2])
#
#     pass


if __name__ == '__main__':
    # test()
    #
    log.info(f"input args: {sys.argv}")
    parser = argparse.ArgumentParser()
    parser.add_argument("-m", "--max-round", type=int, default=2, help='最大迭代轮次')
    parser.add_argument("-i", "--interval_of_sed", type=int, default=300, help='每轮间隔时间(秒)')
    args = parser.parse_args()

    main(max_round=args.max_round, interval_of_sed=args.interval_of_sed)
    pass