shop_live_scraper/live_promoti...

#!/usr/bin/python3
# coding=utf-8
# -*- coding: utf-8 -*-

import time
import json
import threading
import sys
import random

from rds_model.rds_live_commodity_detail_request_list import RdsLiveCommodityDetailRequestList
from libs.live_commodity_detail_v1 import LiveCommodityDetailV1
from libs.mysql_dy_live_ifsolt import MysqlDyLiveCommodity
from log.print_log import PrintLog

start_time = time.time()

def scrape(request_data, heat):
    rds_list = RdsLiveCommodityDetailRequestList()
    request_split = request_data.split('@')

    if len(request_split) < 5:
        print('数据格式不对' + request_data)
        sys.exit(0)

    promotion_id = request_split[0]
    product_id = request_split[1]
    uid = request_split[2]
    room_id = request_split[3]
    exec_time = int(request_split[4])

    if int(time.time()) < exec_time:
        sp_ns = exec_time - int(time.time())
        time.sleep(sp_ns)


    PrintLog.print(
        time.strftime("%H:%M:%S", time.localtime()) + ' '
           + time.strftime("%H:%M:%S", time.localtime(exec_time)) + ' '
           + str(heat) + ' ' + promotion_id
    )

    try:

        commodity_detail = LiveCommodityDetailV1.get_data(product_id)

        if (commodity_detail is None) or (commodity_detail == '') or ('real_sell_num' in commodity_detail)==False :
            rds_list.record_v1_score(0)
            PrintLog.print(
                time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ' 数据为空 ' + promotion_id
            )

            if heat==0:
                if random.randint(0, 100) > 0:
                    rds_list.add_request_params(request_data, heat)
                    print('首次失败重新插入 ' + request_data)
                    sys.exit(0)

            if heat>0:
                if random.randint(0, 9) > 0:
                    rds_list.add_request_params(request_data, heat)
                    print(str(heat) + '非首次失败重新插入 ' + request_data)
                    sys.exit(0)

        if heat < 2:
            goodsDb = MysqlDyLiveCommodity()
            pix = str(uid)[-1]
            goodsInfo = goodsDb.get_goods_info(room_id, product_id, pix)

            dataid,sold_out_at,revise_price,created_at = goodsInfo

            if revise_price is not None:
                sys.exit(0)

            if sold_out_at == '0000-00-00 00:00:00' or sold_out_at == '1970-01-01 08:00:00':
                add_t = 300
                news_time = int(time.time()) + add_t
                next_rds = promotion_id + '@' + product_id + '@' + uid + '@' + room_id + '@' + str(news_time) 
                rds_list.push_middle_list(next_rds)
            else :
                add_t = 1200
                news_time = int(time.time()) + add_t
                next_rds = promotion_id + '@' + product_id + '@' + uid + '@' + room_id + '@' + str(news_time) 
                rds_list.push_last_list(next_rds)

        if (commodity_detail is None) or (commodity_detail == '') or ('real_sell_num' in commodity_detail)==False:
            sys.exit(0)

        data = json.dumps({
            "data": commodity_detail,
            "extra": {
                'room_id': room_id,
                'promotion_id': promotion_id,
                'product_id': product_id,
                'uid': uid,
                'heat': heat
            }
        })

        rds_list.record_v1_score(1)
        rds_list.push_v1_data_list(data)
    except Exception as e:
        rds_list.record_v1_score(0)
        rds_list.add_request_params(request_data, heat)
        PrintLog.print(
            time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + '抛出异常！' + str(e) + '\n'
            + request_data
        )

    sys.exit(0)


if __name__ == "__main__":

    import warnings
    warnings.filterwarnings("ignore")

    heat = int(sys.argv[1])
    threading_count = int(sys.argv[2])
    rds = RdsLiveCommodityDetailRequestList()

    print(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ' ' + str(heat) + ' 开始执行，商品销量队列长度：' + str(rds.get_len(heat)))

    while True:
        sys.stdout.flush()

        # 减去主线程
        active_count = threading.active_count() - 1

        increment = threading_count - active_count

        while increment > 0:

            sys.stdout.flush()

            request_data = rds.get_request_params(heat)

            if request_data is None:
                time.sleep(0.1)
                break

            task = threading.Thread(target=scrape, args=(request_data, heat))
            task.start()  # 准备就绪,等待cpu执行
            increment = increment - 1

        current_time = time.time()

        if current_time - start_time > 1800:
            print(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ' 主方法执行终止')
            sys.exit(0)

        time.sleep(0.01)