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, get_yesterday_ds
from utils import map_2_json_str, json_path
from utils.base_utils import tuple_max, get_str_intersection
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'
all_sdk = get_all_client()


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, rabbit_mq):
    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))

    if li is not None and len(li) > 0:
        # insert_many(li, holo_keys, HOLO_TABLE_NAME, holo_client)
        li = [tuple(i.values()) for i in li]
        for i in li:
            rabbit_mq.send_by_fanout("cpa_insert_holo", json.dumps(i, ensure_ascii=False).encode())
            pass

    del_num = 0
    try:
        del_num = col.delete_many({"_id": {"$regex": "^" + deleted_key}}).deleted_count
        del_col.insert_many(data, ordered=False)
    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):
        r_sdk = all_sdk.get_rabbit_mq_sdk()
        while True:
            data_transform(q.get(), r_sdk)


today_ds = get_ds()
yesterday_ds = get_yesterday_ds()
# yesterday_ds = '20221231'

# scan_ds = today_ds[:-2]
scan_ds = get_str_intersection(today_ds, yesterday_ds)


# scan_ds = '2022'


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


import tqdm


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

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

    round_num = 0
    while True:
        round_num += 1

        print('{},第{}遍轮循...'.format(scan_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 tqdm.tqdm(col.find({"_id": {"$regex": "^" + scan_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:
            print('{},第{}遍轮循结束.'.format(scan_ds, round_num))
            time.sleep(interval_of_sed)
            pass
        except:
            pass
        pass

    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)

    while not q.empty():
        log.info(f"遍历未结束，队列剩余：{q.qsize()}")
        time.sleep(300)
        pass

    log.info(f"遍历完成，队列剩余：{q.qsize()}")
    pass