eaf56c1c9e
状态盘就绪态 {kind}_ready 改为从持久的 ingest_state 派生（DB 真相、重启不丢），
不再由心跳读 downloads/ Excel、也不在启动时重置：
- 心跳 _fresh_ingest/_ready_flags/_apply_ready：expected/actual_ready=该类今天入库成功；
undelivered_ready=百世原生(今天入库) / 4 站派生(expected ∧ actual)。
- _persist_to_db 入库后调 _refresh_ready 立即派生（省 30s 心跳等待，与心跳同源）；
4 站 undelivered 连入 expected+actual，故 ingest_state 补记 expected/actual/undelivered 三行。
- _record_business_date 只写 business_date，不再碰 ready。
- 删 run_heartbeat 的 Excel 探测循环、DATA_FILENAMES、probe_data_file、遗物清理。
- state_store 增 set_business_date / set_ready（仅写单字段，不碰彼此）。
- 删 GET /data/{filename}（前端不再下载原始中转 Excel）；/report 保留。
原则：Excel 只作「站点下载→入库」中转，状态/比对一律走 DB。
真机+单元验证：重启后心跳派生（百世今日入库→立即绿、重启不掉灰）；韵达 expected-only
时 undelivered 不亮、补 actual 后亮（派生）；百世原生；_ready_flags 7 例边界全过。
Co-Authored-By: Claude <noreply@anthropic.com>
427 lines
15 KiB
Python
427 lines
15 KiB
Python
# server.py
|
||
#
|
||
# 服务模式入口：FastAPI（主线程，uvicorn/asyncio）+ Playwright worker（独立线程）。
|
||
# 客户端经 HTTP 触发任务、查状态、下载数据；worker 串行执行任务并跑心跳。
|
||
#
|
||
# 线程模型（关键）：
|
||
# - 主线程：uvicorn + FastAPI。路由【绝不】访问 Playwright 对象，只经
|
||
# task_queue（投递任务）+ state_store（查状态/任务）+ 文件系统（下载数据）。
|
||
# - worker 线程：runtime.launch_and_prepare（sync_playwright 在此）+ 任务循环，
|
||
# 独占所有 page 操作；与主线程仅经 Queue + SQLite 通信。
|
||
# 违反"路由不碰 Playwright"会崩（sync 对象跨线程访问）。
|
||
#
|
||
# 运行：python -m inbound_verify.cli.server （或 inbound-verify-server；默认监听 0.0.0.0:8000）
|
||
|
||
import os
|
||
import queue
|
||
import threading
|
||
import time
|
||
from contextlib import asynccontextmanager
|
||
from datetime import datetime, timedelta
|
||
from typing import Dict, Optional
|
||
|
||
import uvicorn
|
||
from apscheduler.schedulers.background import BackgroundScheduler
|
||
from apscheduler.triggers.interval import IntervalTrigger
|
||
from fastapi import FastAPI, HTTPException
|
||
from fastapi.responses import FileResponse
|
||
from pydantic import BaseModel
|
||
|
||
from inbound_verify.paths import OUTPUT_DIR
|
||
from inbound_verify import state_store
|
||
from inbound_verify.runtime import (
|
||
HEARTBEAT_INTERVAL,
|
||
TASK_HANDLERS,
|
||
dispatch_task,
|
||
launch_and_prepare,
|
||
run_heartbeat,
|
||
)
|
||
from inbound_verify import db_compare
|
||
|
||
# 全部站点；百世固定下载当天，不可配置偏移
|
||
ALL_SITES = ["顺心", "百世", "中通", "韵达", "安能"]
|
||
CONFIGURABLE_SITES = {"顺心", "中通", "韵达", "安能"}
|
||
|
||
# 任务队列：元素 (task_id, task_spec)。主线程投递，worker 消费。
|
||
task_queue: "queue.Queue" = queue.Queue()
|
||
# worker 运行状态（主线程只读，worker 写）
|
||
worker_state = {
|
||
"ctx": None,
|
||
"stop": False,
|
||
"thread": None,
|
||
"ready": False, # launch_and_prepare 完成（各站就绪，可接任务）
|
||
"error": None, # worker 启动失败原因
|
||
}
|
||
|
||
# 每日定时下载调度器（进程内；job 只往 task_queue 投任务，不碰 Playwright）
|
||
scheduler = BackgroundScheduler(daemon=True)
|
||
|
||
|
||
def _worker_loop():
|
||
"""worker 线程：启动 Playwright + 等就绪 + 任务循环（执行任务 + 心跳）。"""
|
||
try:
|
||
ctx = launch_and_prepare(foreground=False)
|
||
worker_state["ctx"] = ctx
|
||
worker_state["ready"] = True
|
||
# 【P1-2 重启自愈】worker 就绪后清理上轮遗留的 pending/running 僵尸任务
|
||
cleaned = state_store.fail_stale_tasks()
|
||
if cleaned:
|
||
print(
|
||
f">> [worker] 自愈：清理 {cleaned} 条遗留任务（pending/running → failed）"
|
||
)
|
||
print(">> [worker] 各站就绪，开始接收任务 ...")
|
||
except Exception as e:
|
||
worker_state["error"] = str(e)
|
||
print(f"❌ [worker] 启动失败: {e}")
|
||
return
|
||
|
||
last_heartbeat = 0.0
|
||
while not worker_state["stop"]:
|
||
try:
|
||
task_id, task_spec = task_queue.get(timeout=1)
|
||
except queue.Empty:
|
||
# 空闲时跑心跳
|
||
if time.monotonic() - last_heartbeat >= HEARTBEAT_INTERVAL:
|
||
run_heartbeat(ctx)
|
||
last_heartbeat = time.monotonic()
|
||
continue
|
||
|
||
state_store.update_task(task_id, state_store.TASK_RUNNING)
|
||
print(f">> [worker] 执行任务 #{task_id}: {task_spec}")
|
||
status, error = dispatch_task(ctx, task_spec)
|
||
state_store.update_task(task_id, status, error)
|
||
print(f">> [worker] 任务 #{task_id} 完成: {status} {error or ''}")
|
||
|
||
try:
|
||
ctx.stop()
|
||
except Exception:
|
||
pass
|
||
worker_state["ready"] = False
|
||
print(">> [worker] 已退出。")
|
||
|
||
|
||
def _in_active_window(active_start, active_end):
|
||
"""当前本地时间是否落在激活时段内（避免半夜空跑）。
|
||
- start/end 任一为空 → 不限时段（24h 活跃）
|
||
- start == end（非空）→ 视为全天活跃
|
||
- start < end → 半开区间 [start, end)
|
||
- start > end → 跨午夜（如 22:00-06:00）：now >= start 或 now < end
|
||
"""
|
||
if not active_start or not active_end:
|
||
return True
|
||
now = datetime.now().strftime("%H:%M")
|
||
if active_start == active_end:
|
||
return True
|
||
if active_start < active_end:
|
||
return active_start <= now < active_end
|
||
return now >= active_start or now < active_end
|
||
|
||
|
||
def _enqueue_fetch(site, kind):
|
||
"""周期 job 回调：worker 就绪 + 在激活时段内 + 该(site,kind)无未完成任务时，投递一次抓取任务。
|
||
跑在 APScheduler 线程池线程——绝不碰 Playwright，只经 SQLite + task_queue 通信。"""
|
||
if not worker_state["ready"]:
|
||
return # worker 未就绪：与 POST /tasks 的 409 同义，下个周期补抓
|
||
cfg = state_store.get_fetch_schedule(site, kind)
|
||
if not cfg or not cfg["enabled"]:
|
||
return # 已禁用（job 本应已注销，双重保险）
|
||
if not _in_active_window(cfg["active_start"], cfg["active_end"]):
|
||
return # 不在激活时段，跳过本次 fire
|
||
try:
|
||
tid = state_store.create_task_if_idle(site, kind)
|
||
if tid is None:
|
||
return # 上一次同类任务还没跑完，跳过避免堆积
|
||
task_queue.put((tid, {"site": site, "kind": kind}))
|
||
print(f">> [周期] 投递 {site}/{kind} 任务 #{tid}")
|
||
except Exception as e:
|
||
print(f">> [周期] 投递 {site}/{kind} 失败: {e}")
|
||
|
||
|
||
def _reschedule_fetch(site, kind):
|
||
"""按持久化配置（重新）注册或取消该 (site,kind) 的周期抓取 job（IntervalTrigger）。"""
|
||
job_id = f"{site}_{kind}"
|
||
try:
|
||
scheduler.remove_job(job_id)
|
||
except Exception:
|
||
pass
|
||
cfg = state_store.get_fetch_schedule(site, kind)
|
||
if not cfg or not cfg["enabled"] or cfg["interval_minutes"] < 1:
|
||
return # 未启用或频率非法：不注册（等同取消）
|
||
try:
|
||
scheduler.add_job(
|
||
_enqueue_fetch,
|
||
IntervalTrigger(minutes=cfg["interval_minutes"]),
|
||
args=[site, kind],
|
||
id=job_id,
|
||
replace_existing=True,
|
||
)
|
||
print(
|
||
f">> [周期] 已注册 {site}/{kind}：每 {cfg['interval_minutes']} 分钟"
|
||
f"（激活 {cfg['active_start'] or '不限'}~{cfg['active_end'] or '不限'}）"
|
||
)
|
||
except Exception as e:
|
||
print(f">> [周期] 注册 {site}/{kind} 失败: {e}")
|
||
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(_app):
|
||
"""服务启停：起 worker 线程 / 通知 worker 停。"""
|
||
state_store.init_db() # 先建表/迁移状态库，确保早于 worker 就绪的 /api/status 可用
|
||
for site in ALL_SITES: # 按持久化配置注册各站各 kind 的周期抓取 job
|
||
for kind in state_store.allowed_kinds(site):
|
||
_reschedule_fetch(site, kind)
|
||
scheduler.start()
|
||
print(">> [定时] 调度器已启动")
|
||
t = threading.Thread(target=_worker_loop, daemon=True)
|
||
worker_state["thread"] = t
|
||
t.start()
|
||
yield
|
||
scheduler.shutdown(wait=False)
|
||
print(">> [定时] 调度器已停止")
|
||
worker_state["stop"] = True
|
||
t.join(timeout=10)
|
||
|
||
|
||
app = FastAPI(title="InboundVerify 服务端", lifespan=lifespan)
|
||
|
||
|
||
class TaskRequest(BaseModel):
|
||
site: str
|
||
kind: str
|
||
force: bool = (
|
||
False # 强制重下：忽略已落库去重，重新提交所有班次/交接单的导出任务（默认关）
|
||
)
|
||
date: Optional[str] = None # YYYY-MM-DD；指定则下载该日数据，否则走站点 offset
|
||
|
||
|
||
@app.post("/tasks")
|
||
def create_task(req: TaskRequest):
|
||
"""提交任务 {site, kind, force, date?} → 入队，返回 task_id。"""
|
||
# 【P0】后端未就绪时直接拒绝，避免任务在 worker 启动前入队卡死
|
||
if not worker_state["ready"]:
|
||
raise HTTPException(
|
||
status_code=409, detail="后端尚未就绪，请等待各站点登录完成后再操作"
|
||
)
|
||
if (req.site, req.kind) not in TASK_HANDLERS:
|
||
raise HTTPException(status_code=400, detail=f"无效任务: {req.site}/{req.kind}")
|
||
# 指定日期合法性校验（仅在传了 date 时）
|
||
if req.date:
|
||
try:
|
||
target_date = datetime.strptime(req.date, "%Y-%m-%d").date()
|
||
except ValueError:
|
||
raise HTTPException(
|
||
status_code=400, detail=f"date 格式非法，需 YYYY-MM-DD: {req.date}"
|
||
)
|
||
today = datetime.now().date()
|
||
if target_date > today:
|
||
raise HTTPException(
|
||
status_code=400, detail=f"date 不可为未来日期: {req.date}"
|
||
)
|
||
if target_date < today - timedelta(days=31):
|
||
raise HTTPException(
|
||
status_code=400, detail=f"date 超出 31 天回溯上限: {req.date}"
|
||
)
|
||
if req.site == "百世":
|
||
raise HTTPException(
|
||
status_code=400, detail="百世固定下载当天，不支持指定日期"
|
||
)
|
||
task_id = state_store.create_task(req.site, req.kind)
|
||
spec = {"site": req.site, "kind": req.kind, "force": req.force}
|
||
if req.date:
|
||
spec["date"] = req.date
|
||
task_queue.put((task_id, spec))
|
||
return {"task_id": task_id}
|
||
|
||
|
||
@app.get("/tasks/{task_id}")
|
||
def get_task(task_id: int):
|
||
t = state_store.get_task(task_id)
|
||
if not t:
|
||
raise HTTPException(status_code=404, detail="任务不存在")
|
||
return t
|
||
|
||
|
||
@app.get("/tasks")
|
||
def list_tasks(limit: int = 20):
|
||
return state_store.list_tasks(limit)
|
||
|
||
|
||
# ── DB 比对（基于 PostgreSQL，不依赖 Excel 文件）──
|
||
|
||
|
||
class CompareRequest(BaseModel):
|
||
site: str
|
||
date: str # YYYY-MM-DD
|
||
|
||
|
||
@app.post("/compare")
|
||
def run_compare(req: CompareRequest):
|
||
"""DB 差缺比对：以实到扫描日期为锚点，反推交接批次，展开全量比对。
|
||
返回统计指标 + 差缺明细。
|
||
"""
|
||
# 合法性校验
|
||
if req.site not in db_compare.SITE_COMPARE_CONFIG:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"不支持的站点: {req.site}（支持: {list(db_compare.SITE_COMPARE_CONFIG.keys())}）",
|
||
)
|
||
try:
|
||
target_date = datetime.strptime(req.date, "%Y-%m-%d").date()
|
||
except ValueError:
|
||
raise HTTPException(
|
||
status_code=400, detail=f"date 格式非法，需 YYYY-MM-DD: {req.date}"
|
||
)
|
||
today = datetime.now().date()
|
||
if target_date > today:
|
||
raise HTTPException(status_code=400, detail=f"date 不可为未来日期: {req.date}")
|
||
|
||
result = db_compare.compare_site_date(req.site, req.date)
|
||
if result is None:
|
||
raise HTTPException(
|
||
status_code=404,
|
||
detail=f"{req.site} {req.date}: 当天无实到数据，无法比对",
|
||
)
|
||
|
||
return {
|
||
"site": result.site,
|
||
"date": result.date,
|
||
"batches": result.batches,
|
||
"stats": {
|
||
"waybill_count": result.stats.waybill_count,
|
||
"sf_wb_count": result.stats.sf_wb_count,
|
||
"expected_pieces": result.stats.expected_pieces,
|
||
"arrived_pieces": result.stats.arrived_pieces,
|
||
"undelivered_pieces": result.stats.undelivered_pieces,
|
||
"undelivered_wb": result.stats.undelivered_wb,
|
||
"full_miss": result.stats.full_miss,
|
||
"part_miss": result.stats.part_miss,
|
||
"sf_undelivered": result.stats.sf_undelivered,
|
||
},
|
||
"rows": [
|
||
{
|
||
"handover_no": r.handover_no,
|
||
"waybill_no": r.waybill_no,
|
||
"total_pieces": r.total_pieces,
|
||
"arrived_pieces": r.arrived_pieces,
|
||
"arrived_list": r.arrived_list,
|
||
"is_sf": r.is_sf,
|
||
}
|
||
for r in result.rows
|
||
],
|
||
}
|
||
|
||
|
||
@app.get("/status")
|
||
def get_status():
|
||
"""各站登录态 + 数据态 + 入库态（前端状态盘用），另含 worker 就绪状态。"""
|
||
return {
|
||
"worker_ready": worker_state["ready"],
|
||
"worker_error": worker_state["error"],
|
||
"sites": state_store.get_all_status(),
|
||
"ingest": state_store.get_all_ingest_state(),
|
||
}
|
||
|
||
|
||
class FetchScheduleSpec(BaseModel):
|
||
enabled: bool
|
||
active_start: str = ""
|
||
active_end: str = ""
|
||
interval_minutes: int
|
||
|
||
|
||
class ConfigRequest(BaseModel):
|
||
expected_offset: Optional[int] = None
|
||
actual_offset: Optional[int] = None
|
||
# DEPRECATED：旧"每日定点"字段，保留一个发布周期兼容旧前端请求（收到即忽略）
|
||
schedule_enabled: Optional[bool] = None
|
||
schedule_time: Optional[str] = None
|
||
settings: Optional[Dict[str, str]] = None
|
||
fetch_schedules: Optional[Dict[str, FetchScheduleSpec]] = None
|
||
|
||
|
||
def _default_cfg():
|
||
return {
|
||
"expected_offset": 0,
|
||
"actual_offset": 0,
|
||
"fetch_schedules": {},
|
||
}
|
||
|
||
|
||
@app.get("/config")
|
||
def get_config():
|
||
"""各站配置：应到/实到偏移 + 每日定时（百世偏移恒 0）。"""
|
||
cfg = state_store.get_all_config()
|
||
return {site: cfg.get(site, _default_cfg()) for site in ALL_SITES}
|
||
|
||
|
||
@app.put("/config/{site}")
|
||
def set_config(site: str, req: ConfigRequest):
|
||
if site not in ALL_SITES:
|
||
raise HTTPException(status_code=400, detail=f"未知站点: {site}")
|
||
# 应到/实到偏移（百世锁定当天）
|
||
for kind, val in (
|
||
("expected", req.expected_offset),
|
||
("actual", req.actual_offset),
|
||
):
|
||
if val is not None:
|
||
if site == "百世":
|
||
raise HTTPException(
|
||
status_code=400, detail="百世固定下载当天，不可配置偏移"
|
||
)
|
||
state_store.set_offset(site, kind, val)
|
||
# 周期抓取调度（应到/实到各自独立；百世只允许 undelivered）
|
||
if req.fetch_schedules:
|
||
allowed = state_store.allowed_kinds(site)
|
||
for kind, spec in req.fetch_schedules.items():
|
||
if kind not in allowed:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"站点 {site} 不支持抓取类型 {kind}（允许: {list(allowed)}）",
|
||
)
|
||
interval = max(1, min(1440, int(spec.interval_minutes)))
|
||
state_store.set_fetch_schedule(
|
||
site,
|
||
kind,
|
||
spec.enabled,
|
||
spec.active_start,
|
||
spec.active_end,
|
||
interval,
|
||
)
|
||
_reschedule_fetch(site, kind)
|
||
# 站点专属配置（密码/账号/路径…）
|
||
if req.settings:
|
||
for k, v in req.settings.items():
|
||
state_store.set_setting(site, k, v)
|
||
cfg = state_store.get_all_config().get(site, _default_cfg())
|
||
cfg["settings"] = state_store.get_site_settings(site)
|
||
return cfg
|
||
|
||
|
||
@app.get("/config/{site}/settings")
|
||
def get_site_settings_api(site: str):
|
||
"""单站专属配置（密码/账号/路径…；不进 5s 轮询）。"""
|
||
if site not in ALL_SITES:
|
||
raise HTTPException(status_code=400, detail=f"未知站点: {site}")
|
||
return state_store.get_site_settings(site)
|
||
|
||
|
||
REPORT_FILE = "应到未到数据.xlsx"
|
||
|
||
|
||
@app.get("/report")
|
||
def download_report():
|
||
"""下载 output/应到未到数据.xlsx（跑比对生成；未生成 404）。"""
|
||
path = os.path.join(OUTPUT_DIR, REPORT_FILE)
|
||
if not os.path.isfile(path):
|
||
raise HTTPException(status_code=404, detail="报告尚未生成，请先跑比对")
|
||
return FileResponse(path, filename=REPORT_FILE)
|
||
|
||
|
||
def main():
|
||
"""服务模式入口。传字符串导入路径（规范写法；不开 reload/workers 时进程内 import，行为等价）。"""
|
||
uvicorn.run("inbound_verify.cli.server:app", host="0.0.0.0", port=8000)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|