3c32720985
所有站点 with_retry 在最后一次重试失败、reset 之前自动截图， 保存到 logs/screenshots/。网页站点走 Playwright page.screenshot()， 安能走 CDP Page.captureScreenshot。截图失败绝不阻塞任务流程。 - paths.py: 新增 SCREENSHOT_DIR (BASE_DIR/logs/screenshots/) - runtime.py: 新增 capture_error_screenshot() 工具函数 - .gitignore: 新增 logs/ 忽略规则 Co-Authored-By: Claude <noreply@anthropic.com>
744 lines
31 KiB
Python
744 lines
31 KiB
Python
# runtime.py
|
||
# 阶段1：服务化共享核心。把"启动浏览器 + 各站就绪 + 弹窗 + 心跳初值"
|
||
# (launch_and_prepare)、"执行一条任务"(dispatch_task)、"一轮心跳"(run_heartbeat)
|
||
# 抽出来，供
|
||
# - cli/router.py（交互菜单模式：调试 / 人工操作）
|
||
# - cli/server.py（FastAPI 服务模式：常驻 + 接收 API 指令）
|
||
# 共同复用，避免两处重复维护。
|
||
#
|
||
# 线程模型：launch_and_prepare 内 sync_playwright().start() 必须在"持有 Playwright 的
|
||
# 线程"调用（交互模式=主线程；服务模式=worker 线程）。该线程独占所有 page 操作；
|
||
# 其他线程（如 FastAPI 路由）只能经 task_queue + state_store 与之通信，绝不跨线程
|
||
# 访问 page。
|
||
|
||
import os
|
||
import socket
|
||
import subprocess
|
||
import time
|
||
import urllib.request
|
||
from datetime import date, datetime, timedelta
|
||
|
||
import yaml
|
||
from playwright.sync_api import sync_playwright
|
||
|
||
from inbound_verify.paths import CONFIG_PATH, SCREENSHOT_DIR
|
||
|
||
from inbound_verify import state_store
|
||
from inbound_verify.sites import shunxin, baishi, zto, yunda, anneng
|
||
from inbound_verify import compare # dispatch 的 compare 任务用
|
||
|
||
# 各网页站点首页 URL（单一来源，取自各站点模块 HOME_URL）
|
||
SITES_CONFIG = {
|
||
"顺心": shunxin.HOME_URL,
|
||
"百世": baishi.HOME_URL,
|
||
"中通": zto.HOME_URL,
|
||
"韵达": yunda.HOME_URL,
|
||
}
|
||
|
||
# 站点就绪特征：登录成功进入工作台后的标志性控件
|
||
READY_SELECTORS = {
|
||
"顺心": 'h1:has-text("盟商门户网")',
|
||
"百世": 'h1[title="百世快运"]',
|
||
"中通": '.logo:has-text("网点版")',
|
||
"韵达": '.el-menu-item:has-text("首页")',
|
||
}
|
||
|
||
# 安能是 Electron 桌面应用（不是 Playwright 网页），单独启动
|
||
APP_SITES = {"安能"}
|
||
|
||
# 心跳间隔（秒）
|
||
HEARTBEAT_INTERVAL = 30
|
||
|
||
|
||
# ============================ 安能启动（CDP）============================
|
||
|
||
|
||
def _find_free_port():
|
||
"""让操作系统分配一个空闲端口，避免固定端口冲突。"""
|
||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
||
s.bind(("127.0.0.1", 0))
|
||
return s.getsockname()[1]
|
||
|
||
|
||
def _wait_cdp_up(port, timeout=60.0):
|
||
"""轮询直到 CDP 调试端口就绪（应用启动需要时间）。"""
|
||
deadline = time.monotonic() + timeout
|
||
while time.monotonic() < deadline:
|
||
try:
|
||
with urllib.request.urlopen(
|
||
f"http://localhost:{port}/json/version", timeout=2
|
||
) as resp:
|
||
if resp.status == 200:
|
||
return True
|
||
except Exception:
|
||
pass
|
||
time.sleep(1)
|
||
return False
|
||
|
||
|
||
def launch_anneng(app_path):
|
||
"""以调试模式启动安能 Electron 应用（自动选取空闲端口），返回子进程对象。"""
|
||
# 【环境兼容】WorkBuddy 等 shell 会注入 NODE_OPTIONS（含 --use-system-ca），
|
||
# Electron 内置 Node 拒绝该 flag 导致安能启动即退出（rc=9）。
|
||
# 拉起前从子进程环境里摘掉 NODE_OPTIONS。
|
||
anneng_env = os.environ.copy()
|
||
anneng_env.pop("NODE_OPTIONS", None)
|
||
port = _find_free_port()
|
||
print(f">> 以调试模式启动【安能】应用（端口 {port}）：{app_path}")
|
||
proc = subprocess.Popen(
|
||
[app_path, f"--remote-debugging-port={port}"], env=anneng_env
|
||
)
|
||
anneng.set_cdp_port(port)
|
||
if not _wait_cdp_up(port):
|
||
raise RuntimeError(
|
||
f"安能应用调试端口 {port} 未就绪——可能应用已在运行（单实例），"
|
||
"请先关闭已有的安能窗口再试"
|
||
)
|
||
return proc
|
||
|
||
|
||
# ============================ 探测函数 ============================
|
||
|
||
|
||
def probe_site_login(site_name, pages_map):
|
||
"""探测单站是否登录（复用就绪判据）。任何异常一律返回 False。
|
||
|
||
只在 Playwright 所属线程调用。
|
||
"""
|
||
try:
|
||
if site_name not in pages_map:
|
||
return False
|
||
if site_name == "安能":
|
||
return anneng.anneng_ready()
|
||
if site_name == "顺心":
|
||
return all(
|
||
pg.locator(READY_SELECTORS["顺心"]).is_visible(timeout=500)
|
||
for pg in pages_map["顺心"]
|
||
)
|
||
return (
|
||
pages_map[site_name]
|
||
.locator(READY_SELECTORS[site_name])
|
||
.is_visible(timeout=500)
|
||
)
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
# ============================ 运行上下文 ============================
|
||
|
||
|
||
class RuntimeContext:
|
||
"""launch_and_prepare 的返回值，持有 Playwright 运行所需对象。"""
|
||
|
||
def __init__(
|
||
self,
|
||
pw,
|
||
browser,
|
||
pages_map,
|
||
ready_status,
|
||
anneng_proc,
|
||
sites_to_watch,
|
||
debug_mode,
|
||
debug_target,
|
||
foreground=True,
|
||
):
|
||
self.pw = pw
|
||
self.browser = browser
|
||
self.pages_map = pages_map
|
||
self.ready_status = ready_status
|
||
self.anneng_proc = anneng_proc
|
||
self.sites_to_watch = sites_to_watch
|
||
self.debug_mode = debug_mode
|
||
self.debug_target = debug_target
|
||
# True=任务执行时把 page 置顶（交互调试）；False=后台静默不置顶（服务模式，避免抢焦点）
|
||
self.foreground = foreground
|
||
|
||
def stop(self):
|
||
"""关闭 browser + 安能 + Playwright。退出时调用。"""
|
||
try:
|
||
self.browser.close()
|
||
except Exception:
|
||
pass
|
||
if self.anneng_proc is not None:
|
||
try:
|
||
self.anneng_proc.terminate()
|
||
print("已关闭安能应用。")
|
||
except Exception:
|
||
pass
|
||
try:
|
||
self.pw.stop()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# ============================ 启动 + 就绪 + 弹窗 ============================
|
||
|
||
|
||
def seed_legacy_config():
|
||
"""一次性：把 config.yaml 里的站点配置（百世密码 / 韵达账密 / 安能 exe 路径）
|
||
灌入 state.db。已存在的值不覆盖（前端改过的不动）。"""
|
||
if not os.path.exists(CONFIG_PATH):
|
||
return
|
||
try:
|
||
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
|
||
cfg = yaml.safe_load(f) or {}
|
||
except Exception as e:
|
||
print(f"⚠️ 读取 config.yaml 做 seed 失败: {e}")
|
||
return
|
||
items = [
|
||
("百世", "password", (cfg.get("baishi", {}) or {}).get("password", "")),
|
||
("韵达", "username", (cfg.get("yunda", {}) or {}).get("username", "")),
|
||
("韵达", "password", (cfg.get("yunda", {}) or {}).get("password", "")),
|
||
("安能", "app_path", (cfg.get("anneng", {}) or {}).get("app_path", "")),
|
||
]
|
||
seeded = []
|
||
for site, key, val in items:
|
||
if val and not state_store.get_setting(site, key):
|
||
state_store.set_setting(site, key, str(val))
|
||
seeded.append(f"{site}.{key}")
|
||
if seeded:
|
||
print(f">> [seed] 从 config.yaml 灌入站点配置: {', '.join(seeded)}")
|
||
|
||
|
||
def launch_and_prepare(debug_mode=False, debug_target="", foreground=True):
|
||
"""启动 Playwright + 各站 page + 就绪轮询 + 弹窗清理 + 心跳初值，返回 RuntimeContext。
|
||
|
||
必须在"持有 Playwright 的线程"调用（交互模式主线程 / 服务模式 worker 线程）。
|
||
阻塞至所有站点登录就绪才返回。
|
||
|
||
foreground：True=任务执行时把 page 置顶（交互调试）；False=后台静默不置顶（服务模式，
|
||
避免抢用户焦点）。仅控制任务执行阶段的 bring_to_front；启动登录/初始弹窗清理的置顶
|
||
不受影响（启动时窗口需对用户可见以便登录）。
|
||
"""
|
||
# 0. 状态库建表/迁移 + 从 config.yaml 灌入站点配置（须在 reset_login_states 等之前）
|
||
state_store.init_db()
|
||
seed_legacy_config()
|
||
|
||
# 1. 读 debug 配置（config.yaml；服务模式也生效）+ 安能路径（state.db，seed 已灌入）
|
||
if os.path.exists(CONFIG_PATH):
|
||
try:
|
||
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
|
||
_cfg = yaml.safe_load(f) or {}
|
||
_dbg = _cfg.get("debug", {}) or {}
|
||
if _dbg.get("enabled"):
|
||
debug_mode = True
|
||
debug_target = str(_dbg.get("target_site", "") or "")
|
||
except Exception as e:
|
||
print(f"⚠️ 读取 config.yaml(debug) 失败: {e}")
|
||
anneng_app_path = state_store.get_setting("安能", "app_path")
|
||
|
||
# 2. 确定要挂载的网页站点 + 安能标记
|
||
anneng_active = False
|
||
if debug_mode:
|
||
if debug_target in SITES_CONFIG:
|
||
print(f"\n🛠️ 【调试模式】仅加载目标站点: [{debug_target}]")
|
||
active_sites = {debug_target: SITES_CONFIG[debug_target]}
|
||
elif debug_target == "安能":
|
||
print(f"\n🛠️ 【调试模式】仅加载目标站点: [安能]")
|
||
active_sites = {}
|
||
anneng_active = True
|
||
else:
|
||
active_sites = dict(SITES_CONFIG)
|
||
anneng_active = True
|
||
else:
|
||
active_sites = dict(SITES_CONFIG)
|
||
anneng_active = True
|
||
|
||
if anneng_active and not anneng_app_path:
|
||
print("⚠️ 已启用安能但 config.yaml 未配置 anneng.app_path，将跳过安能。")
|
||
anneng_active = False
|
||
|
||
# 2.5 重置各站登录态为 unknown：登录态是会话级的，避免显示上一会话残留的陈旧登录态。
|
||
# 数据态（文件就绪）会话无关、保留不动；心跳就绪后会重新探测写真实值。
|
||
reset_sites = list(active_sites.keys())
|
||
if anneng_active:
|
||
reset_sites.append("安能")
|
||
state_store.reset_login_states(reset_sites)
|
||
|
||
# 3. 启动 Playwright（不用 with，改 .start()，由 RuntimeContext.stop() 收尾）
|
||
pw = sync_playwright().start()
|
||
# 调试模式临时开启 CDP 端口，便于外部 Playwright（如 Playwright CLI 技能）
|
||
# 通过 connectOverCDP 挂载到已登录的页面读取内容。仅调试模式生效，不影响服务模式。
|
||
_launch_args = ["--remote-debugging-port=9223"] if debug_mode else []
|
||
browser = pw.chromium.launch(headless=False, args=_launch_args)
|
||
context = browser.new_context(viewport={"width": 1920, "height": 1080})
|
||
# 默认禁用麦克风/摄像头：在每个页面/iframe 加载前覆盖 getUserMedia 为“直接拒绝”，
|
||
# 这样站点（如韵达登录/工作台会请求麦克风）调用时立即 NotAllowedError，Chromium 不再
|
||
# 弹出系统授权窗，且麦克风被真正挡住（不是授权给它）。物流工作台无需音视频采集。
|
||
context.add_init_script(
|
||
"(()=>{const d=()=>Promise.reject(new DOMException('Permission disabled','NotAllowedError'));"
|
||
"if(navigator.mediaDevices)navigator.mediaDevices.getUserMedia=d;"
|
||
"for(const k of ['getUserMedia','webkitGetUserMedia','mozGetUserMedia']){"
|
||
"if(typeof navigator[k]==='function')navigator[k]=function(){return d();};}})();"
|
||
)
|
||
|
||
pages_map = {}
|
||
print("\n====================================================")
|
||
print("【启动】正在打开各站点页面...")
|
||
print("====================================================")
|
||
|
||
def _open_page(label, attempts=3):
|
||
"""开一个页面并 goto(url, domcontentloaded)；容忍瞬时 DNS/超时抖动重试，全失败才抛。
|
||
domcontentloaded：DOM 就绪即返回，不等慢资源(广告/图片)的 load 事件，
|
||
避免某站 load 超 30s / 瞬时 DNS 失败导致整个 worker 启动失败。就绪轮询自会判登录态。"""
|
||
last = None
|
||
for i in range(1, attempts + 1):
|
||
pg = context.new_page()
|
||
try:
|
||
pg.goto(url, wait_until="domcontentloaded")
|
||
return pg
|
||
except Exception as e:
|
||
last = e
|
||
try:
|
||
pg.close()
|
||
except Exception:
|
||
pass
|
||
print(f"⚠️ 打开【{label}】第 {i}/{attempts} 次失败: {e}")
|
||
if i < attempts:
|
||
time.sleep(2)
|
||
raise RuntimeError(f"打开【{label}】连续 {attempts} 次失败: {last}")
|
||
|
||
for site_name, url in active_sites.items():
|
||
if site_name == "顺心":
|
||
# 顺心：两个归属地账号在同一窗口各开一个标签页
|
||
sx_pages = []
|
||
for acct in range(1, 3):
|
||
print(f">> 正在启动【顺心】账号{acct}标签页: {url}")
|
||
sx_pages.append(_open_page(f"顺心账号{acct}"))
|
||
pages_map["顺心"] = sx_pages
|
||
else:
|
||
print(f">> 正在启动【{site_name}】页面: {url}")
|
||
pages_map[site_name] = _open_page(site_name)
|
||
|
||
# 4. 安能 Electron
|
||
anneng_proc = None
|
||
if anneng_active:
|
||
try:
|
||
anneng_proc = launch_anneng(anneng_app_path)
|
||
pages_map["安能"] = True # 哨兵：已启动（无 Playwright page）
|
||
except Exception as e:
|
||
print(f"⚠️ 启动安能应用失败，已跳过安能：{e}")
|
||
anneng_active = False
|
||
|
||
# 5. 韵达前置自动登录
|
||
print("\n====================================================")
|
||
print("【登录检测】正在准备各站点登录...")
|
||
print("====================================================")
|
||
if "韵达" in pages_map:
|
||
try:
|
||
pages_map["韵达"].bring_to_front()
|
||
yunda.yunda_login(pages_map["韵达"])
|
||
except Exception as e:
|
||
print(f" ⚠️ 韵达前置自动登录模块发生波动: {e}")
|
||
|
||
# 6. 就绪轮询（复用 probe_site_login）
|
||
ready_status = {site: False for site in active_sites.keys()}
|
||
if anneng_active:
|
||
ready_status["安能"] = False
|
||
print("\n>> 正在轮询各站点就绪状态 (自动登录或手动登录均可)...")
|
||
while not all(ready_status.values()):
|
||
for site_name in list(ready_status.keys()):
|
||
if ready_status[site_name]:
|
||
continue
|
||
if probe_site_login(site_name, pages_map):
|
||
ready_status[site_name] = True
|
||
print(f" ✅ 【{site_name}】已检测到主页，登录就绪。")
|
||
pending = [s for s, r in ready_status.items() if not r]
|
||
if pending:
|
||
print(
|
||
f" ⏳ 等待以下站点完成登录: [{', '.join(pending)}] ... "
|
||
"(请在浏览器/应用中操作)"
|
||
)
|
||
time.sleep(3)
|
||
|
||
# 7. 初始弹窗清理
|
||
print("\n====================================================")
|
||
print("【准备】所有站点已就绪，正在清理初始弹窗...")
|
||
print("====================================================")
|
||
_dismiss_initial_popups(pages_map)
|
||
for s in ("中通", "韵达", "安能"):
|
||
if s in pages_map:
|
||
print(f" ✅ 【{s}】已就绪。")
|
||
|
||
# 8. 心跳初值（就绪轮询刚通过 → 各站视为已登录；init_db 已在启动时完成）
|
||
sites_to_watch = list(ready_status.keys())
|
||
for _site in sites_to_watch:
|
||
state_store.set_login_state(_site, True)
|
||
|
||
return RuntimeContext(
|
||
pw,
|
||
browser,
|
||
pages_map,
|
||
ready_status,
|
||
anneng_proc,
|
||
sites_to_watch,
|
||
debug_mode,
|
||
debug_target,
|
||
foreground,
|
||
)
|
||
|
||
|
||
def _dismiss_initial_popups(pages_map):
|
||
"""顺心 / 百世 / 韵达 初始弹窗清理。"""
|
||
if "顺心" in pages_map:
|
||
for acct, sx_page in enumerate(pages_map["顺心"], start=1):
|
||
try:
|
||
sx_page.bring_to_front()
|
||
print(f">> 正在处理【顺心】账号{acct}弹窗与遮罩...")
|
||
sx_page.locator("a").nth(4).click(timeout=2000)
|
||
sx_page.wait_for_timeout(500)
|
||
sx_page.get_by_role("button", name="Close").click(timeout=2000)
|
||
sx_page.wait_for_timeout(500)
|
||
sx_page.get_by_role("button", name="不再询问").click(timeout=2000)
|
||
print(f" ✅ 【顺心】账号{acct}初始弹窗处理完成。")
|
||
except Exception:
|
||
pass
|
||
if "百世" in pages_map:
|
||
try:
|
||
bs_page = pages_map["百世"]
|
||
bs_page.bring_to_front()
|
||
print(">> 正在处理【百世】初始弹窗（阅读消息 / 配置检查 / 优惠券广告）...")
|
||
# 委托给 baishi 的专用清理：优惠券广告是全屏居中 modal，
|
||
# 关闭键为 .ant-modal-close（纯图标无文字），必须点它才能真正关掉。
|
||
baishi.dismiss_baishi_popups(bs_page)
|
||
print(" ✅ 【百世】初始弹窗处理完成。")
|
||
except Exception:
|
||
pass
|
||
if "韵达" in pages_map:
|
||
try:
|
||
yd_page = pages_map["韵达"]
|
||
yd_page.bring_to_front()
|
||
print(">> 正在检查【韵达】音频设备授权提示...")
|
||
yunda.dismiss_audio_prompt(yd_page)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# ============================ 任务派发 ============================
|
||
|
||
|
||
def _web_handler(site, download_func):
|
||
"""构造网页站任务 handler。
|
||
|
||
foreground（ctx）控制任务执行时是否把 page 置顶：服务模式后台跑不置顶，避免抢用户
|
||
焦点；交互模式置顶便于调试。顺心是 page 列表，置顶标志透传给 shunxin_download。
|
||
"""
|
||
|
||
def handler(ctx, force=False, date=None):
|
||
pg = ctx.pages_map[site]
|
||
if isinstance(pg, list):
|
||
# 顺心双账号：置顶与否交给 shunxin_download 在逐账号循环里按 foreground 决定
|
||
return download_func(pg, foreground=ctx.foreground, force=force, date=date)
|
||
if ctx.foreground:
|
||
pg.bring_to_front()
|
||
return download_func(pg, force=force, date=date)
|
||
|
||
return handler
|
||
|
||
|
||
def _site_undelivered_handler(site):
|
||
"""4 站未到：下应到+实到 → DB 比对 → 写 output/<站>-<日期>-未到数据.xlsx。
|
||
应到全量去重（已落库则跳过导出），因此比对不依赖 Excel 文件，走数据库查询。
|
||
下载成功则返回 True（比对失败不影响任务判定，数据已入库）。"""
|
||
|
||
def handler(ctx, force=False, date=None):
|
||
exp_ok = TASK_HANDLERS[(site, "expected")](ctx, force, date) is not False
|
||
act_ok = (
|
||
(TASK_HANDLERS[(site, "actual")](ctx, force, date) is not False)
|
||
if exp_ok
|
||
else False
|
||
)
|
||
if not exp_ok or not act_ok:
|
||
return False
|
||
|
||
# ── 先入库再比对（修复时序：比对须读到本次下载的数据，
|
||
# 否则首次/force 时 PG 无当天数据，比对返回 None、不产出 Excel）──
|
||
try:
|
||
_record_business_date(site, "undelivered", date)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
from inbound_verify import store # 懒导入，避免成环
|
||
|
||
if store.ingest_enabled():
|
||
store.ingest_task(
|
||
site, "undelivered"
|
||
) # 4 站 = ingest expected + actual
|
||
print(f">> [入库] {site} 前置入库完成")
|
||
except Exception as e:
|
||
print(f">> [入库] {site} 前置入库失败（不影响比对尝试）: {e}")
|
||
|
||
# ── DB 比对（替代旧 Excel 比对）──
|
||
try:
|
||
from inbound_verify import db_compare # 懒导入，避免成环
|
||
|
||
if date:
|
||
target_date = date
|
||
else:
|
||
offset = state_store.get_offset(site, "actual")
|
||
target_date = (datetime.now().date() - timedelta(days=offset)).strftime(
|
||
"%Y-%m-%d"
|
||
)
|
||
|
||
result = db_compare.compare_site_date(site, target_date)
|
||
if result is not None:
|
||
db_compare.write_result_excel(result)
|
||
else:
|
||
print(f">> [未到] {site} {target_date}: 当天无实到数据，跳过比对")
|
||
except Exception as e:
|
||
print(f">> [未到] {site} DB 比对异常（不影响下载结果）: {e}")
|
||
|
||
return True # 下载成功即返回 True，比对失败不影响任务判定
|
||
|
||
return handler
|
||
|
||
|
||
# 「跑比对」= DB 版全站汇总报表（替代旧 compare.main Excel 路径；下载交由各站定时/手动）。
|
||
|
||
|
||
def _run_db_full_report(date=None):
|
||
"""生成 DB 版全站汇总报表（output/应到未到数据.xlsx）。
|
||
懒导入 db_compare，best-effort：失败只告警，返回 True（与旧 lambda 契约一致）。"""
|
||
try:
|
||
from inbound_verify import db_compare
|
||
|
||
db_compare.build_full_report(date)
|
||
except Exception as e:
|
||
print(f">> [跑比对] DB 汇总报表生成失败: {e}")
|
||
return True
|
||
|
||
|
||
TASK_HANDLERS = {
|
||
("顺心", "expected"): _web_handler("顺心", shunxin.shunxin_expected_download),
|
||
("顺心", "actual"): _web_handler("顺心", shunxin.shunxin_actual_download),
|
||
("顺心", "undelivered"): _site_undelivered_handler("顺心"),
|
||
("百世", "undelivered"): _web_handler(
|
||
"百世", baishi.baishi_download_undelivered_data
|
||
),
|
||
("中通", "expected"): _web_handler("中通", zto.zto_expected_download),
|
||
("中通", "actual"): _web_handler("中通", zto.zto_actual_download),
|
||
("中通", "undelivered"): _site_undelivered_handler("中通"),
|
||
("韵达", "expected"): _web_handler("韵达", yunda.yunda_expected_download),
|
||
("韵达", "actual"): _web_handler("韵达", yunda.yunda_actual_download),
|
||
("韵达", "undelivered"): _site_undelivered_handler("韵达"),
|
||
(
|
||
"安能",
|
||
"expected",
|
||
): lambda ctx, force=False, date=None: anneng.anneng_expected_download(
|
||
force=force, date=date
|
||
),
|
||
(
|
||
"安能",
|
||
"actual",
|
||
): lambda ctx, force=False, date=None: anneng.anneng_actual_download(
|
||
force=force, date=date
|
||
),
|
||
("安能", "undelivered"): _site_undelivered_handler("安能"),
|
||
("__compare__", "compare"): lambda ctx, force=False, date=None: _run_db_full_report(
|
||
date
|
||
),
|
||
}
|
||
|
||
|
||
def _record_business_date(site, kind, date=None):
|
||
"""下载成功后，把本次数据的业务日期快照写进状态库（供前端/报告显示「是哪天的数据」）。
|
||
有 date 用 date；否则 = 下载当天 − 该数据对应的日期偏移。__compare__ 无数据概念，跳过。
|
||
|
||
kind → 写入：
|
||
expected/actual：各写自己一列。
|
||
undelivered：百世直供（恒当天）写 undelivered；4 站未到由 _site_undelivered_handler
|
||
内部连带下了 expected+actual（不经 dispatch，无业务日期写入），故此处一并补写
|
||
expected/actual/undelivered 三列——actual 用 actual 偏移、未到跟随 expected 偏移。
|
||
只写业务日期；ready 语义已移交「入库成功」（_persist_to_db 置位），此处不再碰 ready。"""
|
||
if site == "__compare__":
|
||
return
|
||
today = datetime.now().date()
|
||
|
||
def _write(k, biz_or_off):
|
||
# biz_or_off: int=偏移（today−off）；str=已确定业务日期（date）
|
||
biz = (
|
||
(today - timedelta(days=biz_or_off)).strftime("%Y-%m-%d")
|
||
if isinstance(biz_or_off, int)
|
||
else biz_or_off
|
||
)
|
||
try:
|
||
state_store.set_business_date(site, k, biz)
|
||
except Exception as e:
|
||
print(f">> [状态] 写业务日期失败 {site}/{k}: {e}")
|
||
|
||
def off(kind_key):
|
||
return state_store.get_offset(site, kind_key)
|
||
|
||
if kind == "expected":
|
||
_write("expected", date if date else off("expected"))
|
||
elif kind == "actual":
|
||
_write("actual", date if date else off("actual"))
|
||
elif site == "百世":
|
||
_write("undelivered", 0)
|
||
else: # 4 站 undelivered：连带补写 expected/actual/undelivered 三列
|
||
_write("expected", date if date else off("expected"))
|
||
_write("actual", date if date else off("actual"))
|
||
_write("undelivered", date if date else off("expected"))
|
||
|
||
|
||
def _ready_flags(site):
|
||
"""从 PG 业务表派生单站三就绪态（ready = DB 数据真相）。
|
||
|
||
expected/actual = PG 中存在对应 target_date（today − offset）的数据；
|
||
百世 undelivered = baishi_daily_stats 中存在 target_date 的数据；
|
||
4 站 undelivered = expected_ready ∧ actual_ready（派生）。
|
||
PG 不可达时返回全 False（降级安全，不阻塞心跳）。
|
||
|
||
返回 (flags: {kind: bool}, dates: {kind: target_date_str})。
|
||
dates 与 flags 同源——ready=True 时 business_date 即该 target_date，
|
||
彻底消除 ready 与 business_date 不同源导致的日期标签漂移。"""
|
||
from inbound_verify import store # 懒导入：避免模块级循环
|
||
|
||
today = date.today()
|
||
today_str = today.isoformat()
|
||
|
||
if site == "百世":
|
||
has_und, _ = store.has_data(site, "undelivered", today_str)
|
||
return (
|
||
{"expected": False, "actual": False, "undelivered": has_und},
|
||
{"undelivered": today_str},
|
||
)
|
||
|
||
exp_off = state_store.get_offset(site, "expected")
|
||
act_off = state_store.get_offset(site, "actual")
|
||
exp_date = (today - timedelta(days=exp_off)).isoformat()
|
||
act_date = (today - timedelta(days=act_off)).isoformat()
|
||
|
||
has_exp, _ = store.has_data(site, "expected", exp_date)
|
||
has_act, _ = store.has_data(site, "actual", act_date)
|
||
return (
|
||
{"expected": has_exp, "actual": has_act, "undelivered": has_exp and has_act},
|
||
{"expected": exp_date, "actual": act_date, "undelivered": exp_date},
|
||
)
|
||
|
||
|
||
def _apply_ready(site, flags, dates=None):
|
||
"""写入单站就绪态 + 业务日期（同源：ready 与 business_date 均据 PG + offset 派生）。
|
||
ready=True 时同步写入 target_date 作为 business_date，消除不同源导致的日期标签漂移。
|
||
失败仅告警。"""
|
||
for k, rdy in flags.items():
|
||
try:
|
||
state_store.set_ready(site, k, rdy)
|
||
if rdy and dates and dates.get(k):
|
||
state_store.set_business_date(site, k, dates[k])
|
||
except Exception as e:
|
||
print(f">> [状态] 置就绪态失败 {site}/{k}: {e}")
|
||
|
||
|
||
def capture_error_screenshot(page, site, kind, attempt, error):
|
||
"""流程失败时截取当前页面，保存到 logs/screenshots/。
|
||
page: Playwright Page 对象（安能传 None 走 CDP 分支，调用方自行处理）。
|
||
截图失败绝不外抛——只打告警，不干扰任务重试/清场流程。"""
|
||
try:
|
||
os.makedirs(SCREENSHOT_DIR, exist_ok=True)
|
||
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
err_short = (error or "unknown")[:40].replace("/", "_").replace("\\", "_")
|
||
fname = f"{site}_{kind}_{ts}_attempt{attempt}_{err_short}.png"
|
||
path = os.path.join(SCREENSHOT_DIR, fname)
|
||
page.screenshot(path=path, full_page=False)
|
||
print(f"📸 【{site}-{kind}】错误截图已保存: {path}")
|
||
except Exception as se:
|
||
print(f"📸 【{site}-{kind}】截图失败（不影响任务）: {se}")
|
||
|
||
|
||
def _refresh_ready(site):
|
||
"""入库后立即据 PG 派生并写入该站就绪态（省 30s 心跳等待，与心跳同源）。"""
|
||
flags, dates = _ready_flags(site)
|
||
_apply_ready(site, flags, dates)
|
||
|
||
|
||
def _persist_to_db(site, kind):
|
||
"""下载成功后把本次数据入库 PostgreSQL（尽力而为，绝不外抛，不影响任务判定）。
|
||
- __compare__ 无源数据，跳过。
|
||
- auto_ingest=false 时跳过（无 PG/cpolar 的开发机）。
|
||
- 懒导入 store 以回避 import 顺序（store↔compare 与 runtime↔compare 共存）。
|
||
- 结果写 state_store.ingest_state，供 /api/status 反映入库健康。
|
||
所有写库/写状态都包 try/except：失败仅告警，绝不改变 dispatch_task 的 SUCCESS 判定。"""
|
||
if site == "__compare__":
|
||
return
|
||
try:
|
||
from inbound_verify import store # 懒导入：冷路径（每下载一次），回避成环
|
||
except Exception as e:
|
||
print(f">> [warn] 入库模块不可用: {e}")
|
||
return
|
||
try:
|
||
if not store.ingest_enabled(): # 移入 try：config.yaml 缺失/损坏时也不外抛
|
||
print(">> [入库] 已关闭 (auto_ingest=false)，跳过")
|
||
return
|
||
count = store.ingest_task(site, kind)
|
||
# 4 站 undelivered 连带入了 expected+actual：按实际入库的类补记 ingest_state，
|
||
# 否则心跳派生 ready（expected ∧ actual → undelivered）会读到陈旧值。
|
||
logged = (
|
||
["expected", "actual", "undelivered"]
|
||
if kind == "undelivered" and site != "百世"
|
||
else [kind]
|
||
)
|
||
for k in logged:
|
||
state_store.set_ingest_state(site, k, ok=True, count=count)
|
||
_refresh_ready(site) # 入库成功 → 立即据 ingest_state 派生就绪态（与心跳同源）
|
||
print(f">> [入库] {site}/{kind} 成功，{count} 条")
|
||
except Exception as e:
|
||
print(f">> [warn] 入库失败 {site}/{kind}: {e}")
|
||
try:
|
||
state_store.set_ingest_state(site, kind, ok=False, error=str(e))
|
||
except Exception as e2:
|
||
print(f">> [warn] 写入库状态也失败: {e2}")
|
||
|
||
|
||
def dispatch_task(ctx, task_spec):
|
||
"""执行一条任务。task_spec = {"site", "kind"}。返回 (status, error)。
|
||
|
||
掉登录的站点直接判 failed（呼应"掉登录则该站任务全停"）。
|
||
只在 Playwright 所属线程调用。
|
||
"""
|
||
site = task_spec.get("site")
|
||
kind = task_spec.get("kind")
|
||
if site != "__compare__":
|
||
if site not in ctx.pages_map:
|
||
return (state_store.TASK_FAILED, f"站点 {site} 未加载")
|
||
if not probe_site_login(site, ctx.pages_map):
|
||
return (
|
||
state_store.TASK_FAILED,
|
||
f"站点 {site} 未登录，已跳过（需人工重登）",
|
||
)
|
||
handler = TASK_HANDLERS.get((site, kind))
|
||
if handler is None:
|
||
return (state_store.TASK_FAILED, f"未知任务: {site}/{kind}")
|
||
try:
|
||
ret = handler(ctx, bool(task_spec.get("force", False)), task_spec.get("date"))
|
||
if ret is False:
|
||
return (state_store.TASK_FAILED, "任务执行失败（重试耗尽）")
|
||
_record_business_date(site, kind, task_spec.get("date"))
|
||
_persist_to_db(site, kind)
|
||
return (state_store.TASK_SUCCESS, None)
|
||
except Exception as e:
|
||
return (state_store.TASK_FAILED, str(e))
|
||
|
||
|
||
# ============================ 心跳 ============================
|
||
|
||
|
||
def run_heartbeat(ctx):
|
||
"""一轮心跳：探测各站登录态 + 据 PG 业务表派生数据就绪态；登录态变化时提示。
|
||
|
||
ready 直接查询 PG 业务表（expected_record / actual_record / baishi_daily_stats），
|
||
以「目标业务日期是否有数据」为唯一依据，彻底消除 ingest_state 日期比对带来的每日零点重置。
|
||
_refresh_ready 在入库瞬间即据 PG 派生（省 30s 等待），心跳同源复核。
|
||
只在 Playwright 所属线程调用。
|
||
"""
|
||
prev = state_store.get_all_status()
|
||
for site_name in ctx.sites_to_watch:
|
||
logged_in = probe_site_login(site_name, ctx.pages_map)
|
||
prev_login = prev.get(site_name, {}).get("login_state")
|
||
state_store.set_login_state(site_name, logged_in)
|
||
now_login = state_store.LOGIN_IN if logged_in else state_store.LOGIN_OUT
|
||
if prev_login and prev_login not in (now_login, state_store.LOGIN_UNKNOWN):
|
||
print(f"\n ⚠️【{site_name}】登录态变化: {prev_login} → {now_login}")
|
||
flags, dates = _ready_flags(site_name)
|
||
_apply_ready(site_name, flags, dates)
|