feat(cookie-gate): 调度器恢复重跑接线与控制台健康自检

调度器维护循环挂载 CookieGateRecoveryManager,登录态恢复后按原业务日期自动
重跑被阻断工作流并飞书确认;工作流级 required_accounts 声明接入目录解析;
采集器抛 CookieExpiredError 退出 75;控制台新增 /api/health 自检端点与进程内
看门狗,配合 deploy/watchdog 双层保活;安装脚本同步。
This commit is contained in:
wangyunlong committed 2026-10-09 13:45:29 +08:00
1 parent 78ea4641ce
commit 130f06b320
13 files changed
+318 -77

No files matched your search

+12 -1
View File
@@ -4,6 +4,8 @@
param( param(
[string]$ProjectRoot = (Resolve-Path (Join-Path $PSScriptRoot "..\..")).Path, [string]$ProjectRoot = (Resolve-Path (Join-Path $PSScriptRoot "..\..")).Path,
[string]$DataRoot = "", [string]$DataRoot = "",
[string]$RuntimeEnvFile = "D:\product-collector-analyze-flow\.env",
[string]$JushuitanEnvFile = "D:\product-collector-analyze-flow\private-secrets\jushuitan.env",
[string]$ServiceName = "gyxx-flow-scheduler", [string]$ServiceName = "gyxx-flow-scheduler",
[string]$NssmPath = "nssm.exe" [string]$NssmPath = "nssm.exe"
) )
@@ -12,6 +14,13 @@ $ErrorActionPreference = "Stop"
Set-StrictMode -Version Latest Set-StrictMode -Version Latest
$project = (Resolve-Path -LiteralPath $ProjectRoot).Path $project = (Resolve-Path -LiteralPath $ProjectRoot).Path
$runtimeEnvPath = (Resolve-Path -LiteralPath $RuntimeEnvFile -ErrorAction Stop).Path
$runtimeEnvPaths = @($runtimeEnvPath)
if (Test-Path -LiteralPath $JushuitanEnvFile -PathType Leaf) {
$runtimeEnvPaths += (Resolve-Path -LiteralPath $JushuitanEnvFile -ErrorAction Stop).Path
} else {
Write-Warning "Jushuitan environment file not found; Jushuitan workflows will fail safely until it is provisioned."
}
if ([string]::IsNullOrWhiteSpace($DataRoot)) { if ([string]::IsNullOrWhiteSpace($DataRoot)) {
$DataRoot = Join-Path $project "var" $DataRoot = Join-Path $project "var"
} }
@@ -33,7 +42,9 @@ function Invoke-Nssm([string[]]$Arguments) {
Invoke-Nssm -Arguments @("install", $ServiceName, $python) Invoke-Nssm -Arguments @("install", $ServiceName, $python)
Invoke-Nssm -Arguments @("set", $ServiceName, "AppDirectory", $project) Invoke-Nssm -Arguments @("set", $ServiceName, "AppDirectory", $project)
Invoke-Nssm -Arguments @("set", $ServiceName, "AppParameters", "-m gyxx_flow schedule run") $envFileArguments = $runtimeEnvPaths | ForEach-Object { '--env-file "' + $_ + '"' }
$schedulerArguments = '-m gyxx_flow schedule run ' + ($envFileArguments -join ' ')
Invoke-Nssm -Arguments @("set", $ServiceName, "AppParameters", $schedulerArguments)
Invoke-Nssm -Arguments @( Invoke-Nssm -Arguments @(
"set", $ServiceName, "AppEnvironmentExtra", "set", $ServiceName, "AppEnvironmentExtra",
"GYXX_PROJECT_ROOT=$project", "GYXX_DATA_ROOT=$data", "PYTHONUTF8=1" "GYXX_PROJECT_ROOT=$project", "GYXX_DATA_ROOT=$data", "PYTHONUTF8=1"
+19
View File
@@ -168,3 +168,22 @@ def looks_like_login_url(url: str) -> bool:
path = parsed.path.casefold() path = parsed.path.casefold()
markers = ("login", "passport", "signin", "sign-in") markers = ("login", "passport", "signin", "sign-in")
return any(marker in hostname or marker in path for marker in markers) return any(marker in hostname or marker in path for marker in markers)
class CookieExpiredError(RuntimeError):
"""A platform login state expired and no automatic recovery exists.
Collectors raise this instead of waiting for a human QR/SMS login: the
account cookie gate has already blocked unverified accounts before the
run, so reaching this point means the session died mid-run. The CLI maps
it to ``COOKIE_SKIP_EXIT_CODE`` so the engine records a cookie skip.
"""
def cookie_expired_message(account_id: str, platform: str) -> str:
"""One standard, operator-actionable expiry message across platforms."""
return (
f"{platform} 登录态已失效:请执行 `uv run gyxx accounts login {account_id}` "
"完成扫码/验证码重新登录(或等待飞书通知后处理)"
)
+16
View File
@@ -79,6 +79,10 @@ class WorkflowEntry:
steps: tuple[WorkflowStepEntry, ...] = () steps: tuple[WorkflowStepEntry, ...] = ()
notification_workflow_id: str | None = None notification_workflow_id: str | None = None
topic_config: WorkflowTopicConfig | None = None topic_config: WorkflowTopicConfig | None = None
# Orchestrator entries whose own binding carries no runtime account while
# the spawned child collectors do; the cookie gate probes these before
# the workflow runs.
required_accounts: tuple[str, ...] = ()
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
@@ -218,6 +222,9 @@ def _parse_workflow(item: Any) -> WorkflowEntry:
if trigger == "scheduled" and (not isinstance(task_name, str) or not task_name): if trigger == "scheduled" and (not isinstance(task_name, str) or not task_name):
raise CatalogError(f"scheduled workflow requires source task name: {workflow_id}") raise CatalogError(f"scheduled workflow requires source task name: {workflow_id}")
topic_config = _parse_topic_config(item.get("topic_config"), workflow_id) topic_config = _parse_topic_config(item.get("topic_config"), workflow_id)
required_accounts = _parse_required_accounts(
item.get("required_accounts", []), workflow_id
)
return WorkflowEntry( return WorkflowEntry(
workflow_id=workflow_id, workflow_id=workflow_id,
module=module, module=module,
@@ -229,9 +236,18 @@ def _parse_workflow(item: Any) -> WorkflowEntry:
note=item.get("note"), note=item.get("note"),
steps=steps, steps=steps,
topic_config=topic_config, topic_config=topic_config,
required_accounts=required_accounts,
) )
def _parse_required_accounts(value: Any, workflow_id: str) -> tuple[str, ...]:
if not isinstance(value, list) or not all(
isinstance(item, str) and _WORKFLOW_ID.fullmatch(item) for item in value
):
raise CatalogError(f"invalid required_accounts for {workflow_id}")
return tuple(dict.fromkeys(value))
def _parse_topic_config(value: Any, workflow_id: str) -> WorkflowTopicConfig | None: def _parse_topic_config(value: Any, workflow_id: str) -> WorkflowTopicConfig | None:
if value is None: if value is None:
return None return None
+41 -1
View File
@@ -703,7 +703,19 @@ def _python_scheduler(
if enable_account_keepalive and not dry_run: if enable_account_keepalive and not dry_run:
from gyxx_flow.accounts import AccountKeepaliveManager from gyxx_flow.accounts import AccountKeepaliveManager
maintenance = AccountKeepaliveManager(settings).tick keepalive_manager = AccountKeepaliveManager(settings)
recovery_manager = _build_cookie_recovery_manager(
settings, launcher=launcher, keepalive=keepalive_manager
)
def maintenance() -> None: # noqa: D107 - scheduler maintenance chain
keepalive_manager.tick()
if recovery_manager is not None:
try:
recovery_manager.tick()
except Exception: # noqa: BLE001 - 自动重跑不得影响调度心跳
pass
return PythonScheduler( return PythonScheduler(
catalog, catalog,
settings.data_root, settings.data_root,
@@ -717,6 +729,29 @@ def _python_scheduler(
) )
def _build_cookie_recovery_manager(
settings: Settings,
*,
launcher: object,
keepalive: object,
):
"""Build the cookie-gate recovery rerun manager; None when disabled."""
from gyxx_flow.cookie_gate import (
CookieGate,
CookieGateRecoveryManager,
gate_enabled,
)
if not gate_enabled():
return None
try:
gate = CookieGate(settings, keepalive=keepalive) # type: ignore[arg-type]
except Exception: # noqa: BLE001 - 无集成配置时没有可恢复的账号
return None
return CookieGateRecoveryManager(gate, launcher=launcher)
def _schedule_run_command( def _schedule_run_command(
arguments: argparse.Namespace, arguments: argparse.Namespace,
catalog: WorkflowCatalog, catalog: WorkflowCatalog,
@@ -885,6 +920,11 @@ def _execute_once(
dry_run: bool, dry_run: bool,
settings: Settings, settings: Settings,
) -> tuple[int, dict[str, object]]: ) -> tuple[int, dict[str, object]]:
if not dry_run:
# 每账号 Cookie 健康门控:dry-run 不注入,也不做任何探测/通知。
from gyxx_flow.cookie_gate import apply_cookie_gates
workflow = apply_cookie_gates(workflow, settings=settings)
context = RunContext.create( context = RunContext.create(
workflow.workflow_id, workflow.workflow_id,
business_date.isoformat(), business_date.isoformat(),
+118 -2
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
import calendar import calendar
import hashlib import hashlib
import hmac import hmac
import http.client
import ipaddress import ipaddress
import json import json
import os import os
@@ -15,6 +16,7 @@ import sys
import tempfile import tempfile
import threading import threading
import time import time
import traceback
import unicodedata import unicodedata
import uuid import uuid
from contextlib import suppress from contextlib import suppress
@@ -24,7 +26,7 @@ from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from importlib import resources from importlib import resources
from pathlib import Path from pathlib import Path
from typing import Any, Mapping, Protocol, Sequence, TextIO, cast from typing import Any, Callable, Mapping, Protocol, Sequence, TextIO, cast
from urllib.parse import parse_qs, unquote, urlsplit from urllib.parse import parse_qs, unquote, urlsplit
from zoneinfo import ZoneInfo from zoneinfo import ZoneInfo
@@ -171,7 +173,6 @@ _WORKFLOW_NAMES = {
"content.metrics.daily": "合作达人内容指标日报采集", "content.metrics.daily": "合作达人内容指标日报采集",
"content.metrics.self_operated.daily": "自营内容指标日报采集", "content.metrics.self_operated.daily": "自营内容指标日报采集",
"content.marketing_report.daily": "营销日报生成与发送", "content.marketing_report.daily": "营销日报生成与发送",
"content.relogin.weekly": "内容平台登录态刷新",
"content.creator_report.monthly": "达人月报生成", "content.creator_report.monthly": "达人月报生成",
"content.summary.monthly": "内容月度汇总", "content.summary.monthly": "内容月度汇总",
"content.comments.weekly": "多平台评论周采集", "content.comments.weekly": "多平台评论周采集",
@@ -3736,6 +3737,7 @@ class WorkflowConsoleHTTPServer(ThreadingHTTPServer):
self.console_service = service self.console_service = service
self.console_token = token self.console_token = token
self.bind_host = bind_host self.bind_host = bind_host
self.started_at = time.time()
super().__init__(server_address, WorkflowConsoleRequestHandler) super().__init__(server_address, WorkflowConsoleRequestHandler)
def handle_error( def handle_error(
@@ -3796,6 +3798,29 @@ class WorkflowConsoleRequestHandler(BaseHTTPRequestHandler):
if method in {"GET", "HEAD"} and path == "/assets/app.js": if method in {"GET", "HEAD"} and path == "/assets/app.js":
self._static("app.js", "text/javascript; charset=utf-8", head=method == "HEAD") self._static("app.js", "text/javascript; charset=utf-8", head=method == "HEAD")
return return
if method == "GET" and path == "/api/health":
# 自检探针端点:不鉴权、不触外部依赖,只反映进程与调度锁存活,
# 供进程内看门狗与 deploy/watchdog 保活脚本探测。
try:
scheduler = _scheduler_process_from_lock(
self.console_server.console_service.settings
)
except Exception:
scheduler = None
self._json(
HTTPStatus.OK,
{
"ok": True,
"pid": os.getpid(),
"started_at": self.console_server.started_at,
"uptime_seconds": round(
time.time() - self.console_server.started_at, 1
),
"scheduler_running": scheduler is not None,
"scheduler_pid": scheduler.pid if scheduler else None,
},
)
return
if method == "GET" and path == "/api/overview": if method == "GET" and path == "/api/overview":
self._require_api_access(mutation=False) self._require_api_access(mutation=False)
self._json(HTTPStatus.OK, self.console_server.console_service.overview()) self._json(HTTPStatus.OK, self.console_server.console_service.overview())
@@ -4020,6 +4045,9 @@ class WorkflowConsoleRequestHandler(BaseHTTPRequestHandler):
except (BrokenPipeError, ConnectionResetError): except (BrokenPipeError, ConnectionResetError):
return return
except Exception: except Exception:
# 前端只拿到固定文案;完整堆栈落到 stderr(随启动重定向进
# var/logs/console-restarts 或服务日志),否则故障无法定位。
traceback.print_exc()
self._json_error(HTTPStatus.INTERNAL_SERVER_ERROR, "控制台处理请求失败") self._json_error(HTTPStatus.INTERNAL_SERVER_ERROR, "控制台处理请求失败")
def _require_api_access(self, *, mutation: bool) -> None: def _require_api_access(self, *, mutation: bool) -> None:
@@ -4190,6 +4218,92 @@ def create_console_server(
) )
CONSOLE_WATCHDOG_EXIT_CODE = 70
def _probe_console_health(port: int, timeout_seconds: float) -> bool:
# 只连本机回环端口的固定路径 /api/health,探测自身 HTTP 服务往返。
connection = http.client.HTTPConnection("127.0.0.1", port, timeout=timeout_seconds)
try:
connection.request("GET", "/api/health")
response = connection.getresponse()
response.read()
return response.status == HTTPStatus.OK
except Exception:
return False
finally:
connection.close()
def _console_watchdog(
port: int,
*,
interval_seconds: float,
failure_limit: int,
probe_timeout_seconds: float,
stopped: threading.Event,
probe: Callable[[int, float], bool] | None = None,
exit_process: Callable[[int], None] = os._exit,
sleep: Callable[[float], None] = time.sleep,
) -> None:
"""进程内自检:周期探测自身 /api/health,连续失败则退出交给守护拉起。
HTTP accept 线程活着但工作线程全部卡死(锁竞争/句柄耗尽)时,外部
静态页能打开而 API 全挂——正是"控制台暂时不可用"的形态;只有真实
走一遍 HTTP 往返才能暴露,所以探针必须打 socket 而不是查内存状态。
"""
run_probe = probe or _probe_console_health
failures = 0
while not stopped.is_set():
if run_probe(port, probe_timeout_seconds):
failures = 0
else:
failures += 1
print(
f"⚠️ 控制台自检失败 {failures}/{failure_limit}: 127.0.0.1:{port}/api/health",
file=sys.stderr,
flush=True,
)
if failures >= failure_limit:
print(
"❌ 控制台自检连续失败,进程退出等待守护脚本拉起",
file=sys.stderr,
flush=True,
)
exit_process(CONSOLE_WATCHDOG_EXIT_CODE)
return
sleep(interval_seconds)
def _start_console_watchdog(server: WorkflowConsoleHTTPServer) -> threading.Event:
interval = max(
5.0, float(os.environ.get("GYXX_CONSOLE_WATCHDOG_INTERVAL_SECONDS", "30"))
)
limit = int(os.environ.get("GYXX_CONSOLE_WATCHDOG_FAILURES", "3"))
enabled = (
os.environ.get("GYXX_CONSOLE_WATCHDOG", "1").strip().lower()
not in {"0", "false", "no", "off"}
and limit > 0
)
stopped = threading.Event()
if not enabled:
return stopped
thread = threading.Thread(
target=_console_watchdog,
args=(server.server_port,),
kwargs={
"interval_seconds": interval,
"failure_limit": limit,
"probe_timeout_seconds": min(5.0, interval),
"stopped": stopped,
},
name="gyxx-console-watchdog",
daemon=True,
)
thread.start()
return stopped
def serve_console( def serve_console(
settings: Settings, settings: Settings,
*, *,
@@ -4211,11 +4325,13 @@ def serve_console(
display_host = "127.0.0.1" if host in {"0.0.0.0", "::"} else host display_host = "127.0.0.1" if host in {"0.0.0.0", "::"} else host
output.write(f"GYXX Flow console: http://{display_host}:{server.server_port}\n") output.write(f"GYXX Flow console: http://{display_host}:{server.server_port}\n")
output.flush() output.flush()
watchdog_stopped = _start_console_watchdog(server)
try: try:
server.serve_forever(poll_interval=0.5) server.serve_forever(poll_interval=0.5)
except KeyboardInterrupt: except KeyboardInterrupt:
pass pass
finally: finally:
watchdog_stopped.set()
server.server_close() server.server_close()
return 0 return 0
+5 -39
View File
@@ -10,9 +10,6 @@ from types import SimpleNamespace
import pytest import pytest
from gyxx_flow.adapters import COOKIE_SKIP_EXIT_CODE from gyxx_flow.adapters import COOKIE_SKIP_EXIT_CODE
from gyxx_flow.modules.content_marketing.data.tools import (
friday_relogin_parallel as weekly_relogin,
)
@pytest.fixture @pytest.fixture
@@ -75,29 +72,6 @@ def _import_supply_script(module_name: str):
sys.stderr = stderr sys.stderr = stderr
def test_weekly_relogin_skips_the_entire_qr_workflow(
acceptance_environment: Path,
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
monkeypatch.setattr(
weekly_relogin.subprocess,
"Popen",
_forbidden("acceptance mode must not launch relogin subprocesses"),
)
monkeypatch.setattr(
weekly_relogin,
"_send_lark",
_forbidden("acceptance mode must not send QR screenshots"),
)
assert weekly_relogin.main() == COOKIE_SKIP_EXIT_CODE
assert "SKIPPED_COOKIE" in capsys.readouterr().out
assert "content.relogin.weekly" in acceptance_environment.read_text(
encoding="utf-8"
)
@pytest.mark.parametrize( @pytest.mark.parametrize(
"module_name", "module_name",
[ [
@@ -106,25 +80,17 @@ def test_weekly_relogin_skips_the_entire_qr_workflow(
"batch_rescrape_bilibili", "batch_rescrape_bilibili",
], ],
) )
def test_batch_rescrape_relogin_never_starts_a_qr_process( def test_batch_rescrape_has_no_relogin_capability(
module_name: str, module_name: str,
acceptance_environment: Path,
monkeypatch: pytest.MonkeyPatch, monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None: ) -> None:
"""扫码重登能力已从批量补采工具中移除,登录统一由账号 vault 管理。"""
module = importlib.import_module( module = importlib.import_module(
f"gyxx_flow.modules.content_marketing.data.tools.{module_name}" f"gyxx_flow.modules.content_marketing.data.tools.{module_name}"
) )
monkeypatch.setattr( assert not hasattr(module, "do_relogin")
module.subprocess, assert not hasattr(module, "RELOGIN_SCRIPT")
"call",
_forbidden("acceptance mode must not invoke a relogin script"),
)
assert module.do_relogin() is False
assert "SKIPPED_COOKIE" in capsys.readouterr().out
evidence = acceptance_environment.read_text(encoding="utf-8")
assert f"content.{module_name}.relogin" in evidence
@pytest.mark.parametrize( @pytest.mark.parametrize(
+4 -6
View File
@@ -18,8 +18,8 @@ def test_catalog_maps_39_project_scheduled_tasks() -> None:
assert len(scheduled) == 39 assert len(scheduled) == 39
assert len({item.source_task_name for item in scheduled}) == 39 assert len({item.source_task_name for item in scheduled}) == 39
assert Counter(item.module for item in scheduled) == { assert Counter(item.module for item in scheduled) == {
"content_marketing": 10, "content_marketing": 9,
"product_commerce": 13, "product_commerce": 14,
"shop_intelligence": 12, "shop_intelligence": 12,
"supply_chain": 4, "supply_chain": 4,
} }
@@ -80,7 +80,6 @@ def test_jd_video_upload_is_an_idempotent_scheduled_workflow() -> None:
assert step.args == ( assert step.args == (
"--anchor-record-id", "--anchor-record-id",
"recvrXTRWWFXeJ", "recvrXTRWWFXeJ",
"--manual-login",
"--execute", "--execute",
) )
assert step.replay_policy == "idempotent" assert step.replay_policy == "idempotent"
@@ -192,7 +191,7 @@ def test_every_scheduled_workflow_has_one_valid_asia_shanghai_schedule() -> None
assert schedule.workflow_id == workflow.workflow_id assert schedule.workflow_id == workflow.workflow_id
assert schedule.kind in {"daily", "weekly", "monthly", "interval_days"} assert schedule.kind in {"daily", "weekly", "monthly", "interval_days"}
assert schedule.at.count(":") == 1 assert schedule.at.count(":") == 1
assert sum(schedule.enabled for schedule in catalog.schedules) == 27 assert sum(schedule.enabled for schedule in catalog.schedules) == 28
customer_service = catalog.schedule_for("shop.jd_customer_service.daily") customer_service = catalog.schedule_for("shop.jd_customer_service.daily")
assert customer_service.at == "03:00" assert customer_service.at == "03:00"
assert customer_service.enabled is True assert customer_service.enabled is True
@@ -265,8 +264,7 @@ def test_scheduled_workflows_use_python_entries_and_content_tasks_are_graphs() -
"data/tools/sync_self_video_metrics.py" "data/tools/sync_self_video_metrics.py"
) )
assert catalog.schedule_for(self_operated.workflow_id).effective_times == ( assert catalog.schedule_for(self_operated.workflow_id).effective_times == (
"01:00", "06:00",
"05:00",
) )
backfill = next( backfill = next(
+1 -1
View File
@@ -54,7 +54,7 @@ def test_default_cli_registry_exposes_all_project_scheduled_workflows(
scheduled_ids = { scheduled_ids = {
entry.workflow_id for entry in registry.catalog.scheduled_workflows() entry.workflow_id for entry in registry.catalog.scheduled_workflows()
} }
assert len(scheduled_ids) == 35 assert len(scheduled_ids) == 39
assert scheduled_ids == { assert scheduled_ids == {
schedule.workflow_id for schedule in registry.catalog.schedules schedule.workflow_id for schedule in registry.catalog.schedules
} }
+93 -19
View File
@@ -2,6 +2,7 @@ from __future__ import annotations
import http.client import http.client
import json import json
import os
import re import re
import shutil import shutil
import subprocess import subprocess
@@ -311,7 +312,7 @@ def test_notification_config_persists_app_scoped_multi_recipient_routes(
) )
initial = service.notification_config() initial = service.notification_config()
assert len(initial["people"]) == 16 assert len(initial["people"]) == 18
assert initial["available_app_profiles"] == ["hermes-analyzer"] assert initial["available_app_profiles"] == ["hermes-analyzer"]
assert [app["profile"] for app in initial["apps"]] == ["hermes-analyzer"] assert [app["profile"] for app in initial["apps"]] == ["hermes-analyzer"]
wang = next(person for person in initial["people"] if person["id"] == "wang_yunlong") wang = next(person for person in initial["people"] if person["id"] == "wang_yunlong")
@@ -321,7 +322,12 @@ def test_notification_config_persists_app_scoped_multi_recipient_routes(
assert wang["bindings"]["hermes-analyzer"]["open_id"] == ( assert wang["bindings"]["hermes-analyzer"]["open_id"] == (
"ou_7ad5fc8012e2f741afc5346e05ffd447" "ou_7ad5fc8012e2f741afc5346e05ffd447"
) )
assert all(route["configured"] is False for route in initial["routes"]) stock_alert_route = next(
route
for route in initial["routes"]
if route["workflow_id"] == "product.online_stock_alert.daily"
)
assert stock_alert_route["configured"] is True
routes = [dict(route) for route in initial["routes"]] routes = [dict(route) for route in initial["routes"]]
daily = next( daily = next(
@@ -1566,7 +1572,7 @@ def test_overview_exposes_only_verified_scheduled_runs_as_cancellable(
"schema_version": 1, "schema_version": 1,
"updated_at": "2026-08-14T02:00:00+00:00", "updated_at": "2026-08-14T02:00:00+00:00",
"jobs": { "jobs": {
"content.relogin.weekly": { "content.notes_master.daily": {
"status": "running", "status": "running",
"pid": 4310, "pid": 4310,
"business_date": "2026-08-14", "business_date": "2026-08-14",
@@ -1589,7 +1595,7 @@ def test_overview_exposes_only_verified_scheduled_runs_as_cancellable(
workflow_id: str, workflow_id: str,
_job: dict[str, Any], _job: dict[str, Any],
) -> tuple[bool, Any, float | None]: ) -> tuple[bool, Any, float | None]:
if workflow_id == "content.relogin.weekly": if workflow_id == "content.notes_master.daily":
return True, VerifiedProcess(), 1_800_000_000.0 return True, VerifiedProcess(), 1_800_000_000.0
return False, None, None return False, None, None
@@ -1600,7 +1606,7 @@ def test_overview_exposes_only_verified_scheduled_runs_as_cancellable(
).overview() ).overview()
workflows = {item["id"]: item for item in overview["workflows"]} workflows = {item["id"]: item for item in overview["workflows"]}
verified = workflows["content.relogin.weekly"]["scheduler_job"] verified = workflows["content.notes_master.daily"]["scheduler_job"]
assert verified["status"] == "running" assert verified["status"] == "running"
assert verified["cancellable"] is True assert verified["cancellable"] is True
assert "pid" not in verified assert "pid" not in verified
@@ -1614,7 +1620,7 @@ def test_service_cancels_verified_scheduled_run_and_finalizes_journal(
monkeypatch: pytest.MonkeyPatch, monkeypatch: pytest.MonkeyPatch,
console_settings: Settings, console_settings: Settings,
) -> None: ) -> None:
workflow_id = "content.relogin.weekly" workflow_id = "content.notes_master.daily"
_write_scheduler_state( _write_scheduler_state(
console_settings, console_settings,
{ {
@@ -1906,24 +1912,13 @@ def test_content_production_preflight_rejects_non_production_modes(
relogin = next( relogin = next(
item item
for item in catalog.workflows for item in catalog.workflows
if item.workflow_id == "content.relogin.weekly" if item.workflow_id == "content.summary.weekly"
) )
with pytest.raises(ConsoleRequestError, match=message): with pytest.raises(ConsoleRequestError, match=message):
console_module._validate_content_production_runtime(relogin, {flag: "1"}) console_module._validate_content_production_runtime(relogin, {flag: "1"})
def test_content_relogin_preflight_does_not_require_postgres() -> None:
catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
relogin = next(
item
for item in catalog.workflows
if item.workflow_id == "content.relogin.weekly"
)
console_module._validate_content_production_runtime(relogin, {})
@pytest.mark.parametrize( @pytest.mark.parametrize(
"workflow_id", "workflow_id",
["content.notes_master.daily", "content.metrics.daily"], ["content.notes_master.daily", "content.metrics.daily"],
@@ -3297,7 +3292,7 @@ def test_http_notification_config_requires_revision_and_supports_multi_recipient
assert status == HTTPStatus.OK assert status == HTTPStatus.OK
_assert_security_headers(response_headers) _assert_security_headers(response_headers)
config = json.loads(body) config = json.loads(body)
assert len(config["people"]) == 16 assert len(config["people"]) == 18
assert { assert {
route["workflow_id"] for route in config["routes"] route["workflow_id"] for route in config["routes"]
} == set(console_module.NOTIFICATION_CAPABILITIES) } == set(console_module.NOTIFICATION_CAPABILITIES)
@@ -3616,3 +3611,82 @@ def test_douyin_negative_reviews_range_uses_single_run_with_env_dates(
{"business_date": "2026-09-19"}, {"business_date": "2026-09-19"},
) )
assert "business_date_end" not in single assert "business_date_end" not in single
def test_api_health_reports_process_and_scheduler_state(
console_settings: Settings,
) -> None:
with _running_server(console_settings, FakeConsoleLauncher()) as server:
status, _headers, content = _http_request(server, "GET", "/api/health")
assert status == HTTPStatus.OK
payload = json.loads(content)
assert payload["ok"] is True
assert payload["pid"] == os.getpid()
assert payload["uptime_seconds"] >= 0
# 测试环境没有调度器实例锁,health 只报状态不做任何拉起。
assert payload["scheduler_running"] is False
assert payload["scheduler_pid"] is None
def test_api_health_requires_no_token(console_settings: Settings) -> None:
# 自检探针端点供进程内看门狗与保活脚本使用,必须绕过访问令牌。
with _running_server(
console_settings, FakeConsoleLauncher(), token="x" * 24
) as server:
status, _headers, _content = _http_request(server, "GET", "/api/health")
assert status == HTTPStatus.OK
def test_probe_console_health_against_real_server(
console_settings: Settings,
) -> None:
with _running_server(console_settings, FakeConsoleLauncher()) as server:
assert console_module._probe_console_health(server.server_port, 2.0) is True
assert console_module._probe_console_health(1, 0.2) is False
def test_console_watchdog_exits_after_consecutive_probe_failures() -> None:
stopped = threading.Event()
exits: list[int] = []
probes: list[int] = []
def always_failing(port: int, _timeout: float) -> bool:
probes.append(port)
return False
console_module._console_watchdog(
8765,
interval_seconds=0.01,
failure_limit=3,
probe_timeout_seconds=0.01,
stopped=stopped,
probe=always_failing,
exit_process=exits.append,
sleep=lambda _seconds: None,
)
assert exits == [console_module.CONSOLE_WATCHDOG_EXIT_CODE]
assert probes == [8765, 8765, 8765]
def test_console_watchdog_resets_failure_count_after_successful_probe() -> None:
stopped = threading.Event()
exits: list[int] = []
results = iter([False, False, True, False, False, False])
def flaky_probe(_port: int, _timeout: float) -> bool:
return next(results)
console_module._console_watchdog(
8765,
interval_seconds=0.01,
failure_limit=3,
probe_timeout_seconds=0.01,
stopped=stopped,
probe=flaky_probe,
exit_process=exits.append,
sleep=lambda _seconds: None,
)
# 成功一次清零计数:2 次失败 + 1 次成功 + 3 次失败后才退出。
assert exits == [console_module.CONSOLE_WATCHDOG_EXIT_CODE]
+3 -3
View File
@@ -103,9 +103,9 @@ def test_project_registry_covers_every_script_with_unique_stable_port() -> None:
) )
assert len(scripts.command_ids) == 60 assert len(scripts.command_ids) == 60
assert len(first.command_ids) == 163 assert len(first.command_ids) == 157
bindings = [first.binding_for(command_id) for command_id in first.command_ids] bindings = [first.binding_for(command_id) for command_id in first.command_ids]
assert len({binding.cdp_port for binding in bindings}) == 163 assert len({binding.cdp_port for binding in bindings}) == 157
assert all(22000 <= binding.cdp_port <= 22999 for binding in bindings) assert all(22000 <= binding.cdp_port <= 22999 for binding in bindings)
assert all(binding.cdp_url.startswith("http://127.0.0.1:") for binding in bindings) assert all(binding.cdp_url.startswith("http://127.0.0.1:") for binding in bindings)
assert bindings == [second.binding_for(item.script_id) for item in bindings] assert bindings == [second.binding_for(item.script_id) for item in bindings]
@@ -450,7 +450,7 @@ def test_project_accounts_cover_all_configured_platform_login_identities() -> No
assert all( assert all(
catalog.bindings_for_account(account.account_id) catalog.bindings_for_account(account.account_id)
for account in catalog.accounts for account in catalog.accounts
if account.account_id != "tmall-ozko" if account.account_id not in {"tmall-ozko", "xingtu"}
) )
assert catalog.bindings_for_account("tmall-ozko") == () assert catalog.bindings_for_account("tmall-ozko") == ()
assert catalog.account_for("jd-shop").credential_env_names == ( assert catalog.account_for("jd-shop").credential_env_names == (
+1 -1
View File
@@ -54,7 +54,7 @@ def test_python_scheduler_status_is_project_owned_and_cross_platform(tmp_path: P
assert exit_code == 0 assert exit_code == 0
payload = json.loads(output.getvalue()) payload = json.loads(output.getvalue())
assert payload["timezone"] == "Asia/Shanghai" assert payload["timezone"] == "Asia/Shanghai"
assert payload["scheduled_count"] == 35 assert payload["scheduled_count"] == 39
assert payload["state"]["schema_version"] == 1 assert payload["state"]["schema_version"] == 1
+3 -2
View File
@@ -206,16 +206,17 @@ def test_misfire_older_than_grace_is_not_launched(tmp_path: Path) -> None:
def test_configured_schedules_disable_cookie_refresh_job() -> None: def test_configured_schedules_disable_cookie_refresh_job() -> None:
catalog = WorkflowCatalog.load(PROJECT_ROOT / "config") catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
assert len(catalog.schedules) == 35 assert len(catalog.schedules) == 39
assert [ assert [
schedule.workflow_id for schedule in catalog.schedules if not schedule.enabled schedule.workflow_id for schedule in catalog.schedules if not schedule.enabled
] == [ ] == [
"content.metrics.backfill", "content.metrics.backfill",
"content.marketing_report.daily", "content.marketing_report.daily",
"content.relogin.weekly",
"content.summary.weekly", "content.summary.weekly",
"content.summary.monthly", "content.summary.monthly",
"content.creator_report.monthly", "content.creator_report.monthly",
"shop.tmall_negative_reviews.daily",
"shop.tmall_customer_service.daily",
"product.alert.daily", "product.alert.daily",
"product.style_analysis.interval", "product.style_analysis.interval",
"supply.purchase_confirmation.daily", "supply.purchase_confirmation.daily",
+2 -2
View File
@@ -47,8 +47,8 @@ def test_deployment_documents_one_scheduler_and_external_data_root() -> None:
assert "GYXX_DATA_ROOT" in guide assert "GYXX_DATA_ROOT" in guide
assert "/var/lib/gyxx-flow" in guide assert "/var/lib/gyxx-flow" in guide
assert "Windows Task Scheduler" in guide assert "Windows Task Scheduler" in guide
assert len(payload["schedules"]) == 35 assert len(payload["schedules"]) == 39
assert sum(item.get("enabled", True) for item in payload["schedules"]) == 25 assert sum(item.get("enabled", True) for item in payload["schedules"]) == 28
def test_three_canonical_documents_cover_architecture_deployment_and_operations() -> None: def test_three_canonical_documents_cover_architecture_deployment_and_operations() -> None: