""" 公共工具函数 — 消除 V1 中 _ordered_unique、_session_stamp、policy normalization 等 在各模块中 3~5 份重复定义的代码。 """ from __future__ import annotations import re from datetime import datetime, timezone from typing import Any def ordered_unique(items: list) -> list: """保序去重。支持字符串列表和带 id 字段的字典列表。""" seen: set[str] = set() result = [] for item in items: key = item if isinstance(item, str) else item.get("id", str(item)) if key not in seen: seen.add(key) result.append(item) return result def session_stamp() -> str: """统一的文件系统安全时间戳,所有模块共用。""" return datetime.now(timezone.utc).isoformat().replace(":", "-").replace(".", "-").replace("+", "-") def now_iso() -> str: """ISO 格式 UTC 时间戳,用于 JSON state 文件。""" return datetime.now(timezone.utc).isoformat() def normalize_policy(defaults: dict[str, Any], overrides: dict[str, Any] | None) -> dict[str, Any]: """通用的策略合并:overrides 覆盖 defaults,类型自动转换。""" merged = {**defaults} if overrides: for k, v in overrides.items(): if k in merged: expected_type = type(defaults[k]) try: merged[k] = expected_type(v) if not isinstance(v, expected_type) else v except (ValueError, TypeError): merged[k] = v return merged def sanitize_task_id(task_id: str) -> str: """防止路径注入:仅允许字母数字、下划线、连字符、点号。""" if not re.fullmatch(r"[A-Za-z0-9_\-\.]+", task_id): raise ValueError(f"invalid task_id: {task_id!r}") return task_id def sanitize_marker(marker: str) -> str: """防止 HTML 注释注入。""" if "-->" in marker or "