# -*- coding: utf-8 -*- # Author : Charley # Python : 3.12.10 # Date : 2026/08/03 """得卡 DECA 商家历史成交采集爬虫(独立任务)。 抓取每个商家的「历史成交」groupbuy/merchant/sold-list,INSERT IGNORE 写入 deca_sold_record。 历史成交为终态、只增不改,用 product_code 唯一键去重增量采集。 deca_sold_record 预留 report_state / replay_state,供后续通过 product_code / live_id 采集「拆卡报告」与「视频回放」。 鉴权:本接口需 Authorization Bearer(需 token),复用 deca_daily_spider 的登录/续期逻辑 (优先 refreshToken 续期,避开阿里云验证码)。签名规则同样复用。 """ import sys import time import schedule from loguru import logger from tenacity import retry, stop_after_attempt, wait_fixed from mysql_pool import MySQLConnectionPool import deca_daily_spider as deca # 复用 do_request / ensure_token / make_signature / after_log # 覆盖 deca 导入时装的 logger handler,本任务日志独立到 sold_*.log logger.remove() logger.add("./logs/sold_{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") PAGE_SIZE = 20 # 历史成交每页条数(抓包实测 20) MAX_PAGES = 200 # 单商家历史成交翻页保护上限 def parse_sold(item: dict) -> dict | None: """把历史成交列表项解析成 deca_sold_record 一行。 Args: item (dict): sold-list 的 data.list 项(结构同在售商品列表)。 Returns: dict | None: 与 deca_sold_record 列对应的数据字典;无 code 时返回 None。 """ code = item.get("code") if not code: return None m = item.get("merchant") or {} return { "product_code": code, "merchant_user_id": str(m.get("merchantUserID")) if m.get("merchantUserID") else None, "merchant_name": m.get("merchantName"), "title": item.get("title"), "card_product_title": item.get("cardProductTitle"), "cover_image_url": item.get("coverImageUrl"), "unit_price": item.get("unitPrice"), "min_unit_price": item.get("minUnitPrice"), "max_unit_price": item.get("maxUnitPrice"), "card_count": item.get("cardCount"), "sold_count": item.get("soldCount"), "available_stock": item.get("availableStock"), "groupbuy_status": item.get("groupbuyStatus"), "groupbuy_status_name": item.get("groupbuyStatusName"), "play_type": item.get("playType"), "live_id": item.get("liveId"), "completed_at": item.get("completedAt") or None, # 成交完成时间 "publicity_at": item.get("publicityAt") or None, } def get_sold_list(log, merchant_user_id: str, pool) -> int: """翻页拉取某商家历史成交,INSERT IGNORE 写入 deca_sold_record。 历史成交为终态,用 product_code 唯一键去重,只增不改。 Args: log: 日志对象。 merchant_user_id (str): 商家用户 ID。 pool (MySQLConnectionPool): MySQL 连接池。 Returns: int: 本商家写入的历史成交商品数(已去重)。 """ page = 1 saved = 0 total = None while page <= MAX_PAGES: body = {"merchantUserId": merchant_user_id, "page": page, "pageSize": PAGE_SIZE} try: resp = deca.do_request(log, "/api/v1/app/groupbuy/merchant/sold-list", body, need_auth=True) except Exception as e: log.error(f"商家 {merchant_user_id} 历史成交第 {page} 页请求失败: {e}") break if not resp or resp.get("code") != 0: log.info(f"商家 {merchant_user_id} 历史成交返回异常: {resp.get('msg') if resp else None}") break data = resp.get("data") or {} if total is None: total = data.get("total") items = data.get("list") or [] if not items: break rows = [r for r in (parse_sold(it) for it in items) if r] if rows: pool.insert_many(table="deca_sold_record", data_list=rows, ignore=True) saved += len(rows) # 翻页终止:已覆盖 total,或本页不足一页 if total is not None and page * PAGE_SIZE >= total: break if len(items) < PAGE_SIZE: break page += 1 time.sleep(0.2) if saved: log.info(f"商家 {merchant_user_id} 历史成交入库 {saved} 个(total={total})") return saved def fill_sold_details(log, pool) -> int: """给未补详情(publish_at 为空)的历史成交商品补上架/开售/结束时间。 调 groupbuy/detail 取 publishAt/saleStartAt/saleEndAt。历史成交为终态, 每个商品补一次即可(publish_at 非空后不再重复拉)。 Args: log: 日志对象。 pool (MySQLConnectionPool): MySQL 连接池。 Returns: int: 本轮成功补详情的商品数。 """ rows = pool.select_all( "SELECT product_code FROM deca_sold_record WHERE publish_at IS NULL") or [] log.info(f"待补详情历史成交商品 {len(rows)} 个") filled = 0 for (code,) in rows: try: d = deca.get_product_detail(log, code) if not d: continue pool.update_one( "UPDATE deca_sold_record SET publish_at=%s, sale_start_at=%s, sale_end_at=%s " "WHERE product_code=%s", (d.get("publishAt") or None, d.get("saleStartAt") or None, d.get("saleEndAt") or None, code)) filled += 1 except Exception as e: log.error(f"历史成交商品 {code} 补详情失败: {e}") time.sleep(0.3) # 轻微限速 return filled @retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=deca.after_log) def main_task(log): """遍历库内所有商家,采集历史成交写入 deca_sold_record。 商家名单取自 deca_shop_record(由 deca_daily_spider 维护)。 Args: log: 日志对象。 Raises: RuntimeError: 数据库连接池异常时抛出以触发重试。 """ log.info(f"开始运行 {sys._getframe().f_code.co_name} 商家历史成交采集" + "." * 40) pool = MySQLConnectionPool(log=log) if not pool.check_pool_health(): log.error("数据库连接池异常") raise RuntimeError("数据库连接池异常") try: deca.ensure_token(log) # 预取 token(后续请求自动续期) rows = pool.select_all("SELECT merchant_user_id FROM deca_shop_record") uids = [r[0] for r in rows] if rows else [] log.info(f"待采集商家 {len(uids)} 个") for uid in uids: try: get_sold_list(log, uid, pool) except Exception as e: log.error(f"get_sold_list error(商家 {uid}): {e}") time.sleep(0.3) # 轻微限速 # 补详情:给未补(publish_at 为空)的历史成交商品拉 groupbuy/detail 的上架/开售/结束时间 try: n = fill_sold_details(log, pool) log.info(f"历史成交补详情完成,本轮 {n} 个") except Exception as e: log.error(f"fill_sold_details error: {e}") except Exception as e: log.error(f"{sys._getframe().f_code.co_name} error: {e}") finally: log.info(f"商家历史成交采集 {sys._getframe().f_code.co_name} 运行结束" + "." * 20) def schedule_task(): """定时任务入口:每天 03:00 采集一次商家历史成交。""" main_task(log=logger) # 立即跑一次(调试时取消注释) schedule.every().day.at("03:00").do(main_task, log=logger) while True: schedule.run_pending() time.sleep(1) if __name__ == "__main__": schedule_task()