refactor: move flat modules into inbound_verify package (Tier 1, behavior-identical)
Relocate 12 root .py modules into inbound_verify/ (sites/, cli/ subpackages). Rewrite all internal imports to package-qualified; drop the site_ prefix on the 5 site modules and their 28 call sites. Fix paths.py BASE_DIR to anchor at the project root. Add main() entry wrappers (cli/router, cli/server, store). No behavior change. Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Misaka_Company
348
inbound_verify/cli/router.py
Normal file
348
inbound_verify/cli/router.py
Normal file
@@ -0,0 +1,348 @@
|
||||
# main_router.py
|
||||
#
|
||||
# 交互菜单模式入口（调试 / 人工操作）。核心 Playwright 管理、任务派发、心跳
|
||||
# 已抽到 runtime.py 共享；本文件只保留交互菜单与自动化测试。
|
||||
# 服务模式（常驻 + FastAPI 接收指令）见 server.py。
|
||||
|
||||
import os
|
||||
import queue
|
||||
import threading
|
||||
import time
|
||||
|
||||
import yaml
|
||||
|
||||
from inbound_verify.paths import CONFIG_PATH
|
||||
from inbound_verify.runtime import (
|
||||
APP_SITES,
|
||||
HEARTBEAT_INTERVAL,
|
||||
dispatch_task,
|
||||
launch_and_prepare,
|
||||
run_heartbeat,
|
||||
)
|
||||
|
||||
from inbound_verify import state_store
|
||||
|
||||
# 各站点模块（自动化测试 + 比对用；任务派发在 runtime）
|
||||
from inbound_verify.sites import shunxin, baishi, zto, yunda, anneng
|
||||
from inbound_verify import expected_undelivered
|
||||
|
||||
|
||||
def run_undelivered_compare():
|
||||
"""应到未到比对（全站点）：调用 expected_undelivered，读 downloads/ 下的应到/实到
|
||||
数据，生成 output/应到未到数据.xlsx（汇总报表 + 各站明细）。"""
|
||||
print("\n▶ 开始执行【应到未到比对（全站点）】任务 ...")
|
||||
expected_undelivered.main()
|
||||
|
||||
|
||||
# ====================================================================
|
||||
# 自动化测试入口
|
||||
# ====================================================================
|
||||
# 每个(非百世)站点的交叉测试序列：覆盖两种下载流程之间的全部 4 种相邻转换，
|
||||
# 用于验证无论上一个流程把页面留在什么状态，下一个流程都能正常运行：
|
||||
# 应到->实到、实到->应到、应到->应到、实到->实到
|
||||
CROSS_TEST_SEQUENCE = ["expected", "actual", "expected", "expected", "actual", "actual"]
|
||||
|
||||
|
||||
def run_automation_test(pages_map):
|
||||
"""自动化测试入口：按交叉序列逐个跑通各站点的下载流程，结束后打印统计报告。
|
||||
|
||||
判定规则：流程函数返回 False 或抛出异常记为 FAIL，其余记为 PASS。
|
||||
"""
|
||||
# 站点 -> {流程键: (中文名, 流程函数)}；百世为单流程，单独处理
|
||||
# 自动化测试刻意走各站点的 _impl（单次执行、无兜底重试），以便探测原始失败、
|
||||
# 不被模块内部"失败→重置→重试"机制掩盖。
|
||||
flow_table = {
|
||||
"顺心": {
|
||||
"expected": ("应到", shunxin.shunxin_expected_download_impl),
|
||||
"actual": ("实到", shunxin.shunxin_actual_download_impl),
|
||||
},
|
||||
"中通": {
|
||||
"expected": ("应到", zto.zto_expected_download_impl),
|
||||
"actual": ("实到", zto.zto_actual_download_impl),
|
||||
},
|
||||
"韵达": {
|
||||
"expected": ("应到", yunda.yunda_expected_download_impl),
|
||||
"actual": ("实到", yunda.yunda_actual_download_impl),
|
||||
},
|
||||
"安能": {
|
||||
"expected": ("应到", anneng.anneng_expected_download_impl),
|
||||
"actual": ("实到", anneng.anneng_actual_download_impl),
|
||||
},
|
||||
}
|
||||
|
||||
# 构建测试计划：[(站点, 流程中文名, 流程函数, 绑定page, 归属标签), ...]
|
||||
# 顺心为双账号：两个 page 各自读归属地后跑一遍交叉序列；其余站点单 page。
|
||||
plan = []
|
||||
for site_name, flows in flow_table.items():
|
||||
if site_name == "顺心":
|
||||
if "顺心" not in pages_map:
|
||||
continue
|
||||
for sx_idx, sx_page in enumerate(pages_map["顺心"], start=1):
|
||||
try:
|
||||
tag = shunxin.shunxin_belonging(sx_page)
|
||||
except Exception:
|
||||
tag = f"账号{sx_idx}" # 读不到归属地时用序号占位，不阻断测试
|
||||
for flow_key in CROSS_TEST_SEQUENCE:
|
||||
label, func = flows[flow_key]
|
||||
plan.append((f"顺心·{tag}", label, func, sx_page, tag))
|
||||
continue
|
||||
if site_name not in pages_map:
|
||||
continue
|
||||
bound_page = pages_map[site_name]
|
||||
for flow_key in CROSS_TEST_SEQUENCE:
|
||||
label, func = flows[flow_key]
|
||||
plan.append((site_name, label, func, bound_page, ""))
|
||||
# 百世：单流程，跑一次即可
|
||||
if "百世" in pages_map:
|
||||
plan.append(
|
||||
(
|
||||
"百世",
|
||||
"应到未到",
|
||||
baishi.baishi_download_undelivered_data_impl,
|
||||
pages_map["百世"],
|
||||
"",
|
||||
)
|
||||
)
|
||||
|
||||
if not plan:
|
||||
print("\n⚠️ 当前没有已就绪的站点，无法执行自动化测试。")
|
||||
return
|
||||
|
||||
total = len(plan)
|
||||
print("\n====================================================")
|
||||
print(f"自动化测试开始，共 {total} 个步骤。")
|
||||
print("（双流程站点按应到/实到交叉序列执行，覆盖全部相邻转换）")
|
||||
print("====================================================")
|
||||
|
||||
results = [] # [(站点, 流程, 状态, 耗时秒, 错误信息)]
|
||||
for idx, (site_name, label, func, bound_page, out_tag) in enumerate(plan, start=1):
|
||||
print("\n----------------------------------------------------")
|
||||
print(f"[步骤 {idx}/{total}] 站点【{site_name}】流程【{label}】")
|
||||
print("----------------------------------------------------")
|
||||
start = time.time()
|
||||
status = "PASS"
|
||||
err = ""
|
||||
try:
|
||||
if site_name in APP_SITES:
|
||||
# 安能：Electron 应用，无 Playwright page，函数不收 page 参数
|
||||
ret = func()
|
||||
else:
|
||||
bound_page.bring_to_front()
|
||||
# 顺心 _impl 带 out_tag（归属地）；其余站点 _impl 仅收 page
|
||||
ret = func(bound_page, out_tag=out_tag) if out_tag else func(bound_page)
|
||||
if ret is False:
|
||||
status = "FAIL"
|
||||
err = "流程返回失败状态"
|
||||
except Exception as e:
|
||||
status = "FAIL"
|
||||
err = str(e)
|
||||
elapsed = time.time() - start
|
||||
results.append((site_name, label, status, elapsed, err))
|
||||
print(f">> 步骤结果: {status} (耗时 {elapsed:.1f}s)")
|
||||
|
||||
_print_test_report(results)
|
||||
|
||||
|
||||
def _print_test_report(results):
|
||||
"""打印自动化测试统计报告。"""
|
||||
passed = sum(1 for r in results if r[2] == "PASS")
|
||||
failed = len(results) - passed
|
||||
|
||||
print("\n====================================================")
|
||||
print("自动化测试统计报告")
|
||||
print("====================================================")
|
||||
for i, (site_name, label, status, elapsed, err) in enumerate(results, start=1):
|
||||
mark = "✅" if status == "PASS" else "❌"
|
||||
print(
|
||||
f" {i:>2}. {mark} {status} {site_name} - {label} (耗时 {elapsed:.1f}s)"
|
||||
)
|
||||
if err:
|
||||
note = err if len(err) <= 60 else err[:57] + "..."
|
||||
print(f" 说明: {note}")
|
||||
print("----------------------------------------------------")
|
||||
print(f" 合计 {len(results)} 步：通过 {passed}，失败 {failed}")
|
||||
if failed == 0:
|
||||
print(" ✅ 全部流程跑通。")
|
||||
else:
|
||||
print(" ❌ 存在失败流程，请结合上方说明与运行日志排查。")
|
||||
print("====================================================")
|
||||
|
||||
|
||||
# ====================================================================
|
||||
# 交互菜单模式
|
||||
# ====================================================================
|
||||
|
||||
|
||||
def _read_debug_config():
|
||||
"""读 config.yaml 的 debug 段，返回 (debug_mode, debug_target)。"""
|
||||
debug_mode = False
|
||||
debug_target = ""
|
||||
try:
|
||||
if os.path.exists(CONFIG_PATH):
|
||||
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
|
||||
config = yaml.safe_load(f) or {}
|
||||
debug_mode = (config.get("debug", {}) or {}).get("enabled", False)
|
||||
debug_target = (config.get("debug", {}) or {}).get("target_site", "")
|
||||
except Exception as e:
|
||||
print(f"⚠️ 读取 config.yaml 异常，将使用全量模式启动: {e}")
|
||||
return debug_mode, debug_target
|
||||
|
||||
|
||||
# 菜单编号 → 任务规格（dispatch_task 消费）
|
||||
CHOICE_TO_TASK = {
|
||||
"1": {"site": "顺心", "kind": "expected"},
|
||||
"2": {"site": "顺心", "kind": "actual"},
|
||||
"3": {"site": "百世", "kind": "undelivered"},
|
||||
"4": {"site": "中通", "kind": "expected"},
|
||||
"5": {"site": "中通", "kind": "actual"},
|
||||
"6": {"site": "韵达", "kind": "expected"},
|
||||
"7": {"site": "韵达", "kind": "actual"},
|
||||
"10": {"site": "安能", "kind": "expected"},
|
||||
"11": {"site": "安能", "kind": "actual"},
|
||||
"9": {"site": "__compare__", "kind": "compare"},
|
||||
}
|
||||
|
||||
|
||||
def _interactive_menu_loop(ctx):
|
||||
"""交互菜单循环：input 后台线程 + _await_command + dispatch_task + 心跳 + 状态盘。
|
||||
|
||||
所有 page 操作经 runtime（主线程），满足 Playwright sync 线程安全。
|
||||
"""
|
||||
pages_map = ctx.pages_map
|
||||
sites_to_watch = ctx.sites_to_watch
|
||||
|
||||
def is_site_ready(site_name):
|
||||
if site_name not in pages_map:
|
||||
print(f"\n🚫 站点 [{site_name}] 未加载（当前为调试模式），已跳过。")
|
||||
return False
|
||||
return True
|
||||
|
||||
last_heartbeat = 0.0
|
||||
command_queue = queue.Queue()
|
||||
|
||||
def _await_command():
|
||||
nonlocal last_heartbeat
|
||||
while True:
|
||||
try:
|
||||
return command_queue.get(timeout=0.5)
|
||||
except queue.Empty:
|
||||
if time.monotonic() - last_heartbeat >= HEARTBEAT_INTERVAL:
|
||||
run_heartbeat(ctx)
|
||||
last_heartbeat = time.monotonic()
|
||||
|
||||
def _print_status_board():
|
||||
print("\n====================== 站点状态盘 ======================")
|
||||
status = state_store.get_all_status()
|
||||
if not status:
|
||||
print(" （暂无状态记录）")
|
||||
print("======================================================")
|
||||
return
|
||||
login_text = {
|
||||
state_store.LOGIN_IN: "✅ 已登录",
|
||||
state_store.LOGIN_OUT: "❌ 未登录",
|
||||
state_store.LOGIN_UNKNOWN: "❔ 未知",
|
||||
}
|
||||
for site_name in sites_to_watch:
|
||||
s = status.get(site_name)
|
||||
if not s:
|
||||
continue
|
||||
login_mark = login_text.get(s["login_state"], s["login_state"])
|
||||
exp = f"应到{'✅' if s['expected_ready'] else '—'} {s['expected_generated_at'] or '无'}"
|
||||
act = f"实到{'✅' if s['actual_ready'] else '—'} {s['actual_generated_at'] or '无'}"
|
||||
print(
|
||||
f" 【{site_name}】 {login_mark} | {exp} | {act} "
|
||||
f"| 探测于 {s['login_checked_at']}"
|
||||
)
|
||||
print("======================================================")
|
||||
|
||||
def _input_loop():
|
||||
while True:
|
||||
try:
|
||||
command_queue.put(input())
|
||||
except EOFError:
|
||||
return
|
||||
|
||||
threading.Thread(target=_input_loop, daemon=True).start()
|
||||
|
||||
while True:
|
||||
print("\n====================================================")
|
||||
print(" 物流数据下载主菜单 ")
|
||||
if ctx.debug_mode:
|
||||
print(f" [ 调试模式，仅加载: {ctx.debug_target} ]")
|
||||
print("====================================================")
|
||||
print(" 模块一：【顺心】数据处理流")
|
||||
print(" [1] 执行 - 应到货物数据下载")
|
||||
print(" [2] 执行 - 实到货物数据下载")
|
||||
print("-" * 52)
|
||||
print(" 模块二：【百世】数据处理流")
|
||||
print(" [3] 执行 - 一键提取应到未到异常数据")
|
||||
print("-" * 52)
|
||||
print(" 模块三：【中通】数据处理流")
|
||||
print(" [4] 执行 - 应到货物数据下载")
|
||||
print(" [5] 执行 - 实到货物数据下载")
|
||||
print("-" * 52)
|
||||
print(" 模块四：【韵达】数据处理流")
|
||||
print(" [6] 执行 - 应到货物数据下载")
|
||||
print(" [7] 执行 - 实到货物数据下载")
|
||||
print("-" * 52)
|
||||
print(" 模块五：【安能】数据处理流（Electron 应用）")
|
||||
print(" [10] 执行 - 应到货物数据下载（运单信息）")
|
||||
print(" [11] 执行 - 实到货物数据下载（网点到件扫描）")
|
||||
print("-" * 52)
|
||||
print(" 自动化测试")
|
||||
print(" [8] 执行 - 全站点下载流程自动化测试 (交叉跑通校验)")
|
||||
print("-" * 52)
|
||||
print(" 全局离线数据处理")
|
||||
print(" [9] 执行 - 应到未到比对（全站点汇总，输出 output/应到未到数据.xlsx）")
|
||||
print("-" * 52)
|
||||
print(" 站点状态")
|
||||
print(" [12] 查看 - 各站登录态 / 数据就绪状态")
|
||||
print("-" * 52)
|
||||
print(" [0] 退出系统")
|
||||
print("====================================================")
|
||||
|
||||
print("请输入任务编号并回车: ", end="", flush=True)
|
||||
choice = _await_command()
|
||||
|
||||
try:
|
||||
if choice in CHOICE_TO_TASK:
|
||||
task = CHOICE_TO_TASK[choice]
|
||||
site = task["site"]
|
||||
if site == "__compare__" or is_site_ready(site):
|
||||
status, error = dispatch_task(ctx, task)
|
||||
if status == state_store.TASK_FAILED:
|
||||
print(f"❌ 任务失败: {error}")
|
||||
elif choice == "8":
|
||||
run_automation_test(pages_map)
|
||||
elif choice == "12":
|
||||
_print_status_board()
|
||||
elif choice == "0":
|
||||
break
|
||||
elif choice.strip() != "":
|
||||
print("\n⚠️ 无效输入，请查证后回车。")
|
||||
except Exception as e:
|
||||
print(f"❌ 任务调度异常: {e}")
|
||||
|
||||
|
||||
def run_multi_site_daemon():
|
||||
"""多站点自动化主控流程（交互菜单模式）。
|
||||
|
||||
启动 → 等待各站登录就绪 → 进入交互菜单；退出时关闭浏览器与安能。
|
||||
"""
|
||||
debug_mode, debug_target = _read_debug_config()
|
||||
ctx = launch_and_prepare(debug_mode, debug_target)
|
||||
try:
|
||||
_interactive_menu_loop(ctx)
|
||||
finally:
|
||||
print("\n正在关闭浏览器并退出...")
|
||||
ctx.stop()
|
||||
print("程序已退出。")
|
||||
|
||||
|
||||
def main():
|
||||
"""交互菜单模式入口。"""
|
||||
run_multi_site_daemon()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
300
inbound_verify/cli/server.py
Normal file
300
inbound_verify/cli/server.py
Normal file
@@ -0,0 +1,300 @@
|
||||
# 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 server.py （默认监听 0.0.0.0:8000）
|
||||
|
||||
import os
|
||||
import queue
|
||||
import threading
|
||||
import time
|
||||
from contextlib import asynccontextmanager
|
||||
from typing import Dict, Optional
|
||||
|
||||
import uvicorn
|
||||
from apscheduler.schedulers.background import BackgroundScheduler
|
||||
from apscheduler.triggers.cron import CronTrigger
|
||||
from fastapi import FastAPI, HTTPException
|
||||
from fastapi.responses import FileResponse
|
||||
from pydantic import BaseModel
|
||||
|
||||
from inbound_verify.paths import DOWNLOAD_DIR, OUTPUT_DIR
|
||||
from inbound_verify import state_store
|
||||
from inbound_verify.runtime import (
|
||||
HEARTBEAT_INTERVAL,
|
||||
TASK_HANDLERS,
|
||||
dispatch_task,
|
||||
launch_and_prepare,
|
||||
run_heartbeat,
|
||||
)
|
||||
|
||||
# 全部站点；百世固定下载当天，不可配置偏移
|
||||
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()
|
||||
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 _enqueue_undelivered(site):
|
||||
"""定时 job：把该站 undelivered 任务投到队列（worker 串行处理；本线程不碰 Playwright）。"""
|
||||
try:
|
||||
tid = state_store.create_task(site, "undelivered")
|
||||
task_queue.put((tid, {"site": site, "kind": "undelivered"}))
|
||||
print(f">> [定时] 投递 {site}/undelivered 任务 #{tid}")
|
||||
except Exception as e:
|
||||
print(f">> [定时] 投递 {site} 失败: {e}")
|
||||
|
||||
|
||||
def _reschedule_site(site):
|
||||
"""按持久化配置（重新）注册或取消该站的每日定时 job。"""
|
||||
job_id = f"site_{site}"
|
||||
try:
|
||||
scheduler.remove_job(job_id)
|
||||
except Exception:
|
||||
pass
|
||||
cfg = state_store.get_all_config().get(site)
|
||||
if not cfg or not cfg.get("schedule_enabled") or not cfg.get("schedule_time"):
|
||||
return
|
||||
try:
|
||||
hh, mm = cfg["schedule_time"].split(":")
|
||||
scheduler.add_job(
|
||||
_enqueue_undelivered,
|
||||
CronTrigger(hour=int(hh), minute=int(mm)),
|
||||
args=[site],
|
||||
id=job_id,
|
||||
replace_existing=True,
|
||||
)
|
||||
print(f">> [定时] 已注册 {site} 每日 {cfg['schedule_time']} 下载")
|
||||
except Exception as e:
|
||||
print(f">> [定时] 注册 {site} 失败: {e}")
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(_app):
|
||||
"""服务启停：起 worker 线程 / 通知 worker 停。"""
|
||||
state_store.init_db() # 先建表/迁移状态库，确保早于 worker 就绪的 /api/status 可用
|
||||
for site in ALL_SITES: # 按持久化配置注册各站每日定时 job
|
||||
_reschedule_site(site)
|
||||
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
|
||||
|
||||
|
||||
@app.post("/tasks")
|
||||
def create_task(req: TaskRequest):
|
||||
"""提交任务 {site, kind} → 入队，返回 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}")
|
||||
task_id = state_store.create_task(req.site, req.kind)
|
||||
task_queue.put((task_id, {"site": req.site, "kind": req.kind}))
|
||||
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)
|
||||
|
||||
|
||||
@app.get("/status")
|
||||
def get_status():
|
||||
"""各站登录态 + 数据态（前端状态盘用），另含 worker 就绪状态。"""
|
||||
return {
|
||||
"worker_ready": worker_state["ready"],
|
||||
"worker_error": worker_state["error"],
|
||||
"sites": state_store.get_all_status(),
|
||||
}
|
||||
|
||||
|
||||
class ConfigRequest(BaseModel):
|
||||
expected_offset: Optional[int] = None
|
||||
actual_offset: Optional[int] = None
|
||||
schedule_enabled: Optional[bool] = None
|
||||
schedule_time: Optional[str] = None
|
||||
settings: Optional[Dict[str, str]] = None
|
||||
|
||||
|
||||
def _default_cfg():
|
||||
return {
|
||||
"expected_offset": 0,
|
||||
"actual_offset": 0,
|
||||
"schedule_enabled": False,
|
||||
"schedule_time": "",
|
||||
}
|
||||
|
||||
|
||||
@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}")
|
||||
cur = state_store.get_all_config().get(site, _default_cfg())
|
||||
# 应到/实到偏移（百世锁定当天）
|
||||
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)
|
||||
# 每日定时（所有站可配，含百世）
|
||||
if req.schedule_enabled is not None or req.schedule_time is not None:
|
||||
enabled = (
|
||||
req.schedule_enabled
|
||||
if req.schedule_enabled is not None
|
||||
else cur["schedule_enabled"]
|
||||
)
|
||||
time_str = (
|
||||
req.schedule_time if req.schedule_time is not None else cur["schedule_time"]
|
||||
)
|
||||
state_store.set_schedule(site, enabled, time_str)
|
||||
_reschedule_site(site)
|
||||
# 站点专属配置（密码/账号/路径…）
|
||||
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)
|
||||
|
||||
|
||||
@app.get("/data/{filename}")
|
||||
def download_data(filename: str):
|
||||
"""下载 downloads/ 下的数据文件（防路径穿越）。"""
|
||||
if not filename or "/" in filename or "\\" in filename or ".." in filename:
|
||||
raise HTTPException(status_code=400, detail="非法文件名")
|
||||
path = os.path.join(DOWNLOAD_DIR, filename)
|
||||
# 双重校验：解析后绝对路径仍在 DOWNLOAD_DIR 内
|
||||
if not os.path.abspath(path).startswith(os.path.abspath(DOWNLOAD_DIR) + os.sep):
|
||||
raise HTTPException(status_code=400, detail="非法路径")
|
||||
if not os.path.isfile(path):
|
||||
raise HTTPException(status_code=404, detail="文件不存在")
|
||||
return FileResponse(path, filename=filename)
|
||||
|
||||
|
||||
def main():
|
||||
"""服务模式入口。传字符串导入路径（规范写法；不开 reload/workers 时进程内 import，行为等价）。"""
|
||||
uvicorn.run("inbound_verify.cli.server:app", host="0.0.0.0", port=8000)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user