Compare commits

2 Commits

3 changed files with 99 additions and 21 deletions

View File

@@ -128,7 +128,10 @@ def _enqueue_fetch(site, kind):
if not _in_active_window(cfg["active_start"], cfg["active_end"]):
return # 不在激活时段,跳过本次 fire
try:
tid = state_store.create_task_if_idle(site, kind)
target_date = state_store.resolve_target_date(site, kind)
tid = state_store.create_task_if_idle(
site, kind, trigger="auto", target_date=target_date
)
if tid is None:
return # 上一次同类任务还没跑完,跳过避免堆积
task_queue.put((tid, {"site": site, "kind": kind}))
@@ -225,7 +228,13 @@ def create_task(req: TaskRequest):
raise HTTPException(
status_code=400, detail="百世固定下载当天,不支持指定日期"
)
task_id = state_store.create_task(req.site, req.kind)
task_id = state_store.create_task(
req.site,
req.kind,
trigger="manual",
target_date=state_store.resolve_target_date(req.site, req.kind, req.date),
force=req.force,
)
spec = {"site": req.site, "kind": req.kind, "force": req.force}
if req.date:
spec["date"] = req.date

View File

@@ -76,13 +76,37 @@ def _wait_cdp_up(port, timeout=60.0):
return False
# 【环境兼容】宿主 shellCodex/VS Code 插件、WorkBuddy 等)会向子进程注入一批与业务
# 无关的变量,实测会让安能应用登录后反复弹出“获取试用网点接口报错”:
# - HTTP(S)_PROXY=http://127.0.0.1:8800QuickQ 加速器代理):安能的 wnp.ane56.com
# 接口请求被塞进第三方代理后返回 400/用户未登录;
# - VSCODE_* / CODEX_* / EFC_*VS Code 扩展宿主注入IPC、PID、NLS、ESM 等);
# - NODE_TLS_REJECT_UNAUTHORIZED / DEBUG / RUST_LOG 等宿主调试变量。
# 另ELECTRON_RUN_AS_NODE=1 会把安能当作纯 Node 运行(拒绝 Chromium 参数、启动即
# 退出 rc=9NODE_OPTIONS 同样会干扰。拉起前全部摘掉,尽量还原终端手动启动环境。
_ANNENG_STRIP_PREFIXES = ("VSCODE_", "CODEX_", "EFC_")
_ANNENG_STRIP_EXACT = {
"NODE_OPTIONS",
"ELECTRON_RUN_AS_NODE",
"HTTP_PROXY",
"HTTPS_PROXY",
"ALL_PROXY",
"NO_PROXY",
"NODE_TLS_REJECT_UNAUTHORIZED",
"NODEFAULTCURRENTDIRECTORYINEXEPATH",
"DEBUG",
"RUST_LOG",
"APPLICATION_INSIGHTS_NO_STATSBEAT",
}
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)
anneng_env = {
key: value
for key, value in os.environ.items()
if key not in _ANNENG_STRIP_EXACT and not key.startswith(_ANNENG_STRIP_PREFIXES)
}
port = _find_free_port()
print(f">> 以调试模式启动【安能】应用(端口 {port}{app_path}")
proc = subprocess.Popen(

View File

@@ -7,7 +7,7 @@
import os
import sqlite3
from datetime import datetime
from datetime import datetime, timedelta
from inbound_verify.paths import STATE_DB_PATH
@@ -70,9 +70,22 @@ def init_db():
status TEXT,
started_at TEXT,
finished_at TEXT,
error TEXT
error TEXT,
trigger TEXT NOT NULL DEFAULT '',
target_date TEXT NOT NULL DEFAULT '',
force INTEGER NOT NULL DEFAULT 0
)
""")
# 旧库迁移:补触发方式/目标日期/强制重下三列(新库已含;重复添加抛 OperationalError忽略
for _col, _typedef in [
("trigger", "TEXT NOT NULL DEFAULT ''"),
("target_date", "TEXT NOT NULL DEFAULT ''"),
("force", "INTEGER NOT NULL DEFAULT 0"),
]:
try:
conn.execute(f"ALTER TABLE task_history ADD COLUMN {_col} {_typedef}")
except sqlite3.OperationalError:
pass
conn.execute("""
CREATE TABLE IF NOT EXISTS site_config (
site TEXT PRIMARY KEY,
@@ -357,6 +370,27 @@ def set_offset(site, kind, offset):
return offset
def resolve_target_date(site, kind, date=None):
"""计算一条任务的目标下载日期YYYY-MM-DD供任务日志展示 / 重试回放)。
有 date 用 date否则按站点偏移推算与 runtime._record_business_date 同源):
expected → 应到偏移actual → 实到偏移;百世 undelivered → 当天;
4 站 undelivered → 跟随应到偏移。__compare__ 无数据概念,返回 ''"""
if site == "__compare__":
return ""
if date:
return date
today = datetime.now().date()
if kind == "expected":
return (today - timedelta(days=get_offset(site, "expected"))).strftime(
"%Y-%m-%d"
)
if kind == "actual":
return (today - timedelta(days=get_offset(site, "actual"))).strftime("%Y-%m-%d")
if site == "百世":
return today.strftime("%Y-%m-%d")
return (today - timedelta(days=get_offset(site, "expected"))).strftime("%Y-%m-%d")
def set_schedule(site, enabled, time_str):
"""【DEPRECATED】旧"每日单时点定时"——已被 fetch_schedule 的周期+激活时段模式取代。
保留死代码以防外部残留调用;新代码请用 set_fetch_schedule。"""
@@ -539,26 +573,28 @@ def get_site_settings(site):
# ============================ 任务历史 ============================
def create_task(site, kind):
"""新建一条 pending 任务,返回其 id。"""
def create_task(site, kind, trigger="manual", target_date="", force=False):
"""新建一条 pending 任务(手动触发),返回其 id。trigger='manual'/'auto'
target_date 为该任务的目标下载日期YYYY-MM-DD可为 ''force 是否强制重下。"""
now = _now()
with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn:
cur = conn.execute(
"INSERT INTO task_history (site, kind, status, started_at, finished_at, error) "
"VALUES (?, ?, ?, ?, '', '')",
(site, kind, TASK_PENDING, now),
"INSERT INTO task_history "
"(site, kind, status, started_at, finished_at, error, trigger, target_date, force) "
"VALUES (?, ?, ?, ?, '', '', ?, ?, ?)",
(site, kind, TASK_PENDING, now, trigger, target_date, 1 if force else 0),
)
conn.commit()
return cur.lastrowid
def create_task_if_idle(site, kind):
def create_task_if_idle(site, kind, trigger="auto", target_date=""):
"""周期调度专用:若该 (site,kind) 已有 pending/running 任务则返回 None跳过本次周期
否则建一条 pending 任务返回其 id。单连接内 check-then-insert靠 SQLite 写锁把竞态压到忽略不计。
与 create_task 的区别:手动触发(POST /tasks)用 create_task用户点的必建周期 job 用本函数
——上一次还没跑完时跳过,避免同 (site,kind) 任务堆积。手动建的任务会让紧随其后的周期 fire
判到 inflight 而跳过,天然互斥。"""
判到 inflight 而跳过,天然互斥。trigger='auto'target_date 为目标下载日期YYYY-MM-DD"""
now = _now()
with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn:
row = conn.execute(
@@ -569,9 +605,10 @@ def create_task_if_idle(site, kind):
if row:
return None
cur = conn.execute(
"INSERT INTO task_history (site, kind, status, started_at, finished_at, error) "
"VALUES (?, ?, ?, ?, '', '')",
(site, kind, TASK_PENDING, now),
"INSERT INTO task_history "
"(site, kind, status, started_at, finished_at, error, trigger, target_date, force) "
"VALUES (?, ?, ?, ?, '', '', ?, ?, 0)",
(site, kind, TASK_PENDING, now, trigger, target_date),
)
conn.commit()
return cur.lastrowid
@@ -598,7 +635,8 @@ def get_task(task_id):
"""返回单条任务 dict不存在返回 None。"""
with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn:
row = conn.execute(
"SELECT id, site, kind, status, started_at, finished_at, error "
"SELECT id, site, kind, status, started_at, finished_at, error, "
"trigger, target_date, force "
"FROM task_history WHERE id=?",
(task_id,),
).fetchone()
@@ -612,6 +650,9 @@ def get_task(task_id):
"started_at": row[4],
"finished_at": row[5],
"error": row[6],
"trigger": row[7],
"target_date": row[8],
"force": bool(row[9]),
}
@@ -619,7 +660,8 @@ def list_tasks(limit=20):
"""返回最近 limit 条任务(按 id 倒序)。"""
with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn:
rows = conn.execute(
"SELECT id, site, kind, status, started_at, finished_at, error "
"SELECT id, site, kind, status, started_at, finished_at, error, "
"trigger, target_date, force "
"FROM task_history ORDER BY id DESC LIMIT ?",
(limit,),
).fetchall()
@@ -632,6 +674,9 @@ def list_tasks(limit=20):
"started_at": r[4],
"finished_at": r[5],
"error": r[6],
"trigger": r[7],
"target_date": r[8],
"force": bool(r[9]),
}
for r in rows
]