spider_work/yueka_spider/yk_new_daily_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.10.8
# Date   : 2025/12/22 10:44
import time
from mysql_pool import MySQLConnectionPool
from settings import *


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_all_sold_one_page(log, page: int, last_id, lastSalePrice):
    """
    获取指定页面的已售数据
    :param log: logger对象
    :param page: 页码
    :param last_id: last_id
    :param lastSalePrice: lastSalePrice
    :return: 该页的数据, totalPage, total
    """
    log.info(f"Starting < get_all_sold_one_page > to fetch page {page}")
    url = "https://api.joycard.xyz/api/front/c/product/productShowList"
    #      https://api.joycard.xyz/api/front/c/product/productShowList
    data = {
        "lastId": last_id,
        "lastSalePrice": lastSalePrice,
        "limit": 20,
        "openMode": "",
        "page": page,
        "saleStatus": "2",
        "sort": "0"
    }
    # {
    #   "lastId": 0,
    #   "lastSalePrice": "",
    #   "limit": 20,
    #   "openMode": "",
    #   "page": 1,
    #   "saleStatus": "1",
    #   "sort": "0"
    # }
    
    try:
        # data = json.dumps(data, separators=(',', ':'))
        response = make_request(log, 'POST', url, data=data)
        # print(response)
        if response:
            items = response["data"]["list"]
            total_page = response["data"]["totalPage"]
            total = response["data"]["total"]
            log.info(f"Successfully fetched page {page}: {len(items)} items")
            return items, total_page, total
        else:
            return [], 0, 0
    except requests.exceptions.RequestException as e:
        log.error(f"Error fetching page {page}: {e}")
        raise e
    except ValueError as e:
        log.error(f"Error parsing JSON for page {page}: {e}")
        raise e


def get_shop_detail(log, shop_id):
    """
    获取店铺详情信息
    :param log:
    :param shop_id:
    :return:
    """
    log.info(f"Start fetching shop {shop_id}")
    url = f"https://api.joycard.xyz/api/front/c/merchant/{shop_id}"
    try:
        response = make_request(log, 'GET', url)
        if response:
            item = response["data"]
            fans_num = item.get("fansNum")
            group_num = item.get("salesQuantity")
            create_time = item.get("createTime")
            log.info(f"Successfully fetched shop {shop_id}")
            return fans_num, group_num, create_time
        else:
            return None, None, None
    except Exception as e:
        log.error(f"Error fetching shop {shop_id}: {e}")
        return None, None, None


def parse_shop_items(log, items, sql_pool):
    if not items:
        log.warning(f"Warning {inspect.currentframe().f_code.co_name}: No items found")
        return
    
    info_list = []
    for item in items:
        shop_id = item.get("merchantCode")
        
        fans_num, group_num, create_time = get_shop_detail(log, shop_id)
        
        shop_name = item.get("merchantName")
        shop_info_dict = {
            "shop_id": shop_id,
            "shop_name": shop_name,
            "fans_num": fans_num,
            "group_num": group_num,
            "create_time": create_time
        }
        info_list.append(shop_info_dict)
    
    # 插入数据
    if info_list:
        sql_pool.insert_many(table="yueka_shop_record", data_list=info_list, ignore=True)


def get_product(log, items, sql_pool, last_product_id):
    if not items:
        log.warning(f"Warning {inspect.currentframe().f_code.co_name}: No items found")
        return True
    
    should_stop = False
    info_list = []
    for item in items:
        product_id = item.get("code")
        if not product_id:
            log.warning(f"Warning {inspect.currentframe().f_code.co_name}: No product_id found")
            continue
        info_list.append(
            {
                "product_id": product_id,
            }
        )
        
        # 判断是否是昨天的最后一条id, 如果是  则停止翻页 / 只有当 last_product_id 不为 None 时才判断是否停止翻页
        if last_product_id and product_id == last_product_id:
            log.info(
                f"----------------- The product_id {product_id} is the last product_id:{last_product_id}, stop fetching -----------------")
            should_stop = True
            
    # 插入数据
    if info_list:
        sql_pool.insert_many(table="yueka_product_record", data_list=info_list, ignore=True)
    
    # 如果 items 数量小于 20，说明已经到达最后一页
    if len(items) < 20:
        log.info(
            f"----------------- {len(items)} items found, less than 20, stop fetching -----------------")
        should_stop = True
    
    return should_stop


def get_all_sold_data(log, sql_pool, last_product_id):
    """
    获取 全部类别的已售数据
    :param sql_pool: MySQL连接池对象
    :param log: logger对象
    :param last_product_id: last_product_id，如果为 None 则表示从头开始采集
    """
    page = 1
    max_page = 500
    last_id = 0
    lastSalePrice = ''
    
    while page <= max_page:
        items, total_page, total = get_all_sold_one_page(log, page, last_id, lastSalePrice)
        if not items:
            break
        
        # 处理 items 数据
        parse_shop_items(log, items, sql_pool)
        
        stop_page = get_product(log, items, sql_pool, last_product_id)
        if stop_page:
            log.info(
                f"----------------- The product_id {last_product_id} is the last product_id, stop fetching -----------------")
            break
        
        # 更新lastId为最后一条的userId
        last_id = items[-1].get("id")
        lastSalePrice = items[-1].get("unitPriceStr")
        if not last_id:
            log.error("API response missing userId in last item, cannot paginate")
            break
        if not lastSalePrice:
            log.error("API response missing lastSalePrice in last item, cannot paginate")
            break
        
        page += 1
        
        log.info(f"Finished fetching all data. Total pages: {total_page}, total items: {total}")


@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
def yueka_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:
        token = sql_pool.select_one("SELECT token FROM yueka_token")
        token = token[0]
        try:
            # 获取最后一条pid的数据
            last_product_id_result = sql_pool.select_one(
                "SELECT product_id FROM yueka_product_record ORDER BY finish_time DESC LIMIT 1")
            # 如果表中没有数据，last_product_id_result 为 None
            last_product_id = last_product_id_result[0] if last_product_id_result else None
            
            get_all_sold_data(log, sql_pool, last_product_id)
        except Exception as e:
            log.error(f"Error fetching last_product_id: {e}")
        time.sleep(5)
        
        # 获取商品详情
        try:
            get_product_detail_list(log, sql_pool, token)
        except Exception as e:
            log.error(f"Error fetching product_detail_list: {e}")
        time.sleep(5)
        
        # 获取商品玩家
        try:
            get_players(log, sql_pool, token)
        except Exception as e:
            log.error(f"Error fetching players: {e}")
        time.sleep(5)
        
        # 获取拆卡报告
        try:
            get_reports(log, sql_pool, token)
        except Exception as e:
            log.error(f"Error fetching reports: {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} 运行结束,等待下一轮的采集任务............')


if __name__ == '__main__':
    yueka_main(logger)
    # get_all_sold_one_page(logger, 1, 0, '')