spider_work/qiandao_spider/qd_card_spider/福袋补充.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.10.8
# Date   : 2025/6/30 17:59
# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.10.8
# Date   : 2025/6/19 14:12
import time
from datetime import datetime

import utils
import random
import inspect
import pymysql
import schedule
from loguru import logger
from mysql_pool import MySQLConnectionPool
from tenacity import retry, stop_after_attempt, wait_fixed


logger.remove()
logger.add("./logs/bag_{time:YYYYMMDD}.log", encoding='utf-8', rotation="00:00",
           format="[{time:YYYY-MM-DD HH:mm:ss.SSS}] {level} {message}",
           level="DEBUG", retention="7 day")


def after_log(retry_state):
    """
    retry 回调
    :param retry_state: RetryCallState 对象
    """
    # 检查 args 是否存在且不为空
    if retry_state.args and len(retry_state.args) > 0:
        log = retry_state.args[0]  # 获取传入的 logger
    else:
        log = logger  # 使用全局 logger

    if retry_state.outcome.failed:
        log.warning(
            f"Function '{retry_state.fn.__name__}', Attempt {retry_state.attempt_number} Times")
    else:
        log.info(f"Function '{retry_state.fn.__name__}', Attempt {retry_state.attempt_number} succeeded")


def get_luckybag_list(log, sql_pool, sql_luckybag_list):
    page = 1
    max_page = 500
    total_count = 0
    while page <= max_page:
        len_item, totalCount = get_luckybag_one_page(log, page, sql_pool,sql_luckybag_list)

        if len_item < 10:
            log.debug(f"--------------- page {page}, len_item: {len_item} totalCount: {totalCount} ---------------")
            break

        total_count += len_item
        if total_count >= int(totalCount):
            log.debug(f"total_count: {total_count} totalCount: {totalCount}")
            break

        page += 1

        time.sleep(random.uniform(0.1, 1))


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_luckybag_one_page(log, page, sql_pool,sql_luckybag_list):
    log.debug(f" {inspect.currentframe().f_code.co_name}  for page:{page}.....")
    url = "https://api.qiandao.cn/c2c-web/v1/luckybag/list"
    params = {
        "limit": "10",
        # "offset": "0",
        "offset": str((page - 1) * 10),
        "hostId": "228946998408215136",
        "orderBy": "updatedAt.desc",
        "filterFollow": "false",
        "tagIds": "1429578",
        "typeIds": "1405184,1132249,1000375,49695,1443553,1002041"
    }
    resp_json = utils.request_get_data(log, url, params)
    if not resp_json:
        log.error("get_luckybag_one_page error")
        raise RuntimeError("get_luckybag_one_page error")

    rows = resp_json.get("data", {}).get("rows", [])
    total_count = resp_json.get("data", {}).get("count", 0)
    try:
        parse_luckybag_data(log, rows, sql_pool,sql_luckybag_list)
    except Exception as e:
        log.error(f"parse_luckybag_data error: {e}")

    return len(rows), total_count


def parse_luckybag_data(log, items, sql_pool,sql_luckybag_list):
    # info_list = []
    for item in items:
        luckybag_id = item.get("id")
        if luckybag_id in sql_luckybag_list:
            log.info(f"luckybag_id:{luckybag_id} is exist, skip.......")
            continue

        name = item.get("name")
        original_price = item.get("price")
        ice_breaking_price = item.get("iceBreakingPrice")
        seller_id = item.get("sellerId")
        seller_name = item.get("sellerName")
        remain_count = item.get("remainCount")  # 剩下数量
        total_count = item.get("totalCount")  # 总数

        createdAt = item.get("createdAt")
        created_at = utils.transform_ms(log, createdAt)

        updatedAt = item.get("updatedAt")
        updated_at = utils.transform_ms(log, updatedAt)

        info_dict = {
            "luckybag_id": luckybag_id,
            "luckybag_name": name,
            "original_price": original_price,
            "ice_breaking_price": ice_breaking_price,
            "seller_id": seller_id,
            "seller_name": seller_name,
            "remain_count": remain_count,
            "total_count": total_count,
            "created_at": created_at,
            "updated_at": updated_at
        }
        # print(info_dict)
        # info_list.append(info_dict)
        try:
            sql_pool.insert_one_or_dict(
                table="qiandao_luckybag_list_record",
                data=info_dict
            )
        except Exception as e:
            log.error(f" {inspect.currentframe().f_code.co_name} 插入失败:", e)

        sql_luckybag_list.append(luckybag_id)

    # if info_list:
    #     try:
    #         sql_pool.insert_many(table="qiandao_luckybag_list_record", data_list=info_list)
    #     except pymysql.err.IntegrityError as e:
    #         if "Duplicate entry" in str(e):
    #             logger.warning("存在重复的 luckybag_id，跳过插入")
    #         else:
    #             raise e


# -------------------------------------------------------------------------------------------------------------------


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_luckybag_detail(log, luckyBagId: str):
    """
    获取福袋详情
    :param luckyBagId: luckyBagId
    :param log: logger对象
    """
    url = "https://api.qiandao.cn/c2c-web/v1/luckybag"
    params = {
        # "luckyBagId": "876352952506283967"
        "luckyBagId": luckyBagId
    }
    resp_json = utils.request_get_data(log, url, params)
    if not resp_json:
        log.error("get_luckybag_detail error")
        raise RuntimeError("get_luckybag_detail error")
    return resp_json


def parse_luckybag_detail_basic_info_data(log, resp_json, sql_pool, luckyBagId):
    try:
        basic_info = resp_json.get("data", {}).get("basicInfo", {})
        sell_end_time = basic_info.get("sellEndTime")
        sell_end_time = utils.transform_ms(log, sell_end_time)
        # moderate_status = basic_info.get("moderateStatus")
        status = basic_info.get("status")

        # 更新 qiandao_luckybag_list_record 表
        updata_dict = {
            "sell_end_time": sell_end_time,
            # "moderate_status": moderate_status,
            "status": status
        }
        # print(f"updata_dict:{updata_dict}")
        sql_pool.update_one_or_dict(
            table="qiandao_luckybag_list_record",
            data=updata_dict,
            condition={
                "luckybag_id": luckyBagId
            }
        )
    except Exception as e:
        log.error(f"{inspect.currentframe().f_code.co_name} for basic_info error: {e}")


def parse_luckybag_detail_product_list_data(log, resp_json, sql_pool, luckybag_id):
    try:
        product_list = resp_json.get("data", {}).get("products", [])
        # info_product_list = []
        for product in product_list:
            product_id = product.get("id")
            product_name = product.get("name")
            image = product.get("image")
            reference_price = product.get("referencePrice")  # 参考价格
            spu_id = product.get("spuId")
            spec_list = product.get("specValues", [{}])
            spec_name_list = [spec.get("name") for spec in spec_list]
            spec_name = "".join(spec_name_list)  # 规格名称

            spec_value_list = [spec.get("value") for spec in spec_list]
            spec_value = "".join(spec_value_list)  # 规格值

            remain_count = product.get("remainCount")  # 剩余数量
            total_count = product.get("totalCount")  # 总数

            product_dict = {
                "luckybag_id": luckybag_id,
                "product_id": product_id,
                "product_name": product_name,
                "image": image,
                "reference_price": reference_price,
                "spu_id": spu_id,
                "spec_name": spec_name,
                "spec_value": spec_value,
                "remain_count": remain_count,
                "total_count": total_count
            }
            # print(f"product_dict:{product_dict}")
            try:
                sql_pool.insert_one_or_dict(table="qiandao_luckybag_products_record", data=product_dict)
                sql_pool.update_one_or_dict(
                    table="qiandao_luckybag_list_record",
                    data={
                        "detail_state": 1
                    },
                    condition={
                        "luckybag_id": luckybag_id
                    }
                )
            except Exception as e:
                log.error(f"{inspect.currentframe().f_code.co_name} for product_list error: {e}")
            # info_product_list.append(product_dict)

        # if info_product_list:
        #     try:
        #         sql_pool.insert_many(table="qiandao_luckybag_products_record", data_list=info_product_list)
        #         sql_pool.update_one_or_dict(
        #             table="qiandao_luckybag_list_record",
        #             data={
        #                 "detail_state": 1
        #             },
        #             condition={
        #                 "luckybag_id": luckybag_id
        #             }
        #         )
        #     except pymysql.err.IntegrityError as e:
        #         if "Duplicate entry" in str(e):
        #             logger.warning("存在重复的 product_id，跳过插入")
        #         else:
        #             raise e

    except Exception as e:
        log.error(f"{inspect.currentframe().f_code.co_name} for product_list error: {e}")


def parse_luckybag_detail_order_list_data(log, resp_json, sql_pool, luckybag_id):
    try:
        order_list = []
        # 先查询该福袋 购买数据的最大时间
        max_luckybay_time = sql_pool.select_one(
            query="SELECT MAX(order_created_at) FROM qiandao_luckybag_orders_record WHERE luckybag_id = %s",
            args=(luckybag_id,))
        max_luckybay_time = max_luckybay_time[0] if max_luckybay_time else None
        records = resp_json.get("data", {}).get("records", [])
        for record in records:
            product_id = record.get("productId")
            # product_name = record.get("productName")
            serial_number = record.get("serialNumber")  # 序列号
            order_id = record.get("orderId")

            trading_time = record.get("orderCreatedAt")
            trading_time_str = utils.transform_ms(log, trading_time)
            if not trading_time_str:
                log.debug(f"Invalid trading time: {trading_time_str}")
                continue  # 跳过无效时间

            # 字符串 -> datetime
            trading_time = datetime.strptime(trading_time_str, "%Y-%m-%d %H:%M:%S")

            if max_luckybay_time and trading_time <= max_luckybay_time:
                log.debug(f"trading_time: {trading_time_str} <= max_luckybay_time: {max_luckybay_time}, 跳过旧数据")
                break


            buyerNickname = record.get("buyerNickname")
            order_dict = {
                "luckybag_id": luckybag_id,
                "product_id": product_id,
                "order_id": order_id,
                "serial_number": serial_number,
                "order_created_at": trading_time_str,
                "buyer_nickname": buyerNickname
            }
            # print(f"order_dict:{order_dict}")
            # try:
            #     sql_pool.insert_one_or_dict(table="qiandao_luckybag_orders_record", data=order_dict)
            #     sql_pool.update_one_or_dict(
            #         table="qiandao_luckybag_list_record",
            #         data={
            #             "order_state": 1
            #         },
            #         condition={
            #             "luckybag_id": luckybag_id
            #         }
            #     )
            # except Exception as e:
            #     log.error(f"{inspect.currentframe().f_code.co_name} for order_list error: {e}")

            order_list.append(order_dict)

        if order_list:
            try:
                sql_pool.insert_many(table="qiandao_luckybag_orders_record", data_list=order_list)
                sql_pool.update_one_or_dict(
                    table="qiandao_luckybag_list_record",
                    data={
                        "order_state": 1
                    },
                    condition={
                        "luckybag_id": luckybag_id
                    }
                )
            except pymysql.err.IntegrityError as e:
                if "Duplicate entry" in str(e):
                    logger.warning("存在重复的 order_id，跳过插入")
                else:
                    # raise e
                    log.warning(f"{str(e)[:200]}")
                    sql_pool.update_one_or_dict(
                        table="qiandao_luckybag_list_record",
                        data={
                            "order_state": 2
                        },
                        condition={
                            "luckybag_id": luckybag_id
                        }
                    )

    except Exception as e:
        log.error(f"{inspect.currentframe().f_code.co_name} for order_list error: {e}")


def get_detail_main(log, sql_pool):
    """
    1. 将 qiandao_luckybag_list_record 表中 detail_state = 0 的数据进行更新
        插入 product list 数据 共用一个状态

    2. 将 qiandao_luckybag_list_record 表中 order_state = 0 并且 status 不是 'ON_SALE' 的数据进行更新

    :param log: logger对象
    :param sql_pool: MySQLConnectionPool对象
    """
    sql_detail_list = sql_pool.select_all("SELECT luckybag_id FROM qiandao_luckybag_list_record WHERE order_state = 0")
    sql_detail_list = [item[0] for item in sql_detail_list]

    for luckyBagId in sql_detail_list:
        try:
            resp_json = get_luckybag_detail(log, luckyBagId)
        except Exception as e:
            log.error(f"{inspect.currentframe().f_code.co_name} for sql_detail_list: {luckyBagId}, error: {e}")
            resp_json = {}

        # parse_luckybag_detail_basic_info_data(log, resp_json, sql_pool, luckyBagId)
        # parse_luckybag_detail_product_list_data(log, resp_json, sql_pool, luckyBagId)
        parse_luckybag_detail_order_list_data(log, resp_json, sql_pool, luckyBagId)

    # ---------------------------------------------------------------------------------------
    # order 信息待后续分析 status 后 再运行
    # sql_order_list = sql_pool.select_all("SELECT luckybag_id FROM qiandao_luckybag_list_record WHERE order_state = 0 AND status != 'ON_SALE'")
    # sql_order_list = [item[0] for item in sql_order_list]
    #
    # for luckyBagId2 in sql_order_list:
    #     try:
    #         resp_json = get_luckybag_detail(log, luckyBagId2)
    #     except Exception as e:
    #         log.error(f"{inspect.currentframe().f_code.co_name} for sql_order_list: {luckyBagId2}, error: {e}")
    #         resp_json = {}
    #
    #     parse_luckybag_detail_order_list_data(log, resp_json, sql_pool, luckyBagId2)


@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
def qd_lb_main(log):
    """
    主函数
    :param log: logger对象
    """
    log.info(
        f'开始运行 {inspect.currentframe().f_code.co_name} 爬虫任务....................................................')

    # 配置 MySQL 连接池
    sql_pool = MySQLConnectionPool(log=log)
    if not sql_pool.check_pool_health():
        log.error("数据库连接池异常")
        raise RuntimeError("数据库连接池异常")

    try:
        # 获取 福袋 列表
        log.info("-------------------------------------- 开始 获取 福袋 列表 --------------------------------------")
        # sql_luckybag_list = sql_pool.select_all("SELECT luckybag_id FROM qiandao_luckybag_list_record")
        # sql_luckybag_list = [item[0] for item in sql_luckybag_list]
        # try:
        #     get_luckybag_list(log, sql_pool,sql_luckybag_list)
        # except Exception as e:
        #     log.error(f"Error fetching get_luckybag_list: {e}")
        #
        # sql_luckybag_list.clear()
        #
        # time.sleep(5)

        # 获取 福袋 详情
        log.info("-------------------------------------- 开始获取 福袋 详情 --------------------------------------")
        try:
            get_detail_main(log, sql_pool)
        except Exception as e:
            log.error(f"Error fetching get_detail_main: {e}")

    except Exception as e:
        log.error(f'{inspect.currentframe().f_code.co_name} error: {e}')
    finally:
        log.info(f'爬虫程序 {inspect.currentframe().f_code.co_name} 运行结束,等待下一轮的采集任务............')

        # EmailSender().send(subject="【千岛 - 爬虫通知】今日任务已完成",
        #                    content="数据采集和处理已全部完成，请查收结果。\n\n                  ------ 来自 Python 爬虫系统。")




if __name__ == '__main__':
    qd_lb_main(logger)