diff --git a/src/gyxx_flow/accounts.py b/src/gyxx_flow/accounts.py index e57d93a..fcaf3cd 100644 --- a/src/gyxx_flow/accounts.py +++ b/src/gyxx_flow/accounts.py @@ -223,6 +223,8 @@ class AccountKeepaliveManager: self._next_due: dict[str, float] = {} self._locks = LockManager(settings.data_root / "state" / "locks") self._login_runner = login_runner or self._run_configured_login + self._audit_store: Any | None = None + self._audit_seen: dict[str, tuple[str, float]] = {} def tick(self) -> list[dict[str, object]]: """Run due account refreshes once; never starts work in dry-run mode.""" @@ -586,8 +588,63 @@ class AccountKeepaliveManager: if result.get("status") == "success": payload["last_success_at"] = timestamp atomic_write_json(path, payload) + self._audit_keepalive(account, result) return dict(result) + _AUDIT_STATE_BY_PROBE = { + "success": "healthy", + "expired": "expired", + "not_logged_in": "not_logged_in", + "busy": "busy", + "disabled": "disabled", + } + + def _audit_keepalive(self, account: Any, result: dict[str, object]) -> None: + """Record probe outcomes so the real cookie lifetime becomes measurable. + + The resident scheduler probes every ``keepalive_interval_seconds`` + (10 minutes by default), which is what turns "logins seem to last about + a week" into a measured distribution. Repeated identical outcomes + within an hour collapse into one row, so the event table tracks state + changes rather than polling noise. + """ + + try: + from gyxx_flow.ops.audit import snapshot_for_account + + probe = str(result.get("status", "") or "error") + state = self._AUDIT_STATE_BY_PROBE.get(probe, "probe_error") + key = str(account.account_id) + previous = self._audit_seen.get(key) + if previous is not None: + previous_state, previous_at = previous + if previous_state == state and (self._clock() - previous_at) < 3600: + return + store = self._audit() + if store is None or not store.enabled: + return + if store.record_cookie_event( + account_id=key, + state=state, + snapshot=snapshot_for_account(account), + probe_status=probe, + probe_source="keepalive", + detail=str(result.get("error_type", "") or ""), + ): + self._audit_seen[key] = (state, self._clock()) + except Exception: # noqa: BLE001 - telemetry must never break keepalive + return + + def _audit(self) -> Any | None: + if self._audit_store is None: + try: + from gyxx_flow.ops.audit import AuditStore + + self._audit_store = AuditStore(self.settings.data_root) + except Exception: # noqa: BLE001 + return None + return self._audit_store + def _vault_cookie_valid(account: Any, store: BrowserCookieStore) -> bool: if not account.cookie_file.exists(): diff --git a/src/gyxx_flow/cli.py b/src/gyxx_flow/cli.py index 4833b98..5f7d50e 100644 --- a/src/gyxx_flow/cli.py +++ b/src/gyxx_flow/cli.py @@ -229,6 +229,36 @@ def build_parser() -> argparse.ArgumentParser: status_parser = acceptance_commands.add_parser("status") status_parser.add_argument("--json", action="store_true") status_parser.add_argument("--output", type=Path) + + audit_parser = commands.add_parser( + "audit", help="query the PostgreSQL operational audit tables" + ) + audit_commands = audit_parser.add_subparsers(dest="audit_command", required=True) + + audit_runs = audit_commands.add_parser("runs", help="recent workflow runs") + audit_runs.add_argument("--workflow") + audit_runs.add_argument("--status") + audit_runs.add_argument("--date", dest="business_date") + audit_runs.add_argument("--limit", type=int, default=30) + audit_runs.add_argument("--json", action="store_true") + + audit_cookies = audit_commands.add_parser( + "cookies", help="recent cookie observations" + ) + audit_cookies.add_argument("--account") + audit_cookies.add_argument("--state") + audit_cookies.add_argument("--limit", type=int, default=30) + audit_cookies.add_argument("--json", action="store_true") + + audit_stats = audit_commands.add_parser( + "cookie-stats", help="measured cookie lifetime per account" + ) + audit_stats.add_argument("--account") + audit_stats.add_argument("--json", action="store_true") + + audit_commands.add_parser( + "schema", help="print the DDL that creates the audit tables" + ) return parser @@ -304,6 +334,8 @@ def main( and arguments.acceptance_command == "status" ): return _acceptance_status_command(arguments, resolved_settings, output) + if arguments.command == "audit": + return _audit_command(arguments, resolved_settings, output) if arguments.command == "scripts" and arguments.scripts_command == "list": return _list_scripts(arguments, output) if arguments.command == "scripts" and arguments.scripts_command == "run": @@ -549,6 +581,152 @@ def _acceptance_status_command( return EXIT_SUCCESS if report.is_complete else EXIT_ACCEPTANCE_INCOMPLETE +def _format_seconds(value: object) -> str: + """Human-readable duration for the audit tables (None -> '-').""" + + if value is None: + return "-" + try: + seconds = int(value) + except (TypeError, ValueError): + return "-" + if seconds >= 86400: + return f"{seconds / 86400:.1f}d" + if seconds >= 3600: + return f"{seconds / 3600:.1f}h" + if seconds >= 60: + return f"{seconds / 60:.0f}m" + return f"{seconds}s" + + +def _format_timestamp(value: object) -> str: + if value is None: + return "-" + text = value.isoformat() if hasattr(value, "isoformat") else str(value) + return text.replace("T", " ")[:19] + + +def _audit_command( + arguments: argparse.Namespace, settings: Settings, output: TextIO +) -> int: + from gyxx_flow.ops.audit import SCHEMA_SQL, AuditStore + + command = arguments.audit_command + if command == "schema": + output.write(SCHEMA_SQL.strip() + "\n") + return EXIT_SUCCESS + + store = AuditStore(settings.data_root) + if not store.enabled: + output.write("审计已关闭（GYXX_OPS_AUDIT=off）\n") + return EXIT_CONFIGURATION + + if command == "runs": + rows = store.query_runs( + workflow_id=arguments.workflow, + status=arguments.status, + business_date=arguments.business_date, + limit=arguments.limit, + ) + elif command == "cookies": + rows = store.query_cookie_events( + account_id=arguments.account, + state=arguments.state, + limit=arguments.limit, + ) + elif command == "cookie-stats": + rows = ( + store.query_cookie_lifetimes(account_id=arguments.account) + if arguments.account + else store.query_cookie_stats() + ) + else: # pragma: no cover - argparse 已限制取值 + raise CliConfigurationError(f"unsupported audit command: {command}") + + if store.last_error and not rows: + output.write(f"审计库不可用：{store.last_error}\n") + return EXIT_CONFIGURATION + + if getattr(arguments, "json", False): + _write_json(output, {"rows": rows}) + return EXIT_SUCCESS + + if not rows: + output.write("暂无审计记录\n") + return EXIT_SUCCESS + + if command == "runs": + output.write( + f"{'started':19s} {'workflow':40s} {'date':10s} {'mode':7s} " + f"{'status':8s} {'dur':>6s} steps\n" + ) + for row in rows: + steps = ( + f"{row.get('step_success', 0)}/{row.get('step_total', 0)}" + f" failed={row.get('step_failed', 0)}" + ) + output.write( + f"{_format_timestamp(row.get('started_at')):19s} " + f"{str(row.get('workflow_id', ''))[:40]:40s} " + f"{_format_timestamp(row.get('business_date'))[:10]:10s} " + f"{str(row.get('mode', '')):7s} " + f"{str(row.get('status', '')):8s} " + f"{_format_seconds(row.get('duration_seconds')):>6s} {steps}\n" + ) + if row.get("error"): + output.write(f"{'':21s}error: {str(row['error'])[:160]}\n") + return EXIT_SUCCESS + + if command == "cookies": + output.write( + f"{'observed':19s} {'account':24s} {'state':14s} {'probe':14s} " + f"{'fingerprint':12s} expires_at\n" + ) + for row in rows: + output.write( + f"{_format_timestamp(row.get('observed_at')):19s} " + f"{str(row.get('account_id', ''))[:24]:24s} " + f"{str(row.get('state', '')):14s} " + f"{str(row.get('probe_status', '')):14s} " + f"{str(row.get('fingerprint', ''))[:12]:12s} " + f"{_format_timestamp(row.get('declared_expires_at'))}\n" + ) + return EXIT_SUCCESS + + # cookie-stats / 单账号 lifetime + if arguments.account: + output.write( + f"{'fingerprint':12s} {'first_seen':19s} {'last_healthy':19s} " + f"{'failed_at':19s} {'lifetime':>9s} {'declared':19s}\n" + ) + for row in rows: + output.write( + f"{str(row.get('fingerprint', ''))[:12]:12s} " + f"{_format_timestamp(row.get('first_seen_at')):19s} " + f"{_format_timestamp(row.get('last_healthy_at')):19s} " + f"{_format_timestamp(row.get('first_failed_at')):19s} " + f"{_format_seconds(row.get('lifetime_seconds')):>9s} " + f"{_format_timestamp(row.get('declared_expires_at')):19s}\n" + ) + return EXIT_SUCCESS + + output.write( + f"{'account':24s} {'sessions':>8s} {'expired':>7s} {'avg':>7s} " + f"{'median':>7s} {'min':>7s} {'max':>7s} last_failed\n" + ) + for row in rows: + output.write( + f"{str(row.get('account_id', ''))[:24]:24s} " + f"{row.get('sessions', 0):>8d} {row.get('expired_sessions', 0):>7d} " + f"{_format_seconds(row.get('avg_lifetime_seconds')):>7s} " + f"{_format_seconds(row.get('median_lifetime_seconds')):>7s} " + f"{_format_seconds(row.get('min_lifetime_seconds')):>7s} " + f"{_format_seconds(row.get('max_lifetime_seconds')):>7s} " + f"{_format_timestamp(row.get('last_failed_at'))}\n" + ) + return EXIT_SUCCESS + + def _list_workflows( arguments: argparse.Namespace, registry: WorkflowRegistry, diff --git a/src/gyxx_flow/cookie_gate.py b/src/gyxx_flow/cookie_gate.py index 6cb4b19..a40a8c1 100644 --- a/src/gyxx_flow/cookie_gate.py +++ b/src/gyxx_flow/cookie_gate.py @@ -147,6 +147,7 @@ class CookieGate: self._now = now or (lambda: datetime.now(timezone.utc)) self._sender = sender or send_lark_bot_message self._workflow_catalog: WorkflowCatalog | None = None + self._audit_store: Any | None = None # ------------------------------------------------------------------ # Account resolution @@ -236,7 +237,7 @@ class CookieGate: 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") + self._write_state(account, STATE_HEALTHY, probe="fresh-keepalive", context=context) return GateVerdict( account_id=account_id, mode=mode, @@ -269,6 +270,7 @@ class CookieGate: STATE_HEALTHY, probe="success", recovered=recovered, + context=context, ) message = ( f"账号 {account_id} 登录态有效（自动账密重登成功）" @@ -367,6 +369,20 @@ class CookieGate: state.get("pending_reruns"), context, _now_iso(self._now) ) self._save_state(account, state) + try: + keepalive_status = _read_keepalive_status(account) or {} + except Exception: # noqa: BLE001 - state file is advisory + keepalive_status = {} + self._audit_cookie( + account, + "not_logged_in" + if str(keepalive_status.get("status", "")) == "not_logged_in" + else "expired", + probe_status=str(keepalive_status.get("status", "expired")), + probe_source="gate", + detail=detail, + context=context, + ) message = _expiry_message(account_id, mode, context, detail) if notified: message += " 已通过飞书通知登录负责人。" @@ -387,7 +403,17 @@ class CookieGate: *, probe: str, recovered: bool = False, + context: RunContext | None = None, ) -> None: + if state_name == STATE_HEALTHY: + # 健康心跳要落在每一次判定上，才能算准「最后一次可用」的时间。 + self._audit_cookie( + account, + "healthy", + probe_status=probe, + probe_source="gate", + context=context, + ) state = self._read_state(account) if ( state_name == STATE_HEALTHY @@ -431,6 +457,64 @@ class CookieGate: except OSError: # pragma: no cover - state file is advisory only pass + # ------------------------------------------------------------------ + # Cookie telemetry (best-effort; never blocks or fails the gate) + # ------------------------------------------------------------------ + + def _audit(self) -> Any | None: + if self._audit_store is None: + try: + from gyxx_flow.ops.audit import AuditStore + + self._audit_store = AuditStore( + self.settings.data_root, + environment=self._environment, + ) + except Exception: # noqa: BLE001 - telemetry is optional + return None + return self._audit_store + + def _audit_cookie( + self, + account: Any, + state: str, + *, + probe_status: str, + probe_source: str, + detail: str = "", + context: RunContext | None = None, + ) -> None: + """Append one cookie observation. + + Stores a value-free fingerprint plus the earliest declared expiry of the + required cookies, so failure patterns (how long a login really lasts, + and whether it dies before or after the declared expiry) can be derived + from history instead of guessed. Errors are swallowed: gating must + never depend on the audit database. + """ + + try: + store = self._audit() + if store is None or not store.enabled: + return + store.record_cookie_event( + account_id=str(getattr(account, "account_id", "") or ""), + state=state, + snapshot=self._cookie_snapshot(account), + probe_status=probe_status, + probe_source=probe_source, + detail=detail, + workflow_id=str(getattr(context, "workflow_id", "") or ""), + run_id=str(getattr(context, "run_id", "") or ""), + ) + except Exception: # noqa: BLE001 - telemetry is advisory only + return + + def _cookie_snapshot(self, account: Any) -> Any | None: + from gyxx_flow.ops.audit import snapshot_for_account + + return snapshot_for_account(account) + def _within_notify_interval(self, state: Mapping[str, object]) -> bool: raw = state.get("last_notified_at") if not isinstance(raw, str): diff --git a/src/gyxx_flow/ops/AUDIT.md b/src/gyxx_flow/ops/AUDIT.md new file mode 100644 index 0000000..20875e6 --- /dev/null +++ b/src/gyxx_flow/ops/AUDIT.md @@ -0,0 +1,117 @@ +# 运维审计库（ops_* 表） + +工作流执行历史与登录 cookie 遥测的 PostgreSQL 落库。本地 `var/runs/**/run.json` +仍是权威源，这里做的是**跨机器可查、可长期留存的关系型镜像**。 + +## 为什么需要 + +- 本地 journal 只在部署机磁盘上：换机、扩容、清理后历史就断了。 +- `var/logs` 是文本，做不了「近 30 天哪些工作流失败最多」这类查询。 +- cookie 状态原先只存在 `keepalive_status.json`（只有最新值），无法回答 + 「登录态到底能活几天」。 + +## 表与视图 + +| 对象 | 粒度 | 用途 | +|---|---|---| +| `ops_workflow_runs` | 一次运行一行 | 状态、模式、耗时、步骤计数、错误、journal 路径、主机 | +| `ops_workflow_run_steps` | 一次尝试一行 | 含重试；`exit_code`、错误、耗时 | +| `ops_account_cookie_events` | 一次观测一行 | 事件流；只存指纹，不存 cookie 明文 | +| `ops_account_cookie_lifetimes`（视图） | 一份 cookie 一行 | 从首次出现到失效的寿命 | +| `ops_account_cookie_stats`（视图） | 一个账号一行 | 平均 / 中位 / 最短 / 最长寿命 | + +## 自动建表 + +首次写入时执行 `SCHEMA_SQL`（幂等 DDL），无需手工迁移。也可以用： + +```bash +uv run gyxx audit schema > 001_ops_audit.sql # 导出 DDL 交 DBA 预建 +psql "$GYXX_POSTGRES_DSN" -f 001_ops_audit.sql +``` + +## 查询 + +```bash +uv run gyxx audit runs --workflow content.metrics.self_operated.daily --limit 20 +uv run gyxx audit runs --status failed --limit 50 +uv run gyxx audit cookies --account douyin-shop --limit 50 +uv run gyxx audit cookie-stats # 各账号失效规律 +uv run gyxx audit cookie-stats --account douyin-shop # 单账号每次的寿命明细 +``` + +都支持 `--json`，方便接到面板或告警脚本。 + +直接 SQL 也一样： + +```sql +-- 最近 7 天失败最多的 10 个工作流 +SELECT workflow_id, COUNT(*) AS failures +FROM ops_workflow_runs +WHERE status = 'failed' AND started_at > NOW() - INTERVAL '7 days' +GROUP BY workflow_id ORDER BY failures DESC LIMIT 10; + +-- 每次失败卡在哪个步骤 +SELECT r.workflow_id, s.step_id, COUNT(*) AS n +FROM ops_workflow_run_steps s +JOIN ops_workflow_runs r USING (run_id) +WHERE s.status = 'failed' AND r.started_at > NOW() - INTERVAL '7 days' +GROUP BY 1, 2 ORDER BY n DESC; + +-- 哪个账号的登录态最短命（预警阈值参考） +SELECT account_id, sessions, expired_sessions, + avg_lifetime_seconds / 86400.0 AS avg_days, + min_lifetime_seconds / 86400.0 AS min_days, + last_failed_at +FROM ops_account_cookie_stats ORDER BY min_lifetime_seconds NULLS LAST; +``` + +## 怎么读「失效规律」 + +一次 cookie 代由 `fingerprint` 标识（需求域名/名称/值哈希排序后取摘要，与文件顺序无关）： + +- `first_healthy_at` → `first_failed_at` = `lifetime_seconds`：**实际可用时长** +- `healthy_to_failure_seconds`：最后一次探活成功到发现失效，反映**探测精度** +- `declared_expires_at`：cookie 自带的最早过期时间 + +把 `lifetime_seconds` 和 `declared_expires_at` 对比，能区分三种情况： + +1. 实际寿命接近声明寿命 → 平台按声明失效，可按 `declared_expires_at` 提前预警； +2. 实际远短于声明 → 服务端提前吊销（常见于多端登录、风控）； +3. `declared_expires_at` 为空（session cookie）→ 只能靠观测，`min_lifetime_seconds` + 就是安全刷新周期的上界。 + +`fingerprint` 变化表示换了一份 cookie（重新登录），不算失效，也不会把两段寿命混在一起。 + +## 采集点 + +| 来源 | 触发 | `probe_source` | +|---|---|---| +| `CookieGate.check` | 每次工作流运行前的门禁判定 | `gate` | +| `AccountKeepaliveManager` | 常驻调度器每 `keepalive_interval_seconds`（默认 600s） | `keepalive` | +| `RunIndex.index_journal` | 每次运行开始/结束/异常 | — | + +keepalive 是主要的精度来源：默认 10 分钟一次，同类状态一小时内只落一行， +状态翻转立即落行，所以「首次失效时间」误差不超过一个探活周期。 + +## 容错 + +审计**永不阻断业务**： + +- 数据库不可达 / 缺 DSN / 建表失败 → 写入返回 `False`，调用方继续执行； +- 连续失败 3 次后熔断冷却（默认 300s），避免每个工作流都卡在连接超时； +- 失败记录写到 `/logs/ops-audit/audit-failures.log`。 + +## 开关 + +| 变量 | 默认 | 说明 | +|---|---|---| +| `GYXX_OPS_AUDIT` | 开 | 设 `off` 关闭全部审计写入 | +| `GYXX_OPS_AUDIT_HOST` | 主机名 | 多机部署时写入手工标识 | +| `GYXX_OPS_AUDIT_FAILURE_COOLDOWN` | 300 | 熔断冷却秒数 | + +## 不存什么 + +- **cookie 值**：只存哈希指纹；排查具体 cookie 请走 vault 文件。 +- **日志正文**：`ops_workflow_runs` 只存 `journal_path` 指向本地文件； + 完整堆栈仍在 `var/logs/console-restarts/`。 +- **凭据**：DSN 只从 `GYXX_POSTGRES_DSN`（或 `PG_*`）读，不落表。 diff --git a/src/gyxx_flow/ops/__init__.py b/src/gyxx_flow/ops/__init__.py index 38c9fc1..210243d 100644 --- a/src/gyxx_flow/ops/__init__.py +++ b/src/gyxx_flow/ops/__init__.py @@ -1,5 +1,6 @@ -"""Operational run indexing and durable outbox services.""" +"""Operational run indexing, audit telemetry and durable outbox services.""" +from .audit import AuditStore, CookieSnapshot, summarize_cookies from .effects import ( EffectAlreadyApplied, EffectClaim, @@ -10,6 +11,8 @@ from .index import RunIndex, RunRecord from .outbox import IdempotencyConflict, Outbox, OutboxMessage __all__ = [ + "AuditStore", + "CookieSnapshot", "IdempotencyConflict", "EffectAlreadyApplied", "EffectClaim", @@ -19,4 +22,5 @@ __all__ = [ "OutboxMessage", "RunIndex", "RunRecord", + "summarize_cookies", ] diff --git a/src/gyxx_flow/ops/audit.py b/src/gyxx_flow/ops/audit.py new file mode 100644 index 0000000..359c9d3 --- /dev/null +++ b/src/gyxx_flow/ops/audit.py @@ -0,0 +1,901 @@ +"""PostgreSQL-backed operational audit: workflow run history and cookie telemetry. + +设计原则 +-------- +1. **可观测性不拖累业务**：审计写入永远不向调用方抛异常。数据库不可达时 + 只记录一条 failure 日志并熔断冷却，工作流照常执行。 +2. **本地 journal 仍是权威源**：本模块是镜像/汇总，供跨机器查询与长期留存； + `var/runs/**/run.json` 不会被替代。 +3. **不落明文凭据**：cookie 只存指纹（域名+名称+值哈希），值本身永不入库。 + +环境开关 +-------- +``GYXX_OPS_AUDIT=off`` 关闭审计（默认开启） +``GYXX_OPS_AUDIT_HOST`` 覆盖主机标识（默认 socket.gethostname()） +``GYXX_OPS_AUDIT_FAILURE_COOLDOWN`` 熔断冷却秒数（默认 300） +``GYXX_POSTGRES_DSN`` 与动态配置共用同一 DSN +""" + +from __future__ import annotations + +import hashlib +import json +import os +import socket +from contextlib import contextmanager +from dataclasses import dataclass +from datetime import date, datetime, timezone +from pathlib import Path +from typing import Any, Iterator, Mapping, Sequence + +SCHEMA_VERSION = 1 + +OPS_AUDIT_ENV = "GYXX_OPS_AUDIT" +OPS_AUDIT_HOST_ENV = "GYXX_OPS_AUDIT_HOST" +OPS_AUDIT_COOLDOWN_ENV = "GYXX_OPS_AUDIT_FAILURE_COOLDOWN" + +_DEFAULT_COOLDOWN_SECONDS = 300 +_MAX_CONSECUTIVE_FAILURES = 3 +_MAX_TEXT_CHARS = 4000 +_SCHEMA_LOCK_KEY = "gyxx:ops-audit:migration" + +HEALTHY_STATES = frozenset({"healthy"}) +FAILED_STATES = frozenset({"expired", "not_logged_in"}) +_COOKIE_STATES = frozenset( + {"healthy", "expired", "not_logged_in", "probe_error", "busy", "disabled"} +) + +SCHEMA_SQL = """ +CREATE TABLE IF NOT EXISTS ops_workflow_runs ( + run_id TEXT PRIMARY KEY, + workflow_id TEXT NOT NULL, + business_date DATE NOT NULL, + mode TEXT NOT NULL DEFAULT 'unknown', + shadow BOOLEAN NOT NULL DEFAULT FALSE, + status TEXT NOT NULL, + started_at TIMESTAMPTZ, + ended_at TIMESTAMPTZ, + duration_seconds INTEGER, + error TEXT, + step_total INTEGER NOT NULL DEFAULT 0, + step_success INTEGER NOT NULL DEFAULT 0, + step_failed INTEGER NOT NULL DEFAULT 0, + step_skipped INTEGER NOT NULL DEFAULT 0, + step_running INTEGER NOT NULL DEFAULT 0, + journal_path TEXT NOT NULL DEFAULT '', + host TEXT NOT NULL DEFAULT '', + indexed_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS ops_workflow_runs_workflow_idx + ON ops_workflow_runs (workflow_id, started_at DESC); +CREATE INDEX IF NOT EXISTS ops_workflow_runs_status_idx + ON ops_workflow_runs (status, started_at DESC); +CREATE INDEX IF NOT EXISTS ops_workflow_runs_business_date_idx + ON ops_workflow_runs (business_date, workflow_id); + +CREATE TABLE IF NOT EXISTS ops_workflow_run_steps ( + run_id TEXT NOT NULL + REFERENCES ops_workflow_runs (run_id) ON DELETE CASCADE, + step_id TEXT NOT NULL, + attempt INTEGER NOT NULL DEFAULT 1, + status TEXT NOT NULL DEFAULT '', + exit_code INTEGER, + started_at TIMESTAMPTZ, + ended_at TIMESTAMPTZ, + duration_seconds INTEGER, + error TEXT, + PRIMARY KEY (run_id, step_id, attempt) +); + +CREATE INDEX IF NOT EXISTS ops_workflow_run_steps_step_idx + ON ops_workflow_run_steps (step_id, status); + +CREATE TABLE IF NOT EXISTS ops_account_cookie_events ( + id BIGSERIAL PRIMARY KEY, + account_id TEXT NOT NULL, + observed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + state TEXT NOT NULL, + probe_status TEXT NOT NULL DEFAULT '', + probe_source TEXT NOT NULL DEFAULT '', + cookie_count INTEGER, + required_count INTEGER, + fingerprint TEXT NOT NULL DEFAULT '', + declared_expires_at TIMESTAMPTZ, + detail TEXT NOT NULL DEFAULT '', + workflow_id TEXT NOT NULL DEFAULT '', + run_id TEXT NOT NULL DEFAULT '', + host TEXT NOT NULL DEFAULT '' +); + +CREATE INDEX IF NOT EXISTS ops_account_cookie_events_account_idx + ON ops_account_cookie_events (account_id, observed_at DESC); +CREATE INDEX IF NOT EXISTS ops_account_cookie_events_state_idx + ON ops_account_cookie_events (state, observed_at DESC); +CREATE INDEX IF NOT EXISTS ops_account_cookie_events_fingerprint_idx + ON ops_account_cookie_events (account_id, fingerprint); + +-- 一段「cookie 代」的生命周期：同一 fingerprint 从首次出现到失效。 +-- first_failed_at 只取最后一次 healthy 之后的失败，避免恢复后再次失败时口径前移。 +CREATE OR REPLACE VIEW ops_account_cookie_lifetimes AS +WITH per_fingerprint AS ( + SELECT + account_id, + fingerprint, + MIN(observed_at) AS first_seen_at, + MIN(observed_at) FILTER (WHERE state = 'healthy') AS first_healthy_at, + MAX(observed_at) FILTER (WHERE state = 'healthy') AS last_healthy_at, + MAX(declared_expires_at) AS declared_expires_at + FROM ops_account_cookie_events + WHERE fingerprint <> '' + GROUP BY account_id, fingerprint +) +SELECT + p.account_id, + p.fingerprint, + p.first_seen_at, + p.first_healthy_at, + p.last_healthy_at, + p.declared_expires_at, + f.first_failed_at, + (f.first_failed_at IS NOT NULL) AS expired, + CASE + WHEN p.first_healthy_at IS NOT NULL AND p.last_healthy_at IS NOT NULL + THEN EXTRACT(EPOCH FROM (p.last_healthy_at - p.first_healthy_at))::BIGINT + END AS healthy_span_seconds, + CASE + WHEN p.first_healthy_at IS NOT NULL AND f.first_failed_at IS NOT NULL + THEN EXTRACT(EPOCH FROM (f.first_failed_at - p.first_healthy_at))::BIGINT + END AS lifetime_seconds, + CASE + WHEN p.last_healthy_at IS NOT NULL AND f.first_failed_at IS NOT NULL + THEN EXTRACT(EPOCH FROM (f.first_failed_at - p.last_healthy_at))::BIGINT + END AS healthy_to_failure_seconds +FROM per_fingerprint p +LEFT JOIN LATERAL ( + SELECT MIN(e.observed_at) AS first_failed_at + FROM ops_account_cookie_events e + WHERE e.account_id = p.account_id + AND e.fingerprint = p.fingerprint + AND e.state IN ('expired', 'not_logged_in') + AND (p.last_healthy_at IS NULL OR e.observed_at > p.last_healthy_at) +) f ON TRUE; + +-- 按账号汇总的失效规律：平均/中位/最短/最长寿命，直接支撑「多久失效一次」。 +CREATE OR REPLACE VIEW ops_account_cookie_stats AS +SELECT + account_id, + COUNT(*) AS sessions, + COUNT(*) FILTER (WHERE expired) AS expired_sessions, + AVG(lifetime_seconds)::BIGINT AS avg_lifetime_seconds, + PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY lifetime_seconds)::BIGINT + AS median_lifetime_seconds, + MIN(lifetime_seconds) AS min_lifetime_seconds, + MAX(lifetime_seconds) AS max_lifetime_seconds, + MAX(first_failed_at) AS last_failed_at, + MAX(last_healthy_at) AS last_healthy_at +FROM ops_account_cookie_lifetimes +WHERE lifetime_seconds IS NOT NULL +GROUP BY account_id; + +CREATE TABLE IF NOT EXISTS ops_audit_schema ( + schema_version INTEGER PRIMARY KEY, + applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); +""" + + +# ---------------------------------------------------------------------- +# 纯函数：cookie 快照 +# ---------------------------------------------------------------------- + + +@dataclass(frozen=True, slots=True) +class CookieSnapshot: + """一次静态检查得到的 cookie 画像，不含任何明文值。""" + + cookie_count: int + required_count: int + fingerprint: str + declared_expires_at: datetime | None + + def to_dict(self) -> dict[str, Any]: + return { + "cookie_count": self.cookie_count, + "required_count": self.required_count, + "fingerprint": self.fingerprint, + "declared_expires_at": ( + self.declared_expires_at.isoformat() + if self.declared_expires_at is not None + else None + ), + } + + +def _domain_matches(domain: str, required_domains: Sequence[str]) -> bool: + normalized = str(domain or "").lstrip(".").casefold() + if not normalized: + return False + return any( + normalized == required.casefold() or normalized.endswith("." + required.casefold()) + for required in required_domains + ) + + +def _expires_at(cookie: Mapping[str, Any]) -> datetime | None: + raw = cookie.get("expires") + if raw is None or isinstance(raw, bool): + return None + try: + value = float(raw) + except (TypeError, ValueError): + return None + # Playwright 用 -1 表示 session cookie；0 与负数都视为「不声明」。 + if value <= 0: + return None + try: + return datetime.fromtimestamp(value, tz=timezone.utc) + except (OverflowError, OSError, ValueError): + return None + + +def summarize_cookies( + cookies: Sequence[Mapping[str, Any]], + *, + required_domains: Sequence[str] = (), + required_names: Sequence[str] = (), +) -> CookieSnapshot: + """Build a value-free snapshot: counts, fingerprint and earliest declared expiry. + + 指纹只看「必需 cookie」的域名/名称/值哈希，因此登录态真正换了一份 cookie + 时指纹会变，而无关 cookie 的增删不会造成抖动。 + """ + + required_domains = tuple(item for item in required_domains if item) + required_names = tuple(item for item in required_names if item) + components: list[str] = [] + required_count = 0 + earliest: datetime | None = None + + for cookie in cookies: + if not isinstance(cookie, Mapping): + continue + name = str(cookie.get("name", "")) + domain = str(cookie.get("domain", "")) + is_required = bool(required_domains) and _domain_matches(domain, required_domains) + if required_names: + is_required = is_required and name in required_names + elif not required_domains: + is_required = True + if not is_required: + continue + required_count += 1 + value = str(cookie.get("value", "")) + value_hash = hashlib.sha256(value.encode("utf-8", errors="replace")).hexdigest()[:16] + components.append(f"{domain.lstrip('.')}|{name}|{value_hash}") + expiry = _expires_at(cookie) + if expiry is not None and (earliest is None or expiry < earliest): + earliest = expiry + + # 浏览器写入 cookie 的顺序并不稳定；排序后哈希才能保证同一份登录态得到同一指纹。 + digest = hashlib.sha256() + for component in sorted(components): + digest.update(f"{component}\n".encode("utf-8")) + + return CookieSnapshot( + cookie_count=len(cookies), + required_count=required_count, + fingerprint=digest.hexdigest()[:32] if required_count else "", + declared_expires_at=earliest, + ) + + +def snapshot_for_account(account: Any) -> CookieSnapshot | None: + """Read an account's cookie vault and summarise it, or ``None`` when unreadable. + + Shared by the gate and the keepalive manager so both record the exact same + fingerprint for the same vault contents. + """ + + cookie_file = getattr(account, "cookie_file", None) + if cookie_file is None: + return None + try: + from gyxx_flow.adapters.browser import BrowserCookieStore + + store = BrowserCookieStore( + cookie_file, + getattr(account, "storage_state_file", cookie_file), + ) + cookies = store.load_cookies() + except Exception: # noqa: BLE001 - a missing/corrupt vault is itself a signal + return None + return summarize_cookies( + cookies, + required_domains=tuple(getattr(account, "required_cookie_domains", ()) or ()), + required_names=tuple(getattr(account, "required_cookie_names", ()) or ()), + ) + + +# ---------------------------------------------------------------------- +# 时间与文本工具 +# ---------------------------------------------------------------------- + + +def _parse_datetime(value: Any) -> datetime | None: + if isinstance(value, datetime): + return value if value.tzinfo else value.replace(tzinfo=timezone.utc) + if not isinstance(value, str) or not value.strip(): + return None + text = value.strip().replace("Z", "+00:00") + try: + parsed = datetime.fromisoformat(text) + except ValueError: + return None + return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) + + +def _parse_date(value: Any) -> date | None: + if isinstance(value, date) and not isinstance(value, datetime): + return value + if not isinstance(value, str): + return None + try: + return date.fromisoformat(value) + except ValueError: + return None + + +def _seconds_between(start: datetime | None, end: datetime | None) -> int | None: + if start is None or end is None: + return None + delta = (end - start).total_seconds() + return max(0, int(delta)) + + +def _clip(value: Any, limit: int = _MAX_TEXT_CHARS) -> str: + if value is None: + return "" + text = value if isinstance(value, str) else str(value) + return text[:limit] + + +def split_step_key(key: str) -> tuple[str, int]: + """``sync.attempt-2`` -> ``("sync", 2)``; 无法解析时原样当作 step_id。""" + + if ".attempt-" in key: + name, _, raw = key.rpartition(".attempt-") + try: + attempt = int(raw) + except ValueError: + return key, 1 + if name and attempt > 0: + return name, attempt + return key, 1 + + +# ---------------------------------------------------------------------- +# 审计存储 +# ---------------------------------------------------------------------- + + +class AuditStore: + """Idempotent writer for the ``ops_*`` tables. + + 所有公开写入方法都返回 bool，并且**永不抛出**异常；数据库问题只会导致本次 + 审计丢失，不会影响工作流执行。 + """ + + def __init__( + self, + data_root: Path | str, + *, + environment: Mapping[str, str] | None = None, + connect: Any | None = None, + clock: Any | None = None, + ) -> None: + self._data_root = Path(data_root).expanduser().resolve() + self._environment = dict(os.environ if environment is None else environment) + self._connect_impl = connect + self._clock = clock or (lambda: datetime.now(timezone.utc)) + self._schema_ready = False + self._consecutive_failures = 0 + self._cooldown_until = 0.0 + self._last_error = "" + self._failure_log = self._data_root / "logs" / "ops-audit" / "audit-failures.log" + + # -- 状态 --------------------------------------------------------- + + @property + def enabled(self) -> bool: + raw = str(self._environment.get(OPS_AUDIT_ENV, "")).strip().lower() + return raw not in {"0", "off", "false", "no", "disabled"} + + @property + def last_error(self) -> str: + return self._last_error + + def _host(self) -> str: + override = str(self._environment.get(OPS_AUDIT_HOST_ENV, "")).strip() + if override: + return override + try: + return socket.gethostname()[:128] + except OSError: # pragma: no cover - 极端环境 + return "" + + def _cooldown_seconds(self) -> int: + raw = str(self._environment.get(OPS_AUDIT_COOLDOWN_ENV, "")).strip() + try: + return max(0, int(raw)) + except ValueError: + return _DEFAULT_COOLDOWN_SECONDS + + def _in_cooldown(self) -> bool: + import time + + return time.monotonic() < self._cooldown_until + + # -- 连接 --------------------------------------------------------- + + @contextmanager + def _connection(self) -> Iterator[Any]: + if self._connect_impl is not None: + conn = self._connect_impl() + try: + yield conn + finally: + close = getattr(conn, "close", None) + if callable(close): + close() + return + + import psycopg # 延迟导入：未安装 psycopg 时整个模块仍可导入 + from psycopg.rows import dict_row + + dsn, kwargs = database_kwargs(self._environment) + if dsn: + conn = psycopg.connect(dsn, row_factory=dict_row) + else: + conn = psycopg.connect(**kwargs, row_factory=dict_row) + try: + yield conn + finally: + conn.close() + + def _ensure_schema(self, conn: Any) -> None: + """Create the audit objects once, serialising the first DDL pass. + + Several workflows can start at the same moment; running CREATE/COMMENT + concurrently lets sessions take DDL locks in different orders and can + deadlock. The advisory lock is only taken when the schema is missing, + so steady-state processes skip the round trip entirely. + """ + + if self._schema_ready: + return + with conn.cursor() as cursor: + cursor.execute( + "SELECT 1 FROM pg_class " + "WHERE relname = 'ops_audit_schema' AND relkind = 'r'" + ) + initialised = cursor.fetchone() is not None + if not initialised: + with conn.cursor() as cursor: + cursor.execute( + "SELECT pg_advisory_xact_lock(hashtext(%s))", + (_SCHEMA_LOCK_KEY,), + ) + cursor.execute(SCHEMA_SQL) + cursor.execute( + "INSERT INTO ops_audit_schema (schema_version) VALUES (%s) " + "ON CONFLICT (schema_version) DO NOTHING", + (SCHEMA_VERSION,), + ) + conn.commit() + self._schema_ready = True + + # -- 失败处理 ----------------------------------------------------- + + def _record_failure(self, operation: str, exc: BaseException) -> None: + import time + + self._consecutive_failures += 1 + self._last_error = f"{type(exc).__name__}: {exc}"[:300] + if self._consecutive_failures >= _MAX_CONSECUTIVE_FAILURES: + self._cooldown_until = time.monotonic() + self._cooldown_seconds() + try: + self._failure_log.parent.mkdir(parents=True, exist_ok=True) + payload = { + "at": self._clock().isoformat(timespec="seconds"), + "operation": operation, + "error": self._last_error, + "consecutive_failures": self._consecutive_failures, + } + with self._failure_log.open("a", encoding="utf-8") as stream: + stream.write(json.dumps(payload, ensure_ascii=False) + "\n") + except OSError: + pass + + def _run_guarded(self, operation: str, action: Any) -> bool: + if not self.enabled or self._in_cooldown(): + return False + try: + with self._connection() as conn: + self._ensure_schema(conn) + action(conn) + self._consecutive_failures = 0 + return True + except Exception as exc: # noqa: BLE001 - 审计绝不影响业务 + self._record_failure(operation, exc) + return False + + # -- 运行记录 ----------------------------------------------------- + + def record_run( + self, + record: Mapping[str, Any], + steps: Mapping[str, Any] | None = None, + ) -> bool: + """Upsert one run and its steps from a journal payload or RunRecord dict.""" + + workflow_id = _clip(record.get("workflow_id"), 200) + run_id = _clip(record.get("run_id"), 200) + business_date = _parse_date(record.get("business_date")) + if not workflow_id or not run_id or business_date is None: + return False + status = _clip(record.get("status"), 32) or "unknown" + mode = _clip(record.get("mode"), 32) or "unknown" + started_at = _parse_datetime(record.get("started_at")) + ended_at = _parse_datetime(record.get("ended_at")) + step_rows = self._step_rows(run_id, steps) + counts = self._step_counts(steps, record) + + def action(conn: Any) -> None: + with conn.cursor() as cursor: + cursor.execute( + """ + INSERT INTO ops_workflow_runs ( + run_id, workflow_id, business_date, mode, shadow, status, + started_at, ended_at, duration_seconds, error, + step_total, step_success, step_failed, step_skipped, + step_running, journal_path, host, indexed_at + ) VALUES ( + %(run_id)s, %(workflow_id)s, %(business_date)s, %(mode)s, + %(shadow)s, %(status)s, %(started_at)s, %(ended_at)s, + %(duration_seconds)s, %(error)s, %(step_total)s, + %(step_success)s, %(step_failed)s, %(step_skipped)s, + %(step_running)s, %(journal_path)s, %(host)s, NOW() + ) + ON CONFLICT (run_id) DO UPDATE SET + status = EXCLUDED.status, + ended_at = EXCLUDED.ended_at, + duration_seconds = EXCLUDED.duration_seconds, + error = EXCLUDED.error, + step_total = EXCLUDED.step_total, + step_success = EXCLUDED.step_success, + step_failed = EXCLUDED.step_failed, + step_skipped = EXCLUDED.step_skipped, + step_running = EXCLUDED.step_running, + journal_path = EXCLUDED.journal_path, + indexed_at = NOW() + """, + { + "run_id": run_id, + "workflow_id": workflow_id, + "business_date": business_date, + "mode": mode, + "shadow": bool(record.get("shadow", False)), + "status": status, + "started_at": started_at, + "ended_at": ended_at, + "duration_seconds": _seconds_between(started_at, ended_at), + "error": _clip(record.get("error")), + "step_total": counts["total"], + "step_success": counts["success"], + "step_failed": counts["failed"], + "step_skipped": counts["skipped"], + "step_running": counts["running"], + "journal_path": _clip(record.get("journal_path"), 1000), + "host": self._host(), + }, + ) + if step_rows: + cursor.executemany( + """ + INSERT INTO ops_workflow_run_steps ( + run_id, step_id, attempt, status, exit_code, + started_at, ended_at, duration_seconds, error + ) VALUES ( + %(run_id)s, %(step_id)s, %(attempt)s, %(status)s, + %(exit_code)s, %(started_at)s, %(ended_at)s, + %(duration_seconds)s, %(error)s + ) + ON CONFLICT (run_id, step_id, attempt) DO UPDATE SET + status = EXCLUDED.status, + exit_code = EXCLUDED.exit_code, + started_at = COALESCE( + ops_workflow_run_steps.started_at, + EXCLUDED.started_at + ), + ended_at = EXCLUDED.ended_at, + duration_seconds = EXCLUDED.duration_seconds, + error = EXCLUDED.error + """, + step_rows, + ) + conn.commit() + + return self._run_guarded("record_run", action) + + @staticmethod + def _step_rows( + run_id: str, steps: Mapping[str, Any] | None + ) -> list[dict[str, Any]]: + if not isinstance(steps, Mapping): + return [] + rows: list[dict[str, Any]] = [] + for key, raw in steps.items(): + if not isinstance(raw, Mapping): + continue + step_id, attempt = split_step_key(str(key)) + attempt_value = raw.get("attempt") + if isinstance(attempt_value, int) and not isinstance(attempt_value, bool): + attempt = max(1, attempt_value) + started_at = _parse_datetime(raw.get("started_at")) + ended_at = _parse_datetime(raw.get("ended_at")) + exit_code = raw.get("exit_code") + rows.append( + { + "run_id": run_id, + "step_id": _clip(step_id, 200), + "attempt": attempt, + "status": _clip(raw.get("status"), 32), + "exit_code": exit_code if isinstance(exit_code, int) else None, + "started_at": started_at, + "ended_at": ended_at, + "duration_seconds": _seconds_between(started_at, ended_at), + "error": _clip(raw.get("error")), + } + ) + return rows + + @staticmethod + def _step_counts( + steps: Mapping[str, Any] | None, record: Mapping[str, Any] + ) -> dict[str, int]: + counts = {"total": 0, "success": 0, "failed": 0, "skipped": 0, "running": 0} + if isinstance(steps, Mapping) and steps: + for raw in steps.values(): + if not isinstance(raw, Mapping): + continue + status = str(raw.get("status", "")) + counts["total"] += 1 + if status in counts: + counts[status] += 1 + return counts + # 回退到 RunRecord 的 step_counts 投影 + projection = record.get("step_counts") + if isinstance(projection, Mapping): + for name in ("success", "failed", "skipped", "running"): + value = projection.get(name) + if isinstance(value, int) and not isinstance(value, bool): + counts[name] = max(0, value) + counts["total"] = counts["success"] + counts["failed"] + counts["skipped"] + counts["running"] + return counts + + # -- cookie 事件 --------------------------------------------------- + + def record_cookie_event( + self, + *, + account_id: str, + state: str, + snapshot: CookieSnapshot | None = None, + probe_status: str = "", + probe_source: str = "", + detail: str = "", + workflow_id: str = "", + run_id: str = "", + observed_at: datetime | None = None, + ) -> bool: + """Append one observation. 状态必须是白名单值，避免脏数据污染统计。""" + + account_id = _clip(account_id, 200) + if not account_id: + return False + state = str(state or "").strip() + if state not in _COOKIE_STATES: + raise ValueError(f"invalid cookie state: {state!r}") + + def action(conn: Any) -> None: + with conn.cursor() as cursor: + cursor.execute( + """ + INSERT INTO ops_account_cookie_events ( + account_id, observed_at, state, probe_status, probe_source, + cookie_count, required_count, fingerprint, declared_expires_at, + detail, workflow_id, run_id, host + ) VALUES ( + %(account_id)s, %(observed_at)s, %(state)s, %(probe_status)s, + %(probe_source)s, %(cookie_count)s, %(required_count)s, + %(fingerprint)s, %(declared_expires_at)s, %(detail)s, + %(workflow_id)s, %(run_id)s, %(host)s + ) + """, + { + "account_id": account_id, + "observed_at": observed_at or self._clock(), + "state": state, + "probe_status": _clip(probe_status, 64), + "probe_source": _clip(probe_source, 64), + "cookie_count": snapshot.cookie_count if snapshot else None, + "required_count": snapshot.required_count if snapshot else None, + "fingerprint": snapshot.fingerprint if snapshot else "", + "declared_expires_at": ( + snapshot.declared_expires_at if snapshot else None + ), + "detail": _clip(detail), + "workflow_id": _clip(workflow_id, 200), + "run_id": _clip(run_id, 200), + "host": self._host(), + }, + ) + conn.commit() + + return self._run_guarded("record_cookie_event", action) + + # -- 查询（供 CLI） ------------------------------------------------ + + def query_runs( + self, + *, + workflow_id: str | None = None, + status: str | None = None, + business_date: str | None = None, + limit: int = 50, + ) -> list[dict[str, Any]]: + conditions: list[str] = [] + params: dict[str, Any] = {"limit": max(1, min(int(limit), 500))} + if workflow_id: + conditions.append("workflow_id = %(workflow_id)s") + params["workflow_id"] = workflow_id + if status: + conditions.append("status = %(status)s") + params["status"] = status + if business_date: + parsed = _parse_date(business_date) + if parsed is not None: + conditions.append("business_date = %(business_date)s") + params["business_date"] = parsed + where = f"WHERE {' AND '.join(conditions)}" if conditions else "" + sql = f""" + SELECT * FROM ops_workflow_runs + {where} + ORDER BY started_at DESC NULLS LAST, run_id DESC + LIMIT %(limit)s + """ + return self._fetch(sql, params) + + def query_steps(self, run_id: str) -> list[dict[str, Any]]: + return self._fetch( + """ + SELECT * FROM ops_workflow_run_steps + WHERE run_id = %(run_id)s + ORDER BY step_id, attempt + """, + {"run_id": run_id}, + ) + + def query_cookie_events( + self, + *, + account_id: str | None = None, + state: str | None = None, + limit: int = 50, + ) -> list[dict[str, Any]]: + conditions: list[str] = [] + params: dict[str, Any] = {"limit": max(1, min(int(limit), 500))} + if account_id: + conditions.append("account_id = %(account_id)s") + params["account_id"] = account_id + if state: + conditions.append("state = %(state)s") + params["state"] = state + where = f"WHERE {' AND '.join(conditions)}" if conditions else "" + sql = f""" + SELECT * FROM ops_account_cookie_events + {where} + ORDER BY observed_at DESC + LIMIT %(limit)s + """ + return self._fetch(sql, params) + + def query_cookie_lifetimes(self, *, account_id: str | None = None) -> list[dict[str, Any]]: + if account_id: + return self._fetch( + """ + SELECT * FROM ops_account_cookie_lifetimes + WHERE account_id = %(account_id)s + ORDER BY first_seen_at DESC + """, + {"account_id": account_id}, + ) + return self._fetch( + "SELECT * FROM ops_account_cookie_lifetimes ORDER BY first_seen_at DESC", + {}, + ) + + def query_cookie_stats(self) -> list[dict[str, Any]]: + return self._fetch( + "SELECT * FROM ops_account_cookie_stats ORDER BY account_id", + {}, + ) + + def _fetch(self, sql: str, params: Mapping[str, Any]) -> list[dict[str, Any]]: + if not self.enabled or self._in_cooldown(): + return [] + try: + with self._connection() as conn: + self._ensure_schema(conn) + with conn.cursor() as cursor: + cursor.execute(sql, params) + rows = cursor.fetchall() + self._consecutive_failures = 0 + return [dict(row) for row in rows] + except Exception as exc: # noqa: BLE001 + self._record_failure("query", exc) + return [] + + +# ---------------------------------------------------------------------- +# 连接参数（与 PostgresWorkflowConfigStore 同口径） +# ---------------------------------------------------------------------- + + +def database_kwargs(environment: Mapping[str, str]) -> tuple[str | None, dict[str, Any]]: + values = _apply_runtime_policy(environment) + dsn = str(values.get("GYXX_POSTGRES_DSN", "")).strip() + if dsn: + return dsn, {} + required = ("PG_HOST", "PG_PORT", "PG_DB", "PG_USER", "PG_PASSWORD") + missing = [name for name in required if not str(values.get(name, "")).strip()] + if missing: + raise RuntimeError("审计数据库配置缺失: " + ", ".join(missing)) + try: + port = int(values["PG_PORT"]) + except (TypeError, ValueError) as exc: + raise RuntimeError("PG_PORT 必须是整数") from exc + return None, { + "host": str(values["PG_HOST"]).strip(), + "port": port, + "dbname": str(values["PG_DB"]).strip(), + "user": str(values["PG_USER"]).strip(), + "password": str(values["PG_PASSWORD"]), + } + + +def _apply_runtime_policy(environment: Mapping[str, str]) -> Mapping[str, str]: + """复用 RuntimeServicePolicy，保证与控制台/工作流看到同一个数据库地址。""" + + try: + from gyxx_flow.adapters import RuntimeServicePolicy + except Exception: # pragma: no cover - 循环导入兜底 + return environment + try: + return RuntimeServicePolicy().apply(dict(environment)) + except Exception: # pragma: no cover + return environment + + +def default_store(data_root: Path | str | None = None) -> AuditStore: + """Best-effort store resolved from the standard data root.""" + + if data_root is None: + try: + from gyxx_flow.core.config import Settings + + data_root = Settings.from_env().data_root + except Exception: # pragma: no cover + data_root = Path.cwd() / "var" + return AuditStore(data_root) diff --git a/src/gyxx_flow/ops/index.py b/src/gyxx_flow/ops/index.py index 53d4d67..0c14301 100644 --- a/src/gyxx_flow/ops/index.py +++ b/src/gyxx_flow/ops/index.py @@ -86,6 +86,7 @@ class RunIndex: resource = f"ops:run-index:{_digest(record.run_id)}" with self._locks.acquire(resource, owner=_owner(), timeout_seconds=30): atomic_write_json(destination, record.to_dict()) + _mirror_run_to_audit(self._data_root, record, raw) return record def get(self, run_id: str) -> RunRecord | None: @@ -255,3 +256,42 @@ def _digest(value: str) -> str: import hashlib return hashlib.sha256(value.encode("utf-8")).hexdigest()[:32] + + +_AUDIT_STORES: dict[str, Any] = {} + + +def _audit_store_for(data_root: Path) -> Any: + """Reuse one store per data root so schema bootstrap and the circuit breaker + survive the short-lived ``RunIndex`` instances created per run.""" + + key = str(data_root) + store = _AUDIT_STORES.get(key) + if store is None: + from gyxx_flow.ops.audit import AuditStore + + store = AuditStore(data_root) + _AUDIT_STORES[key] = store + return store + + +def _mirror_run_to_audit( + data_root: Path, + record: RunRecord, + journal_payload: Any, +) -> None: + """Best-effort mirror of a run into the PostgreSQL audit tables. + + Auditing is strictly additive: any failure here (missing DSN, unreachable + database, unexpected schema state) is swallowed so that journal indexing and + the workflow itself are never affected. + """ + + try: + store = _audit_store_for(data_root) + if not store.enabled: + return + steps = journal_payload.get("steps") if isinstance(journal_payload, dict) else None + store.record_run(record.to_dict(), steps) + except Exception: # noqa: BLE001 - audit mirroring must never raise + return diff --git a/src/gyxx_flow/ops/migrations/001_ops_audit.sql b/src/gyxx_flow/ops/migrations/001_ops_audit.sql new file mode 100644 index 0000000..0a5eb9c --- /dev/null +++ b/src/gyxx_flow/ops/migrations/001_ops_audit.sql @@ -0,0 +1,152 @@ +-- Migration 001: operational audit tables (workflow runs + login cookie telemetry). +-- +-- 本文件由 src/gyxx_flow/ops/audit.py 的 SCHEMA_SQL 导出，请勿手工编辑； +-- 运行时首次写入会自动执行同一份 DDL（幂等），此处仅供 DBA 审阅或预建。 +-- +-- 用法: psql "$GYXX_POSTGRES_DSN" -f 001_ops_audit.sql +-- +-- 说明: +-- ops_workflow_runs 一次工作流运行一行，run.json 的关系型投影 +-- ops_workflow_run_steps 每次尝试一行（含重试），保留 exit_code 与错误 +-- ops_account_cookie_events cookie 观测事件流，只存指纹不存明文值 +-- ops_account_cookie_lifetimes 一处登录态从首次出现到失效（视图） +-- ops_account_cookie_stats 每账号失效规律汇总（视图） +-- +-- 日志正文仍留在 var/logs 下，数据库只存结构化元数据与指向路径。 + +CREATE TABLE IF NOT EXISTS ops_workflow_runs ( + run_id TEXT PRIMARY KEY, + workflow_id TEXT NOT NULL, + business_date DATE NOT NULL, + mode TEXT NOT NULL DEFAULT 'unknown', + shadow BOOLEAN NOT NULL DEFAULT FALSE, + status TEXT NOT NULL, + started_at TIMESTAMPTZ, + ended_at TIMESTAMPTZ, + duration_seconds INTEGER, + error TEXT, + step_total INTEGER NOT NULL DEFAULT 0, + step_success INTEGER NOT NULL DEFAULT 0, + step_failed INTEGER NOT NULL DEFAULT 0, + step_skipped INTEGER NOT NULL DEFAULT 0, + step_running INTEGER NOT NULL DEFAULT 0, + journal_path TEXT NOT NULL DEFAULT '', + host TEXT NOT NULL DEFAULT '', + indexed_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS ops_workflow_runs_workflow_idx + ON ops_workflow_runs (workflow_id, started_at DESC); +CREATE INDEX IF NOT EXISTS ops_workflow_runs_status_idx + ON ops_workflow_runs (status, started_at DESC); +CREATE INDEX IF NOT EXISTS ops_workflow_runs_business_date_idx + ON ops_workflow_runs (business_date, workflow_id); + +CREATE TABLE IF NOT EXISTS ops_workflow_run_steps ( + run_id TEXT NOT NULL + REFERENCES ops_workflow_runs (run_id) ON DELETE CASCADE, + step_id TEXT NOT NULL, + attempt INTEGER NOT NULL DEFAULT 1, + status TEXT NOT NULL DEFAULT '', + exit_code INTEGER, + started_at TIMESTAMPTZ, + ended_at TIMESTAMPTZ, + duration_seconds INTEGER, + error TEXT, + PRIMARY KEY (run_id, step_id, attempt) +); + +CREATE INDEX IF NOT EXISTS ops_workflow_run_steps_step_idx + ON ops_workflow_run_steps (step_id, status); + +CREATE TABLE IF NOT EXISTS ops_account_cookie_events ( + id BIGSERIAL PRIMARY KEY, + account_id TEXT NOT NULL, + observed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + state TEXT NOT NULL, + probe_status TEXT NOT NULL DEFAULT '', + probe_source TEXT NOT NULL DEFAULT '', + cookie_count INTEGER, + required_count INTEGER, + fingerprint TEXT NOT NULL DEFAULT '', + declared_expires_at TIMESTAMPTZ, + detail TEXT NOT NULL DEFAULT '', + workflow_id TEXT NOT NULL DEFAULT '', + run_id TEXT NOT NULL DEFAULT '', + host TEXT NOT NULL DEFAULT '' +); + +CREATE INDEX IF NOT EXISTS ops_account_cookie_events_account_idx + ON ops_account_cookie_events (account_id, observed_at DESC); +CREATE INDEX IF NOT EXISTS ops_account_cookie_events_state_idx + ON ops_account_cookie_events (state, observed_at DESC); +CREATE INDEX IF NOT EXISTS ops_account_cookie_events_fingerprint_idx + ON ops_account_cookie_events (account_id, fingerprint); + +-- 一段「cookie 代」的生命周期：同一 fingerprint 从首次出现到失效。 +-- first_failed_at 只取最后一次 healthy 之后的失败，避免恢复后再次失败时口径前移。 +CREATE OR REPLACE VIEW ops_account_cookie_lifetimes AS +WITH per_fingerprint AS ( + SELECT + account_id, + fingerprint, + MIN(observed_at) AS first_seen_at, + MIN(observed_at) FILTER (WHERE state = 'healthy') AS first_healthy_at, + MAX(observed_at) FILTER (WHERE state = 'healthy') AS last_healthy_at, + MAX(declared_expires_at) AS declared_expires_at + FROM ops_account_cookie_events + WHERE fingerprint <> '' + GROUP BY account_id, fingerprint +) +SELECT + p.account_id, + p.fingerprint, + p.first_seen_at, + p.first_healthy_at, + p.last_healthy_at, + p.declared_expires_at, + f.first_failed_at, + (f.first_failed_at IS NOT NULL) AS expired, + CASE + WHEN p.first_healthy_at IS NOT NULL AND p.last_healthy_at IS NOT NULL + THEN EXTRACT(EPOCH FROM (p.last_healthy_at - p.first_healthy_at))::BIGINT + END AS healthy_span_seconds, + CASE + WHEN p.first_healthy_at IS NOT NULL AND f.first_failed_at IS NOT NULL + THEN EXTRACT(EPOCH FROM (f.first_failed_at - p.first_healthy_at))::BIGINT + END AS lifetime_seconds, + CASE + WHEN p.last_healthy_at IS NOT NULL AND f.first_failed_at IS NOT NULL + THEN EXTRACT(EPOCH FROM (f.first_failed_at - p.last_healthy_at))::BIGINT + END AS healthy_to_failure_seconds +FROM per_fingerprint p +LEFT JOIN LATERAL ( + SELECT MIN(e.observed_at) AS first_failed_at + FROM ops_account_cookie_events e + WHERE e.account_id = p.account_id + AND e.fingerprint = p.fingerprint + AND e.state IN ('expired', 'not_logged_in') + AND (p.last_healthy_at IS NULL OR e.observed_at > p.last_healthy_at) +) f ON TRUE; + +-- 按账号汇总的失效规律：平均/中位/最短/最长寿命，直接支撑「多久失效一次」。 +CREATE OR REPLACE VIEW ops_account_cookie_stats AS +SELECT + account_id, + COUNT(*) AS sessions, + COUNT(*) FILTER (WHERE expired) AS expired_sessions, + AVG(lifetime_seconds)::BIGINT AS avg_lifetime_seconds, + PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY lifetime_seconds)::BIGINT + AS median_lifetime_seconds, + MIN(lifetime_seconds) AS min_lifetime_seconds, + MAX(lifetime_seconds) AS max_lifetime_seconds, + MAX(first_failed_at) AS last_failed_at, + MAX(last_healthy_at) AS last_healthy_at +FROM ops_account_cookie_lifetimes +WHERE lifetime_seconds IS NOT NULL +GROUP BY account_id; + +CREATE TABLE IF NOT EXISTS ops_audit_schema ( + schema_version INTEGER PRIMARY KEY, + applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); diff --git a/tests/test_ops_audit.py b/tests/test_ops_audit.py new file mode 100644 index 0000000..b5b2bc0 --- /dev/null +++ b/tests/test_ops_audit.py @@ -0,0 +1,359 @@ +from __future__ import annotations + +import json +from datetime import date, datetime, timezone +from pathlib import Path +from typing import Any + +import pytest + +from gyxx_flow.ops.audit import ( + SCHEMA_SQL, + AuditStore, + split_step_key, + summarize_cookies, +) +from gyxx_flow.ops.index import RunIndex + + +DOUYIN_COOKIES: list[dict[str, Any]] = [ + { + "name": "sessionid", + "value": "AAA", + "domain": ".jinritemai.com", + "expires": 1766000000.0, + }, + { + "name": "sid_tt", + "value": "BBB", + "domain": ".jinritemai.com", + "expires": 1766100000.0, + }, + { + "name": "unrelated", + "value": "CCC", + "domain": ".example.com", + "expires": 1700000000.0, + }, + { + "name": "session", + "value": "DDD", + "domain": ".jinritemai.com", + "expires": -1, + }, +] + + +class _FakeCursor: + def __init__(self, log: list[Any], rows: list[dict[str, Any]]) -> None: + self._log = log + self._rows = rows + + def __enter__(self) -> "_FakeCursor": + return self + + def __exit__(self, *_: object) -> bool: + return False + + def execute(self, sql: str, params: Any = None) -> None: + self._log.append(("execute", sql, params)) + + def executemany(self, sql: str, rows: Any) -> None: + self._log.append(("executemany", sql, list(rows))) + + def fetchall(self) -> list[dict[str, Any]]: + return self._rows + + def fetchone(self) -> dict[str, Any] | None: + # 默认模拟「ops_audit_schema 尚不存在」，从而走首次建表分支。 + return None + + def close(self) -> None: + return None + + +class _FakeConnection: + def __init__(self, log: list[Any], rows: list[dict[str, Any]] | None = None) -> None: + self._log = log + self._rows = rows or [] + + def cursor(self) -> _FakeCursor: + return _FakeCursor(self._log, self._rows) + + def commit(self) -> None: + self._log.append(("commit",)) + + def close(self) -> None: + self._log.append(("close",)) + + +def _store_with_fake( + tmp_path: Path, + *, + environment: dict[str, str] | None = None, + rows: list[dict[str, Any]] | None = None, +) -> tuple[AuditStore, list[Any]]: + log: list[Any] = [] + store = AuditStore( + tmp_path, + environment=environment or {"GYXX_POSTGRES_DSN": "postgresql://fake/audit"}, + connect=lambda: _FakeConnection(log, rows), + ) + return store, log + + +# ---------------------------------------------------------------------- +# cookie 快照 +# ---------------------------------------------------------------------- + + +def test_summarize_cookies_keeps_only_required_cookies() -> None: + snapshot = summarize_cookies( + DOUYIN_COOKIES, + required_domains=("jinritemai.com",), + required_names=("sessionid", "sid_tt", "session"), + ) + + assert snapshot.cookie_count == 4 + assert snapshot.required_count == 3 + # 最早的必需 cookie 过期时间：1766000000，而不是无关 cookie 的 1700000000 + assert snapshot.declared_expires_at == datetime.fromtimestamp( + 1766000000.0, tz=timezone.utc + ) + + +def test_summarize_cookies_ignores_session_cookies_for_expiry() -> None: + snapshot = summarize_cookies( + DOUYIN_COOKIES, + required_domains=("jinritemai.com",), + required_names=("session",), + ) + + assert snapshot.required_count == 1 + assert snapshot.declared_expires_at is None + + +def test_summarize_cookies_fingerprint_tracks_value_changes_only() -> None: + baseline = summarize_cookies(DOUYIN_COOKIES, required_domains=("jinritemai.com",)) + reordered = summarize_cookies( + list(reversed(DOUYIN_COOKIES)), required_domains=("jinritemai.com",) + ) + replaced = [dict(cookie) for cookie in DOUYIN_COOKIES] + replaced[0]["value"] = "ROTATED" + + assert baseline.fingerprint # 非空 + assert baseline.fingerprint == reordered.fingerprint + assert baseline.fingerprint != summarize_cookies( + replaced, required_domains=("jinritemai.com",) + ).fingerprint + + +def test_summarize_cookies_without_requirements_counts_everything() -> None: + snapshot = summarize_cookies(DOUYIN_COOKIES) + + assert snapshot.required_count == 4 + assert snapshot.declared_expires_at == datetime.fromtimestamp( + 1700000000.0, tz=timezone.utc + ) + + +def test_summarize_cookies_has_no_plaintext_value_in_fingerprint() -> None: + snapshot = summarize_cookies(DOUYIN_COOKIES, required_domains=("jinritemai.com",)) + + assert "AAA" not in snapshot.fingerprint + assert len(snapshot.fingerprint) == 32 + + +def test_split_step_key_handles_retry_suffix() -> None: + assert split_step_key("sync.attempt-2") == ("sync", 2) + assert split_step_key("collect_collaborators") == ("collect_collaborators", 1) + assert split_step_key("a.attempt-x") == ("a.attempt-x", 1) + + +# ---------------------------------------------------------------------- +# 写入装配 +# ---------------------------------------------------------------------- + + +def test_record_run_upserts_run_and_steps(tmp_path: Path) -> None: + store, log = _store_with_fake(tmp_path) + + ok = store.record_run( + { + "workflow_id": "content.metrics.daily", + "run_id": "content.metrics.daily__20261018__abc", + "business_date": "2026-10-18", + "status": "success", + "mode": "dry_run", + "shadow": False, + "started_at": "2026-10-19T01:41:43+00:00", + "ended_at": "2026-10-19T01:41:48+00:00", + "error": None, + "journal_path": "runs/content.metrics.daily/run.json", + }, + { + "sync.attempt-1": { + "status": "success", + "exit_code": 0, + "started_at": "2026-10-19T01:41:44+00:00", + "ended_at": "2026-10-19T01:41:46+00:00", + }, + "collect.attempt-1": {"status": "failed", "exit_code": 4, "error": "boom"}, + }, + ) + + assert ok is True + statements = [entry for entry in log if entry[0] == "execute"] + assert any("CREATE TABLE IF NOT EXISTS ops_workflow_runs" in item[1] for item in statements) + + insert = next(item for item in statements if "INSERT INTO ops_workflow_runs" in item[1]) + params = insert[2] + assert params["business_date"] == date(2026, 10, 18) + assert params["duration_seconds"] == 5 + assert params["step_total"] == 2 + assert params["step_success"] == 1 + assert params["step_failed"] == 1 + + batch = next(item for item in log if item[0] == "executemany") + rows = {item["step_id"]: item for item in batch[2]} + assert rows["sync"]["attempt"] == 1 + assert rows["sync"]["duration_seconds"] == 2 + assert rows["collect"]["exit_code"] == 4 + assert rows["collect"]["error"] == "boom" + + +def test_record_run_rejects_incomplete_payload(tmp_path: Path) -> None: + store, log = _store_with_fake(tmp_path) + + assert store.record_run({"workflow_id": "x"}) is False + assert not [entry for entry in log if entry[0] == "execute"] + + +def test_record_cookie_event_writes_value_free_snapshot(tmp_path: Path) -> None: + store, log = _store_with_fake(tmp_path) + snapshot = summarize_cookies(DOUYIN_COOKIES, required_domains=("jinritemai.com",)) + + ok = store.record_cookie_event( + account_id="douyin-shop", + state="healthy", + snapshot=snapshot, + probe_status="success", + probe_source="keepalive", + workflow_id="content.metrics.daily", + run_id="run-1", + ) + + assert ok is True + insert = next( + item + for item in log + if item[0] == "execute" and "INSERT INTO ops_account_cookie_events" in item[1] + ) + params = insert[2] + assert params["account_id"] == "douyin-shop" + assert params["state"] == "healthy" + assert params["fingerprint"] == snapshot.fingerprint + assert params["declared_expires_at"] == snapshot.declared_expires_at + # 明文值不得出现在任何参数里 + assert "AAA" not in json.dumps(params, default=str) + + +def test_record_cookie_event_rejects_unknown_state(tmp_path: Path) -> None: + store, _ = _store_with_fake(tmp_path) + + with pytest.raises(ValueError): + store.record_cookie_event(account_id="a", state="definitely-not-valid") + + +# ---------------------------------------------------------------------- +# 降级与熔断 +# ---------------------------------------------------------------------- + + +def test_missing_dsn_never_raises_and_logs_failure(tmp_path: Path) -> None: + store = AuditStore(tmp_path, environment={}) + + assert store.record_run( + { + "workflow_id": "content.metrics.daily", + "run_id": "r1", + "business_date": "2026-10-18", + "status": "success", + } + ) is False + assert "审计数据库配置缺失" in store.last_error + failure_log = tmp_path / "logs" / "ops-audit" / "audit-failures.log" + assert failure_log.exists() + assert json.loads(failure_log.read_text(encoding="utf-8").splitlines()[0])["operation"] == ( + "record_run" + ) + + +def test_audit_can_be_disabled_without_touching_the_database(tmp_path: Path) -> None: + store, log = _store_with_fake( + tmp_path, environment={"GYXX_POSTGRES_DSN": "postgresql://fake", "GYXX_OPS_AUDIT": "off"} + ) + + assert store.enabled is False + assert store.record_cookie_event(account_id="a", state="healthy") is False + assert log == [] + + +def test_consecutive_failures_open_the_circuit(tmp_path: Path) -> None: + attempts = 0 + + def _boom() -> Any: + nonlocal attempts + attempts += 1 + raise OSError("database unreachable") + + store = AuditStore( + tmp_path, + environment={"GYXX_POSTGRES_DSN": "postgresql://fake", "GYXX_OPS_AUDIT_FAILURE_COOLDOWN": "60"}, + connect=_boom, + ) + payload = { + "workflow_id": "w", + "run_id": "r", + "business_date": "2026-10-18", + "status": "success", + } + + for _ in range(3): + store.record_run(payload) + assert attempts == 3 + + # 熔断生效后不再尝试连接，避免每个工作流都卡在 TCP 超时上 + assert store.record_run(payload) is False + assert attempts == 3 + + +# ---------------------------------------------------------------------- +# 与运行索引的集成 +# ---------------------------------------------------------------------- + + +def test_run_index_still_works_when_audit_is_unavailable(tmp_path: Path) -> None: + from gyxx_flow.core.context import RunContext + from gyxx_flow.core.layout import DataLayout + from gyxx_flow.core.records import RunJournal + + layout = DataLayout(tmp_path) + context = RunContext.create("content.metrics.daily", "2026-10-18") + journal = RunJournal.create(layout, context, mode="dry_run") + journal.finalize("success") + + record = RunIndex(tmp_path).index_journal(journal) + + assert record.workflow_id == "content.metrics.daily" + assert record.status == "success" + # 运行索引文件仍然写出，审计失败不影响主链路 + assert (tmp_path / "state" / "ops" / "run-index" / f"{record.run_id}.json").exists() + + +def test_schema_sql_defines_both_audit_surfaces() -> None: + assert "ops_workflow_runs" in SCHEMA_SQL + assert "ops_workflow_run_steps" in SCHEMA_SQL + assert "ops_account_cookie_events" in SCHEMA_SQL + assert "ops_account_cookie_lifetimes" in SCHEMA_SQL + assert "ops_account_cookie_stats" in SCHEMA_SQL