diff --git a/deploy/watchdog/gyxx_keepalive.py b/deploy/watchdog/gyxx_keepalive.py new file mode 100644 index 0000000..4806dce --- /dev/null +++ b/deploy/watchdog/gyxx_keepalive.py @@ -0,0 +1,336 @@ +# -*- coding: utf-8 -*- +"""控制台与调度器的进程外保活（登录自启动，随用户会话常驻）。 + +职责（不承载任何业务调度规则，业务调度仍由 `gyxx schedule run` 唯一负责）： +1. 控制台：探测 http://127.0.0.1:8765/api/health——进程不在则拉起； + 进程在但探测连续失败（wedged）则结束该进程树后拉起。控制台进程内 + 另有看门狗（console.py `_console_watchdog`）自检失败会主动退出，此处 + 负责把退出后的空窗补上，两层互为补充。 +2. 调度器：读 `var/state/scheduler/service.lock` 的 PID 判断存活，锁主 + 已死则按 install.ps1 同款参数拉起 `python -m gyxx_flow schedule run`； + 调度器自身单实例锁保证不会双跑。 +3. 节流：每个目标有冷却间隔；短时间反复重启触发熔断，只记录不再拉起， + 避免故障风暴。 + +用法： + gyxx_keepalive.py # 常驻循环（Startup 的 vbs 以 pythonw 启动） + gyxx_keepalive.py --once # 单轮探测+处置后退出，供验收/手动巡检 + +日志：var/logs/keepalive/keepalive.log（追加）。 +自身单实例：var/state/keepalive.lock（PID 存活检测，重复启动直接退出）。 +""" + +from __future__ import annotations + +import argparse +import ctypes +import http.client +import json +import os +import subprocess +import sys +import time +from datetime import datetime, timezone +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +ENV_FILES = [ + Path(os.getenv("GYXX_RUNTIME_ENV_FILE", r"D:\product-collector-analyze-flow\.env")), + Path( + os.getenv( + "GYXX_RUNTIME_JUSHUITAN_ENV_FILE", + r"D:\product-collector-analyze-flow\private-secrets\jushuitan.env", + ) + ), +] +CONSOLE_HOST = "127.0.0.1" +CONSOLE_PORT = int(os.getenv("GYXX_CONSOLE_PORT", "8765")) +LOOP_INTERVAL_SECONDS = float(os.getenv("GYXX_KEEPALIVE_INTERVAL_SECONDS", "30")) +PROBE_TIMEOUT_SECONDS = 5.0 +PROBE_FAILURE_LIMIT = 3 +RESTART_COOLDOWN_SECONDS = float(os.getenv("GYXX_KEEPALIVE_COOLDOWN_SECONDS", "120")) +TRIP_MAX_RESTARTS = 5 +TRIP_WINDOW_SECONDS = 30 * 60 +TRIP_PAUSE_SECONDS = 60 * 60 + +DATA_ROOT = PROJECT_ROOT / "var" +LOG_PATH = DATA_ROOT / "logs" / "keepalive" / "keepalive.log" +LOCK_PATH = DATA_ROOT / "state" / "keepalive.lock" +CONSOLE_LOG_DIR = DATA_ROOT / "logs" / "console-restarts" +SCHEDULER_LOG_DIR = DATA_ROOT / "logs" / "scheduler-service" +SCHEDULER_SERVICE_LOCK = DATA_ROOT / "state" / "scheduler" / "service.lock" + + +def _log(event: str, detail: str = "") -> None: + stamp = datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M:%S") + line = f"{stamp} {event} {detail}".rstrip() + LOG_PATH.parent.mkdir(parents=True, exist_ok=True) + with LOG_PATH.open("a", encoding="utf-8", newline="\n") as stream: + stream.write(line + "\n") + print(line, flush=True) + + +def _pid_alive(pid: int) -> bool: + if pid <= 0: + return False + kernel32 = ctypes.windll.kernel32 + handle = kernel32.OpenProcess(0x1000, False, pid) # PROCESS_QUERY_LIMITED_INFORMATION + if not handle: + return False + try: + exit_code = ctypes.c_ulong() + if kernel32.GetExitCodeProcess(handle, ctypes.byref(exit_code)): + return exit_code.value == 259 # STILL_ACTIVE + return False + finally: + kernel32.CloseHandle(handle) + + +def _acquire_self_lock() -> bool: + LOCK_PATH.parent.mkdir(parents=True, exist_ok=True) + if LOCK_PATH.is_file(): + try: + existing = json.loads(LOCK_PATH.read_text(encoding="utf-8")) + if _pid_alive(int(existing.get("pid", 0))): + return False + except Exception: + pass + LOCK_PATH.write_text( + json.dumps({"pid": os.getpid(), "started_at": datetime.now(timezone.utc).isoformat()}), + encoding="utf-8", + ) + return True + + +def _console_env_files() -> list[Path]: + return [path for path in ENV_FILES if path.is_file()] + + +def _spawn(argv: list[str], log_dir: Path, log_prefix: str) -> int: + log_dir.mkdir(parents=True, exist_ok=True) + stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + log_path = log_dir / f"{log_prefix}-{stamp}.log" + kwargs: dict[str, object] = {} + if os.name == "nt": + kwargs["creationflags"] = ( + subprocess.CREATE_NO_WINDOW | subprocess.CREATE_NEW_PROCESS_GROUP + ) + else: + kwargs["start_new_session"] = True + with log_path.open("a", encoding="utf-8", newline="\n") as stream: + process = subprocess.Popen( + argv, + cwd=str(PROJECT_ROOT), + env=os.environ, + stdin=subprocess.DEVNULL, + stdout=stream, + stderr=subprocess.STDOUT, + shell=False, + **kwargs, + ) + return process.pid + + +def _console_argv() -> list[str]: + argv = [ + str(PROJECT_ROOT / ".venv" / "Scripts" / "python.exe"), + "-m", + "gyxx_flow", + "console", + "--host", + CONSOLE_HOST, + "--port", + str(CONSOLE_PORT), + ] + for env_file in _console_env_files(): + argv.extend(("--env-file", str(env_file))) + return argv + + +def _scheduler_argv() -> list[str]: + argv = [ + str(PROJECT_ROOT / ".venv" / "Scripts" / "python.exe"), + "-m", + "gyxx_flow", + "schedule", + "run", + ] + for env_file in _console_env_files(): + argv.extend(("--env-file", str(env_file))) + return argv + + +def _probe_console() -> bool: + try: + connection = http.client.HTTPConnection(CONSOLE_HOST, CONSOLE_PORT, timeout=PROBE_TIMEOUT_SECONDS) + try: + connection.request("GET", "/api/health") + response = connection.getresponse() + response.read() + return response.status == 200 + finally: + connection.close() + except Exception: + return False + + +def _port_owner_pid(port: int) -> int | None: + try: + output = subprocess.run( + ["netstat", "-ano", "-p", "tcp"], + capture_output=True, + text=True, + timeout=10, + ).stdout + except Exception: + return None + needle = f"{CONSOLE_HOST}:{port}" + for line in output.splitlines(): + parts = line.split() + if ( + len(parts) >= 5 + and parts[0].upper() == "TCP" + and parts[1] == needle + and parts[3].upper() == "LISTENING" + ): + try: + return int(parts[4]) + except ValueError: + return None + return None + + +def _kill_pid_tree(pid: int) -> None: + if pid <= 0 or pid == os.getpid(): + return + subprocess.run( + ["taskkill", "/PID", str(pid), "/T", "/F"], + capture_output=True, + timeout=30, + ) + + +def _scheduler_lock_pid() -> int | None: + try: + payload = json.loads(SCHEDULER_SERVICE_LOCK.read_text(encoding="utf-8")) + return int(payload.get("pid", 0)) or None + except Exception: + return None + + +class _TargetState: + """记录单目标的连续失败计数、冷却与熔断状态。""" + + def __init__(self, name: str) -> None: + self.name = name + self.failures = 0 + self.last_restart_at = 0.0 + self.recent_restarts: list[float] = [] + self.tripped_at: float | None = None + + def tripped(self, now: float) -> bool: + if self.tripped_at is None: + return False + if now - self.tripped_at >= TRIP_PAUSE_SECONDS: + self.tripped_at = None + self.recent_restarts.clear() + _log("keepalive-trip-reset", f"target={self.name} 熔断窗口结束恢复守护") + return False + return True + + def may_restart(self, now: float) -> bool: + if self.tripped(now): + return False + if now - self.last_restart_at < RESTART_COOLDOWN_SECONDS: + return False + self.recent_restarts = [ + stamp for stamp in self.recent_restarts if now - stamp < TRIP_WINDOW_SECONDS + ] + if len(self.recent_restarts) >= TRIP_MAX_RESTARTS: + self.tripped_at = now + _log( + "keepalive-tripped", + f"target={self.name} {TRIP_MAX_RESTARTS} 次重启/{TRIP_WINDOW_SECONDS // 60}分钟 " + f"暂停拉起{TRIP_PAUSE_SECONDS // 60}分钟，请人工检查日志", + ) + return False + return True + + def record_restart(self, now: float) -> None: + self.last_restart_at = now + self.recent_restarts.append(now) + + +def run_cycle(console: _TargetState, scheduler: _TargetState) -> None: + now = time.time() + + # ---- 控制台 ---- + if _probe_console(): + console.failures = 0 + else: + console.failures += 1 + owner = _port_owner_pid(CONSOLE_PORT) + _log( + "console-probe-failed", + f"{console.failures}/{PROBE_FAILURE_LIMIT} port_owner_pid={owner}", + ) + if console.failures >= PROBE_FAILURE_LIMIT and console.may_restart(now): + if owner is not None and _pid_alive(owner): + _log("console-kill-wedged", f"pid={owner} 探测持续失败，结束进程树") + _kill_pid_tree(owner) + time.sleep(3) + pid = _spawn(_console_argv(), CONSOLE_LOG_DIR, "keepalive-console") + console.record_restart(time.time()) + console.failures = 0 + _log("console-restarted", f"pid={pid}") + + # ---- 调度器 ---- + scheduler_pid = _scheduler_lock_pid() + scheduler_alive = scheduler_pid is not None and _pid_alive(scheduler_pid) + if not scheduler_alive and scheduler.may_restart(now): + pid = _spawn(_scheduler_argv(), SCHEDULER_LOG_DIR, "keepalive-start") + scheduler.record_restart(time.time()) + _log( + "scheduler-restarted", + f"pid={pid} prev_lock_pid={scheduler_pid}", + ) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="GYXX Flow 控制台/调度器保活") + parser.add_argument("--once", action="store_true", help="单轮探测+处置后退出") + arguments = parser.parse_args(argv) + + if not _acquire_self_lock(): + _log("keepalive-duplicate-exit", "已有保活进程在运行") + return 0 + + console = _TargetState("console") + scheduler = _TargetState("scheduler") + missing_env = [str(path) for path in ENV_FILES if not path.is_file()] + _log( + "keepalive-start", + f"pid={os.getpid()} once={arguments.once}" + + (f" missing_env={missing_env}" if missing_env else ""), + ) + try: + while True: + run_cycle(console, scheduler) + if arguments.once: + _log("keepalive-once-done", "") + return 0 + time.sleep(LOOP_INTERVAL_SECONDS) + except KeyboardInterrupt: + return 0 + finally: + try: + if LOCK_PATH.is_file(): + payload = json.loads(LOCK_PATH.read_text(encoding="utf-8")) + if int(payload.get("pid", 0)) == os.getpid(): + LOCK_PATH.unlink() + except Exception: + pass + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/deploy/watchdog/start-keepalive.vbs b/deploy/watchdog/start-keepalive.vbs new file mode 100644 index 0000000..2fca36a --- /dev/null +++ b/deploy/watchdog/start-keepalive.vbs @@ -0,0 +1,5 @@ +' GYXX Flow 保活登录自启动：无窗口启动 deploy/watchdog/gyxx_keepalive.py， +' 由它负责拉起控制台（含进程内自检看门狗）与调度器。 +' 安装：复制到 %APPDATA%\Microsoft\Windows\Start Menu\Programs\Startup\ +CreateObject("WScript.Shell").Run _ + """D:\gyxx-flow\.venv\Scripts\pythonw.exe"" ""D:\gyxx-flow\deploy\watchdog\gyxx_keepalive.py""", 0, False diff --git a/src/gyxx_flow/cookie_gate.py b/src/gyxx_flow/cookie_gate.py new file mode 100644 index 0000000..6cb4b19 --- /dev/null +++ b/src/gyxx_flow/cookie_gate.py @@ -0,0 +1,882 @@ +"""Per-account cookie health gate injected before workflow collection steps. + +Every runtime account a workflow depends on is checked before the collection +steps that use it may run. Accounts fall into three gate modes derived from +``config/runtime-bindings.json``: + +* ``blocking`` -- QR-code / SMS-only accounts (no credential environment and + no configured login command). The gate verifies the login state and, when + it has expired, fails the gate step so every dependent step is skipped with + ``dependency-failed`` and sends a Feishu alert asking the login owner to + re-login interactively via ``gyxx accounts login ``. +* ``recover`` -- accounts with a vault-managed credential login command + (``login_mode`` ``C``). The keepalive manager re-runs the credential login + automatically; only a failed recovery blocks the workflow and alerts. +* ``defer`` -- accounts whose member scripts embed username/password login + flows. These keep their script-level login behaviour unchanged and are not + gated at all. + +The per-account decision is persisted as a small state machine under +``state/accounts//cookie_state.json`` (``healthy`` <-> ``awaiting_login``) +so repeat failures alert at most once per notify interval instead of once per +workflow run. +""" + +from __future__ import annotations + +import json +import os +from collections.abc import Callable +from dataclasses import dataclass, replace +from datetime import date, datetime, timezone +from pathlib import Path +from typing import Any, Mapping + +from gyxx_flow.accounts import ( + AccountKeepaliveManager, + _keepalive_requires_relogin, + _read_keepalive_status, + _vault_cookie_valid, +) +from gyxx_flow.adapters.browser import BrowserCookieStore +from gyxx_flow.adapters.feishu_messages import send_lark_bot_message +from gyxx_flow.adapters.integration import RuntimeIntegrationCatalog +from gyxx_flow.catalog import WorkflowCatalog +from gyxx_flow.core.artifacts import atomic_write_json +from gyxx_flow.core.config import Settings +from gyxx_flow.core.context import RunContext +from gyxx_flow.notification_routing import ( + ANALYZER_ACCEPTANCE_RECIPIENT_OPEN_ID, + ANALYZER_NOTIFICATION_APP_PROFILE, + NotificationRoutingStore, +) +from gyxx_flow.workflow.model import StepDefinition, WorkflowDefinition +from gyxx_flow.workflow.steps import StepExecution + +GATE_MODE_BLOCKING = "blocking" +GATE_MODE_RECOVER = "recover" +GATE_MODE_DEFER = "defer" + +# 飞书通知的默认接收人：按手机号在 config/notification-routing.json 的 +# people 段解析 open_id；解析失败时回退到同名 analyzer 常量收件人。 +DEFAULT_LOGIN_ALERT_MOBILE = "+8613663785145" + +GYXX_COOKIE_GATE = "GYXX_COOKIE_GATE" +GYXX_COOKIE_GATE_MAX_STATUS_AGE_SECONDS = "GYXX_COOKIE_GATE_MAX_STATUS_AGE_SECONDS" +GYXX_LOGIN_ALERT_OPEN_IDS = "GYXX_LOGIN_ALERT_OPEN_IDS" +GYXX_LOGIN_ALERT_APP_PROFILE = "GYXX_LOGIN_ALERT_APP_PROFILE" +GYXX_LOGIN_ALERT_NOTIFY_INTERVAL_SECONDS = "GYXX_LOGIN_ALERT_NOTIFY_INTERVAL_SECONDS" + +_GATE_STEP_PREFIX = "cookie-gate" +_GATE_STATE_FILENAME = "cookie_state.json" +_GATE_TIMEOUT_SECONDS = 1800.0 +_DEFAULT_MAX_STATUS_AGE_SECONDS = 3600.0 +_MIN_MAX_STATUS_AGE_SECONDS = 300.0 +_DEFAULT_NOTIFY_INTERVAL_SECONDS = 1800.0 +_DISABLED_VALUES = frozenset({"0", "false", "off", "no", "disabled"}) +# 自动重跑：待重跑清单上限、健康探活节流、同账号重跑启动冷却。 +_MAX_PENDING_RERUNS = 20 +_RECOVERY_PROBE_INTERVAL_SECONDS = 300.0 +_RECOVERY_LAUNCH_COOLDOWN_SECONDS = 300.0 +_RECOVERY_HISTORY_LIMIT = 20 + +STATE_HEALTHY = "healthy" +STATE_AWAITING_LOGIN = "awaiting_login" + + +@dataclass(frozen=True, slots=True) +class GateVerdict: + """One account's gate decision, recorded in the run journal on failure.""" + + account_id: str + mode: str + blocked: bool + message: str + recovered: bool = False + notified: bool = False + + +@dataclass(frozen=True, slots=True) +class CookieGateStepAction: + """ExecutableStep that turns one account verdict into a step outcome.""" + + gate: "CookieGate" + account_id: str + + def execute( + self, + *, + context: RunContext, + timeout_seconds: float | None, + dry_run: bool, + ) -> StepExecution: + if dry_run: # pragma: no cover - the engine skips dry-run steps first + return StepExecution(exit_code=0, skipped=True, reason="dry-run") + verdict = self.gate.check(self.account_id, context=context) + if verdict.blocked: + return StepExecution( + exit_code=1, + error=verdict.message, + reason="cookie-expired", + ) + return StepExecution(exit_code=0) + + +class CookieGate: + """Resolve, probe, and gate the accounts one workflow depends on.""" + + def __init__( + self, + settings: Settings, + *, + catalog: RuntimeIntegrationCatalog | None = None, + keepalive: AccountKeepaliveManager | None = None, + environment: Mapping[str, str] | None = None, + now: Callable[[], datetime] | None = None, + sender: Callable[..., dict[str, Any]] | None = None, + ) -> None: # noqa: D107 - simple wiring + self.settings = settings + self.catalog = catalog or RuntimeIntegrationCatalog.load_default( + project_root=settings.project_root, + data_root=settings.data_root, + ) + self.keepalive = keepalive or AccountKeepaliveManager( + settings, catalog=self.catalog + ) + self._environment = dict(os.environ if environment is None else environment) + self._now = now or (lambda: datetime.now(timezone.utc)) + self._sender = sender or send_lark_bot_message + self._workflow_catalog: WorkflowCatalog | None = None + + # ------------------------------------------------------------------ + # Account resolution + # ------------------------------------------------------------------ + + def gate_mode(self, account_id: str) -> str: + account = self.catalog.account_for(account_id) + if str(account.login_command_id or "").strip(): + return GATE_MODE_RECOVER + if account.credential_env_names or account.login_mode == "B": + return GATE_MODE_DEFER + return GATE_MODE_BLOCKING + + def collect_workflow_accounts( + self, + workflow: WorkflowDefinition, + ) -> dict[str, set[str]]: + """Map each gated account to the step ids that depend on it. + + Accounts resolve two ways: directly, from each step binding's + ``account``; and declaratively, from the catalog workflow's + ``required_accounts`` (used by orchestrator entries whose own binding + has no account while the spawned child collectors do). + """ + + accounts: dict[str, set[str]] = {} + for step in workflow.steps: + account_id = self._account_for_step(step) + if account_id: + accounts.setdefault(account_id, set()).add(step.step_id) + for account_id in self._declared_required_accounts(workflow.workflow_id): + self.catalog.account_for(account_id) # unknown id: loud config error + accounts.setdefault(account_id, set()).update( + step.step_id for step in workflow.steps + ) + # Defer-mode accounts keep their script-level credential login flows + # and are intentionally not gated. + return { + account_id: step_ids + for account_id, step_ids in accounts.items() + if self.gate_mode(account_id) != GATE_MODE_DEFER + } + + def _account_for_step(self, step: StepDefinition) -> str | None: + action = step.action + entry = getattr(action, "entry", None) + module_id = getattr(entry, "module", "") + command_entry = getattr(action, "command_entry", None) or getattr( + entry, "entry", "" + ) + if not module_id or not command_entry: + return None + try: + binding = self.catalog.binding_for(f"{module_id}:{command_entry}") + except Exception: # noqa: BLE001 - unbound entries run ungated + return None + return binding.account or None + + def workflow_catalog(self) -> WorkflowCatalog: + """Load (once) the catalog used for declared account and rerun lookups.""" + + if self._workflow_catalog is None: + try: + self._workflow_catalog = WorkflowCatalog.load( + self.settings.project_root / "config" + ) + except Exception: # noqa: BLE001 - catalog issues surface elsewhere + self._workflow_catalog = WorkflowCatalog( + timezone="UTC", workflows=(), schedules=() + ) + return self._workflow_catalog + + def _declared_required_accounts(self, workflow_id: str) -> tuple[str, ...]: + for entry in self.workflow_catalog().workflows: + if entry.workflow_id == workflow_id: + return tuple(entry.required_accounts) + return () + + # ------------------------------------------------------------------ + # State machine + # ------------------------------------------------------------------ + + def check(self, account_id: str, *, context: RunContext) -> GateVerdict: + account = self.catalog.account_for(account_id) + mode = self.gate_mode(account_id) + store = BrowserCookieStore(account.cookie_file, account.storage_state_file) + static_valid = _vault_cookie_valid(account, store) + status = _read_keepalive_status(account) + if static_valid and self._keepalive_success_is_fresh(status): + self._write_state(account, STATE_HEALTHY, probe="fresh-keepalive") + return GateVerdict( + account_id=account_id, + mode=mode, + blocked=False, + message=f"账号 {account_id} 登录态有效（keepalive 近期验证通过）", + ) + if self._acceptance_enabled(): + # 验收模式不开浏览器：按静态证据判定，且不投递飞书通知。 + if static_valid and not _keepalive_requires_relogin(status): + return GateVerdict( + account_id=account_id, + mode=mode, + blocked=False, + message=f"账号 {account_id} 静态 cookie 校验通过（验收模式）", + ) + return self._expire( + account, + mode, + context, + detail="验收模式下静态 cookie 校验未通过", + notify=False, + ) + + outcome = self.keepalive.refresh_account(account_id) + probe_status = str(outcome.get("status", "error")) + if probe_status == "success": + recovered = bool(outcome.get("login_recovered")) + self._write_state( + account, + STATE_HEALTHY, + probe="success", + recovered=recovered, + ) + message = ( + f"账号 {account_id} 登录态有效（自动账密重登成功）" + if recovered + else f"账号 {account_id} 登录态有效（探活通过）" + ) + return GateVerdict( + account_id=account_id, + mode=mode, + blocked=False, + message=message, + recovered=recovered, + ) + if probe_status == "busy": + if static_valid and not _keepalive_requires_relogin(status): + return GateVerdict( + account_id=account_id, + mode=mode, + blocked=False, + message=f"账号 {account_id} 探活锁忙，静态 cookie 校验通过", + ) + return self._expire( + account, + mode, + context, + detail="探活锁被占用且静态 cookie 校验未通过", + ) + if probe_status in {"expired", "not_logged_in"}: + evidence = _recovery_evidence(_read_keepalive_status(account)) + detail = "cookie 已过期，无法正常采集" + if probe_status == "not_logged_in" and evidence: + detail = f"登录态缺失；{evidence}" + elif evidence: + detail = f"cookie 已过期；{evidence}" + return self._expire(account, mode, context, detail=detail) + # timeout / error / http_error / disabled：探测结果不可信时回退静态证据。 + if static_valid and not _keepalive_requires_relogin(status): + return GateVerdict( + account_id=account_id, + mode=mode, + blocked=False, + message=f"账号 {account_id} 探活异常（{probe_status}），静态 cookie 校验通过", + ) + return self._expire( + account, + mode, + context, + detail=f"探活异常（{probe_status}）且静态 cookie 校验未通过", + ) + + # ------------------------------------------------------------------ + # Expiry, notification, persisted state + # ------------------------------------------------------------------ + + def _expire( + self, + account: Any, + mode: str, + context: RunContext, + *, + detail: str, + notify: bool | None = None, + ) -> GateVerdict: + account_id = str(account.account_id) + if notify is None: + notify = not context.shadow and not self._acceptance_enabled() + state = self._read_state(account) + notified = False + notification_note = "" + if notify and not self._within_notify_interval(state): + recipients = self._alert_recipients() + message = _expiry_message(account_id, mode, context, detail) + for open_id, profile in recipients: + try: + self._sender( + user_id=open_id, + text=message, + idempotency_key=f"cookie-gate:{account_id}:{context.run_id}", + profile=profile, + ) + notified = True + except Exception as exc: # noqa: BLE001 - alerting must not mask the verdict + notification_note = f"{type(exc).__name__}: {exc}"[:300] + if recipients and notified: + state["last_notified_at"] = _now_iso(self._now) + state.pop("notification_error", None) + elif notification_note: + state["notification_error"] = notification_note + state["state"] = STATE_AWAITING_LOGIN + state["updated_at"] = _now_iso(self._now) + state["last_probe_status"] = "expired" + state["last_blocked_run"] = context.run_id + if not context.shadow and not self._acceptance_enabled(): + # 记录待自动重跑清单：登录态恢复后由调度器维护循环按原业务日期重跑。 + state["pending_reruns"] = _record_pending_rerun( + state.get("pending_reruns"), context, _now_iso(self._now) + ) + self._save_state(account, state) + message = _expiry_message(account_id, mode, context, detail) + if notified: + message += " 已通过飞书通知登录负责人。" + elif notification_note: + message += f" 飞书通知发送失败（{notification_note}）。" + return GateVerdict( + account_id=account_id, + mode=mode, + blocked=True, + message=message, + notified=notified, + ) + + def _write_state( + self, + account: Any, + state_name: str, + *, + probe: str, + recovered: bool = False, + ) -> None: + state = self._read_state(account) + if ( + state_name == STATE_HEALTHY + and state.get("state") == STATE_HEALTHY + and state.get("last_probe_status") in {"fresh-keepalive", "success"} + and not recovered + ): + return # 持续健康时避免每次运行都重写状态文件 + state["state"] = state_name + state["updated_at"] = _now_iso(self._now) + state["last_probe_status"] = probe + if state_name == STATE_HEALTHY: + state["last_healthy_at"] = _now_iso(self._now) + if recovered: + state["last_recovered_at"] = _now_iso(self._now) + state.pop("notification_error", None) + self._save_state(account, state) + + def _state_path(self, account: Any) -> Path: + return account.cookie_file.with_name(_GATE_STATE_FILENAME) + + def _read_state(self, account: Any) -> dict[str, object]: + path = self._state_path(account) + if not path.exists(): + return {"account_id": str(account.account_id)} + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {"account_id": str(account.account_id)} + if not isinstance(payload, dict): + return {"account_id": str(account.account_id)} + return dict(payload) + + def _save_state(self, account: Any, state: Mapping[str, object]) -> None: + payload = { + "account_id": str(account.account_id), + **state, + } + try: + atomic_write_json(self._state_path(account), payload) + except OSError: # pragma: no cover - state file is advisory only + pass + + def _within_notify_interval(self, state: Mapping[str, object]) -> bool: + raw = state.get("last_notified_at") + if not isinstance(raw, str): + return False + try: + last = datetime.fromisoformat(raw) + except ValueError: + return False + now = self._now() + if last.tzinfo is None: + last = last.replace(tzinfo=timezone.utc) + interval = _env_int( + self._environment, + GYXX_LOGIN_ALERT_NOTIFY_INTERVAL_SECONDS, + _DEFAULT_NOTIFY_INTERVAL_SECONDS, + minimum=0, + ) + return (now - last).total_seconds() < interval + + def _keepalive_success_is_fresh(self, status: Mapping[str, object] | None) -> bool: + if not isinstance(status, Mapping) or status.get("status") != "success": + return False + raw = status.get("last_success_at") + if not isinstance(raw, str): + return False + try: + last = datetime.fromisoformat(raw) + except ValueError: + return False + now = self._now() + if last.tzinfo is None: + last = last.replace(tzinfo=timezone.utc) + max_age = _env_float( + self._environment, + GYXX_COOKIE_GATE_MAX_STATUS_AGE_SECONDS, + _DEFAULT_MAX_STATUS_AGE_SECONDS, + minimum=_MIN_MAX_STATUS_AGE_SECONDS, + ) + return (now - last).total_seconds() <= max_age + + def _alert_recipients(self) -> list[tuple[str, str]]: + profile = ( + self._environment.get(GYXX_LOGIN_ALERT_APP_PROFILE, "").strip() + or ANALYZER_NOTIFICATION_APP_PROFILE + ) + override = [ + value.strip() + for value in self._environment.get(GYXX_LOGIN_ALERT_OPEN_IDS, "").split(",") + if value.strip() + ] + if override: + return [(open_id, profile) for open_id in override] + try: + payload, _revision = NotificationRoutingStore(self.settings).snapshot() + for person in payload["people"].values(): + if person.get("mobile") == DEFAULT_LOGIN_ALERT_MOBILE: + binding = person.get("bindings", {}).get(profile) + if binding: + return [(str(binding["open_id"]), profile)] + except Exception: # noqa: BLE001 - fall back to the constant recipient + pass + return [(ANALYZER_ACCEPTANCE_RECIPIENT_OPEN_ID, profile)] + + def _acceptance_enabled(self) -> bool: + value = self._environment.get("GYXX_WORKFLOW_ACCEPTANCE", "") + return bool(value and value.strip().casefold() in {"1", "true", "yes", "on"}) + + +def gate_enabled(environment: Mapping[str, str] | None = None) -> bool: + values = os.environ if environment is None else environment + return values.get(GYXX_COOKIE_GATE, "").strip().casefold() not in _DISABLED_VALUES + + +def apply_cookie_gates( + workflow: WorkflowDefinition, + *, + settings: Settings, + environment: Mapping[str, str] | None = None, + gate: CookieGate | None = None, +) -> WorkflowDefinition: + """Inject one critical gate step per gated account in front of its users. + + Gate steps are graph sources; each step that uses an account gains the + account's gate as an extra dependency. A failed gate therefore skips its + dependents with ``dependency-failed`` through the regular engine, while + unrelated branches keep running. Operator slices (``--rerun-step`` / + ``--resume-from``) are produced before this injection and stay gate-free. + """ + + if not workflow.steps: + return workflow + if gate is None: + try: + gate = CookieGate( + settings, + environment=os.environ if environment is None else environment, + ) + except Exception: # noqa: BLE001 - 无集成配置时不门控；步骤执行仍会显式报错 + return workflow + account_steps = gate.collect_workflow_accounts(workflow) + if not account_steps: + return workflow + existing_ids = {step.step_id for step in workflow.steps} + gate_steps: list[StepDefinition] = [] + for account_id in sorted(account_steps): + step_id = f"{_GATE_STEP_PREFIX}.{account_id}" + if step_id in existing_ids: + continue + gate_steps.append( + StepDefinition( + step_id, + CookieGateStepAction(gate, account_id), + timeout_seconds=_GATE_TIMEOUT_SECONDS, + max_attempts=1, + ) + ) + if not gate_steps: + return workflow + patched: list[StepDefinition] = [] + for step in workflow.steps: + dependencies = list(dict.fromkeys(step.depends_on)) + changed = False + for account_id, step_ids in account_steps.items(): + gate_id = f"{_GATE_STEP_PREFIX}.{account_id}" + if step.step_id in step_ids and gate_id not in dependencies: + dependencies.append(gate_id) + changed = True + if changed: + step = replace(step, depends_on=tuple(dependencies)) + patched.append(step) + return WorkflowDefinition( + workflow.workflow_id, + (*gate_steps, *patched), + ) + + +class CookieGateRecoveryManager: + """Re-run workflows blocked by expired cookies once the login recovers. + + Runs inside the resident scheduler's maintenance loop. For every account + with a non-empty ``pending_reruns`` list, it re-verifies the login state + (fresh keepalive success first, then a throttled live probe); on recovery + it relaunches each recorded workflow with its original business date + through the scheduler's own launcher, clears the pending list, and sends a + Feishu confirmation to the login owner. + """ + + def __init__( + self, + gate: CookieGate, + *, + launcher: Any, + probe_interval_seconds: float = _RECOVERY_PROBE_INTERVAL_SECONDS, + launch_cooldown_seconds: float = _RECOVERY_LAUNCH_COOLDOWN_SECONDS, + ) -> None: + self.gate = gate + self._launcher = launcher + self._probe_interval = probe_interval_seconds + self._launch_cooldown = launch_cooldown_seconds + + def tick(self) -> list[dict[str, object]]: + """Attempt recovery reruns once; returns redacted outcomes per account.""" + + outcomes: list[dict[str, object]] = [] + for account in self.gate.catalog.accounts: + state = self.gate._read_state(account) + pending = state.get("pending_reruns") + if not isinstance(pending, list) or not pending: + continue + now = self.gate._now() + if not self._launch_cooldown_passed(state, now): + continue + if not self._account_recovered(account, state, now): + continue + launched, remaining = self._launch_pending(account, pending, now) + state["pending_reruns"] = remaining + state["updated_at"] = _now_iso(self.gate._now) + state["last_recovered_at"] = state["updated_at"] + if not remaining: + state["state"] = STATE_HEALTHY + state["last_probe_status"] = "recovered-rerun" + if launched: + state["last_rerun_launch_at"] = state["updated_at"] + history = [ + item + for item in state.get("rerun_history", []) + if isinstance(item, dict) + ] + history.extend(launched) + state["rerun_history"] = history[-_RECOVERY_HISTORY_LIMIT:] + self.gate._save_state(account, state) + if launched: + self._notify_recovered(str(account.account_id), launched) + outcomes.append( + { + "account_id": str(account.account_id), + "launched": launched, + "pending": len(remaining), + } + ) + return outcomes + + # ------------------------------------------------------------------ + # internals + # ------------------------------------------------------------------ + + def _launch_cooldown_passed( + self, state: Mapping[str, object], now: datetime + ) -> bool: + raw = state.get("last_rerun_launch_at") + if not isinstance(raw, str): + return True + try: + last = datetime.fromisoformat(raw) + except ValueError: + return True + if last.tzinfo is None: + last = last.replace(tzinfo=timezone.utc) + return (now - last).total_seconds() >= self._launch_cooldown + + def _account_recovered( + self, + account: Any, + state: dict[str, object], + now: datetime, + ) -> bool: + """Fresh keepalive success first; otherwise one throttled live probe.""" + + store = BrowserCookieStore(account.cookie_file, account.storage_state_file) + if _vault_cookie_valid(account, store) and self.gate._keepalive_success_is_fresh( + _read_keepalive_status(account) + ): + return True + raw = state.get("last_recovery_probe_at") + if isinstance(raw, str): + try: + last_probe = datetime.fromisoformat(raw) + except ValueError: + last_probe = None + if last_probe is not None: + if last_probe.tzinfo is None: + last_probe = last_probe.replace(tzinfo=timezone.utc) + if (now - last_probe).total_seconds() < self._probe_interval: + return False + state["last_recovery_probe_at"] = _now_iso(self.gate._now) + self.gate._save_state(account, state) + try: + outcome = self.gate.keepalive.refresh_account(str(account.account_id)) + except Exception: # noqa: BLE001 - 恢复探活失败留给下一轮 + return False + return str(outcome.get("status", "error")) == "success" + + def _launch_pending( + self, + account: Any, + pending: list[object], + now: datetime, + ) -> tuple[list[dict[str, object]], list[dict[str, object]]]: + known_ids = { + entry.workflow_id for entry in self.gate.workflow_catalog().workflows + } + launched: list[dict[str, object]] = [] + remaining: list[dict[str, object]] = [] + for item in pending: + if not isinstance(item, dict): + continue # 控制台脚本等临时运行 id：不在 catalog 中，直接丢弃 + workflow_id = item.get("workflow_id") + business_date = item.get("business_date") + if not isinstance(workflow_id, str) or not isinstance(business_date, str): + continue + if workflow_id not in known_ids: + continue + try: + process = self._launcher.launch( + workflow_id, now, date.fromisoformat(business_date) + ) + except Exception as exc: # noqa: BLE001 - 保留 pending，下一轮重试 + remaining.append( + { + "workflow_id": workflow_id, + "business_date": business_date, + "error": f"{type(exc).__name__}: {exc}"[:200], + } + ) + continue + close = getattr(self._launcher, "close", None) + if callable(close): + try: + close(process) + except Exception: # noqa: BLE001 - 关闭日志句柄失败不影响重跑 + pass + launched.append( + { + "workflow_id": workflow_id, + "business_date": business_date, + "pid": int(getattr(process, "pid", 0) or 0), + "launched_at": _now_iso(self.gate._now), + } + ) + return launched, remaining + + def _notify_recovered( + self, + account_id: str, + launched: list[dict[str, object]], + ) -> None: + lines = [ + f"- {item['workflow_id']}（业务日期 {item['business_date']}）" + for item in launched + ] + message = ( + "【GYXX 登录态已恢复】账号 " + f"{account_id} 登录态已恢复，已自动重跑被阻断的工作流：\n" + + "\n".join(lines) + + "\n如本次重跑仍有失败，请查看运行日志或工作流控制台。" + ) + key = "-".join( + [account_id, *(str(item["workflow_id"]) for item in launched)] + ) + for open_id, profile in self.gate._alert_recipients(): + try: + self.gate._sender( + user_id=open_id, + text=message, + idempotency_key=f"cookie-recovery:{key}", + profile=profile, + ) + except Exception: # noqa: BLE001 - 确认通知失败不影响已发起的重跑 + continue + + +def _record_pending_rerun( + current: object, + context: RunContext, + blocked_at: str, +) -> list[dict[str, object]]: + """Append one blocked run to the pending list, deduped and bounded.""" + + pending: list[dict[str, object]] = [ + dict(item) + for item in (current if isinstance(current, list) else []) + if isinstance(item, dict) + ] + entry = { + "workflow_id": context.workflow_id, + "business_date": context.business_date.isoformat(), + "blocked_run_id": context.run_id, + "blocked_at": blocked_at, + } + pending = [ + item + for item in pending + if not ( + item.get("workflow_id") == entry["workflow_id"] + and item.get("business_date") == entry["business_date"] + ) + ] + pending.append(entry) + return pending[-_MAX_PENDING_RERUNS:] + + +def _recovery_evidence(status: Mapping[str, object] | None) -> str: + if not isinstance(status, Mapping): + return "" + if not status.get("login_attempted"): + return "" + reason = status.get("login_reason") + error_type = status.get("login_error_type") + exit_code = status.get("login_exit_code") + parts = ["自动账密重登已尝试但未成功"] + if isinstance(reason, str) and reason: + parts.append(reason) + if isinstance(error_type, str) and error_type: + parts.append(f"错误={error_type}") + if isinstance(exit_code, int) and not isinstance(exit_code, bool): + parts.append(f"退出码={exit_code}") + return "（" + "，".join(parts) + "）" + + +def _expiry_message( + account_id: str, + mode: str, + context: RunContext, + detail: str, +) -> str: + mode_note = ( + "账号密码自动重登失败，需人工处理（如验证码）" + if mode == GATE_MODE_RECOVER + else "该平台仅支持扫码/验证码登录，无法自动恢复" + ) + return ( + "【GYXX 登录态失效】工作流被阻断，需要人工重新登录\n" + f"工作流: {context.workflow_id}（业务日期 {context.business_date.isoformat()}）\n" + f"账号: {account_id}\n" + f"原因: {detail}；{mode_note}\n" + f"请在服务器上执行: uv run gyxx accounts login {account_id}\n" + "（将打开浏览器窗口，完成扫码或验证码登录后自动保存并同步到全部成员脚本；" + "受影响的采集步骤本次已跳过，登录完成后下次运行自动恢复）" + ) + + +def _now_iso(now: Callable[[], datetime]) -> str: + return now().astimezone(timezone.utc).isoformat(timespec="seconds") + + +def _env_int( + environment: Mapping[str, str], + name: str, + default: int, + *, + minimum: int, +) -> int: + raw = environment.get(name, "").strip() + try: + value = int(raw) + except (TypeError, ValueError): + return default + return max(minimum, value) + + +def _env_float( + environment: Mapping[str, str], + name: str, + default: float, + *, + minimum: float, +) -> float: + raw = environment.get(name, "").strip() + try: + value = float(raw) + except (TypeError, ValueError): + return default + return max(minimum, value) + + +__all__ = [ + "DEFAULT_LOGIN_ALERT_MOBILE", + "GATE_MODE_BLOCKING", + "GATE_MODE_DEFER", + "GATE_MODE_RECOVER", + "GYXX_COOKIE_GATE", + "GYXX_LOGIN_ALERT_OPEN_IDS", + "CookieGate", + "CookieGateRecoveryManager", + "CookieGateStepAction", + "GateVerdict", + "apply_cookie_gates", + "gate_enabled", +]