onsale_alert_spider.py 51 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074
  1. # -*- coding: utf-8 -*-
  2. # Author : Charley
  3. # Python : 3.12.10
  4. # Date : 2026/08/04
  5. """得卡 DECA · 指定商家在售商品提醒(新品上架 + 进度过半,常驻轮询,支持多商家)。
  6. 监控 MERCHANTS 列表中各商家的在售商品(每项 id/name/tag,加店只追加一行),三类提醒(各发一条独立消息):
  7. 1. 新商品上架:出现库里没有的新 product_code。
  8. 2. 进度过半:售卖进度 sold_count/card_count 首次达到 HALF_THRESHOLD。
  9. 去重靠 deca_onsale_alert_record 两个标记位(new_notified/half_notified),每类每商品只提醒一次。
  10. 发送渠道 SEND_CHANNEL 可切(2026/08/11 起默认企业微信):
  11. - "qywx": 企业微信群机器人(markdown_v2)——发到 WEBHOOK_URL 配置的群,标题加粗、不带链接。
  12. - "pc" : PC 版微信(wxauto4 文本)发给 WX_TARGET——纯文本文案,随时可切回。
  13. 消息不带商品链接(2026/08/11):得卡商品分享落地页(share-detail 接口)只认「登录态(带 token)
  14. 生成的 shareCode」,免 token 详情返回的 shareCode 一律被判失效;为不引入登录态、降账号风控,
  15. 提醒消息直接不挂链接。三类提醒都发:新商品上架 / 拼团进度过半 / 一车结束战报。
  16. 运行时段:仅每天 RUN_START~次日 06:00 轮询(RUN_START/RUN_END 控制),其余时间休眠到下次开窗再跑。
  17. RUN_START 默认 20:30,可用命令行传参覆盖(见下),窗口起点同时是「新上架」时间门槛:只提醒
  18. publishAt 晚于该起点的商品。
  19. 从根目录运行:
  20. python onsale_alert_spider.py # 默认 20:30 开始
  21. python onsale_alert_spider.py 17:00 # 改为 17:00 开始:17:00 后的新上架才提醒
  22. python onsale_alert_spider.py --start 17:00
  23. (默认企微渠道,需 WEBHOOK_URL 配好)
  24. """
  25. import sys
  26. import time
  27. import random
  28. import argparse
  29. from datetime import datetime, time as dtime, timedelta
  30. # 挂靠新项目根:sys.path 指向 common、CWD 对齐新根(application.yml / logs / 账号池 DB 生效)
  31. import os
  32. _ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
  33. sys.path.insert(0, os.path.join(_ROOT, "common"))
  34. os.chdir(_ROOT)
  35. from loguru import logger
  36. from tenacity import retry, stop_after_attempt, wait_fixed
  37. from mysql_pool import MySQLConnectionPool
  38. import deca_sold_core as core
  39. try:
  40. import deca_wechat # PC 微信(wxauto4),已弃用;仅 SEND_CHANNEL=="pc" 时需要
  41. except ImportError:
  42. deca_wechat = None
  43. from auto_send_wx_msg import send_wechat_group_msg # 企微机器人发送(默认渠道)
  44. # 复用 daily 的免 token 全站在售拉取(home/search,只拉不落库);alert 独立自采、不依赖 buy_record 落库
  45. from deca_on_sale_daily_spider import parse_product
  46. logger.remove()
  47. logger.add("./logs/onsale_alert_{time:YYYYMMDD}.log", encoding="utf-8", rotation="00:00",
  48. format="[{time:YYYY-MM-DD HH:mm:ss.SSS}] {level} {message}",
  49. level="DEBUG", retention="7 day")
  50. # ==================== 配置 ====================
  51. # 监控商家列表(可扩展:加店只追加一行)。两店同属「魔都兄弟」,靠 tag 在提醒标题里区分:
  52. # id :商家用户ID(merchantUserId) name:商家全名(消息正文用) tag:短标签(标题【】前缀,区分同名商家)
  53. MERCHANTS = [
  54. {"id": "881226408", "name": "魔都兄弟球星卡", "tag": "球星卡"}, # 老店
  55. {"id": "606370597", "name": "魔都兄弟综合体育", "tag": "综合体育"}, # 2026/09/08 新增,与球星卡同属一家、算两个店铺
  56. ]
  57. HALF_THRESHOLD = 0.5 # 进度过半阈值(0.5=50%)
  58. PROGRESS_TABLE = "deca_onsale_product_progress_record" # 进度时间序列(server B 的 buy_record 每 60s 变化才写),算阶段耗时用
  59. SEND_CHANNEL = "qywx" # 发送渠道:qywx=企业微信群机器人(默认) / pc=PC版微信(wxauto4,可切回)
  60. # WX_TARGET = "backup" # PC 微信发送目标(好友备注名/群名),改成实际接收人
  61. WX_TARGET = "得卡-通知" # PC 微信发送目标(好友备注名/群名),改成实际接收人
  62. # COLD_START_PUSH 已废弃(2026/08/08):改为「只提醒本轮窗口起点(20:30)后新上架」,冷启动不再发全量快照
  63. MIN_INTERVAL_SEC = 60 # 轮询间隔随机下限(秒)
  64. MAX_INTERVAL_SEC = 90 # 轮询间隔随机上限(秒),每轮在 [下限,上限] 取随机数,打散规律降风控
  65. RUN_START = dtime(20, 30) # 运行窗口开始默认 20:30;可由命令行传参覆盖(见 _parse_args),同时作为「新上架」时间门槛
  66. RUN_END = dtime(6, 0) # 运行窗口结束:次日 06:00(窗口跨午夜;2026/08/15 由 03:00 延到 06:00,昨天 5 点还在播)
  67. MAX_PROD_PAGES = 3 # 单商家在售翻页上限(服务端 *-list 硬限最近 3 页;在售普遍远 <60 个,靠 total/末页早停)
  68. T_ALERT = "deca_onsale_alert_record"
  69. T_PROD = core.T_PROD
  70. ON_SALE_PATH = "/api/v1/app/groupbuy/merchant/on-sale-list" # 商家在售商品列表(need_auth)
  71. DETAIL_PATH = "/api/v1/app/groupbuy/detail" # 商品详情(取 publishAt 判新品)
  72. # 本进程启动后「见过在售」的 product_code 集合,按商家 id 隔离(模块级内存,进程重启即清空):
  73. # 结构 {merchant_id: set(product_code)}——各商家互不干扰,避免不同店的 code 混在一起误判结束。
  74. # 结束战报只认对应商家 seen 里的 code——即只播报「程序运行之后」才结束的车;程序启动前就已结束的历史车不补发。
  75. # 与持久化的 ended_notified 列配合:seen 决定「管不管」,ended_notified 决定「发没发过」。
  76. _seen_onsale: dict = {}
  77. def _get_detail_data(log, code: str) -> dict:
  78. """打商品详情接口(免 token)返回 data 层。
  79. Args:
  80. log: 日志对象。
  81. code (str): 商品编码(product_code)。
  82. Returns:
  83. dict: 详情 data 层;请求失败返回空 dict。
  84. """
  85. try:
  86. resp = core.do_request(log, DETAIL_PATH, {"code": code}, need_auth=False) # 详情接口免登录,实测不带token也返回
  87. return (resp or {}).get("data") or {}
  88. except Exception as e:
  89. log.warning(f"取详情失败({code}): {e}")
  90. return {}
  91. def get_publish_at(log, code: str) -> str | None:
  92. """调详情接口取该商品的上架时间(data.publishAt),供新品判断用。
  93. Args:
  94. log: 日志对象。
  95. code (str): 商品编码(product_code)。
  96. Returns:
  97. str | None: publishAt 原文如 "2026-08-08 20:31:00";取不到返回 None。
  98. """
  99. return _get_detail_data(log, code).get("publishAt")
  100. def _window_start(now: datetime) -> datetime:
  101. """本轮运行窗口的起点(最近一个已过去的 RUN_START,默认 20:30),作为「新上架」时间门槛。
  102. RUN_START 可由命令行传参覆盖,故此处读全局值而非写死时刻。
  103. Args:
  104. now (datetime): 当前时间。
  105. Returns:
  106. datetime: 当天时刻 >= RUN_START → 今天 RUN_START;否则 → 昨天 RUN_START(跨午夜窗口取窗口起点那天)。
  107. """
  108. if now.time() >= RUN_START:
  109. return datetime.combine(now.date(), RUN_START)
  110. return datetime.combine(now.date() - timedelta(days=1), RUN_START)
  111. def _is_new_arrival(publish_at_text: str, window_start: datetime) -> bool:
  112. """判断商品上架时间(publishAt)是否晚于窗口起点,即本轮才新上架。
  113. Args:
  114. publish_at_text (str): 详情 publishAt,格式 "%Y-%m-%d %H:%M:%S",可能为空。
  115. window_start (datetime): 本轮窗口起点(20:30)。
  116. Returns:
  117. bool: True=上架时间 >= 窗口起点(本轮新上架);空值/解析失败按 False(保守,不误报老货)。
  118. """
  119. if not publish_at_text:
  120. return False
  121. try:
  122. pub = datetime.strptime(publish_at_text.strip(), "%Y-%m-%d %H:%M:%S")
  123. except (ValueError, TypeError):
  124. return False
  125. return pub >= window_start
  126. def _fmt_money(v) -> str:
  127. """把价格数值格式化为紧凑字符串:整数去掉小数(80.00→80),非整去尾零(1.50→1.5)。
  128. Args:
  129. v: 价格(Decimal/float/int/str)。
  130. Returns:
  131. str: 紧凑价格字符串。
  132. """
  133. f = float(v)
  134. if f == int(f):
  135. return str(int(f))
  136. return f"{f:g}"
  137. def _fmt_price(unit, mn, mx) -> str:
  138. """构造价格文案:区间价(min~max)优先,其次单价,与 App 详情页口径一致。
  139. 得卡「单箱选队」等商品 unitPrice 常为 0,真实价落在 minUnitPrice~maxUnitPrice 区间
  140. (如详情页「¥1.50 ~ 80」),故区间存在且非单点时优先展示区间,避免误显 ¥0.00。
  141. Args:
  142. unit: 单价 unitPrice(可能为 0/None)。
  143. mn: 最低单价 minUnitPrice(可能为 None)。
  144. mx: 最高单价 maxUnitPrice(可能为 None)。
  145. Returns:
  146. str: 价格文案,如 "¥1.5~80" / "¥400" / "¥-"。
  147. """
  148. if mn is not None and mx is not None and float(mn) != float(mx):
  149. return f"¥{_fmt_money(mn)}~{_fmt_money(mx)}" # 区间价(单箱选队等)
  150. if unit is not None and float(unit) > 0:
  151. return f"¥{_fmt_money(unit)}" # 正常单价
  152. if mn is not None and float(mn) > 0:
  153. return f"¥{_fmt_money(mn)}" # unit 为 0/None 时退回 min
  154. return "¥-"
  155. def _fmt_item(r: dict, kind: str, pct: float = None, plain: bool = False) -> str:
  156. """构造一条通知条目:标题 + 价格 + 份数/进度 + 余·共(不带链接)。
  157. 2026/08/11 起消息不挂商品链接(share-detail 只认带 token 生成的 shareCode,免 token 一律失效,
  158. 为不引入登录态直接去链接);markdown 渠道把标题加粗,纯文本渠道直出标题。
  159. Args:
  160. r (dict): parse_product 产出的商品字典。
  161. kind (str): "new"=新品(显示份数) / "half"=过半(显示进度%) / "onsale"=当前在售快照(显示进度%)。
  162. pct (float, optional): 售卖进度百分比(0~100),kind 为 "half"/"onsale" 时用。Defaults to None。
  163. plain (bool, optional): True=纯文本(PC微信)/False=markdown(企微)。Defaults to False。
  164. Returns:
  165. str: 一条通知文案(两行)。
  166. """
  167. # 标题行:markdown 渠道加粗突出,纯文本渠道直出
  168. title_line = r["title"] if plain else f"**{r['title']}**"
  169. # 价格:区间价优先(单箱选队等 unitPrice=0 的商品用 min~max),避免误显 ¥0.00
  170. price_text = _fmt_price(r.get("unit_price"), r.get("min_unit_price"), r.get("max_unit_price"))
  171. # 信息行:新品显示份数,过半/当前在售显示进度百分比
  172. if kind in ("half", "onsale"):
  173. info = (f"💰 {price_text} | 📈 进度{pct:.0f}% | "
  174. f"🎯 余{r['available_stock']}/{r['card_count']}")
  175. else:
  176. info = (f"💰 {price_text} | 📦 {r['card_count']}份 | "
  177. f"🎯 余{r['available_stock']}/{r['card_count']}")
  178. return f"{title_line}\n{info}"
  179. def _dispatch(log, items: list, title: str) -> bool:
  180. """按 SEND_CHANNEL 把一批条目发出去(pc=PC微信整段文本 / qywx=企微 markdown)。
  181. Args:
  182. log: 日志对象。
  183. items (list[str]): 已构造好的通知条目列表。
  184. title (str): 消息标题。
  185. Returns:
  186. bool: 发送成功返回 True;无条目或发送失败返回 False(调用方据此决定是否置位提醒标记)。
  187. """
  188. if not items:
  189. return False
  190. if SEND_CHANNEL == "pc":
  191. # PC 微信:标题 + 编号清单,拼成一整段纯文本,一条消息发给 WX_TARGET
  192. body = [f"{i}. {it}" for i, it in enumerate(items, 1)]
  193. text = title + "\n\n" + "\n----------------------------------\n".join(body)
  194. return bool(deca_wechat.send_text(text, who=WX_TARGET))
  195. # 企微机器人:成功返回 dict、失败返回 None
  196. return bool(send_wechat_group_msg(log=log, items=items, title=title))
  197. def _pc_section(title: str, items: list) -> str:
  198. """把一组条目拼成 PC 微信纯文本的一个分区(小标题 + 编号清单)。
  199. Args:
  200. title (str): 分区小标题(如「新商品上架」)。
  201. items (list[str]): 已构造好的通知条目列表。
  202. Returns:
  203. str: 该分区的纯文本(含小标题与编号清单);items 为空返回空串。
  204. """
  205. if not items:
  206. return ""
  207. body = [f"{i}. {it}" for i, it in enumerate(items, 1)]
  208. return f"{title}({len(items)}款)\n" + "\n----------------------------------\n".join(body)
  209. def _dispatch_pc_combined(log, mname: str, new_items: list, half_items: list, tag: str) -> bool:
  210. """PC 微信:把新品与过半两类合并为一条纯文本消息发出(减少操作,一次发完)。
  211. Args:
  212. log: 日志对象。
  213. mname (str): 商家名称,用于消息大标题。
  214. new_items (list[str]): 新品上架通知条目。
  215. half_items (list[str]): 进度过半通知条目。
  216. tag (str): 商家短标签(如「球星卡」),加到大标题【】前缀区分同名商家。
  217. Returns:
  218. bool: 发送成功返回 True;无内容或发送失败返回 False(调用方据此决定是否置位新品标记)。
  219. """
  220. sections = []
  221. new_sec = _pc_section("【新商品上架】", new_items)
  222. half_sec = _pc_section("【拼团进度过半】", half_items)
  223. if new_sec:
  224. sections.append(new_sec)
  225. if half_sec:
  226. sections.append(half_sec)
  227. if not sections:
  228. return False
  229. header = f"【{tag}】得卡 · {mname} 在售提醒"
  230. text = header + "\n\n" + "\n\n==================================\n\n".join(sections)
  231. return bool(deca_wechat.send_text(text, who=WX_TARGET))
  232. def _progress(sold, card) -> float:
  233. """计算售卖进度(0~1),卡片份数缺失或为 0 时返回 0。
  234. Args:
  235. sold (int | None): 已售份数。
  236. card (int | None): 总份数。
  237. Returns:
  238. float: 进度比例 sold/card;无法计算时返回 0.0。
  239. """
  240. if not card or card <= 0 or sold is None:
  241. return 0.0
  242. return sold / card
  243. def fetch_onsale(log, merchant_id: str) -> tuple[list, bool]:
  244. """带 token 抓指定商家的在售商品(逐页 on-sale-list,走直连)。
  245. 2026/09/05:站方把免 token 的全站 home/search 砍成首屏 20、全站免 token 翻页失效,可能刷不到本商家的车;
  246. 故改为直接调本商家 on-sale-list(需 token)分页拉取,parse_product 解析后按 product_code 去重返回。
  247. token 走直连保持账号↔IP 稳定;详情接口仍免 token。
  248. Args:
  249. log: 日志对象。
  250. merchant_id (str): 商家用户ID(merchantUserId),来自 MERCHANTS 列表。
  251. Returns:
  252. tuple[list, bool]: (本商家在售商品字典列表[parse_product 结果,含 title/unit_price/card_count/
  253. sold_count/available_stock/merchant_name 等], 是否正常翻到底)。
  254. bool=False 表示中途异常/未取全,调用方据此放弃本轮下架/结束对账,避免误判。
  255. """
  256. rows = []
  257. page = 1
  258. total = None
  259. ok = False
  260. while page <= MAX_PROD_PAGES:
  261. body = {"merchantUserId": merchant_id, "page": page, "pageSize": 20}
  262. try:
  263. resp = core.do_request(log, ON_SALE_PATH, body, need_auth=True, use_proxy=False)
  264. except Exception as e:
  265. log.error(f"商家 {merchant_id} 在售第 {page} 页请求失败: {e}")
  266. break
  267. if not resp or resp.get("code") != 0:
  268. log.info(f"商家 {merchant_id} 在售返回异常: {resp.get('msg') if resp else None}")
  269. break
  270. data = resp.get("data") or {}
  271. if total is None:
  272. total = data.get("total")
  273. items = data.get("list") or []
  274. rows.extend(r for r in (parse_product(it) for it in items) if r)
  275. if (total is not None and page * 20 >= total) or len(items) < 20: # 采满 total 或末页
  276. ok = True
  277. break
  278. page += 1
  279. uniq = {p["product_code"]: p for p in rows} # 按 code 去重
  280. log.info(f"商家 {merchant_id} 当前在售商品 {len(uniq)} 个(on-sale-list 带 token,ok={ok})")
  281. return list(uniq.values()), ok
  282. def _insert_alert(pool, r: dict, progress_pct: float, new_notified: int,
  283. half_notified: int):
  284. """新增一条在售监控记录(share_code 列 2026/08/11 起不再写入,消息已去链接)。
  285. Args:
  286. pool (MySQLConnectionPool): MySQL 连接池。
  287. r (dict): parse_product 产出的商品字典。
  288. progress_pct (float): 售卖进度百分比(0~100)。
  289. new_notified (int): 新品提醒标记位 0/1。
  290. half_notified (int): 过半提醒标记位 0/1。
  291. """
  292. sql = (f"INSERT IGNORE INTO {T_ALERT} "
  293. "(product_code, merchant_user_id, merchant_name, title, unit_price, card_count, "
  294. "sold_count, progress, available_stock, groupbuy_status_name, "
  295. "new_notified, half_notified) "
  296. "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)")
  297. args = (r["product_code"], r["merchant_user_id"], r["merchant_name"], r["title"],
  298. r["unit_price"], r["card_count"], r["sold_count"], round(progress_pct, 2),
  299. r["available_stock"], r["groupbuy_status_name"],
  300. new_notified, half_notified)
  301. pool.insert_many(query=sql, args_list=[args])
  302. def _update_progress(pool, r: dict, progress_pct: float, half_notified: int):
  303. """更新已有记录的售卖进度等动态字段(及可选的过半标记位)。
  304. Args:
  305. pool (MySQLConnectionPool): MySQL 连接池。
  306. r (dict): parse_product 产出的商品字典。
  307. progress_pct (float): 售卖进度百分比(0~100)。
  308. half_notified (int): 过半提醒标记位 0/1(命中过半时置 1)。
  309. """
  310. sql = (f"UPDATE {T_ALERT} SET sold_count=%s, progress=%s, available_stock=%s, "
  311. "groupbuy_status_name=%s, half_notified=%s WHERE product_code=%s")
  312. pool.update_one(sql, (r["sold_count"], round(progress_pct, 2), r["available_stock"],
  313. r["groupbuy_status_name"], half_notified, r["product_code"]))
  314. def _mark_new_notified(pool, codes: list):
  315. """把一批商品的 new_notified 置 1(仅在新品提醒发送成功后调用)。
  316. Args:
  317. pool (MySQLConnectionPool): MySQL 连接池。
  318. codes (list[str]): 待置位的 product_code 列表;为空直接返回。
  319. """
  320. if not codes:
  321. return
  322. placeholders = ",".join(["%s"] * len(codes))
  323. pool.update_one(
  324. f"UPDATE {T_ALERT} SET new_notified=1 WHERE product_code IN ({placeholders})",
  325. tuple(codes))
  326. def _is_sale_ended(data: dict, now_ts: int) -> tuple[bool, str]:
  327. """据商品详情 data 判断该车是否售卖结束(售罄 或 已过销售结束时间)。
  328. 与 buy_record_spider.is_sale_ended 同逻辑,内联于此避免 import 采集脚本引入其模块级副作用。
  329. 详情请求失败时 data 为空 dict,两条件均不命中返回未结束——保守,防抓取抖动误判下架。
  330. 预售/未开卖的车:详情接口 availableStock 也返回 0(库存未分配)、soldCount=0、saleStartAt 在未来,
  331. 仅凭 availableStock<=0 会把它误判为「售罄」→ 发出假的「一车结束」战报。直播实时放量卖的车同理:
  332. 卖到一半 availableStock 也会瞬时抖成 0。故 ① saleStartAt 未到直接判未结束;② 售罄改判 soldCount>=totalCardCount
  333. (卖满才算),彻底不看 availableStock(2026/09/04:GB26090464871 卖 2/31 时 availableStock=0 曾致 buy_record 误杀)。
  334. Args:
  335. data (dict): 商品详情接口(groupbuy/detail) data 层,可能为空 dict。
  336. now_ts (int): 当前时间戳(秒)。
  337. Returns:
  338. tuple[bool, str]: (是否已结束, 原因文本);未结束时原因为空串。
  339. """
  340. # ① 开卖时间未到 → 预售态,绝不算结束(预售车 availableStock=0 是「未分配」而非「卖光」)
  341. start_text = data.get("saleStartAt")
  342. if start_text:
  343. try:
  344. start_ts = time.mktime(time.strptime(start_text, "%Y-%m-%d %H:%M:%S"))
  345. if now_ts < start_ts:
  346. return False, ""
  347. except (ValueError, OverflowError):
  348. pass
  349. # ② 售罄:改用「已售 >= 总份数」判定,不再用 availableStock。
  350. # 直播实时放量卖的车,availableStock 是「当前放出、还没被抢的量」而非总剩余,卖到一半也会瞬时抖成 0,
  351. # 旧逻辑凭 availableStock<=0 会把仍在售的车误判售罄发假战报(2026/09/04 GB26090464871 卖 2/31 时 availableStock=0)。
  352. # 本函数只在「车已离开在售列表」后作二次确认,从严只认卖满,与下方「结束即全部售出」口径一致。
  353. sold = data.get("soldCount")
  354. total = data.get("totalCardCount")
  355. if sold is not None and total is not None and total > 0 and sold >= total:
  356. return True, f"售罄(soldCount={sold}/{total})"
  357. end_text = data.get("saleEndAt")
  358. if end_text:
  359. try:
  360. end_ts = time.mktime(time.strptime(end_text, "%Y-%m-%d %H:%M:%S"))
  361. if now_ts >= end_ts:
  362. return True, f"已过结束时间(saleEndAt={end_text})"
  363. except (ValueError, OverflowError):
  364. pass
  365. return False, ""
  366. def _fmt_ended(title: str, card, price_text: str, buyers: int, plain: bool = False) -> str:
  367. """构造一条「一车结束」战报正文:标题 + 价格 + 售出总件数 + 参与拆卡人数。
  368. 车结束(拼团成交/售罄)即全部份数售出,故「售出件数」直接取总份数 card_count;
  369. 参与拆卡人数为该车 deca_buy_record 按 user_id 去重的买家数。
  370. Args:
  371. title (str): 商品标题。
  372. card (int | None): 总份数 cardCount,即售出总件数。
  373. price_text (str): 已格式化的价格文案(_fmt_price 产出,区间价优先)。
  374. buyers (int): 去重购买人数(参与拆卡人数)。
  375. plain (bool, optional): True=纯文本(PC微信)/False=markdown(企微,标题加粗)。Defaults to False。
  376. Returns:
  377. str: 战报正文(三行)。
  378. """
  379. card_txt = card if card is not None else "?"
  380. title_line = title if plain else f"**{title}**"
  381. return (f"{title_line}\n"
  382. f"💰 {price_text} | 🎯 售出 {card_txt} 件\n"
  383. f"👥 {buyers} 人参与拆卡")
  384. def _send_ended(log, mname: str, item_text: str, tag: str) -> bool:
  385. """按 SEND_CHANNEL 发送一条「一车结束」战报(每辆车一条独立消息)。
  386. Args:
  387. log: 日志对象。
  388. mname (str): 商家名称,用于消息标题。
  389. item_text (str): _fmt_ended 产出的战报正文。
  390. tag (str): 商家短标签(如「球星卡」),加到标题【】前缀区分同名商家。
  391. Returns:
  392. bool: 发送成功返回 True;失败返回 False(调用方据此决定是否置 ended_notified)。
  393. """
  394. header = f"🏁【{tag}】得卡 · {mname} 一车结束"
  395. if SEND_CHANNEL == "pc":
  396. return bool(deca_wechat.send_text(header + "\n\n" + item_text, who=WX_TARGET))
  397. return bool(send_wechat_group_msg(log=log, items=[item_text], title=header))
  398. def _mark_ended_notified(pool, code: str):
  399. """把一辆车的 ended_notified 置 1(仅在结束战报发送成功后调用)。
  400. Args:
  401. pool (MySQLConnectionPool): MySQL 连接池。
  402. code (str): 商品编码(product_code)。
  403. """
  404. pool.update_one(f"UPDATE {T_ALERT} SET ended_notified=1 WHERE product_code=%s", (code,))
  405. def _count_buyers(pool, code: str) -> int:
  406. """查 deca_buy_record 中该车按 user_id 去重的购买人数(即参与拆卡人数)。
  407. 数据由 buy_record_spider(独立进程)持续采集;若该车未被采到则返回 0。
  408. Args:
  409. pool (MySQLConnectionPool): MySQL 连接池。
  410. code (str): 商品编码(product_code)。
  411. Returns:
  412. int: 去重购买人数;无记录或查询异常返回 0。
  413. """
  414. rows = pool.select_all(
  415. "SELECT COUNT(DISTINCT user_id) FROM deca_buy_record WHERE product_code=%s", code)
  416. if rows and rows[0] and rows[0][0] is not None:
  417. return int(rows[0][0])
  418. return 0
  419. def _get_last_buy_time(pool, code: str):
  420. """取该车购买记录里最新一笔的 purchased_at。"""
  421. rows = pool.select_all(
  422. "SELECT MAX(purchased_at) FROM deca_buy_record WHERE product_code=%s", code)
  423. if rows and rows[0] and rows[0][0] is not None:
  424. return rows[0][0]
  425. return None
  426. def _parse_dt(value) -> datetime | None:
  427. """把接口/数据库时间值转成 datetime。"""
  428. if isinstance(value, datetime):
  429. return value
  430. if not value:
  431. return None
  432. try:
  433. return datetime.strptime(str(value), "%Y-%m-%d %H:%M:%S")
  434. except ValueError:
  435. return None
  436. def _count_overlap_buyers(pool, code_a: str, code_b: str) -> int:
  437. """统计两辆车去重买家交集人数。"""
  438. rows = pool.select_all(
  439. "SELECT COUNT(DISTINCT a.user_id) "
  440. "FROM deca_buy_record a "
  441. "WHERE a.product_code=%s "
  442. "AND EXISTS (SELECT 1 FROM deca_buy_record b "
  443. " WHERE b.product_code=%s AND b.user_id=a.user_id)",
  444. (code_a, code_b))
  445. if rows and rows[0] and rows[0][0] is not None:
  446. return int(rows[0][0])
  447. return 0
  448. def _count_repeat_buyers_today(pool, merchant_user_id: str, code: str,
  449. window_start: str, window_end: str) -> int:
  450. """统计当前车买家里,当日(运行窗口内)重复买过本商家其他车的人数。"""
  451. rows = pool.select_all(
  452. "SELECT COUNT(DISTINCT b.user_id) "
  453. "FROM deca_buy_record b "
  454. "WHERE b.product_code=%s "
  455. "AND EXISTS (SELECT 1 FROM deca_buy_record x "
  456. " WHERE x.merchant_user_id=%s "
  457. " AND x.user_id=b.user_id "
  458. " AND x.product_code<>%s "
  459. " AND x.purchased_at>=%s AND x.purchased_at<=%s)",
  460. (code, merchant_user_id, code, window_start, window_end))
  461. if rows and rows[0] and rows[0][0] is not None:
  462. return int(rows[0][0])
  463. return 0
  464. def _fetch_previous_realtime(pool, merchant_user_id: str, code: str, before_text: str,
  465. series_name: str | None = None) -> dict | None:
  466. """按购买记录实时找某商家在当前车之前结束的上一辆车;可选限制同 series_name。"""
  467. if not before_text:
  468. return None
  469. sql = (
  470. "SELECT b.product_code, MAX(b.title) AS title, MAX(b.purchased_at) AS ended_at, "
  471. " COALESCE(NULLIF(MAX(p.series_name), ''), NULLIF(MAX(o.series_name), '')) AS series_name "
  472. "FROM deca_buy_record b "
  473. f"LEFT JOIN {T_PROD} p ON p.product_code=b.product_code "
  474. "LEFT JOIN deca_onsale_product_record o ON o.product_code=b.product_code "
  475. "WHERE b.merchant_user_id=%s AND b.product_code<>%s AND b.purchased_at<%s")
  476. args = [merchant_user_id, code, before_text]
  477. if series_name:
  478. sql += (
  479. " AND COALESCE(NULLIF(p.series_name, ''), NULLIF(o.series_name, ''), '')=%s")
  480. args.append(series_name)
  481. sql += " GROUP BY b.product_code ORDER BY ended_at DESC LIMIT 1"
  482. rows = pool.select_all(sql, tuple(args))
  483. if not rows:
  484. return None
  485. prev_code, title, ended_at, series = rows[0]
  486. return {
  487. "product_code": prev_code,
  488. "title": title,
  489. "ended_at": ended_at,
  490. "series_name": series,
  491. }
  492. def _biz_window(dt: datetime) -> tuple[str, str]:
  493. """把某时刻归入其成交业务日窗口 [D-1 13:00, D 06:00](口径与 daily_report 一致,2026/09/01 起点由 17:00 提前到 13:00)。
  494. 夜市成交(13:00~次日06:00)算一个业务日:时刻 ≥13:00 归当日窗口起点,<06:00 归昨日窗口。战报只在
  495. 20:30~06:00 发、组齐时刻(取 buy_record 最后一单近似)必落该区间,两分支都正确。
  496. Args:
  497. dt (datetime): 参照时刻(车的组齐时刻)。
  498. Returns:
  499. tuple[str, str]: (窗口起, 窗口止),均 'YYYY-MM-DD HH:MM:SS'。
  500. """
  501. if dt.time() >= dtime(13, 0):
  502. start = datetime.combine(dt.date(), dtime(13, 0))
  503. else:
  504. start = datetime.combine(dt.date() - timedelta(days=1), dtime(13, 0))
  505. end = datetime.combine(start.date() + timedelta(days=1), dtime(6, 0))
  506. return start.strftime("%Y-%m-%d %H:%M:%S"), end.strftime("%Y-%m-%d %H:%M:%S")
  507. def _fmt_duration(seconds: int | None) -> str:
  508. """把秒数格式化成“X小时X分钟”或“X分钟”。"""
  509. if seconds is None or seconds < 0:
  510. return "-"
  511. total_minutes = max(0, int(seconds) // 60)
  512. if total_minutes == 0:
  513. return "<1分钟" # 不足 1 分钟(含 0~59 秒):显示「<1分钟」而非「0分钟」,更直观
  514. hours, minutes = divmod(total_minutes, 60)
  515. if hours:
  516. return f"{hours}小时{minutes:02d}分钟"
  517. return f"{minutes}分钟"
  518. def _milestone_used(tx, first_cap, start) -> str | None:
  519. """算「到某进度用时」,首张快照已越过该阈值(坍缩)时诚实留空。
  520. 到 X% 用时 = 首次 progress_pct≥X 的快照时刻 tx − 开售时间 start。仅当 tx 晚于该商品最早快照
  521. first_cap 时(证明我们从低于 X% 观测到穿越 X%)才可信;若首张快照就已越阈值(tx==first_cap)则坍缩留空。
  522. (与 stats/daily_report._milestone_used 同口径。)
  523. Args:
  524. tx (datetime | None): 首次 progress_pct≥阈值的快照时刻;None 表示从未达到。
  525. first_cap (datetime | None): 该商品最早一条快照时刻。
  526. start (datetime | None): 开售时间 sale_start_at。
  527. Returns:
  528. str | None: 可读用时;未达到/坍缩/时间缺失时返回 None。
  529. """
  530. if tx is None or first_cap is None or start is None:
  531. return None
  532. try:
  533. if (tx - first_cap).total_seconds() <= 0: # 首张快照即越阈值 → 坍缩,不可信
  534. return None
  535. secs = int((tx - start).total_seconds())
  536. except Exception:
  537. return None
  538. return _fmt_duration(secs) if secs >= 0 else None
  539. def _milestone_line(pool, code: str, start_dt, thresholds=(25, 50, 75)) -> str | None:
  540. """从进度表算「到 25/50/75% 各用时」,拼成一行 `📊 阶段耗时 …`。
  541. 源 deca_onsale_product_progress_record(首次 pct≥X 的快照时刻 − 开售时间;坍缩留空 —,见 _milestone_used)。
  542. 全部阈值都无值(留空/坍缩)时返回 None,不加该行。
  543. Args:
  544. pool (MySQLConnectionPool): MySQL 连接池。
  545. code (str): 商品编码。
  546. start_dt (datetime | None): 开售时间(sale_start_at);缺失则返回 None。
  547. thresholds (tuple[int], optional): 要展示的阈值。一车结束用 (25,50,75),过半用 (25,50)。
  548. Returns:
  549. str | None: 形如 "📊 阶段耗时 25% 1分钟 · 50% 2分钟 · 75% 3分钟";无任何可信值时返回 None。
  550. """
  551. if not start_dt:
  552. return None
  553. row = pool.select_one(
  554. "SELECT MIN(captured_at), "
  555. " MIN(CASE WHEN progress_pct>=25 THEN captured_at END), "
  556. " MIN(CASE WHEN progress_pct>=50 THEN captured_at END), "
  557. " MIN(CASE WHEN progress_pct>=75 THEN captured_at END) "
  558. f"FROM {PROGRESS_TABLE} WHERE product_code=%s", (code,))
  559. if not row:
  560. return None
  561. first_cap, t25, t50, t75 = row
  562. tmap = {25: t25, 50: t50, 75: t75}
  563. parts, has_value = [], False
  564. for x in thresholds:
  565. used = _milestone_used(tmap.get(x), first_cap, start_dt)
  566. if used:
  567. has_value = True
  568. parts.append(f"{x}% {used if used else '—'}")
  569. return "📊 阶段耗时 " + " · ".join(parts) if has_value else None
  570. def _half_item(log, pool, r: dict, code: str, pct: float, plain: bool) -> str:
  571. """构造一条「过半」通知文案,并在其后附加阶段耗时行(25/50%)。
  572. 过半路径本身没有开售时间,故顺带取一次详情拿 saleStartAt(详情免 token)算阶段耗时;
  573. 坍缩/无进度数据(如极快车)时不附加耗时行。
  574. Args:
  575. log: 日志对象。
  576. pool (MySQLConnectionPool): MySQL 连接池。
  577. r (dict): parse_product 产出的商品字典。
  578. code (str): 商品编码。
  579. pct (float): 当前售卖进度百分比。
  580. plain (bool): True=纯文本(PC微信)/False=markdown(企微)。
  581. Returns:
  582. str: 过半文案(可能含 `📊 阶段耗时` 追加行)。
  583. """
  584. item = _fmt_item(r, "half", pct=pct, plain=plain)
  585. d = _get_detail_data(log, code)
  586. start_dt = _parse_dt(d.get("saleStartAt")) if d else None
  587. ms = _milestone_line(pool, code, start_dt, (25, 50))
  588. return item + "\n" + ms if ms else item
  589. def _detect_and_report_ended(log, pool, existing: dict, onsale_codes: set,
  590. merchant: dict, seen: set):
  591. """检测本进程运行后消失的车、二次确认结束后逐辆发战报并置位 ended_notified。
  592. 候选条件三取交:本进程见过该商家在售(seen) ∩ 本轮已不在在售(onsale_codes) ∩ 未播报过结束
  593. (ended_notified=0)。命中的候选再打一次详情,仅 _is_sale_ended 确认售罄/过结束时间才播报——
  594. 防「全站抓取抖动导致某车临时缺席」被误判为结束。发送成功才置位;失败保持 0,下轮重发不丢。
  595. Args:
  596. log: 日志对象。
  597. pool (MySQLConnectionPool): MySQL 连接池。
  598. existing (dict): 库内该商家记录 {code: {ended/title/mname/unit_price/card/sold/...}}。
  599. onsale_codes (set): 本轮在售的 product_code 集合。
  600. merchant (dict): 当前商家配置 {"id","name","tag"},用于兜底商家名/id 与战报标题标签。
  601. seen (set): 该商家「本进程见过在售」的 product_code 集合(按商家隔离)。
  602. """
  603. now_ts = int(time.time())
  604. for code, info in existing.items():
  605. if info["ended"] == 1:
  606. continue # 已播报过结束,跳过
  607. if code not in seen:
  608. continue # 本进程运行后没见它在售 → 属「之前的车」,不补发
  609. if code in onsale_codes:
  610. continue # 本轮仍在售,未结束
  611. # 疑似消失:二次打详情确认(失败/仍在售都不判结束,下轮再看,防抓取抖动误报)
  612. data = _get_detail_data(log, code)
  613. ended, reason = _is_sale_ended(data, now_ts)
  614. if not ended:
  615. log.info(f"[疑似消失未确认结束] {code} 详情仍在售或取详情失败,跳过(可能抓取抖动) | {info['title']}")
  616. continue
  617. # 【2026/08/31】soldCount==0 的车不发结束战报:这类是「上架后无人问津、被商家撤下或到 saleEndAt
  618. # 到期」的流车,从没成交(既不在在售 is_on_sale=0、也不在已售成交列表),购买记录必空。旧逻辑
  619. # 「结束即全部售出」会把它虚报成「售出<总份数>件 / 0 人参与拆卡」,属误发(主公反馈老是误发)。
  620. # 故 soldCount==0 时静默不发,并置位 ended_notified 停止下轮重复对账/重打详情。
  621. # 只在详情明确返回 soldCount==0 时静默;soldCount 缺失(None)不静默,保守走正常播报,防抓取抖动误吞真售罄车。
  622. sold_now = data.get("soldCount")
  623. if sold_now == 0:
  624. _mark_ended_notified(pool, code)
  625. log.info(f"[结束静默] {code} soldCount=0 流车(无人成交/撤下),不发战报并置位防重扫 | {info['title']}")
  626. continue
  627. end_dt = _parse_dt(_get_last_buy_time(pool, code))
  628. end_text = end_dt.strftime("%Y-%m-%d %H:%M:%S") if end_dt else ""
  629. sale_start_text = data.get("saleStartAt") or ""
  630. sale_start_dt = _parse_dt(sale_start_text)
  631. # 售出件数=总份数(结束即全部售出):优先详情最新 cardCount,缺失回退库内旧值
  632. card = data.get("cardCount")
  633. if card is None:
  634. card = info["card"]
  635. # 价格:区间价(minUnitPrice~maxUnitPrice)优先,详情取不到再回退库内 unit_price
  636. price_text = _fmt_price(data.get("unitPrice"), data.get("minUnitPrice"),
  637. data.get("maxUnitPrice"))
  638. if price_text == "¥-" and info["unit_price"] is not None:
  639. price_text = _fmt_price(info["unit_price"], None, None)
  640. buyers = _count_buyers(pool, code) # 该车去重购买人数(参与拆卡人数)
  641. item_text = _fmt_ended(info["title"], card, price_text, buyers,
  642. plain=(SEND_CHANNEL == "pc"))
  643. extra_lines = []
  644. if end_dt:
  645. if sale_start_dt:
  646. duration_text = _fmt_duration(int((end_dt - sale_start_dt).total_seconds()))
  647. extra_lines.append(f"⏱ 组齐时间 {end_dt:%m-%d %H:%M} | 共花费 {duration_text}")
  648. else:
  649. extra_lines.append(f"⏱ 组齐时间 {end_dt:%m-%d %H:%M} | 共花费 -")
  650. else:
  651. extra_lines.append("⏱ 组齐时间 未知 | 共花费 -")
  652. # 阶段耗时 25/50/75%(100% 即共花费,不重复);源进度表,坍缩/无进度数据则不加此行
  653. ms_line = _milestone_line(pool, code, sale_start_dt, (25, 50, 75))
  654. if ms_line:
  655. extra_lines.append(ms_line)
  656. current_series = (data.get("giftInfo") or {}).get("items") or [{}]
  657. current_series_name = ""
  658. if current_series:
  659. first = current_series[0] or {}
  660. current_series_name = (first.get("seriesName") or "").strip()
  661. mid = info["mid"] or merchant["id"]
  662. # 先「商家上一辆」(商家真正紧邻的上一车,优先级更高),再「同系列上一辆」
  663. merchant_prev = _fetch_previous_realtime(pool, mid, code, end_text)
  664. if merchant_prev:
  665. merchant_overlap = _count_overlap_buyers(pool, code, merchant_prev["product_code"])
  666. extra_lines.append(
  667. f"🔁 商家上一辆 {merchant_prev['title']}(重复 {merchant_overlap} 人)")
  668. else:
  669. extra_lines.append("🔁 商家上一辆 无(重复 0 人)")
  670. same_series_prev = _fetch_previous_realtime(
  671. pool, mid, code, end_text, current_series_name or None)
  672. if same_series_prev:
  673. same_series_overlap = _count_overlap_buyers(
  674. pool, code, same_series_prev["product_code"])
  675. extra_lines.append(
  676. f"🔁 同系列上一辆 {same_series_prev['title']}(重复 {same_series_overlap} 人)")
  677. else:
  678. extra_lines.append("🔁 同系列上一辆 无(重复 0 人)")
  679. # 📅 当日重复购买(改进1,2026/09/01):按「组齐时刻」所在业务日窗口 [D-1 13:00, D 06:00]、仅本商家,
  680. # 口径与 daily_report 一致;修掉原先用 _window_start(datetime.now()) 造成的窗口错位(补发/非当晚结束车会误算 0)。
  681. if end_dt:
  682. win_s, win_e = _biz_window(end_dt)
  683. repeat_today = _count_repeat_buyers_today(pool, mid, code, win_s, win_e)
  684. extra_lines.append(f"📅 当日重复购买 {repeat_today} 人")
  685. else:
  686. extra_lines.append("📅 当日重复购买 - 人")
  687. item_text = item_text + "\n" + "\n".join(extra_lines)
  688. if _send_ended(log, info["mname"] or merchant["name"], item_text, merchant["tag"]):
  689. _mark_ended_notified(pool, code)
  690. log.info(f"[结束战报已发] {code} {reason} 售出{card}件 {buyers}人 | {info['title']}")
  691. else:
  692. log.warning(f"[结束战报发送失败] {code} 保持 ended_notified=0,下轮重发 | {info['title']}")
  693. def _run_merchant(log, pool, merchant: dict):
  694. """跑单个商家一轮:拉在售 → 结束对账(发一车结束战报) → 比对库内状态 → 发新品/过半提醒 → 落库。
  695. 多商家共用同一套逻辑,仅按 merchant 参数化:抓哪家(id)、消息落款(name)、标题标签(tag)、
  696. 以及按商家隔离的「见过在售」集合(_seen_onsale[id])。三类提醒标题统一加 【tag】前缀区分同名商家。
  697. Args:
  698. log: 日志对象。
  699. pool (MySQLConnectionPool): MySQL 连接池。
  700. merchant (dict): 商家配置 {"id","name","tag"},来自 MERCHANTS 列表。
  701. """
  702. mid = merchant["id"]
  703. tag = merchant["tag"]
  704. products, ok = fetch_onsale(log, mid)
  705. if not ok:
  706. log.warning(f"[{tag}] 在售抓取未取全(请求异常),本轮跳过该商家:不做新品/过半/结束判断,避免误判车下架")
  707. return
  708. # 库内该商家已监控商品:含结束标记与结束战报所需的静态字段(标题/商家名/单价/份数)
  709. existing_rows = pool.select_all(
  710. "SELECT product_code, new_notified, half_notified, ended_notified, "
  711. "title, merchant_user_id, merchant_name, unit_price, card_count, sold_count "
  712. f"FROM {T_ALERT} WHERE merchant_user_id=%s", mid) or []
  713. existing = {}
  714. for code, new, half, ended, title, mid_, mname_, uprice, ccount, scount in existing_rows:
  715. existing[code] = {"new": new, "half": half, "ended": ended,
  716. "title": title, "mid": mid_, "mname": mname_, "unit_price": uprice,
  717. "card": ccount, "sold": scount}
  718. is_cold = len(existing) == 0 # 冷启动:库内该商家零记录(仅作日志提示,逻辑与常规轮一致)
  719. if is_cold:
  720. log.info(f"[{tag}] 首次运行:库内该商家零记录,只提醒本轮窗口起点({RUN_START:%H:%M})后新上架的商品,老货静默建档")
  721. seen = _seen_onsale.setdefault(mid, set()) # 该商家「本进程见过在售」的 code 集合(按商家隔离)
  722. onsale_codes = {r["product_code"] for r in products} # 本轮在售 code 集合
  723. # 结束对账:只播报本进程运行后见过在售、之后确认结束的车(历史已结束车不补发;置位防重发)
  724. _detect_and_report_ended(log, pool, existing, onsale_codes, merchant, seen)
  725. seen.update(onsale_codes) # 本轮在售并入「见过在售」集合,供下轮结束对账(实现「只管运行后」)
  726. if not products:
  727. log.info(f"[{tag}] 本轮无在售商品,已完成结束对账,跳过新品/过半")
  728. return
  729. plain = SEND_CHANNEL == "pc" # PC 微信用纯文本,企微用 markdown
  730. new_items = [] # 新品上架提醒文案
  731. half_items = [] # 进度过半提醒文案
  732. pending_new = [] # 待发新品的 product_code:仅在提醒发送成功后才置 new_notified=1
  733. window_start = _window_start(datetime.now()) # 「新上架」时间门槛:本轮窗口起点(最近的 RUN_START,默认 20:30)
  734. for r in products:
  735. code = r["product_code"]
  736. ratio = _progress(r["sold_count"], r["card_count"])
  737. pct = ratio * 100
  738. over_half = ratio >= HALF_THRESHOLD
  739. if code not in existing:
  740. # 库里没有的候选:打一次详情拿上架时间,只有 publishAt 晚于窗口起点(RUN_START,默认 20:30)才算「本轮新上架」
  741. publish_at = get_publish_at(log, code)
  742. if _is_new_arrival(publish_at, window_start):
  743. # 真·本轮新上架:new_notified 先记 0,发送成功再置 1(发失败下轮自动重发,不丢)
  744. _insert_alert(pool, r, pct, new_notified=0,
  745. half_notified=1 if over_half else 0)
  746. new_items.append(_fmt_item(r, "new", plain=plain))
  747. pending_new.append(code)
  748. if over_half: # 新品上架即已过半,一并提示(half 逻辑保持原状)
  749. half_items.append(_half_item(log, pool, r, code, pct, plain))
  750. else:
  751. # 上架早于窗口起点(或拿不到上架时间)的老货:静默建档、不提醒(new_notified 直接置 1,避免下轮反复判断)
  752. _insert_alert(pool, r, pct, new_notified=1,
  753. half_notified=1 if over_half else 0)
  754. log.info(f"[{tag}][静默建档] {code} 上架 {publish_at or '未知'} 早于窗口起点 {window_start:%m-%d %H:%M},不提醒 | {r['title']}")
  755. else:
  756. # 已在表:new_notified=0 视为「待发新品」(上次发失败残留 或 人工改回 0),重新纳入新品提醒
  757. if existing[code]["new"] == 0:
  758. new_items.append(_fmt_item(r, "new", plain=plain))
  759. pending_new.append(code)
  760. # 过半逻辑保持原状(half_notified 策略不改动)
  761. already_half = existing[code]["half"] == 1
  762. if over_half and not already_half:
  763. _update_progress(pool, r, pct, half_notified=1)
  764. half_items.append(_half_item(log, pool, r, code, pct, plain))
  765. else:
  766. _update_progress(pool, r, pct, half_notified=1 if already_half else 0)
  767. mname = products[0].get("merchant_name") or merchant["name"]
  768. # 常规轮次推送;sent_ok 记新品提醒是否发送成功,决定要不要置位 new_notified
  769. sent_ok = False
  770. if SEND_CHANNEL == "pc":
  771. # PC 微信:新品 + 过半合并成一条消息发出(减少操作,一次发完)
  772. if new_items or half_items:
  773. log.info(f"[{tag}] 合并推送(pc):新品{len(new_items)}个、过半{len(half_items)}个")
  774. sent_ok = _dispatch_pc_combined(log, mname, new_items, half_items, tag)
  775. else:
  776. # 企微渠道:仍按两类各发一条 markdown(新品这条成功与否决定 sent_ok),标题加【tag】前缀区分同名商家
  777. if new_items:
  778. log.info(f"[{tag}] 新商品上架 {len(new_items)} 个,推送(qywx)")
  779. sent_ok = _dispatch(log, new_items, f"🆕【{tag}】得卡[{mname}] 新商品上架 {len(new_items)} 个")
  780. if half_items:
  781. log.info(f"[{tag}] 进度过半 {len(half_items)} 个,推送(qywx)")
  782. _dispatch(log, half_items, f"🔥【{tag}】得卡[{mname}] 拼团进度过半 {len(half_items)} 个")
  783. # 新品提醒发送成功后才置位 new_notified=1;失败则保持 0,下一轮继续重发(不丢)
  784. if pending_new:
  785. if sent_ok:
  786. _mark_new_notified(pool, pending_new)
  787. log.info(f"[{tag}] 新品提醒发送成功,new_notified 置 1:{len(pending_new)} 个")
  788. else:
  789. log.warning(f"[{tag}] 新品提醒发送失败,new_notified 保持 0,下轮重发:{len(pending_new)} 个")
  790. if not new_items and not half_items:
  791. log.info(f"[{tag}] 本轮无新品、无新达标过半商品")
  792. def run_once(log, pool):
  793. """跑一轮监控:遍历 MERCHANTS 中每个商家各跑一轮(单家异常不拖垮其他家)。
  794. Args:
  795. log: 日志对象。
  796. pool (MySQLConnectionPool): MySQL 连接池。
  797. """
  798. for merchant in MERCHANTS:
  799. try:
  800. _run_merchant(log, pool, merchant)
  801. except Exception as e:
  802. log.error(f"[{merchant['tag']}] 处理异常,跳过该商家本轮: {e}")
  803. @retry(stop=stop_after_attempt(100), wait=wait_fixed(600), after=core.after_log)
  804. def main_task(log):
  805. """在售监控主函数:建连接池 → 跑一轮监控(挂了每 10 分钟重试)。
  806. Args:
  807. log: 日志对象。
  808. Raises:
  809. RuntimeError: 数据库连接池异常时抛出以触发重试。
  810. """
  811. log.info(f"开始运行 {sys._getframe().f_code.co_name} 在售提醒监控" + "." * 40)
  812. pool = MySQLConnectionPool(log=log)
  813. core.init_account_pool(pool, task_tag="onsale_alert") # 启用账号池:need_auth 请求走 20 号池 + 各号专属 IP
  814. if not pool.check_pool_health():
  815. log.error("数据库连接池异常")
  816. raise RuntimeError("数据库连接池异常")
  817. try:
  818. run_once(log, pool)
  819. except Exception as e:
  820. log.error(f"{sys._getframe().f_code.co_name} error: {e}")
  821. finally:
  822. log.info(f"在售提醒监控 {sys._getframe().f_code.co_name} 运行结束,等待下一轮" + "." * 20)
  823. def _in_run_window(now: datetime) -> bool:
  824. """判断当前时刻是否在运行窗口 [RUN_START, 次日 RUN_END) 内。
  825. Args:
  826. now (datetime): 当前时间。
  827. Returns:
  828. bool: 在窗口内返回 True。窗口跨午夜,故「晚于开始 或 早于结束」即算命中。
  829. """
  830. t = now.time()
  831. return t >= RUN_START or t < RUN_END
  832. def _seconds_to_window(now: datetime) -> int:
  833. """计算从 now 到下一次窗口开始(当天 RUN_START)的休眠秒数(仅窗口外调用)。
  834. Args:
  835. now (datetime): 当前时间。
  836. Returns:
  837. int: 需休眠的秒数;若当天 RUN_START 已过则顺延到次日。
  838. """
  839. start = now.replace(hour=RUN_START.hour, minute=RUN_START.minute, second=0, microsecond=0)
  840. if start <= now: # 当天 RUN_START 已过 → 顺延到次日同一时刻
  841. start += timedelta(days=1)
  842. return int((start - now).total_seconds())
  843. def schedule_task():
  844. """常驻循环:仅在每天 RUN_START~次日 RUN_END 运行;窗口内每轮随机间隔轮询,窗口外休眠到下次开窗。"""
  845. while True:
  846. now = datetime.now()
  847. if not _in_run_window(now):
  848. wait = _seconds_to_window(now)
  849. logger.info(f"当前不在运行窗口({RUN_START:%H:%M}~次日{RUN_END:%H:%M}),休眠 {wait}s 到 {RUN_START:%H:%M} 再跑")
  850. time.sleep(wait)
  851. continue
  852. main_task(log=logger)
  853. wait = random.randint(MIN_INTERVAL_SEC, MAX_INTERVAL_SEC) # 每轮随机间隔(秒)
  854. logger.info(f"下一轮 {wait}s 后运行")
  855. time.sleep(wait)
  856. def _parse_start_time(text: str) -> dtime:
  857. """把命令行传入的开始时间文本解析为 datetime.time。
  858. Args:
  859. text (str): 开始时间文本,格式 "HH:MM" 或 "HH:MM:SS",如 "20:30" / "17:00"。
  860. Returns:
  861. dtime: 解析出的 time 对象。
  862. Raises:
  863. argparse.ArgumentTypeError: 格式非法(非 HH:MM[:SS] 或时分秒越界)时抛出,供 argparse 提示用户。
  864. """
  865. text = text.strip()
  866. for fmt in ("%H:%M:%S", "%H:%M"):
  867. try:
  868. return datetime.strptime(text, fmt).time()
  869. except ValueError:
  870. continue
  871. raise argparse.ArgumentTypeError(f"开始时间格式非法:{text!r},应为 HH:MM 或 HH:MM:SS,如 20:30")
  872. def _parse_args() -> argparse.Namespace:
  873. """解析命令行参数,取运行窗口开始时间(默认 20:30)。
  874. 支持位置参数与 --start 两种写法,二者等价,方便直接 `python xxx.py 17:00`。
  875. Returns:
  876. argparse.Namespace: 含 start(datetime.time) 属性;未传时为默认 RUN_START。
  877. """
  878. parser = argparse.ArgumentParser(
  879. description="得卡 DECA 在售提醒:可指定运行窗口开始时间(该时间后的新上架才提醒)")
  880. parser.add_argument(
  881. "start", nargs="?", type=_parse_start_time, default=None,
  882. help="运行窗口开始时间 HH:MM[:SS],默认 20:30;位置参数写法,如 17:00")
  883. parser.add_argument(
  884. "--start", dest="start_opt", type=_parse_start_time, default=None,
  885. help="运行窗口开始时间 HH:MM[:SS],与位置参数等价,如 --start 17:00")
  886. return parser.parse_args()
  887. if __name__ == "__main__":
  888. # logger.add(sys.stderr, level="INFO") # 控制台同步输出,便于观察
  889. _args = _parse_args()
  890. # 位置参数优先,其次 --start,都未传则保持默认 RUN_START(20:30)
  891. _start = _args.start or _args.start_opt
  892. if _start is not None:
  893. RUN_START = _start # 覆盖模块级默认,窗口判定与「新上架」门槛均随之改变
  894. logger.info(f"运行窗口开始时间由命令行指定为 {RUN_START:%H:%M}")
  895. schedule_task()