|
|
@@ -0,0 +1,384 @@
|
|
|
+# -*- coding: utf-8 -*-
|
|
|
+# Author : Charley
|
|
|
+# Python : 3.12.10
|
|
|
+# Date : 2026/08/11
|
|
|
+"""得卡 DECA · 随机团 teams 明细高频采集（选队随机 + 剩余随机）。
|
|
|
+
|
|
|
+背景：
|
|
|
+ 卡牌拼团分两种"随机团"玩法，同一团会经历两阶段：
|
|
|
+ · 选队随机(playTypeName 以「选队随机」开头)：每支球队一价，team-options 接口(带 token)
|
|
|
+ 返回逐队 cardCount / availableStock / unitPrice；已售数 = cardCount − availableStock。
|
|
|
+ · 剩余随机(playTypeName == 剩余随机)：卖不动后转成的兜底阶段。团购价变成"剩余池加权
|
|
|
+ 均价"、买家随机开一张剩余卡。team-options **对剩余随机永远返 29000**、详情 snapshot
|
|
|
+ 冻结在转换那一刻（每队只有 unitPrice/availableStock，无原始 cardCount）。
|
|
|
+
|
|
|
+关键结论：**一旦转成剩余随机，原始 cardCount 就再也没接口能补回**。所以在选队阶段必须持续
|
|
|
+抓 team-options 存表(deca_groupbuy_team_record)，转剩余后我们才能用「存表的原始总价 − 当前
|
|
|
+剩余货值」算出实时已售额。转剩余前没被我们抓到的老团，team_total_amount 留 NULL、报告端
|
|
|
+标注"原始数据缺失"。
|
|
|
+
|
|
|
+一轮做什么：
|
|
|
+ 1) 从 deca_onsale_product_record 拉当前在售·选队随机团 → team-options →
|
|
|
+ upsert 到 teams 表(data_source=team_options，同 (code, team_id) 覆盖成最新，captured_at
|
|
|
+ 记本轮时刻) → 算 Σ 单价×(cardCount − availableStock) 写回 team_total_amount。
|
|
|
+ 2) 从 deca_onsale_product_record 拉当前在售·剩余随机团 → 详情 snapshot 首次存表
|
|
|
+ (data_source=snapshot；冻结不变，重复见到直接跳过) → 若库里有该团选队阶段的原始
|
|
|
+ cardCount，则算 Σ 单价×cardCount − detail.unitPrice × detail.availableStock 写回
|
|
|
+ team_total_amount；否则留 NULL。
|
|
|
+
|
|
|
+登录态：
|
|
|
+ - team-options 强制带 token（不带返 10002）；
|
|
|
+ - 详情接口 groupbuy/detail 免登录。
|
|
|
+ - 与 alert / sold_daily 共用同一份根目录 token.json；续签已加跨进程锁防踩踏
|
|
|
+ （deca_sold_core._token_lock，2026/08/20 根治多进程互相踩废 refreshToken 的问题）。
|
|
|
+
|
|
|
+运行：项目根目录 `python deca_team_spider.py`（2026/08/20 由 on_sale/ 迁回根目录）
|
|
|
+"""
|
|
|
+import os
|
|
|
+import sys
|
|
|
+import time
|
|
|
+from datetime import datetime
|
|
|
+from decimal import Decimal, ROUND_HALF_UP
|
|
|
+
|
|
|
+import schedule
|
|
|
+from loguru import logger
|
|
|
+from tenacity import retry, stop_after_attempt, wait_fixed
|
|
|
+
|
|
|
+# 切到脚本所在目录(项目根)：token.json / application.yml / ./logs 均在此，
|
|
|
+# 保证从任意工作目录启动都能正确定位（2026/08/20 迁回根目录，去掉原「双 dirname + sys.path.insert」hack）
|
|
|
+os.chdir(os.path.dirname(os.path.abspath(__file__)))
|
|
|
+
|
|
|
+import deca_sold_core as core # noqa: E402
|
|
|
+from mysql_pool import MySQLConnectionPool # noqa: E402
|
|
|
+
|
|
|
+# ==================== 配置 ====================
|
|
|
+INTERVAL_SEC = 300 # 采集间隔：5 分钟一轮（选队随机变化较快、又不至于压接口）
|
|
|
+BETWEEN_ITEM_SEC = 0.3 # 单商品之间的最小间隔，避免瞬时限流
|
|
|
+MAX_ITEM_PER_ROUND = 500 # 单轮采集商品数上限（保护；正常远小于此）
|
|
|
+
|
|
|
+ONSALE_TABLE = "deca_onsale_product_record"
|
|
|
+TEAM_TABLE = "deca_groupbuy_team_record"
|
|
|
+
|
|
|
+# ==================== 日志 ====================
|
|
|
+logger.remove()
|
|
|
+logger.add("./logs/team_spider_{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")
|
|
|
+# logger.add(sys.stderr, level="INFO",
|
|
|
+# format="[{time:HH:mm:ss}] {level} {message}")
|
|
|
+
|
|
|
+
|
|
|
+def after_log(retry_state):
|
|
|
+ """tenacity 重试回调，业务函数首参约定为 log。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ retry_state: tenacity 传入的 RetryCallState，含调用参数与结果。
|
|
|
+ """
|
|
|
+ log = retry_state.args[0] if retry_state.args else 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 _round2(x) -> Decimal | None:
|
|
|
+ """Decimal 化并四舍五入到 2 位小数。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ x: 数值(int/float/str/Decimal)，None/异常直接返回 None。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ Decimal | None: 保留 2 位；None 表示不可算。
|
|
|
+ """
|
|
|
+ if x is None:
|
|
|
+ return None
|
|
|
+ try:
|
|
|
+ return Decimal(str(x)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
|
|
|
+ except Exception:
|
|
|
+ return None
|
|
|
+
|
|
|
+
|
|
|
+def fetch_onsale_random(log, pool) -> tuple[list, list]:
|
|
|
+ """从 deca_onsale_product_record 查当前在售·选队随机 / 剩余随机团 code 列表。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ log: 日志对象。
|
|
|
+ pool: MySQL 连接池。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ tuple[list[str], list[str]]: (选队随机 codes, 剩余随机 codes)。
|
|
|
+ """
|
|
|
+ rows = pool.select_all(
|
|
|
+ f"SELECT product_code, play_type_name FROM {ONSALE_TABLE} "
|
|
|
+ f"WHERE is_on_sale=1 AND play_type_name IS NOT NULL "
|
|
|
+ f"LIMIT {MAX_ITEM_PER_ROUND}") or []
|
|
|
+ xd, sy = [], []
|
|
|
+ for code, ptn in rows:
|
|
|
+ if not ptn:
|
|
|
+ continue
|
|
|
+ if "选队随机" in ptn:
|
|
|
+ xd.append(code)
|
|
|
+ elif "剩余随机" in ptn:
|
|
|
+ sy.append(code)
|
|
|
+ log.info(f"[发现] 在售·选队随机 {len(xd)} / 剩余随机 {len(sy)}")
|
|
|
+ return xd, sy
|
|
|
+
|
|
|
+
|
|
|
+@retry(stop=stop_after_attempt(3), wait=wait_fixed(2), after=after_log)
|
|
|
+def fetch_team_options(log, code: str) -> list | None:
|
|
|
+ """打 team-options 接口拿一个团的逐队 teams(带 token)。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ log: 日志对象。
|
|
|
+ code (str): 商品编码。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ list | None: teams 列表；接口报 29000(已转剩余随机)或失败时返回 None（None 是业务态，
|
|
|
+ 不视为需要重试的错误——只有网络级异常才会被 tenacity 拦下重试）。
|
|
|
+ """
|
|
|
+ r = core.do_request(log, "/api/v1/app/groupbuy/team-options", {"code": code}, need_auth=True)
|
|
|
+ if not r or r.get("code") != 0:
|
|
|
+ return None # 29000 = 商家开启剩余随机中，正常业务态
|
|
|
+ return (r.get("data") or {}).get("list") or []
|
|
|
+
|
|
|
+
|
|
|
+@retry(stop=stop_after_attempt(3), wait=wait_fixed(2), after=after_log)
|
|
|
+def fetch_detail(log, code: str) -> dict | None:
|
|
|
+ """打 detail 接口拿详情(免 token)。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ log: 日志对象。
|
|
|
+ code (str): 商品编码。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ dict | None: data 字典；失败返回 None。
|
|
|
+ """
|
|
|
+ d = core.do_request(log, "/api/v1/app/groupbuy/detail", {"code": code}, need_auth=False)
|
|
|
+ return (d or {}).get("data") or None
|
|
|
+
|
|
|
+
|
|
|
+def upsert_team_row(pool, code: str, ptn: str, source: str, team: dict,
|
|
|
+ captured_at: str, snap_total: int | None = None) -> None:
|
|
|
+ """把一条 team 明细 upsert 进 deca_groupbuy_team_record。
|
|
|
+
|
|
|
+ 唯一键 (product_code, team_id, data_source) 冲突时按最新覆盖——
|
|
|
+ 这就把「在售团每轮覆盖成最新」自动做掉了。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ pool: MySQL 连接池。
|
|
|
+ code (str): 商品编码。
|
|
|
+ ptn (str): 采集时的 playTypeName。
|
|
|
+ source (str): 数据来源 'team_options' / 'snapshot'。
|
|
|
+ team (dict): 单条 team 数据(team-options 项 或 snapshot.teams 项)。
|
|
|
+ captured_at (str): 采集时刻 YYYY-MM-DD HH:MM:SS。
|
|
|
+ snap_total (int | None, optional): 剩余随机 snapshot.totalQuantity。Defaults to None。
|
|
|
+ """
|
|
|
+ tid = team.get("teamId")
|
|
|
+ if tid is None:
|
|
|
+ return
|
|
|
+ cc = team.get("cardCount") # 选队阶段有；剩余快照无 → NULL
|
|
|
+ av = team.get("availableStock")
|
|
|
+ up = _round2(team.get("unitPrice"))
|
|
|
+ sold = (cc - av) if (cc is not None and av is not None) else None
|
|
|
+ pool._execute(
|
|
|
+ f"INSERT INTO {TEAM_TABLE} "
|
|
|
+ f"(product_code, play_type_name, data_source, team_id, team_name_en, team_name_zh, "
|
|
|
+ f" team_logo_image_url, unit_price, card_count, available_stock, sold_count, "
|
|
|
+ f" snapshot_total_quantity, captured_at) "
|
|
|
+ f"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) "
|
|
|
+ f"ON DUPLICATE KEY UPDATE "
|
|
|
+ f" play_type_name=VALUES(play_type_name), team_name_en=VALUES(team_name_en), "
|
|
|
+ f" team_name_zh=VALUES(team_name_zh), team_logo_image_url=VALUES(team_logo_image_url), "
|
|
|
+ f" unit_price=VALUES(unit_price), card_count=VALUES(card_count), "
|
|
|
+ f" available_stock=VALUES(available_stock), sold_count=VALUES(sold_count), "
|
|
|
+ f" snapshot_total_quantity=VALUES(snapshot_total_quantity), "
|
|
|
+ f" captured_at=VALUES(captured_at)",
|
|
|
+ (code, ptn, source, tid, team.get("teamNameEn"), team.get("teamNameZh"),
|
|
|
+ team.get("teamLogoImageUrl"), up, cc, av, sold, snap_total, captured_at),
|
|
|
+ commit=True)
|
|
|
+
|
|
|
+
|
|
|
+def has_snapshot(pool, code: str) -> bool:
|
|
|
+ """判断某团是否已存过剩余随机 snapshot(冻结值，只需存一次)。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ pool: MySQL 连接池。
|
|
|
+ code (str): 商品编码。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ bool: True 已存过、可跳过；False 未存。
|
|
|
+ """
|
|
|
+ row = pool.select_one(
|
|
|
+ f"SELECT 1 FROM {TEAM_TABLE} WHERE product_code=%s AND data_source='snapshot' LIMIT 1",
|
|
|
+ (code,))
|
|
|
+ return bool(row)
|
|
|
+
|
|
|
+
|
|
|
+def get_original_total(pool, code: str) -> Decimal | None:
|
|
|
+ """从 teams 表查某团选队阶段抓过的「原始总价」= Σ 单价×cardCount。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ pool: MySQL 连接池。
|
|
|
+ code (str): 商品编码。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ Decimal | None: 原始总价；缺 team-options 记录时返回 None。
|
|
|
+ """
|
|
|
+ row = pool.select_one(
|
|
|
+ f"SELECT SUM(unit_price * card_count) FROM {TEAM_TABLE} "
|
|
|
+ f"WHERE product_code=%s AND data_source='team_options' AND card_count IS NOT NULL",
|
|
|
+ (code,))
|
|
|
+ if not row or row[0] is None:
|
|
|
+ return None
|
|
|
+ return _round2(row[0])
|
|
|
+
|
|
|
+
|
|
|
+def update_team_total(pool, code: str, amount: Decimal | None) -> None:
|
|
|
+ """把算好的团总价写回 deca_onsale_product_record.team_total_amount。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ pool: MySQL 连接池。
|
|
|
+ code (str): 商品编码。
|
|
|
+ amount (Decimal | None): 总价；None 会显式清空(表示"数据缺失")。
|
|
|
+ """
|
|
|
+ pool.update_one(
|
|
|
+ f"UPDATE {ONSALE_TABLE} SET team_total_amount=%s WHERE product_code=%s",
|
|
|
+ (amount, code))
|
|
|
+
|
|
|
+
|
|
|
+def process_xuandui(log, pool, code: str, captured_at: str) -> bool:
|
|
|
+ """处理一个在售·选队随机团：抓 team-options → 存表 → 算总价 → 回写商品。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ log: 日志对象。
|
|
|
+ pool: MySQL 连接池。
|
|
|
+ code (str): 商品编码。
|
|
|
+ captured_at (str): 本轮采集时刻。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ bool: True 成功；False 接口不可用(如刚转成剩余随机)。
|
|
|
+ """
|
|
|
+ teams = fetch_team_options(log, code)
|
|
|
+ if not teams:
|
|
|
+ return False # 可能这一瞬间刚转成剩余随机，本轮跳过；下轮以剩余随机身份进另一分支
|
|
|
+ total = Decimal("0")
|
|
|
+ ptn = "选队随机" # 精确 name 已在商品行 play_type_name；此处只作 teams 表内的存档
|
|
|
+ for t in teams:
|
|
|
+ upsert_team_row(pool, code, ptn, "team_options", t, captured_at)
|
|
|
+ cc = t.get("cardCount") or 0
|
|
|
+ av = t.get("availableStock") or 0
|
|
|
+ up = Decimal(str(t.get("unitPrice") or "0"))
|
|
|
+ total += up * (cc - av)
|
|
|
+ update_team_total(pool, code, _round2(total))
|
|
|
+ return True
|
|
|
+
|
|
|
+
|
|
|
+def process_shengyu(log, pool, code: str, captured_at: str) -> bool:
|
|
|
+ """处理一个在售·剩余随机团：首次存 snapshot；每轮从 detail 取实时剩余算总价回写。
|
|
|
+
|
|
|
+ 公式：team_total_amount = 原始总价(存表 Σ 单价×cardCount) − detail.unitPrice × detail.availableStock。
|
|
|
+ 若库里没有该团选队阶段的 cardCount 记录，team_total_amount 留 NULL 表示"数据缺失"。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ log: 日志对象。
|
|
|
+ pool: MySQL 连接池。
|
|
|
+ code (str): 商品编码。
|
|
|
+ captured_at (str): 本轮采集时刻(仅首次存 snapshot 时用)。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ bool: True 处理成功；False 详情拉失败。
|
|
|
+ """
|
|
|
+ dd = fetch_detail(log, code)
|
|
|
+ if not dd:
|
|
|
+ return False
|
|
|
+ # 1) 首次存 snapshot(冻结不变，只存一次)
|
|
|
+ if not has_snapshot(pool, code):
|
|
|
+ snap = dd.get("remainingRandomTeamSnapshot") or {}
|
|
|
+ for t in (snap.get("teams") or []):
|
|
|
+ upsert_team_row(pool, code, "剩余随机", "snapshot", t, captured_at,
|
|
|
+ snap_total=snap.get("totalQuantity"))
|
|
|
+ # 2) 每轮重算实时总价
|
|
|
+ original = get_original_total(pool, code)
|
|
|
+ if original is None:
|
|
|
+ # 转剩余随机前没被我们抓到过 team-options → 拿不到原始总价，显式留 NULL
|
|
|
+ update_team_total(pool, code, None)
|
|
|
+ log.warning(f"[数据缺失] {code} 无选队阶段 team-options 记录，team_total_amount=NULL")
|
|
|
+ return True
|
|
|
+ up_now = Decimal(str(dd.get("unitPrice") or "0"))
|
|
|
+ av_now = int(dd.get("availableStock") or 0)
|
|
|
+ remaining_value = up_now * av_now
|
|
|
+ total = original - remaining_value
|
|
|
+ if total < 0:
|
|
|
+ # 边界：原始总价是"名义价"，理论上不会 <0；出现即为数据异常，记日志、置 None
|
|
|
+ log.warning(f"[异常] {code} original={original} - remaining={remaining_value} <0，置 NULL")
|
|
|
+ update_team_total(pool, code, None)
|
|
|
+ else:
|
|
|
+ update_team_total(pool, code, _round2(total))
|
|
|
+ return True
|
|
|
+
|
|
|
+
|
|
|
+@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
|
|
|
+def main_task(log):
|
|
|
+ """采集主函数：一轮遍历在售随机团、更新 teams 表与 team_total_amount。
|
|
|
+
|
|
|
+ 挂了每小时重试(无人值守)；单轮内各商品独立 try/except，单个失败不拖垮整轮。
|
|
|
+
|
|
|
+ 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:
|
|
|
+ xd_codes, sy_codes = fetch_onsale_random(log, pool)
|
|
|
+ captured_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
+
|
|
|
+ # 选队随机：每轮全量刷新 team-options
|
|
|
+ ok_xd = fail_xd = 0
|
|
|
+ for c in xd_codes:
|
|
|
+ try:
|
|
|
+ if process_xuandui(log, pool, c, captured_at):
|
|
|
+ ok_xd += 1
|
|
|
+ else:
|
|
|
+ fail_xd += 1
|
|
|
+ except Exception as e:
|
|
|
+ fail_xd += 1
|
|
|
+ log.error(f"[选队随机·失败] {c}: {e}")
|
|
|
+ time.sleep(BETWEEN_ITEM_SEC)
|
|
|
+ log.info(f"[选队随机] 处理完成: 成功 {ok_xd} 失败 {fail_xd}")
|
|
|
+
|
|
|
+ # 剩余随机：每轮从 detail 取实时剩余、更新总价
|
|
|
+ ok_sy = fail_sy = 0
|
|
|
+ for c in sy_codes:
|
|
|
+ try:
|
|
|
+ if process_shengyu(log, pool, c, captured_at):
|
|
|
+ ok_sy += 1
|
|
|
+ else:
|
|
|
+ fail_sy += 1
|
|
|
+ except Exception as e:
|
|
|
+ fail_sy += 1
|
|
|
+ log.error(f"[剩余随机·失败] {c}: {e}")
|
|
|
+ time.sleep(BETWEEN_ITEM_SEC)
|
|
|
+ log.info(f"[剩余随机] 处理完成: 成功 {ok_sy} 失败 {fail_sy}")
|
|
|
+ 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():
|
|
|
+ """定时入口：每 INTERVAL_SEC 秒跑一次 main_task。启动时立即跑一次。"""
|
|
|
+ main_task(log=logger) # 启动即刻跑一轮，避免等 5 分钟才开始
|
|
|
+ schedule.every(INTERVAL_SEC).seconds.do(main_task, log=logger)
|
|
|
+ while True:
|
|
|
+ schedule.run_pending()
|
|
|
+ time.sleep(1)
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ schedule_task()
|
