refactor: implement Plan B architecture modernization and rolling history pruning

- Git Rolling Pruning (Mode A): run_update_and_commit.sh automatically retains
  the latest 50 commits on the runtime branch with graceful fallback to append-only.
- Router SSOT Synchronization: build_router_package.sh supports --sync-from-config
  to automatically project CFST and output parameters into config_router.conf.
- Modular Decoupling: domain_updater.py refactored into high-cohesion rotator/
  sub-packages (config, filter, sources, scoring, storage, notify, main) with
  100% backward-compatible Facade entry point.
- Quality Assurance: expanded test suite to 21 automated regression tests, 100% passing.

ARCHITECTURE_REVIEW_UNIFIED.md

@@ -756,13 +756,48 @@ function extractValue(rawText, format, jsonKeys, maxAgeSec) {
 
 | 测试文件 | 覆盖范围与断言目标 | 用例数 | 执行耗时 | 测试状态 |
 | :--- | :--- | :--- | :--- | :--- |
-| [`tests/test_domain_updater.py`](./tests/test_domain_updater.py) | • P0 极性反转真实数据排序回归与平局决胜<br>• run_notify 标准输出隔离与超时保护<br>• 异常回退包含 domain/ip 键<br>• 黑名单 last_good 禁止回退复活<br>• 本地 cfst 过滤 0.00 下载速度<br>• `--result-json-file` IPC 解耦<br>• validate_config 严格顶层结构校验 | 7 项 | ~0.15s | **✅ 全部通过 (PASS)** |
-| [`tests/test_shell_lifecycle.py`](./tests/test_shell_lifecycle.py) | • `uninstall_debian.sh` 空服务名安全校验（严禁 `rm -rf /var/lib/`）<br>• `build_router_package.sh` 延迟注册 trap 防止误删工作目录<br>• `run_update_and_commit.sh` 采用 `GIT_CONFIG_*` 环境变量传递 Token（argv 零暴露）<br>• 严格区分 `git ls-remote` 退出码 `2` vs `128`<br>• 回退静默跳过噪音 Git commit | 5 项 | ~0.12s | **✅ 全部通过 (PASS)** |
-| [`tests/test_router_scripts.py`](./tests/test_router_scripts.py) | • `router_local_update.sh` `.tmp.$$` + `mv -f` 原子替换写入<br>• 测速结果优先过滤 `download_speed > 0`<br>• 纠偏初审报告关于 `-dt/-dn/-dd` 参数支持情况 | 3 项 | ~0.05s | **✅ 全部通过 (PASS)** |
+| [`tests/test_domain_updater.py`](./tests/test_domain_updater.py) | • P0 极性反转真实数据排序回归与平局决胜<br>• run_notify 标准输出隔离与超时保护<br>• 异常回退包含 domain/ip 键<br>• 黑名单 last_good 禁止回退复活<br>• 本地 cfst 过滤 0.00 下载速度<br>• `--result-json-file` IPC 解耦<br>• validate_config 严格顶层结构校验<br>• **[方案B] rotator 子包解耦与 Facade 契约完整性** | 8 项 | ~0.15s | **✅ 全部通过 (PASS)** |
+| [`tests/test_shell_lifecycle.py`](./tests/test_shell_lifecycle.py) | • `uninstall_debian.sh` 空服务名安全校验（严禁 `rm -rf /var/lib/`）<br>• `build_router_package.sh` 延迟注册 trap 防止误删工作目录<br>• `run_update_and_commit.sh` 采用 `GIT_CONFIG_*` 环境变量传递 Token（argv 零暴露）<br>• 严格区分 `git ls-remote` 退出码 `2` vs `128`<br>• 回退静默跳过噪音 Git commit<br>• **[方案B] 模式A Git 历史滚动修剪（50 commit 滚动保留）端到端验证** | 6 项 | ~0.75s | **✅ 全部通过 (PASS)** |
+| [`tests/test_router_scripts.py`](./tests/test_router_scripts.py) | • `router_local_update.sh` `.tmp.$$` + `mv -f` 原子替换写入<br>• 测速结果优先过滤 `download_speed > 0`<br>• 纠偏初审报告关于 `-dt/-dn/-dd` 参数支持情况<br>• **[方案B] `build_router_package.sh --sync-from-config` SSOT 单源参数自动投影验证** | 4 项 | ~0.50s | **✅ 全部通过 (PASS)** |
 | [`tests/test_substore_operator.py`](./tests/test_substore_operator.py) | • `extractValue()` 严格类型守卫（拒绝非字符串对象强转为 `[object Object]`）<br>• 域名 (RFC 1123) 与 IPv4/IPv6 正则校验（拒绝 HTML 报错与非法字符）<br>• `LOCAL_MAX_AGE_SEC` (24h) 本地路由器源超时自动降级 | 3 项 | ~0.06s | **✅ 全部通过 (PASS)** |
 
 ### 9.2 运行命令
 ```bash
 python3 -m unittest discover tests -v
-# 输出：Ran 18 tests in 0.378s, OK
+# 输出：Ran 21 tests in 1.417s, OK
 ```
+
+---
+
+## 10. 方案 B（大刀阔斧的业务流程瘦身与架构演进）完整落地报告 ✅
+
+经与用户对齐并选定「模式 A（滚动保留最近 50 次提交，兼顾轻量与防误封）」，方案 B 现已 100% 落地并完成全量验证：
+
+### 10.1 模式 A：Git 孤儿分支滚动修剪与优雅降级机制
+- **问题根源**：原调度脚本每次优选都无条件向 `runtime-state` 分支追加 commit，经过数月运行会堆积上万个 commit，导致远程仓库体积虚高、拉取缓慢。
+- **重构落地**：
+  - 在 [`scripts/run_update_and_commit.sh`](./scripts/run_update_and_commit.sh) 引入 `GIT_MAX_STATE_COMMITS`（默认 50 次提交，可由用户自定义）。
+  - 当历史 commit 数超出阈值时，自动提取保留基线 tree，创建 baseline root commit 并执行 `rebase --onto`，精简历史。
+  - **自适应容错**：优先使用 `--force-with-lease` 推送；若远端存在分支保护策略拒绝 force-push，脚本自动捕获并从 untrimmed 备份分支无损回滚，平滑降级为普通追加推送，绝不中断服务。
+  - **跨平台稳健性**：全面消除 `mapfile`（避免 macOS / 旧系统 Bash 3.2 不支持）与 `set -u` 展开空数组崩溃隐患，兼容任意 Linux/Debian/macOS 环境。
+
+### 10.2 Step 2：路由器配置单源驱动（SSOT 联动）
+- **问题根源**：原路由器配置 `config_router.conf` 与服务端/本地配置 `config.local.json` 割裂，若用户调整了测速线程数、测试时间、测速端口或输出文件名，需双重维护。
+- **重构落地**：
+  - 在 [`build_router_package.sh`](./build_router_package.sh) 中新增 `--sync-from-config` 命令行参数支持。
+  - 打包前自动解析 JSON 配置，将 `TOP_N`, `VALUE_TEXT_FILE`, `CFST_DISPLAY_COUNT`, `CFST_IP_FILE`, `CFST_PORT`, `CFST_DISABLE_DOWNLOAD` 等参数动态投影同步至 tarball 内的 `config_router.conf`。
+  - 彻底杜绝双源维护漂移。
+
+### 10.3 Step 3：`domain_updater.py` 领域模块化解耦
+- **问题根源**：原 `domain_updater.py` 膨胀为 1246 行单文件单片脚本，混合了 CLI、HTTP、CSV 解析、评分策略、原子 IO、系统通知等多种无关职责。
+- **重构落地**：
+  - 新建高内聚子包 [`scripts/rotator/`](./scripts/rotator/)：
+    - `storage.py`：原子落盘写入与状态读写
+    - `config.py`：配置严格校验、路径解析与输出映射
+    - `filter.py`：域名与 IP 验证、正则清洗与黑名单排除
+    - `sources.py`：远程 API 抓取与本地 CFST 结果解析
+    - `scoring.py`：健康评分、字典序排序、迟滞防抖与候选列表构造
+    - `notify.py`：通知命令执行与超时防护
+    - `main.py`：主轮换工作流编排与 CLI 参数解析
+  - [`scripts/domain_updater.py`](./scripts/domain_updater.py) 重构为精简的外观门面（Facade），全面重新导出所有符号，保持 100% 向后兼容，现存 systemd、cron 及自动化测试零修改直接运行。
+

CLAUDE.md

@@ -16,8 +16,16 @@
 
 ```
 scripts/
-├── domain_updater.py          # 核心：两种模式共用的 Python 更新器
-├── run_update_and_commit.sh   # Server 模式调度器 + git 提交逻辑
+├── domain_updater.py          # 顶层外观门面（Facade，100% 向后兼容 CLI 与 API）
+├── rotator/                   # 核心领域高内聚模块子包
+│   ├── config.py              # 配置校验、路径解析与输出设定
+│   ├── filter.py              # 域名清洗、正则匹配与黑名单过滤
+│   ├── sources.py             # 远程 API 与本地 cfst 数据源适配
+│   ├── scoring.py             # 评分排名、平局决胜、迟滞防抖与候选构建
+│   ├── storage.py             # 原子落盘写入与状态读写
+│   ├── notify.py              # 通知命令执行与超时防护
+│   └── main.py                # 主轮换工作流编排与 CLI 参数解析
+├── run_update_and_commit.sh   # Server 模式调度器 + Git 自动提交与滚动修剪（模式 A）
 ├── router_local_update.sh     # BusyBox 路由器：运行 cfst 并写输出
 ├── router_local_http.sh       # BusyBox 路由器：httpd 提供运行时文件
 ├── install_debian.sh          # systemd timer/service 安装
@@ -27,12 +35,12 @@ substore/
 └── operator_template.js       # Sub-Store 脚本操作符，消费运行时 JSON
 
 tests/
-├── test_domain_updater.py     # Python 核心选优、极性、回退与配置校验
-├── test_shell_lifecycle.py    # Shell 部署、发布安全与凭证保护
-├── test_router_scripts.py     # 路由器 BusyBox 脚本文件原子性与测速过滤
+├── test_domain_updater.py     # Python 核心选优、极性、回退、配置校验及 Facade 完整性
+├── test_shell_lifecycle.py    # Shell 部署、发布安全、凭证保护及 Git 历史滚动修剪
+├── test_router_scripts.py     # 路由器 BusyBox 脚本文件原子性、测速过滤及 SSOT 打包同步
 └── test_substore_operator.py  # Sub-Store 算子类型安全与时效降级
 
-build_router_package.sh        # 打包路由器部署 tarball → vmess.tar.gz
+build_router_package.sh        # 打包路由器部署 tarball → vmess.tar.gz（支持 --sync-from-config）
 config.server.json             # Server 模式完整配置
 config.local.json              # Python 本地 cfst 模式配置
 config_router.conf             # BusyBox shell 模式配置
@@ -80,8 +88,11 @@ bash scripts/run_update_and_commit.sh --force-commit config.server.json
 sh scripts/router_local_update.sh ./config_router.conf
 sh scripts/router_local_http.sh ./config_router.conf
 
-# 打包路由器部署包
+# 打包路由器部署包（默认打包现有 config_router.conf）
 ./build_router_package.sh
+
+# 打包路由器部署包（SSOT 单源模式：自动从 config.local.json 投影同步测速与输出参数）
+./build_router_package.sh --sync-from-config config.local.json
 ```
 
 ### systemd 安装/卸载
@@ -95,7 +106,7 @@ sudo bash scripts/uninstall_debian.sh
 
 ## Testing
 
-- **全套自动化回归测试**：运行 `python3 -m unittest discover tests -v`（覆盖 4 大模块 16 项用例，秒级执行，零第三方依赖）。
+- **全套自动化回归测试**：运行 `python3 -m unittest discover tests -v`（覆盖 4 大模块 21 项用例，秒级执行，零第三方依赖）。
 - **语法校验**：使用 `py_compile` 与 `bash -n` / `sh -n` 快速检查语法。
 - **路径解析验证**：`python3 scripts/domain_updater.py --config config.server.json --print-output-settings`。
 - **本地测试服务器端时，仅运行 `python3 scripts/domain_updater.py --config config.server.json` 进行输出验证，严禁直接运行 `run_update_and_commit.sh` 调度/Git 推送脚本。**

build_router_package.sh

@@ -17,6 +17,29 @@ BUSYBOX_SRC=${BUSYBOX_SRC:-"$APP_DIR/busybox/busybox-armv7l"}
 CFST_SRC_DIR=${CFST_SRC_DIR:-"$APP_DIR/cfst_linux_armv5"}
 
 VERBOSE=${VERBOSE:-1}
+SYNC_FROM_CONFIG=${SYNC_FROM_CONFIG:-}
+WRITE_BACK=${WRITE_BACK:-0}
+
+while [ $# -gt 0 ]; do
+  case "$1" in
+    --sync-from=*|--sync-from-config=*|--sync-from-json=*)
+      SYNC_FROM_CONFIG="${1#*=}"
+      shift
+      ;;
+    --sync-from|--sync-from-config|--sync-from-json)
+      [ $# -ge 2 ] || { printf '[error] %s requires a file path argument\n' "$1" >&2; exit 1; }
+      SYNC_FROM_CONFIG="$2"
+      shift 2
+      ;;
+    --write-back)
+      WRITE_BACK=1
+      shift
+      ;;
+    *)
+      shift
+      ;;
+  esac
+done
 
 log() {
   if [ "${VERBOSE}" = "1" ]; then
@@ -109,13 +132,123 @@ if [ -n "${bb_from_conf_raw:-}" ]; then
   fi
 fi
 
+sync_conf_from_json() {
+  src_conf="$1"
+  json_file="$2"
+  dst_conf="$3"
+
+  if ! command -v python3 >/dev/null 2>&1; then
+    warn "python3 not found, skip syncing router config from JSON"
+    cp "$src_conf" "$dst_conf"
+    return 0
+  fi
+
+  python3 - "$src_conf" "$json_file" "$dst_conf" <<'EOF'
+import json
+import re
+import sys
+
+src_conf_path, json_path, dst_conf_path = sys.argv[1], sys.argv[2], sys.argv[3]
+
+with open(json_path, "r", encoding="utf-8") as f:
+    cfg = json.load(f)
+
+updates = {}
+out_cfg = cfg.get("output", {})
+if "selected_value_file" in out_cfg:
+    updates["VALUE_TEXT_FILE"] = str(out_cfg["selected_value_file"])
+if "selected_value_json" in out_cfg:
+    updates["VALUE_JSON_FILE"] = str(out_cfg["selected_value_json"])
+if "selected_value_json_key" in out_cfg:
+    updates["VALUE_JSON_KEY"] = str(out_cfg["selected_value_json_key"])
+if "state_last_good_key" in out_cfg:
+    updates["STATE_LAST_GOOD_KEY"] = str(out_cfg["state_last_good_key"])
+if "export_vars_file" in out_cfg:
+    updates["EXPORT_VARS_FILE"] = str(out_cfg["export_vars_file"])
+if "substore_value_key" in out_cfg:
+    updates["EXPORT_VALUE_KEY"] = str(out_cfg["substore_value_key"])
+
+sel_cfg = cfg.get("selection", {})
+if "top_n" in sel_cfg:
+    updates["TOP_N"] = str(sel_cfg["top_n"])
+
+cfst_cfg = cfg.get("cfst_local", {})
+run_args = cfst_cfg.get("run_args", [])
+if isinstance(run_args, list):
+    arg_map = {
+        "-p": "CFST_DISPLAY_COUNT",
+        "-f": "CFST_IP_FILE",
+        "-n": "CFST_THREADS",
+        "-t": "CFST_TEST_COUNT",
+        "-dn": "CFST_DOWNLOAD_COUNT",
+        "-dt": "CFST_DOWNLOAD_TIME",
+        "-tp": "CFST_PORT",
+        "-url": "CFST_URL",
+        "-cfcolo": "CFST_CFCOLO",
+        "-tl": "CFST_LATENCY_LIMIT",
+        "-tll": "CFST_LATENCY_LOWER",
+        "-sl": "CFST_LOSS_LIMIT",
+        "-speedlimit": "CFST_SPEED_LIMIT",
+    }
+    i = 0
+    while i < len(run_args):
+        flag = str(run_args[i]).strip()
+        if flag in arg_map and i + 1 < len(run_args):
+            updates[arg_map[flag]] = str(run_args[i + 1]).strip()
+            i += 2
+        elif flag == "-httping":
+            updates["CFST_HTTPING"] = "1"
+            i += 1
+        elif flag == "-dd":
+            updates["CFST_DISABLE_DOWNLOAD"] = "1"
+            i += 1
+        elif flag == "-allip":
+            updates["CFST_ALL_IP"] = "1"
+            i += 1
+        else:
+            i += 1
+
+with open(src_conf_path, "r", encoding="utf-8") as f:
+    conf_lines = f.readlines()
+
+new_lines = []
+modified_keys = set()
+for line in conf_lines:
+    m = re.match(r'^([A-Za-z0-9_]+)="(.*)"', line.strip())
+    if m:
+        key = m.group(1)
+        if key in updates:
+            new_lines.append(f'{key}="{updates[key]}"\n')
+            modified_keys.add(key)
+            continue
+    new_lines.append(line)
+
+for key, val in updates.items():
+    if key not in modified_keys:
+        new_lines.append(f'{key}="{val}"\n')
+
+with open(dst_conf_path, "w", encoding="utf-8") as f:
+    f.writelines(new_lines)
+EOF
+}
+
 if ! check_busybox_httpd "$BUSYBOX_SRC"; then
   warn "continuing packaging, but router HTTP may fail later"
 fi
 
 mkdir -p "$TMP_DIR/$PKG_ROOT_NAME/scripts" "$TMP_DIR/$PKG_ROOT_NAME/cfst"
 
-cp "$ROUTER_CONF" "$TMP_DIR/$PKG_ROOT_NAME/config_router.conf"
+if [ -n "$SYNC_FROM_CONFIG" ]; then
+  require_file "$SYNC_FROM_CONFIG"
+  log "[pkg] syncing router config from $SYNC_FROM_CONFIG"
+  sync_conf_from_json "$ROUTER_CONF" "$SYNC_FROM_CONFIG" "$TMP_DIR/$PKG_ROOT_NAME/config_router.conf"
+  if [ "$WRITE_BACK" = "1" ]; then
+    cp "$TMP_DIR/$PKG_ROOT_NAME/config_router.conf" "$ROUTER_CONF"
+    log "[pkg] written synced config back to $ROUTER_CONF"
+  fi
+else
+  cp "$ROUTER_CONF" "$TMP_DIR/$PKG_ROOT_NAME/config_router.conf"
+fi
 cp "$HTTP_SCRIPT" "$TMP_DIR/$PKG_ROOT_NAME/scripts/router_local_http.sh"
 cp "$UPDATE_SCRIPT" "$TMP_DIR/$PKG_ROOT_NAME/scripts/router_local_update.sh"
 cp "$BUSYBOX_SRC" "$TMP_DIR/$PKG_ROOT_NAME/busybox_armv7l"

scripts/domain_updater.py

@@ -1,1245 +1,121 @@
 #!/usr/bin/env python3
-import argparse
-import csv
-import datetime as dt
-import functools
-import json
-import math
-import os
-import re
-import subprocess
-import sys
-import tempfile
-import urllib.parse
-import urllib.request
-
-
-DOMAIN_RE = re.compile(r"^(?=.{1,253}$)(?!-)[A-Za-z0-9-]{1,63}(?<!-)(\.(?!-)[A-Za-z0-9-]{1,63}(?<!-))+$")
-IPV4_RE = re.compile(r"^(?:25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(?:\.(?:25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}$")
-
-
-def utc_now_iso():
-    return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
-
-
-def read_json_file(path, default=None):
-    if default is None:
-        default = {}
-    if not os.path.exists(path):
-        return default
-    try:
-        with open(path, "r", encoding="utf-8") as f:
-            return json.load(f)
-    except (ValueError, json.JSONDecodeError):
-        return default
-
-
-def _atomic_write(path, payload_bytes):
-    """Write `payload_bytes` to `path` atomically.
-
-    Strategy: write to a sibling temp file, fsync, then os.replace.
-    Concurrent readers either see the previous version or the new
-    one — never a truncated/empty file. Required for state.json,
-    which is the sole fallback source on the next run.
-    """
-    parent = os.path.dirname(path) or "."
-    os.makedirs(parent, exist_ok=True)
-    fd, tmp_path = tempfile.mkstemp(
-        prefix=".{}.".format(os.path.basename(path)),
-        suffix=".tmp",
-        dir=parent,
-    )
-    try:
-        with os.fdopen(fd, "wb") as f:
-            f.write(payload_bytes)
-            f.flush()
-            try:
-                os.fsync(f.fileno())
-            except OSError:
-                # fsync may fail on some filesystems (e.g. tmpfs without
-                # backing store); the rename below still gives atomicity
-                # within a single filesystem.
-                pass
-        os.replace(tmp_path, path)
-    except Exception:
-        try:
-            os.unlink(tmp_path)
-        except OSError:
-            pass
-        raise
-
-
-def write_json_file(path, data):
-    payload = json.dumps(data, ensure_ascii=True, indent=2)
-    _atomic_write(path, payload.encode("utf-8"))
-
-
-def write_text_file(path, data):
-    _atomic_write(path, data.encode("utf-8"))
-
-
-def build_url(base_url, params):
-    if not params:
-        return base_url
-    parsed = urllib.parse.urlparse(base_url)
-    current = urllib.parse.parse_qs(parsed.query)
-    for k, v in params.items():
-        current[k] = [str(v)]
-    query = urllib.parse.urlencode(current, doseq=True)
-    return urllib.parse.urlunparse(parsed._replace(query=query))
-
-
-def resolve_path(base_dir, path_value):
-    path_text = str(path_value or "").strip()
-    if not path_text:
-        return ""
-    if os.path.isabs(path_text):
-        return os.path.normpath(path_text)
-    return os.path.normpath(os.path.join(base_dir, path_text))
-
-
-def get_source_type(cfg):
-    source_cfg = cfg.get("source", {})
-    if isinstance(source_cfg, dict):
-        source_type = str(source_cfg.get("type", "api")).strip().lower()
-        if source_type:
-            return source_type
-    return "api"
-
-
-def fetch_api_json(cfg):
-    api = cfg["api"]
-    url = build_url(api["url"], api.get("params", {}))
-    method = api.get("method", "GET").upper()
-    headers = api.get("headers", {})
-    timeout = int(api.get("timeout_sec", 10))
-    body_obj = api.get("body")
-    body = None
-    if body_obj is not None:
-        body = json.dumps(body_obj).encode("utf-8")
-        headers = {**headers, "Content-Type": "application/json"}
-
-    req = urllib.request.Request(url=url, data=body, headers=headers, method=method)
-    with urllib.request.urlopen(req, timeout=timeout) as resp:
-        raw = resp.read().decode("utf-8", errors="replace")
-    return json.loads(raw)
-
-
-def load_cfst_rows(cfg, config_path_abs):
-    cfst_cfg = cfg.get("cfst_local", {})
-    config_dir = os.path.dirname(config_path_abs)
-
-    work_dir = resolve_path(config_dir, cfst_cfg.get("work_dir", "./cfst"))
-    binary_path = resolve_path(work_dir, cfst_cfg.get("binary", "./cfst"))
-    result_file = resolve_path(work_dir, cfst_cfg.get("result_file", "result.csv"))
-    encoding = str(cfst_cfg.get("encoding", "utf-8")).strip() or "utf-8"
-    skip_run = bool(cfst_cfg.get("skip_run", False))
-    timeout_sec = int(cfst_cfg.get("run_timeout_sec", 600))
-
-    run_args = cfst_cfg.get("run_args", ["-o", os.path.basename(result_file)])
-    if not isinstance(run_args, list):
-        raise ValueError("cfst_local.run_args must be an array")
-    command = [binary_path] + [str(x) for x in run_args]
-
-    if not skip_run:
-        completed = subprocess.run(
-            command,
-            cwd=work_dir,
-            check=False,
-            capture_output=True,
-            text=True,
-            encoding=encoding,
-            errors="replace",
-            timeout=timeout_sec,
-        )
-        if completed.returncode != 0:
-            stderr = (completed.stderr or "").strip()
-            stdout = (completed.stdout or "").strip()
-            details = stderr or stdout or f"exit code {completed.returncode}"
-            raise RuntimeError(f"cfst run failed: {details}")
-
-    if not os.path.exists(result_file):
-        raise RuntimeError(f"cfst result file not found: {result_file}")
-
-    with open(result_file, "r", encoding=encoding, errors="replace", newline="") as f:
-        reader = csv.reader(f)
-        rows = [row for row in reader if any(str(col).strip() for col in row)]
-
-    header_rows = int(cfst_cfg.get("header_rows", 1))
-    if len(rows) <= header_rows:
-        raise RuntimeError("cfst result has no data rows")
-
-    columns_cfg = cfst_cfg.get("columns", {})
-    if not isinstance(columns_cfg, dict):
-        raise ValueError("cfst_local.columns must be an object")
-
-    def col_index(name, default_index):
-        raw = columns_cfg.get(name, default_index)
-        try:
-            idx = int(raw)
-        except Exception as exc:
-            raise ValueError(f"cfst_local.columns.{name} must be an integer") from exc
-        if idx < 0:
-            raise ValueError(f"cfst_local.columns.{name} must be >= 0")
-        return idx
-
-    ip_idx = col_index("ip", 0)
-    sent_idx = col_index("sent", 1)
-    received_idx = col_index("received", 2)
-    loss_idx = col_index("loss_rate", 3)
-    latency_idx = col_index("avg_latency", 4)
-    speed_idx = col_index("download_speed", 5)
-    region_idx = col_index("region", 6)
-
-    out = []
-    for row in rows[header_rows:]:
-        if ip_idx >= len(row):
-            continue
-        domain = normalize_domain(row[ip_idx])
-        if not domain:
-            continue
-        out.append(
-            {
-                "domain": domain,
-                "ip": domain,
-                "sent": row[sent_idx].strip() if sent_idx < len(row) else "",
-                "received": row[received_idx].strip() if received_idx < len(row) else "",
-                "loss_rate": row[loss_idx].strip() if loss_idx < len(row) else "",
-                "avg_latency": row[latency_idx].strip() if latency_idx < len(row) else "",
-                "download_speed": row[speed_idx].strip() if speed_idx < len(row) else "",
-                "region": row[region_idx].strip() if region_idx < len(row) else "",
-            }
-        )
-
-    if not out:
-        raise RuntimeError("cfst result parsed to zero valid rows")
-    return out
-
-
-def flatten_values(value):
-    out = []
-    if isinstance(value, str):
-        out.append(value)
-    elif isinstance(value, list):
-        for item in value:
-            out.extend(flatten_values(item))
-    elif isinstance(value, dict):
-        for item in value.values():
-            out.extend(flatten_values(item))
-    return out
-
-
-def get_by_json_path(data, path):
-    cur = data
-    for part in path.split("."):
-        if isinstance(cur, dict) and part in cur:
-            cur = cur[part]
-        else:
-            return None
-    return cur
-
-
-def get_values_by_path(data, path):
-    parts = path.split(".")
-
-    def walk(cur, idx):
-        if idx >= len(parts):
-            return [cur]
-
-        part = parts[idx]
-        if part.endswith("[]"):
-            key = part[:-2]
-            if isinstance(cur, dict):
-                arr = cur.get(key)
-            else:
-                arr = None
-            if not isinstance(arr, list):
-                return []
-
-            out = []
-            for item in arr:
-                out.extend(walk(item, idx + 1))
-            return out
-
-        if isinstance(cur, dict) and part in cur:
-            return walk(cur[part], idx + 1)
-        return []
-
-    return walk(data, 0)
-
-
-def parse_domains(payload, parser_cfg):
-    domains = []
-
-    for p in parser_cfg.get("field_paths", []):
-        values = get_values_by_path(payload, p)
-        domains.extend(flatten_values(values))
-
-    for p in parser_cfg.get("json_paths", []):
-        v = get_by_json_path(payload, p)
-        if v is not None:
-            domains.extend(flatten_values(v))
-
-    if not domains:
-        regex_s = parser_cfg.get("regex", r"[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}")
-        text = json.dumps(payload, ensure_ascii=True)
-        domains.extend(re.findall(regex_s, text))
-
-    clean = []
-    seen = set()
-    for d in domains:
-        d = str(d).strip().lower().rstrip(".")
-        if (DOMAIN_RE.match(d) or IPV4_RE.match(d)) and d not in seen:
-            seen.add(d)
-            clean.append(d)
-    return clean
-
-
-def parse_timezone(tz_raw):
-    if tz_raw is None:
-        return dt.timezone.utc
-    s = str(tz_raw).strip().upper()
-    if s in {"", "UTC", "Z", "+00:00", "+0000"}:
-        return dt.timezone.utc
-
-    m = re.match(r"^([+-])(\d{2}):?(\d{2})$", s)
-    if not m:
-        raise ValueError(f"invalid created_time_timezone: {tz_raw}")
-
-    sign = 1 if m.group(1) == "+" else -1
-    hh = int(m.group(2))
-    mm = int(m.group(3))
-    if hh > 23 or mm > 59:
-        raise ValueError(f"invalid created_time_timezone offset: {tz_raw}")
-    return dt.timezone(sign * dt.timedelta(hours=hh, minutes=mm))
-
-
-def parse_created_time(value, formats, timezone):
-    if value is None:
-        return None
-
-    s = str(value).strip()
-    if not s:
-        return None
-
-    for fmt in formats:
-        try:
-            parsed = dt.datetime.strptime(s, fmt)
-            if parsed.tzinfo is None:
-                parsed = parsed.replace(tzinfo=timezone)
-            return parsed.astimezone(dt.timezone.utc)
-        except Exception:
-            continue
-
-    try:
-        iso_text = s.replace("Z", "+00:00")
-        parsed = dt.datetime.fromisoformat(iso_text)
-        if parsed.tzinfo is None:
-            parsed = parsed.replace(tzinfo=timezone)
-        return parsed.astimezone(dt.timezone.utc)
-    except Exception:
-        return None
-
-
-def normalize_domain(value):
-    if value is None:
-        return ""
-    return str(value).strip().lower().rstrip(".")
-
-
-def to_float_or_none(value):
-    try:
-        f = float(value)
-        if not math.isfinite(f):
-            return None
-        return f
-    except Exception:
-        return None
-
-
-def resolve_field(record, field_name, field_map):
-    path = field_map.get(field_name)
-    if not path:
-        raise ValueError(f"field '{field_name}' is not registered in record_mapping.field_map")
-    if not isinstance(record, dict):
-        return None
-    return get_by_json_path(record, path)
-
-
-def extract_records(payload, record_mapping):
-    records_path = str(record_mapping.get("records_path", "")).strip()
-    raw = get_values_by_path(payload, records_path)
-    return [x for x in raw if isinstance(x, dict)]
-
-
-def validate_config(cfg):
-    source_type = get_source_type(cfg)
-    if source_type not in {"api", "cfst_local"}:
-        raise ValueError("source.type must be 'api' or 'cfst_local'")
-
-    output_cfg = cfg.get("output", {})
-    if output_cfg and not isinstance(output_cfg, dict):
-        raise ValueError("output must be an object")
-
-    if source_type == "cfst_local":
-        cfst_cfg = cfg.get("cfst_local")
-        if not isinstance(cfst_cfg, dict):
-            raise ValueError("cfst_local is required and must be an object when source.type=cfst_local")
-
-        work_dir = str(cfst_cfg.get("work_dir", "")).strip()
-        if not work_dir:
-            raise ValueError("cfst_local.work_dir is required")
-
-        binary = str(cfst_cfg.get("binary", "")).strip()
-        if not binary:
-            raise ValueError("cfst_local.binary is required")
-
-        result_file = str(cfst_cfg.get("result_file", "")).strip()
-        if not result_file:
-            raise ValueError("cfst_local.result_file is required")
-
-        run_args = cfst_cfg.get("run_args", [])
-        if not isinstance(run_args, list):
-            raise ValueError("cfst_local.run_args must be an array")
-
-        columns_cfg = cfst_cfg.get("columns", {})
-        if columns_cfg and not isinstance(columns_cfg, dict):
-            raise ValueError("cfst_local.columns must be an object")
-        return
-
-    # source.type == "api" 校验
-    api_cfg = cfg.get("api")
-    if not isinstance(api_cfg, dict):
-        raise ValueError("api is required and must be an object when source.type=api")
-    api_url = str(api_cfg.get("url", "")).strip()
-    if not api_url:
-        raise ValueError("api.url is required when source.type=api")
-
-    parser_cfg = cfg.get("parser")
-    if parser_cfg is not None and not isinstance(parser_cfg, dict):
-        raise ValueError("parser must be an object")
-
-    selection_cfg = cfg.get("selection")
-    if selection_cfg is not None and not isinstance(selection_cfg, dict):
-        raise ValueError("selection must be an object")
-
-    domain_filter_cfg = cfg.get("domain_filter")
-    if domain_filter_cfg is not None and not isinstance(domain_filter_cfg, dict):
-        raise ValueError("domain_filter must be an object")
-
-    record_mapping = cfg.get("record_mapping")
-    if not isinstance(record_mapping, dict):
-        raise ValueError("record_mapping is required and must be an object")
-
-    records_path = str(record_mapping.get("records_path", "")).strip()
-    if not records_path:
-        raise ValueError("record_mapping.records_path is required")
-
-    field_map = record_mapping.get("field_map")
-    if not isinstance(field_map, dict) or not field_map:
-        raise ValueError("record_mapping.field_map is required and must be a non-empty object")
-
-    for key, path in field_map.items():
-        if not str(key).strip() or not str(path).strip():
-            raise ValueError("record_mapping.field_map contains empty field name or path")
-
-    for required in ["domain", "created_at"]:
-        if required not in field_map:
-            raise ValueError(f"record_mapping.field_map.{required} is required")
-
-    created_time_formats = record_mapping.get("created_time_formats")
-    if not isinstance(created_time_formats, list) or not created_time_formats:
-        raise ValueError("record_mapping.created_time_formats is required and must be a non-empty array")
-
-    for fmt in created_time_formats:
-        if not str(fmt).strip():
-            raise ValueError("record_mapping.created_time_formats contains empty format")
-
-    parse_timezone(record_mapping.get("created_time_timezone", "UTC"))
-
-    def ensure_field_registered(field_name, where):
-        if field_name not in field_map:
-            raise ValueError(f"{where}: field '{field_name}' is not in record_mapping.field_map")
-
-    record_filter = cfg.get("record_filter", {})
-    if record_filter.get("enabled", False):
-        rules = record_filter.get("exclude_if_any", [])
-        if not isinstance(rules, list):
-            raise ValueError("record_filter.exclude_if_any must be an array")
-
-        for i, rule in enumerate(rules):
-            if not isinstance(rule, dict):
-                raise ValueError(f"record_filter.exclude_if_any[{i}] must be an object")
-            field_name = str(rule.get("field", "")).strip()
-            if not field_name:
-                raise ValueError(f"record_filter.exclude_if_any[{i}].field is required")
-            ensure_field_registered(field_name, f"record_filter.exclude_if_any[{i}]")
-
-            has_matcher = any(k in rule for k in ["contains", "equals", "regex"])
-            if not has_matcher:
-                raise ValueError(f"record_filter.exclude_if_any[{i}] must include one of contains/equals/regex")
-
-    scoring = cfg.get("scoring", {})
-    if scoring.get("enabled", False):
-        strategy = str(scoring.get("strategy", "")).strip()
-        if strategy not in {"weighted_average", "lexicographic"}:
-            raise ValueError("scoring.strategy must be 'weighted_average' or 'lexicographic'")
-
-        within_hours = to_float_or_none(scoring.get("within_hours", 0))
-        if within_hours is None or within_hours < 0:
-            raise ValueError("scoring.within_hours must be a non-negative number (0 = disabled)")
-
-        if strategy == "weighted_average":
-            weighted_fields = scoring.get("weighted_fields")
-            if not isinstance(weighted_fields, list) or not weighted_fields:
-                raise ValueError("scoring.weighted_fields is required for weighted_average strategy")
-
-            for i, item in enumerate(weighted_fields):
-                if not isinstance(item, dict):
-                    raise ValueError(f"scoring.weighted_fields[{i}] must be an object")
-                field_name = str(item.get("field", "")).strip()
-                if not field_name:
-                    raise ValueError(f"scoring.weighted_fields[{i}].field is required")
-                ensure_field_registered(field_name, f"scoring.weighted_fields[{i}]")
-
-                weight = to_float_or_none(item.get("weight"))
-                if weight is None or weight <= 0:
-                    raise ValueError(f"scoring.weighted_fields[{i}].weight must be > 0")
-
-        if strategy == "lexicographic":
-            lex_fields = scoring.get("lexicographic_fields")
-            if not isinstance(lex_fields, list) or not lex_fields:
-                raise ValueError("scoring.lexicographic_fields is required for lexicographic strategy")
-
-            for i, item in enumerate(lex_fields):
-                if isinstance(item, str):
-                    field_name = item.strip()
-                    order = ""
-                elif isinstance(item, dict):
-                    field_name = str(item.get("field", "")).strip()
-                    order = str(item.get("order", "")).strip().lower()
-                else:
-                    raise ValueError(f"scoring.lexicographic_fields[{i}] must be string or object")
-
-                if not field_name:
-                    raise ValueError(f"scoring.lexicographic_fields[{i}] field is required")
-                ensure_field_registered(field_name, f"scoring.lexicographic_fields[{i}]")
-                if order and order not in {"asc", "desc"}:
-                    raise ValueError(f"scoring.lexicographic_fields[{i}].order must be asc or desc")
-
-        tie_breakers = scoring.get("tie_breakers", [])
-        if tie_breakers is not None:
-            if not isinstance(tie_breakers, list):
-                raise ValueError("scoring.tie_breakers must be an array")
-            for i, item in enumerate(tie_breakers):
-                if not isinstance(item, dict):
-                    raise ValueError(f"scoring.tie_breakers[{i}] must be an object")
-                field_name = str(item.get("field", "")).strip()
-                order = str(item.get("order", "")).strip().lower()
-                if not field_name:
-                    raise ValueError(f"scoring.tie_breakers[{i}].field is required")
-                if order not in {"asc", "desc"}:
-                    raise ValueError(f"scoring.tie_breakers[{i}].order must be asc or desc")
-                ensure_field_registered(field_name, f"scoring.tie_breakers[{i}]")
-
+"""domain_updater.py
 
-def rule_matches(value, rule):
-    if value is None or not isinstance(rule, dict):
-        return False
+Facade entry point for vmess-domain-rotator.
+Delegates implementation to the modular rotator package while maintaining 100% backward compatibility
+for command line arguments and external python callers.
+"""
 
-    values = flatten_values(value)
-    if not values:
-        values = [value]
-
-    case_sensitive = bool(rule.get("case_sensitive", False))
-
-    if "contains" in rule:
-        needle = str(rule.get("contains", ""))
-        if not needle:
-            return False
-        for item in values:
-            hay = str(item)
-            if case_sensitive:
-                if needle in hay:
-                    return True
-            else:
-                if needle.lower() in hay.lower():
-                    return True
-        return False
-
-    if "equals" in rule:
-        target = str(rule.get("equals", ""))
-        for item in values:
-            item_s = str(item)
-            if case_sensitive:
-                if item_s == target:
-                    return True
-            else:
-                if item_s.lower() == target.lower():
-                    return True
-        return False
-
-    if "regex" in rule:
-        pattern = str(rule.get("regex", ""))
-        if not pattern:
-            return False
-        flags = 0 if case_sensitive else re.IGNORECASE
-        try:
-            rx = re.compile(pattern, flags)
-        except Exception:
-            return False
-        for item in values:
-            if rx.search(str(item)):
-                return True
-        return False
-
-    return False
-
-
-def collect_excluded_domains(records, field_map, record_filter_cfg):
-    if not record_filter_cfg.get("enabled", False):
-        return set()
-
-    rules = record_filter_cfg.get("exclude_if_any", [])
-    if not rules:
-        return set()
-
-    blocked = set()
-    for record in records:
-        domain = normalize_domain(resolve_field(record, "domain", field_map))
-        if not domain:
-            continue
-
-        for rule in rules:
-            field_name = str(rule.get("field", "")).strip()
-            if not field_name:
-                continue
-            value = resolve_field(record, field_name, field_map)
-            if rule_matches(value, rule):
-                blocked.add(domain)
-                break
-
-    return blocked
-
-
-def build_lexicographic_descriptors(scoring_cfg, prefer_lower):
-    out = []
-    for item in scoring_cfg.get("lexicographic_fields", []):
-        if isinstance(item, str):
-            field_name = item.strip()
-            order = "asc" if prefer_lower else "desc"
-        else:
-            field_name = str(item.get("field", "")).strip()
-            order = str(item.get("order", "")).strip().lower()
-            if not order:
-                order = "asc" if prefer_lower else "desc"
-        out.append({"field": field_name, "order": order})
-    return out
-
-
-def parse_scored_records(records, field_map, record_mapping_cfg, scoring_cfg):
-    if not scoring_cfg.get("enabled", False):
-        return []
-
-    strategy = str(scoring_cfg.get("strategy", "weighted_average")).strip()
-    prefer_lower = bool(scoring_cfg.get("prefer_lower", False))
-
-    timezone = parse_timezone(record_mapping_cfg.get("created_time_timezone", "UTC"))
-    time_formats = [str(x) for x in record_mapping_cfg.get("created_time_formats", [])]
-
-    weighted_fields = scoring_cfg.get("weighted_fields", []) if strategy == "weighted_average" else []
-    lex_descriptors = build_lexicographic_descriptors(scoring_cfg, prefer_lower) if strategy == "lexicographic" else []
-
-    needed_fields = set()
-    for item in weighted_fields:
-        needed_fields.add(str(item.get("field", "")).strip())
-    for item in lex_descriptors:
-        needed_fields.add(str(item.get("field", "")).strip())
-    for item in scoring_cfg.get("tie_breakers", []):
-        needed_fields.add(str(item.get("field", "")).strip())
-    needed_fields.discard("domain")
-    needed_fields.discard("created_at")
-
-    out = []
-    for record in records:
-        domain = normalize_domain(resolve_field(record, "domain", field_map))
-        if not domain:
-            continue
-
-        created_raw = resolve_field(record, "created_at", field_map)
-        created_at = parse_created_time(created_raw, time_formats, timezone)
-
-        field_values = {}
-        for field_name in needed_fields:
-            field_values[field_name] = resolve_field(record, field_name, field_map)
-
-        score_value = None
-        scores = []
-        lex_values = []
-
-        if strategy == "weighted_average":
-            total = 0.0
-            total_weight = 0.0
-            missing = False
-            for item in weighted_fields:
-                field_name = str(item.get("field", "")).strip()
-                weight = float(item.get("weight"))
-                raw_v = resolve_field(record, field_name, field_map)
-                val = to_float_or_none(raw_v)
-                scores.append(val)
-                if val is None:
-                    missing = True
-                    continue
-                # 支持字段级单独指定极性（prefer_lower 或 order="desc"/"asc"）
-                field_lower = item.get("prefer_lower", prefer_lower)
-                if str(item.get("order", "")).lower() == "desc":
-                    field_lower = False
-                elif str(item.get("order", "")).lower() == "asc":
-                    field_lower = True
-                effective_val = val if (bool(field_lower) == prefer_lower) else -val
-                total += effective_val * weight
-                total_weight += weight
-
-            if not missing and total_weight > 0:
-                score_value = total / total_weight
-
-        if strategy == "lexicographic":
-            for item in lex_descriptors:
-                field_name = item["field"]
-                order = item["order"]
-                raw_v = resolve_field(record, field_name, field_map)
-                num_v = to_float_or_none(raw_v)
-                v = num_v if num_v is not None else raw_v
-                lex_values.append({"field": field_name, "value": v, "order": order})
-                scores.append(v)
-
-        out.append(
-            {
-                "domain": domain,
-                "created_at": created_at,
-                "created_raw": created_raw,
-                "scores": scores,
-                "score_value": score_value,
-                "lex_values": lex_values,
-                "field_values": field_values,
-                "raw_record": record,
-            }
-        )
-    return out
-
-
-def cmp_scalar(a, b, order):
-    a_none = a is None
-    b_none = b is None
-    if a_none and b_none:
-        return 0
-    if a_none:
-        return 1
-    if b_none:
-        return -1
-
-    if isinstance(a, dt.datetime):
-        a = a.timestamp()
-    if isinstance(b, dt.datetime):
-        b = b.timestamp()
-
-    a_num = to_float_or_none(a)
-    b_num = to_float_or_none(b)
-    if a_num is not None and b_num is not None:
-        if a_num < b_num:
-            base = -1
-        elif a_num > b_num:
-            base = 1
-        else:
-            base = 0
-    else:
-        a_s = str(a).lower()
-        b_s = str(b).lower()
-        if a_s < b_s:
-            base = -1
-        elif a_s > b_s:
-            base = 1
-        else:
-            base = 0
-
-    return base if order == "asc" else -base
-
-
-def get_sort_field_value(record, field_name):
-    if field_name == "domain":
-        return record.get("domain")
-    if field_name == "created_at":
-        return record.get("created_at")
-    return record.get("field_values", {}).get(field_name)
-
-
-def rank_scored_records(records, scoring_cfg):
-    if not records:
-        return []
-
-    within_hours = float(scoring_cfg.get("within_hours", 0))
-    strategy = str(scoring_cfg.get("strategy", "weighted_average")).strip()
-    prefer_lower = bool(scoring_cfg.get("prefer_lower", False))
-    tie_breakers = scoring_cfg.get("tie_breakers", [])
-
-    if within_hours > 0:
-        now = dt.datetime.now(dt.timezone.utc)
-        cutoff = now - dt.timedelta(hours=within_hours)
-        recent = [r for r in records if r.get("created_at") is not None and r["created_at"] >= cutoff]
-        candidates = recent if recent else records
-    else:
-        candidates = records
-
-    default_lex_order = "asc" if prefer_lower else "desc"
-
-    def compare(a, b):
-        if strategy == "weighted_average":
-            order = "asc" if prefer_lower else "desc"
-            c = cmp_scalar(a.get("score_value"), b.get("score_value"), order)
-            if c != 0:
-                return c
-        elif strategy == "lexicographic":
-            a_lex = a.get("lex_values", [])
-            b_lex = b.get("lex_values", [])
-            n = max(len(a_lex), len(b_lex))
-            for i in range(n):
-                av = a_lex[i]["value"] if i < len(a_lex) else None
-                bv = b_lex[i]["value"] if i < len(b_lex) else None
-                order = default_lex_order
-                if i < len(a_lex) and a_lex[i].get("order"):
-                    order = a_lex[i]["order"]
-                c = cmp_scalar(av, bv, order)
-                if c != 0:
-                    return c
-
-        for item in tie_breakers:
-            field_name = str(item.get("field", "")).strip()
-            order = str(item.get("order", "asc")).strip().lower()
-            av = get_sort_field_value(a, field_name)
-            bv = get_sort_field_value(b, field_name)
-            c = cmp_scalar(av, bv, order)
-            if c != 0:
-                return c
-
-        return cmp_scalar(a.get("domain"), b.get("domain"), "asc")
-
-    return sorted(candidates, key=functools.cmp_to_key(compare))
-
-
-def apply_filter(domains, filter_cfg):
-    include_suffixes = [s.lower() for s in filter_cfg.get("include_suffixes", []) if s]
-    exclude_regex = [re.compile(x) for x in filter_cfg.get("exclude_regex", []) if x]
-
-    out = []
-    for d in domains:
-        if include_suffixes and not any(d.endswith(s) for s in include_suffixes):
-            continue
-        if any(rx.search(d) for rx in exclude_regex):
-            continue
-        out.append(d)
-    return out
-
-
-def choose_top_candidate_domains(filtered_domains, top_n, ranked_scored):
-    if ranked_scored:
-        domains_by_score = [x["domain"] for x in ranked_scored]
-        return domains_by_score[:top_n]
-    return filtered_domains[:top_n]
-
-
-def text_or_blank(value):
-    if value is None:
-        return ""
-    return str(value).strip()
-
-
-def set_if_nonempty_text(obj, key, value):
-    text = text_or_blank(value)
-    if text:
-        obj[key] = text
-
-
-def base_top_candidate(domain, source_type):
-    candidate = {"domain": domain, "source_type": source_type}
-    if IPV4_RE.match(domain):
-        candidate["ip"] = domain
-    return candidate
-
-
-def maybe_resolve_field(record, field_name, field_map):
-    if not isinstance(record, dict):
-        return None
-    if field_name not in field_map:
-        return None
-    return resolve_field(record, field_name, field_map)
-
-
-def build_cfst_candidate(row):
-    domain = row.get("domain", "")
-    candidate = base_top_candidate(domain=domain, source_type="cfst_local")
-    set_if_nonempty_text(candidate, "ip", row.get("ip") or domain)
-    set_if_nonempty_text(candidate, "loss_rate", row.get("loss_rate"))
-    set_if_nonempty_text(candidate, "avg_latency", row.get("avg_latency"))
-    set_if_nonempty_text(candidate, "download_speed", row.get("download_speed"))
-    set_if_nonempty_text(candidate, "region", row.get("region"))
-    return candidate
-
-
-def build_api_candidate(domain, record, field_map, scored_record=None):
-    candidate = base_top_candidate(domain=domain, source_type="api")
-
-    if record:
-        set_if_nonempty_text(candidate, "created_raw", maybe_resolve_field(record, "created_at", field_map))
-        set_if_nonempty_text(candidate, "avg_latency", maybe_resolve_field(record, "avg_latency", field_map))
-        set_if_nonempty_text(candidate, "loss_rate", maybe_resolve_field(record, "avg_pkg_lost_rate", field_map))
-        set_if_nonempty_text(candidate, "download_speed", maybe_resolve_field(record, "download_speed", field_map))
-        location_country = text_or_blank(maybe_resolve_field(record, "location_country", field_map))
-        location_city = text_or_blank(maybe_resolve_field(record, "location_city", field_map))
-        region = "/".join([x for x in [location_country, location_city] if x])
-        set_if_nonempty_text(candidate, "region", region)
-
-    if scored_record:
-        if scored_record.get("score_value") is not None:
-            candidate["score_value"] = scored_record.get("score_value")
-        scores = list(scored_record.get("scores", []))
-        if scores:
-            candidate["scores"] = scores
-        if "created_raw" not in candidate:
-            set_if_nonempty_text(candidate, "created_raw", scored_record.get("created_raw"))
-
-    return candidate
-
-
-def build_top_candidates(source_type, candidate_domains, cfst_rows=None, records=None, field_map=None, ranked_scored=None):
-    if source_type == "cfst_local":
-        row_map = {}
-        for row in cfst_rows or []:
-            domain = row.get("domain", "")
-            if domain and domain not in row_map:
-                row_map[domain] = row
-        return [build_cfst_candidate(row_map.get(d, {"domain": d, "ip": d})) for d in candidate_domains]
-
-    ranked_map = {}
-    for item in ranked_scored or []:
-        domain = item.get("domain", "")
-        if domain and domain not in ranked_map:
-            ranked_map[domain] = item
-
-    record_map = {}
-    for record in records or []:
-        domain = normalize_domain(maybe_resolve_field(record, "domain", field_map or {}))
-        if domain and domain not in record_map:
-            record_map[domain] = record
-
-    out = []
-    for domain in candidate_domains:
-        scored_record = ranked_map.get(domain)
-        record = None
-        if scored_record:
-            record = scored_record.get("raw_record")
-        if record is None:
-            record = record_map.get(domain)
-        out.append(build_api_candidate(domain, record, field_map or {}, scored_record=scored_record))
-    return out
-
-
-def run_notify(cmd, domain, status, error=""):
-    if not cmd:
-        return
-    env = os.environ.copy()
-    env["AUTODOMAIN"] = str(domain or "")
-    env["AUTODOMAIN_STATUS"] = str(status or "")
-    env["AUTODOMAIN_ERROR"] = str(error or "")
-    try:
-        subprocess.run(
-            ["/bin/sh", "-c", cmd],
-            env=env,
-            stdout=sys.stderr,
-            stderr=sys.stderr,
-            timeout=15,
-            check=False,
-        )
-    except Exception as exc:
-        print(f"WARN: notify command execution failed: {exc}", file=sys.stderr)
-
-
-def choose_domain(filtered_domains, top_n, ranked_scored, last_good="", switch_margin_pct=0.0):
-    if ranked_scored:
-        top_domains = choose_top_candidate_domains(filtered_domains, top_n, ranked_scored)
-        if top_domains:
-            best_domain = top_domains[0]
-            # 迟滞防抖：当新最优候选相比当前仍在候选池中的 last_good 领先幅度不足 switch_margin_pct 时，保持 last_good
-            if switch_margin_pct > 0 and last_good and best_domain != last_good:
-                last_cand = next((r for r in ranked_scored if r.get("domain") == last_good), None)
-                best_cand = next((r for r in ranked_scored if r.get("domain") == best_domain), None)
-                if last_cand and best_cand:
-                    best_score = to_float_or_none(best_cand.get("score_value"))
-                    last_score = to_float_or_none(last_cand.get("score_value"))
-                    if best_score is not None and last_score is not None and last_score != 0:
-                        improvement_pct = abs((last_score - best_score) / last_score) * 100.0
-                        if improvement_pct < switch_margin_pct:
-                            top_reordered = [last_good] + [d for d in top_domains if d != last_good]
-                            return last_good, top_reordered[:top_n]
-            return best_domain, top_domains
-
-    if filtered_domains:
-        return filtered_domains[0], filtered_domains[:top_n]
-    return None, []
-
-
-def build_output_settings(output_cfg, config_path_abs):
-    # H3: legacy keys (current_domain_file/current_domain_json/substore_vars_file)
-    # were ambiguous and drifted between server/local configs. Refuse to start
-    # so operators migrate to the unified key set documented in CLAUDE.md.
-    legacy_key_map = {
-        "current_domain_file": "selected_value_file",
-        "current_domain_json": "selected_value_json",
-        "substore_vars_file": "export_vars_file",
-    }
-    legacy_in_use = [k for k in legacy_key_map if k in output_cfg]
-    if legacy_in_use:
-        renames = ", ".join(
-            "{} -> {}".format(k, legacy_key_map[k]) for k in legacy_in_use
-        )
-        raise ValueError(
-            "deprecated output keys in config: {}. Rename them: {}.".format(
-                ", ".join(legacy_in_use), renames
-            )
-        )
-
-    runtime_dir_cfg = output_cfg.get("runtime_dir", "./runtime")
-    runtime_dir = resolve_path(os.path.dirname(config_path_abs), runtime_dir_cfg)
-
-    selected_text_name = output_cfg.get("selected_value_file", "current_domain.txt")
-    selected_json_name = output_cfg.get("selected_value_json", "current_domain.json")
-    state_name = output_cfg.get("state_file", "state.json")
-    vars_name = output_cfg.get("export_vars_file", "substore_vars.json")
-
-    return {
-        "runtime_dir": runtime_dir,
-        "selected_text_path": os.path.join(runtime_dir, selected_text_name),
-        "selected_json_path": os.path.join(runtime_dir, selected_json_name),
-        "state_path": os.path.join(runtime_dir, state_name),
-        "vars_path": os.path.join(runtime_dir, vars_name),
-        "selected_json_key": str(output_cfg.get("selected_value_json_key", "domain")).strip() or "domain",
-        "state_last_good_key": str(output_cfg.get("state_last_good_key", "last_good_domain")).strip() or "last_good_domain",
-        "vars_value_key": str(output_cfg.get("substore_value_key", "AUTO_DOMAIN")).strip() or "AUTO_DOMAIN",
-    }
-
-
-def print_output_settings(config_path_abs, cfg):
-    output_cfg = cfg.get("output", {})
-    settings = build_output_settings(output_cfg, config_path_abs)
-    print(json.dumps(settings, ensure_ascii=True))
-
-
-def main():
-    ap = argparse.ArgumentParser(description="Auto select preferred endpoint value")
-    ap.add_argument("--config", default="config.server.json", help="Path to config JSON")
-    ap.add_argument(
-        "--print-output-settings",
-        action="store_true",
-        help="Print resolved output settings as JSON and exit",
-    )
-    ap.add_argument(
-        "--result-json-file",
-        default="",
-        help="Optional path to output machine-readable JSON result (decouples IPC from stdout)",
-    )
-    args = ap.parse_args()
-
-    config_path_abs = os.path.abspath(args.config)
-    if not os.path.exists(config_path_abs):
-        print(json.dumps({"status": "error", "error": f"config file not found: {config_path_abs}"}, ensure_ascii=True), file=sys.stderr)
-        sys.exit(1)
-
-    cfg = read_json_file(config_path_abs)
-
-    try:
-        validate_config(cfg)
-    except Exception as e:
-        print(json.dumps({"status": "error", "error": f"invalid config: {e}"}, ensure_ascii=True), file=sys.stderr)
-        sys.exit(1)
-
-    if args.print_output_settings:
-        print_output_settings(config_path_abs, cfg)
-        return
-
-    output_cfg = cfg.get("output", {})
-    output_settings = build_output_settings(output_cfg, config_path_abs)
-    notify_cfg = cfg.get("notify", {})
-    selected_text_file = output_settings["selected_text_path"]
-    selected_json_file = output_settings["selected_json_path"]
-    state_file = output_settings["state_path"]
-    vars_file = output_settings["vars_path"]
-    selected_json_key = output_settings["selected_json_key"]
-    state_last_good_key = output_settings["state_last_good_key"]
-    vars_value_key = output_settings["vars_value_key"]
-
-    state = read_json_file(state_file, default={})
-    last_good = state.get(state_last_good_key, "")
-    source_type = get_source_type(cfg)
-
-    try:
-        top_n = int(cfg.get("selection", {}).get("top_n", 3))
-        payload = None
-
-        if source_type == "cfst_local":
-            cfst_rows = load_cfst_rows(cfg, config_path_abs)
-            parsed = [row["domain"] for row in cfst_rows]
-            filtered = apply_filter(parsed, cfg.get("domain_filter", {}))
-            filtered_set = set(filtered)
-            cfst_rows = [row for row in cfst_rows if row["domain"] in filtered_set]
-            if not cfst_rows:
-                raise RuntimeError("No valid IP available from cfst result after filtering")
-
-            # 若未显式禁用下载测速（未配置 -dd），优先过滤掉测速失败的 0.00 下载速度记录
-            cfst_cfg = cfg.get("cfst_local", {})
-            disable_download = "-dd" in [str(a).strip() for a in cfst_cfg.get("run_args", [])]
-            valid_speed_rows = [
-                row for row in cfst_rows
-                if to_float_or_none(row.get("download_speed")) is not None
-                and float(to_float_or_none(row.get("download_speed"))) > 0.0
-            ]
-            if not disable_download and not valid_speed_rows and last_good:
-                selected = None
-                candidate_domains = []
-                top_candidates = []
-            else:
-                eligible_rows = valid_speed_rows if (not disable_download and valid_speed_rows) else cfst_rows
-                selected = eligible_rows[0]["domain"]
-                candidate_domains = [row["domain"] for row in eligible_rows[:top_n]]
-                top_candidates = build_top_candidates("cfst_local", candidate_domains, cfst_rows=eligible_rows)
-        else:
-            payload = fetch_api_json(cfg)
-            parsed = parse_domains(payload, cfg.get("parser", {}))
-            filtered = apply_filter(parsed, cfg.get("domain_filter", {}))
-
-            record_mapping_cfg = cfg.get("record_mapping", {})
-            field_map = record_mapping_cfg.get("field_map", {})
-            records = extract_records(payload, record_mapping_cfg)
-
-            record_filter_cfg = cfg.get("record_filter", {})
-            blocked_domains = collect_excluded_domains(records, field_map, record_filter_cfg)
-            if blocked_domains:
-                filtered = [d for d in filtered if d not in blocked_domains]
-
-            scoring_cfg = cfg.get("scoring", {})
-            scored_records = parse_scored_records(records, field_map, record_mapping_cfg, scoring_cfg)
-            filtered_set = set(filtered)
-            scored_records = [r for r in scored_records if r["domain"] in filtered_set]
-            switch_margin_pct = to_float_or_none(scoring_cfg.get("switch_margin_pct")) or 0.0
-            selected, candidate_domains = choose_domain(
-                filtered, top_n, ranked_scored, last_good=last_good, switch_margin_pct=switch_margin_pct
-            )
-            top_candidates = build_top_candidates(
-                "api",
-                candidate_domains,
-                records=records,
-                field_map=field_map,
-                ranked_scored=ranked_scored,
-            )
-
-        status = "ok"
-        if not selected and last_good:
-            # 复检 last_good: 确保已被拉黑的故障节点不会借回退复活
-            if source_type == "api" and apply_filter([last_good], cfg.get("domain_filter", {})) != [last_good]:
-                last_good = ""
-            else:
-                selected = last_good
-                status = "fallback_last_good"
-        if not selected:
-            if source_type == "cfst_local":
-                raise RuntimeError("No valid IP available from cfst and no fallback in state")
-            raise RuntimeError("No valid domain available from API and no fallback in state")
-
-        write_text_file(selected_text_file, selected + "\n")
-
-        current_json = {
-            selected_json_key: selected,
-            "updated_at": utc_now_iso(),
-            "status": status,
-            "source_type": source_type,
-            "source_count": len(parsed),
-            "top_candidates": top_candidates,
-        }
-        write_json_file(selected_json_file, current_json)
-        write_json_file(
-            vars_file,
-            {
-                vars_value_key: selected,
-                "UPDATED_AT": current_json["updated_at"],
-                "STATUS": status,
-            },
-        )
-
-        new_state = {
-            "updated_at": current_json["updated_at"],
-            state_last_good_key: selected,
-            "status": status,
-            "source_count": len(parsed),
-            "source_type": source_type,
-        }
-        write_json_file(state_file, new_state)
-
-        run_notify(notify_cfg.get("command", ""), selected, status)
-        if args.result_json_file:
-            write_json_file(args.result_json_file, current_json)
-        print(json.dumps(current_json, ensure_ascii=True))
-
-    except Exception as e:
-        now = utc_now_iso()
-        err_state = {
-            "updated_at": now,
-            "status": "error",
-            "error": str(e),
-            state_last_good_key: last_good,
-            "source_type": source_type,
-        }
-        write_json_file(state_file, err_state)
-
-        if last_good and source_type == "api":
-            if apply_filter([last_good], cfg.get("domain_filter", {})) != [last_good]:
-                last_good = ""
-
-        if last_good:
-            write_text_file(selected_text_file, last_good + "\n")
-            write_json_file(
-                selected_json_file,
-                {
-                    selected_json_key: last_good,
-                    "updated_at": now,
-                    "status": "error_use_last_good",
-                    "error": str(e),
-                    "source_type": source_type,
-                },
-            )
-            write_json_file(
-                vars_file,
-                {
-                    vars_value_key: last_good,
-                    "UPDATED_AT": now,
-                    "STATUS": "error_use_last_good",
-                },
-            )
-            fallback_res = {
-                selected_json_key: last_good,
-                "status": "error_use_last_good",
-                "error": str(e),
-            }
-            run_notify(notify_cfg.get("command", ""), last_good, "error_use_last_good")
-            if args.result_json_file:
-                write_json_file(args.result_json_file, fallback_res)
-            print(json.dumps(fallback_res, ensure_ascii=True))
-            return
-
-        err_res = {"status": "error", "error": str(e)}
-        run_notify(notify_cfg.get("command", ""), "", "error")
-        if args.result_json_file:
-            write_json_file(args.result_json_file, err_res)
-        print(json.dumps(err_res, ensure_ascii=True), file=sys.stderr)
-        sys.exit(1)
+import sys
+import os
 
+# 确保可以直接导入同目录下的 rotator 包
+SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
+if SCRIPT_DIR not in sys.path:
+    sys.path.insert(0, SCRIPT_DIR)
+
+from rotator.config import (
+    build_output_settings,
+    build_url,
+    get_source_type,
+    parse_timezone,
+    print_output_settings,
+    resolve_path,
+    to_float_or_none,
+    validate_config,
+)
+from rotator.filter import (
+    DOMAIN_RE,
+    IPV4_RE,
+    apply_filter,
+    collect_excluded_domains,
+    flatten_values,
+    get_by_json_path,
+    get_values_by_path,
+    normalize_domain,
+    resolve_field,
+    rule_matches,
+)
+from rotator.main import main, run_updater
+from rotator.notify import run_notify
+from rotator.scoring import (
+    base_top_candidate,
+    build_api_candidate,
+    build_cfst_candidate,
+    build_lexicographic_descriptors,
+    build_top_candidates,
+    choose_domain,
+    choose_top_candidate_domains,
+    cmp_scalar,
+    get_sort_field_value,
+    maybe_resolve_field,
+    parse_created_time,
+    parse_scored_records,
+    rank_scored_records,
+    set_if_nonempty_text,
+    text_or_blank,
+)
+from rotator.sources import (
+    extract_records,
+    fetch_api_json,
+    load_cfst_rows,
+    parse_domains,
+)
+from rotator.storage import (
+    _atomic_write,
+    read_json_file,
+    utc_now_iso,
+    write_json_file,
+    write_text_file,
+)
+
+__all__ = [
+    "DOMAIN_RE",
+    "IPV4_RE",
+    "_atomic_write",
+    "apply_filter",
+    "base_top_candidate",
+    "build_api_candidate",
+    "build_cfst_candidate",
+    "build_lexicographic_descriptors",
+    "build_output_settings",
+    "build_top_candidates",
+    "build_url",
+    "choose_domain",
+    "choose_top_candidate_domains",
+    "cmp_scalar",
+    "collect_excluded_domains",
+    "extract_records",
+    "fetch_api_json",
+    "flatten_values",
+    "get_by_json_path",
+    "get_sort_field_value",
+    "get_source_type",
+    "get_values_by_path",
+    "load_cfst_rows",
+    "main",
+    "maybe_resolve_field",
+    "normalize_domain",
+    "parse_created_time",
+    "parse_domains",
+    "parse_scored_records",
+    "parse_timezone",
+    "print_output_settings",
+    "rank_scored_records",
+    "read_json_file",
+    "resolve_field",
+    "resolve_path",
+    "rule_matches",
+    "run_notify",
+    "run_updater",
+    "set_if_nonempty_text",
+    "text_or_blank",
+    "to_float_or_none",
+    "utc_now_iso",
+    "validate_config",
+    "write_json_file",
+    "write_text_file",
+]
 
 if __name__ == "__main__":
     main()

scripts/rotator/__init__.py

@@ -0,0 +1,6 @@
+"""vmess-domain-rotator package
+
+Core domain rotation and CFST endpoint selector modules.
+"""
+
+__version__ = "2.0.0"

scripts/rotator/config.py

@@ -0,0 +1,281 @@
+import datetime as dt
+import json
+import math
+import os
+import re
+import urllib.parse
+
+
+def build_url(base_url, params):
+    if not params:
+        return base_url
+    parsed = urllib.parse.urlparse(base_url)
+    current = urllib.parse.parse_qs(parsed.query)
+    for k, v in params.items():
+        current[k] = [str(v)]
+    query = urllib.parse.urlencode(current, doseq=True)
+    return urllib.parse.urlunparse(parsed._replace(query=query))
+
+
+def resolve_path(base_dir, path_value):
+    path_text = str(path_value or "").strip()
+    if not path_text:
+        return ""
+    if os.path.isabs(path_text):
+        return os.path.normpath(path_text)
+    return os.path.normpath(os.path.join(base_dir, path_text))
+
+
+def get_source_type(cfg):
+    source_cfg = cfg.get("source", {})
+    if isinstance(source_cfg, dict):
+        source_type = str(source_cfg.get("type", "api")).strip().lower()
+        if source_type:
+            return source_type
+    return "api"
+
+
+def parse_timezone(tz_raw):
+    if tz_raw is None:
+        return dt.timezone.utc
+    s = str(tz_raw).strip().upper()
+    if s in {"", "UTC", "Z", "+00:00", "+0000"}:
+        return dt.timezone.utc
+
+    m = re.match(r"^([+-])(\d{2}):?(\d{2})$", s)
+    if not m:
+        raise ValueError(f"invalid created_time_timezone: {tz_raw}")
+
+    sign = 1 if m.group(1) == "+" else -1
+    hh = int(m.group(2))
+    mm = int(m.group(3))
+    if hh > 23 or mm > 59:
+        raise ValueError(f"invalid created_time_timezone offset: {tz_raw}")
+    return dt.timezone(sign * dt.timedelta(hours=hh, minutes=mm))
+
+
+def to_float_or_none(value):
+    try:
+        f = float(value)
+        if not math.isfinite(f):
+            return None
+        return f
+    except Exception:
+        return None
+
+
+def build_output_settings(output_cfg, config_path_abs):
+    # legacy keys (current_domain_file/current_domain_json/substore_vars_file)
+    # were ambiguous and drifted between server/local configs. Refuse to start
+    # so operators migrate to the unified key set documented in CLAUDE.md.
+    legacy_key_map = {
+        "current_domain_file": "selected_value_file",
+        "current_domain_json": "selected_value_json",
+        "substore_vars_file": "export_vars_file",
+    }
+    legacy_in_use = [k for k in legacy_key_map if k in output_cfg]
+    if legacy_in_use:
+        renames = ", ".join(
+            "{} -> {}".format(k, legacy_key_map[k]) for k in legacy_in_use
+        )
+        raise ValueError(
+            "deprecated output keys in config: {}. Rename them: {}.".format(
+                ", ".join(legacy_in_use), renames
+            )
+        )
+
+    runtime_dir_cfg = output_cfg.get("runtime_dir", "./runtime")
+    runtime_dir = resolve_path(os.path.dirname(config_path_abs), runtime_dir_cfg)
+
+    selected_text_name = output_cfg.get("selected_value_file", "current_domain.txt")
+    selected_json_name = output_cfg.get("selected_value_json", "current_domain.json")
+    state_name = output_cfg.get("state_file", "state.json")
+    vars_name = output_cfg.get("export_vars_file", "substore_vars.json")
+
+    return {
+        "runtime_dir": runtime_dir,
+        "selected_text_path": os.path.join(runtime_dir, selected_text_name),
+        "selected_json_path": os.path.join(runtime_dir, selected_json_name),
+        "state_path": os.path.join(runtime_dir, state_name),
+        "vars_path": os.path.join(runtime_dir, vars_name),
+        "selected_json_key": str(output_cfg.get("selected_value_json_key", "domain")).strip() or "domain",
+        "state_last_good_key": str(output_cfg.get("state_last_good_key", "last_good_domain")).strip() or "last_good_domain",
+        "vars_value_key": str(output_cfg.get("substore_value_key", "AUTO_DOMAIN")).strip() or "AUTO_DOMAIN",
+    }
+
+
+def print_output_settings(config_path_abs, cfg):
+    output_cfg = cfg.get("output", {})
+    settings = build_output_settings(output_cfg, config_path_abs)
+    print(json.dumps(settings, ensure_ascii=True))
+
+
+def validate_config(cfg):
+    source_type = get_source_type(cfg)
+    if source_type not in {"api", "cfst_local"}:
+        raise ValueError("source.type must be 'api' or 'cfst_local'")
+
+    output_cfg = cfg.get("output", {})
+    if output_cfg and not isinstance(output_cfg, dict):
+        raise ValueError("output must be an object")
+
+    if source_type == "cfst_local":
+        cfst_cfg = cfg.get("cfst_local")
+        if not isinstance(cfst_cfg, dict):
+            raise ValueError("cfst_local is required and must be an object when source.type=cfst_local")
+
+        work_dir = str(cfst_cfg.get("work_dir", "")).strip()
+        if not work_dir:
+            raise ValueError("cfst_local.work_dir is required")
+
+        binary = str(cfst_cfg.get("binary", "")).strip()
+        if not binary:
+            raise ValueError("cfst_local.binary is required")
+
+        result_file = str(cfst_cfg.get("result_file", "")).strip()
+        if not result_file:
+            raise ValueError("cfst_local.result_file is required")
+
+        run_args = cfst_cfg.get("run_args", [])
+        if not isinstance(run_args, list):
+            raise ValueError("cfst_local.run_args must be an array")
+
+        columns_cfg = cfst_cfg.get("columns", {})
+        if columns_cfg and not isinstance(columns_cfg, dict):
+            raise ValueError("cfst_local.columns must be an object")
+        return
+
+    # source.type == "api" 校验
+    api_cfg = cfg.get("api")
+    if not isinstance(api_cfg, dict):
+        raise ValueError("api is required and must be an object when source.type=api")
+    api_url = str(api_cfg.get("url", "")).strip()
+    if not api_url:
+        raise ValueError("api.url is required when source.type=api")
+
+    parser_cfg = cfg.get("parser")
+    if parser_cfg is not None and not isinstance(parser_cfg, dict):
+        raise ValueError("parser must be an object")
+
+    selection_cfg = cfg.get("selection")
+    if selection_cfg is not None and not isinstance(selection_cfg, dict):
+        raise ValueError("selection must be an object")
+
+    domain_filter_cfg = cfg.get("domain_filter")
+    if domain_filter_cfg is not None and not isinstance(domain_filter_cfg, dict):
+        raise ValueError("domain_filter must be an object")
+
+    record_mapping = cfg.get("record_mapping")
+    if not isinstance(record_mapping, dict):
+        raise ValueError("record_mapping is required and must be an object")
+
+    records_path = str(record_mapping.get("records_path", "")).strip()
+    if not records_path:
+        raise ValueError("record_mapping.records_path is required")
+
+    field_map = record_mapping.get("field_map")
+    if not isinstance(field_map, dict) or not field_map:
+        raise ValueError("record_mapping.field_map is required and must be a non-empty object")
+
+    for key, path in field_map.items():
+        if not str(key).strip() or not str(path).strip():
+            raise ValueError("record_mapping.field_map contains empty field name or path")
+
+    for required in ["domain", "created_at"]:
+        if required not in field_map:
+            raise ValueError(f"record_mapping.field_map.{required} is required")
+
+    created_time_formats = record_mapping.get("created_time_formats")
+    if not isinstance(created_time_formats, list) or not created_time_formats:
+        raise ValueError("record_mapping.created_time_formats is required and must be a non-empty array")
+
+    for fmt in created_time_formats:
+        if not str(fmt).strip():
+            raise ValueError("record_mapping.created_time_formats contains empty format")
+
+    parse_timezone(record_mapping.get("created_time_timezone", "UTC"))
+
+    def ensure_field_registered(field_name, where):
+        if field_name not in field_map:
+            raise ValueError(f"{where}: field '{field_name}' is not in record_mapping.field_map")
+
+    record_filter = cfg.get("record_filter", {})
+    if record_filter.get("enabled", False):
+        rules = record_filter.get("exclude_if_any", [])
+        if not isinstance(rules, list):
+            raise ValueError("record_filter.exclude_if_any must be an array")
+
+        for i, rule in enumerate(rules):
+            if not isinstance(rule, dict):
+                raise ValueError(f"record_filter.exclude_if_any[{i}] must be an object")
+            field_name = str(rule.get("field", "")).strip()
+            if not field_name:
+                raise ValueError(f"record_filter.exclude_if_any[{i}].field is required")
+            ensure_field_registered(field_name, f"record_filter.exclude_if_any[{i}]")
+
+            has_matcher = any(k in rule for k in ["contains", "equals", "regex"])
+            if not has_matcher:
+                raise ValueError(f"record_filter.exclude_if_any[{i}] must include one of contains/equals/regex")
+
+    scoring = cfg.get("scoring", {})
+    if scoring.get("enabled", False):
+        strategy = str(scoring.get("strategy", "")).strip()
+        if strategy not in {"weighted_average", "lexicographic"}:
+            raise ValueError("scoring.strategy must be 'weighted_average' or 'lexicographic'")
+
+        within_hours = to_float_or_none(scoring.get("within_hours", 0))
+        if within_hours is None or within_hours < 0:
+            raise ValueError("scoring.within_hours must be a non-negative number (0 = disabled)")
+
+        if strategy == "weighted_average":
+            weighted_fields = scoring.get("weighted_fields")
+            if not isinstance(weighted_fields, list) or not weighted_fields:
+                raise ValueError("scoring.weighted_fields is required for weighted_average strategy")
+
+            for i, item in enumerate(weighted_fields):
+                if not isinstance(item, dict):
+                    raise ValueError(f"scoring.weighted_fields[{i}] must be an object")
+                field_name = str(item.get("field", "")).strip()
+                if not field_name:
+                    raise ValueError(f"scoring.weighted_fields[{i}].field is required")
+                ensure_field_registered(field_name, f"scoring.weighted_fields[{i}]")
+
+                weight = to_float_or_none(item.get("weight"))
+                if weight is None or weight <= 0:
+                    raise ValueError(f"scoring.weighted_fields[{i}].weight must be > 0")
+
+        if strategy == "lexicographic":
+            lex_fields = scoring.get("lexicographic_fields")
+            if not isinstance(lex_fields, list) or not lex_fields:
+                raise ValueError("scoring.lexicographic_fields is required for lexicographic strategy")
+
+            for i, item in enumerate(lex_fields):
+                if isinstance(item, str):
+                    field_name = item.strip()
+                    order = ""
+                elif isinstance(item, dict):
+                    field_name = str(item.get("field", "")).strip()
+                    order = str(item.get("order", "")).strip().lower()
+                else:
+                    raise ValueError(f"scoring.lexicographic_fields[{i}] must be string or object")
+
+                if not field_name:
+                    raise ValueError(f"scoring.lexicographic_fields[{i}] field is required")
+                ensure_field_registered(field_name, f"scoring.lexicographic_fields[{i}]")
+                if order and order not in {"asc", "desc"}:
+                    raise ValueError(f"scoring.lexicographic_fields[{i}].order must be asc or desc")
+
+        tie_breakers = scoring.get("tie_breakers", [])
+        if tie_breakers is not None:
+            if not isinstance(tie_breakers, list):
+                raise ValueError("scoring.tie_breakers must be an array")
+            for i, item in enumerate(tie_breakers):
+                if not isinstance(item, dict):
+                    raise ValueError(f"scoring.tie_breakers[{i}] must be an object")
+                field_name = str(item.get("field", "")).strip()
+                order = str(item.get("order", "")).strip().lower()
+                if not field_name:
+                    raise ValueError(f"scoring.tie_breakers[{i}].field is required")
+                if order not in {"asc", "desc"}:
+                    raise ValueError(f"scoring.tie_breakers[{i}].order must be asc or desc")
+                ensure_field_registered(field_name, f"scoring.tie_breakers[{i}]")

scripts/rotator/filter.py

@@ -0,0 +1,164 @@
+import re
+
+DOMAIN_RE = re.compile(r"^(?=.{1,253}$)(?!-)[A-Za-z0-9-]{1,63}(?<!-)(\.(?!-)[A-Za-z0-9-]{1,63}(?<!-))+$")
+IPV4_RE = re.compile(r"^(?:25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(?:\.(?:25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}$")
+
+
+def normalize_domain(value):
+    if value is None:
+        return ""
+    return str(value).strip().lower().rstrip(".")
+
+
+def flatten_values(value):
+    out = []
+    if isinstance(value, str):
+        out.append(value)
+    elif isinstance(value, list):
+        for item in value:
+            out.extend(flatten_values(item))
+    elif isinstance(value, dict):
+        for item in value.values():
+            out.extend(flatten_values(item))
+    return out
+
+
+def get_by_json_path(data, path):
+    cur = data
+    for part in path.split("."):
+        if isinstance(cur, dict) and part in cur:
+            cur = cur[part]
+        else:
+            return None
+    return cur
+
+
+def get_values_by_path(data, path):
+    parts = path.split(".")
+
+    def walk(cur, idx):
+        if idx >= len(parts):
+            return [cur]
+
+        part = parts[idx]
+        if part.endswith("[]"):
+            key = part[:-2]
+            if isinstance(cur, dict):
+                arr = cur.get(key)
+            else:
+                arr = None
+            if not isinstance(arr, list):
+                return []
+
+            out = []
+            for item in arr:
+                out.extend(walk(item, idx + 1))
+            return out
+
+        if isinstance(cur, dict) and part in cur:
+            return walk(cur[part], idx + 1)
+        return []
+
+    return walk(data, 0)
+
+
+def resolve_field(record, field_name, field_map):
+    path = field_map.get(field_name)
+    if not path:
+        raise ValueError(f"field '{field_name}' is not registered in record_mapping.field_map")
+    if not isinstance(record, dict):
+        return None
+    return get_by_json_path(record, path)
+
+
+def rule_matches(value, rule):
+    if value is None or not isinstance(rule, dict):
+        return False
+
+    values = flatten_values(value)
+    if not values:
+        values = [value]
+
+    case_sensitive = bool(rule.get("case_sensitive", False))
+
+    if "contains" in rule:
+        needle = str(rule.get("contains", ""))
+        if not needle:
+            return False
+        for item in values:
+            hay = str(item)
+            if case_sensitive:
+                if needle in hay:
+                    return True
+            else:
+                if needle.lower() in hay.lower():
+                    return True
+        return False
+
+    if "equals" in rule:
+        target = str(rule.get("equals", ""))
+        for item in values:
+            item_s = str(item)
+            if case_sensitive:
+                if item_s == target:
+                    return True
+            else:
+                if item_s.lower() == target.lower():
+                    return True
+        return False
+
+    if "regex" in rule:
+        pattern = str(rule.get("regex", ""))
+        if not pattern:
+            return False
+        flags = 0 if case_sensitive else re.IGNORECASE
+        try:
+            rx = re.compile(pattern, flags)
+        except Exception:
+            return False
+        for item in values:
+            if rx.search(str(item)):
+                return True
+        return False
+
+    return False
+
+
+def collect_excluded_domains(records, field_map, record_filter_cfg):
+    if not record_filter_cfg.get("enabled", False):
+        return set()
+
+    rules = record_filter_cfg.get("exclude_if_any", [])
+    if not rules:
+        return set()
+
+    blocked = set()
+    for record in records:
+        domain = normalize_domain(resolve_field(record, "domain", field_map))
+        if not domain:
+            continue
+
+        for rule in rules:
+            field_name = str(rule.get("field", "")).strip()
+            if not field_name:
+                continue
+            value = resolve_field(record, field_name, field_map)
+            if rule_matches(value, rule):
+                blocked.add(domain)
+                break
+
+    return blocked
+
+
+def apply_filter(domains, filter_cfg):
+    include_suffixes = [s.lower() for s in filter_cfg.get("include_suffixes", []) if s]
+    exclude_regex = [re.compile(x) for x in filter_cfg.get("exclude_regex", []) if x]
+
+    out = []
+    for d in domains:
+        if include_suffixes and not any(d.endswith(s) for s in include_suffixes):
+            continue
+        if any(rx.search(d) for rx in exclude_regex):
+            continue
+        out.append(d)
+    return out

scripts/rotator/main.py

@@ -0,0 +1,253 @@
+import argparse
+import json
+import os
+import sys
+
+from .config import (
+    build_output_settings,
+    get_source_type,
+    print_output_settings,
+    to_float_or_none,
+    validate_config,
+)
+from .filter import apply_filter, collect_excluded_domains
+from .notify import run_notify
+from .scoring import (
+    build_top_candidates,
+    choose_domain,
+    parse_scored_records,
+    rank_scored_records,
+)
+from .sources import extract_records, fetch_api_json, load_cfst_rows, parse_domains
+from .storage import (
+    read_json_file,
+    utc_now_iso,
+    write_json_file,
+    write_text_file,
+)
+
+
+def run_updater(config_path_abs, print_settings_only=False, result_json_file=""):
+    """Execute main rotation workflow with strict error handling and state persistence."""
+    if not os.path.exists(config_path_abs):
+        err_msg = {"status": "error", "error": f"config file not found: {config_path_abs}"}
+        print(json.dumps(err_msg, ensure_ascii=True), file=sys.stderr)
+        return 1
+
+    cfg = read_json_file(config_path_abs)
+
+    try:
+        validate_config(cfg)
+    except Exception as e:
+        err_msg = {"status": "error", "error": f"invalid config: {e}"}
+        print(json.dumps(err_msg, ensure_ascii=True), file=sys.stderr)
+        return 1
+
+    if print_settings_only:
+        print_output_settings(config_path_abs, cfg)
+        return 0
+
+    output_cfg = cfg.get("output", {})
+    output_settings = build_output_settings(output_cfg, config_path_abs)
+    notify_cfg = cfg.get("notify", {})
+    selected_text_file = output_settings["selected_text_path"]
+    selected_json_file = output_settings["selected_json_path"]
+    state_file = output_settings["state_path"]
+    vars_file = output_settings["vars_path"]
+    selected_json_key = output_settings["selected_json_key"]
+    state_last_good_key = output_settings["state_last_good_key"]
+    vars_value_key = output_settings["vars_value_key"]
+
+    state = read_json_file(state_file, default={})
+    last_good = state.get(state_last_good_key, "")
+    source_type = get_source_type(cfg)
+
+    try:
+        top_n = int(cfg.get("selection", {}).get("top_n", 3))
+
+        if source_type == "cfst_local":
+            cfst_rows = load_cfst_rows(cfg, config_path_abs)
+            parsed = [row["domain"] for row in cfst_rows]
+            filtered = apply_filter(parsed, cfg.get("domain_filter", {}))
+            filtered_set = set(filtered)
+            cfst_rows = [row for row in cfst_rows if row["domain"] in filtered_set]
+            if not cfst_rows:
+                raise RuntimeError("No valid IP available from cfst result after filtering")
+
+            # 若未显式禁用下载测速（未配置 -dd），优先过滤掉测速失败的 0.00 下载速度记录
+            cfst_cfg = cfg.get("cfst_local", {})
+            disable_download = "-dd" in [str(a).strip() for a in cfst_cfg.get("run_args", [])]
+            valid_speed_rows = [
+                row for row in cfst_rows
+                if to_float_or_none(row.get("download_speed")) is not None
+                and float(to_float_or_none(row.get("download_speed"))) > 0.0
+            ]
+            if not disable_download and not valid_speed_rows and last_good:
+                selected = None
+                candidate_domains = []
+                top_candidates = []
+            else:
+                eligible_rows = valid_speed_rows if (not disable_download and valid_speed_rows) else cfst_rows
+                selected = eligible_rows[0]["domain"]
+                candidate_domains = [row["domain"] for row in eligible_rows[:top_n]]
+                top_candidates = build_top_candidates("cfst_local", candidate_domains, cfst_rows=eligible_rows)
+        else:
+            payload = fetch_api_json(cfg)
+            parsed = parse_domains(payload, cfg.get("parser", {}))
+            filtered = apply_filter(parsed, cfg.get("domain_filter", {}))
+
+            record_mapping_cfg = cfg.get("record_mapping", {})
+            field_map = record_mapping_cfg.get("field_map", {})
+            records = extract_records(payload, record_mapping_cfg)
+
+            record_filter_cfg = cfg.get("record_filter", {})
+            blocked_domains = collect_excluded_domains(records, field_map, record_filter_cfg)
+            if blocked_domains:
+                filtered = [d for d in filtered if d not in blocked_domains]
+
+            scoring_cfg = cfg.get("scoring", {})
+            scored_records = parse_scored_records(records, field_map, record_mapping_cfg, scoring_cfg)
+            filtered_set = set(filtered)
+            scored_records = [r for r in scored_records if r["domain"] in filtered_set]
+            ranked_scored = rank_scored_records(scored_records, scoring_cfg)
+            switch_margin_pct = to_float_or_none(scoring_cfg.get("switch_margin_pct")) or 0.0
+            selected, candidate_domains = choose_domain(
+                filtered, top_n, ranked_scored, last_good=last_good, switch_margin_pct=switch_margin_pct
+            )
+            top_candidates = build_top_candidates(
+                "api",
+                candidate_domains,
+                records=records,
+                field_map=field_map,
+                ranked_scored=ranked_scored,
+            )
+
+        status = "ok"
+        if not selected and last_good:
+            # 复检 last_good: 确保已被拉黑的故障节点不会借回退复活
+            if source_type == "api" and apply_filter([last_good], cfg.get("domain_filter", {})) != [last_good]:
+                last_good = ""
+            else:
+                selected = last_good
+                status = "fallback_last_good"
+        if not selected:
+            if source_type == "cfst_local":
+                raise RuntimeError("No valid IP available from cfst and no fallback in state")
+            raise RuntimeError("No valid domain available from API and no fallback in state")
+
+        write_text_file(selected_text_file, selected + "\n")
+
+        current_json = {
+            selected_json_key: selected,
+            "updated_at": utc_now_iso(),
+            "status": status,
+            "source_type": source_type,
+            "source_count": len(parsed),
+            "top_candidates": top_candidates,
+        }
+        write_json_file(selected_json_file, current_json)
+        write_json_file(
+            vars_file,
+            {
+                vars_value_key: selected,
+                "UPDATED_AT": current_json["updated_at"],
+                "STATUS": status,
+            },
+        )
+
+        new_state = {
+            "updated_at": current_json["updated_at"],
+            state_last_good_key: selected,
+            "status": status,
+            "source_count": len(parsed),
+            "source_type": source_type,
+        }
+        write_json_file(state_file, new_state)
+
+        run_notify(notify_cfg.get("command", ""), selected, status)
+        if result_json_file:
+            write_json_file(result_json_file, current_json)
+        print(json.dumps(current_json, ensure_ascii=True))
+        return 0
+
+    except Exception as e:
+        now = utc_now_iso()
+        err_state = {
+            "updated_at": now,
+            "status": "error",
+            "error": str(e),
+            state_last_good_key: last_good,
+            "source_type": source_type,
+        }
+        write_json_file(state_file, err_state)
+
+        if last_good and source_type == "api":
+            if apply_filter([last_good], cfg.get("domain_filter", {})) != [last_good]:
+                last_good = ""
+
+        if last_good:
+            write_text_file(selected_text_file, last_good + "\n")
+            write_json_file(
+                selected_json_file,
+                {
+                    selected_json_key: last_good,
+                    "updated_at": now,
+                    "status": "error_use_last_good",
+                    "error": str(e),
+                    "source_type": source_type,
+                },
+            )
+            write_json_file(
+                vars_file,
+                {
+                    vars_value_key: last_good,
+                    "UPDATED_AT": now,
+                    "STATUS": "error_use_last_good",
+                },
+            )
+            fallback_res = {
+                selected_json_key: last_good,
+                "status": "error_use_last_good",
+                "error": str(e),
+            }
+            run_notify(notify_cfg.get("command", ""), last_good, "error_use_last_good")
+            if result_json_file:
+                write_json_file(result_json_file, fallback_res)
+            print(json.dumps(fallback_res, ensure_ascii=True))
+            return 0
+
+        err_res = {"status": "error", "error": str(e)}
+        run_notify(notify_cfg.get("command", ""), "", "error")
+        if result_json_file:
+            write_json_file(result_json_file, err_res)
+        print(json.dumps(err_res, ensure_ascii=True), file=sys.stderr)
+        return 1
+
+
+def main():
+    ap = argparse.ArgumentParser(description="Auto select preferred endpoint value")
+    ap.add_argument("--config", default="config.server.json", help="Path to config JSON")
+    ap.add_argument(
+        "--print-output-settings",
+        action="store_true",
+        help="Print resolved output settings as JSON and exit",
+    )
+    ap.add_argument(
+        "--result-json-file",
+        default="",
+        help="Optional path to output machine-readable JSON result (decouples IPC from stdout)",
+    )
+    args = ap.parse_args()
+
+    config_path_abs = os.path.abspath(args.config)
+    rc = run_updater(
+        config_path_abs=config_path_abs,
+        print_settings_only=args.print_output_settings,
+        result_json_file=args.result_json_file,
+    )
+    if rc != 0:
+        sys.exit(rc)
+
+
+if __name__ == "__main__":
+    main()

scripts/rotator/notify.py

@@ -0,0 +1,24 @@
+import os
+import subprocess
+import sys
+
+
+def run_notify(cmd, domain, status, error=""):
+    """Run notification command with stdout/stderr redirected to stderr and timeout protection."""
+    if not cmd:
+        return
+    env = os.environ.copy()
+    env["AUTODOMAIN"] = str(domain or "")
+    env["AUTODOMAIN_STATUS"] = str(status or "")
+    env["AUTODOMAIN_ERROR"] = str(error or "")
+    try:
+        subprocess.run(
+            ["/bin/sh", "-c", cmd],
+            env=env,
+            stdout=sys.stderr,
+            stderr=sys.stderr,
+            timeout=15,
+            check=False,
+        )
+    except Exception as exc:
+        print(f"WARN: notify command execution failed: {exc}", file=sys.stderr)

scripts/rotator/scoring.py

@@ -0,0 +1,361 @@
+import datetime as dt
+import functools
+
+from .config import parse_timezone, to_float_or_none
+from .filter import IPV4_RE, normalize_domain, resolve_field
+
+
+def parse_created_time(value, formats, timezone):
+    if value is None:
+        return None
+
+    s = str(value).strip()
+    if not s:
+        return None
+
+    for fmt in formats:
+        try:
+            parsed = dt.datetime.strptime(s, fmt)
+            if parsed.tzinfo is None:
+                parsed = parsed.replace(tzinfo=timezone)
+            return parsed.astimezone(dt.timezone.utc)
+        except Exception:
+            continue
+
+    try:
+        iso_text = s.replace("Z", "+00:00")
+        parsed = dt.datetime.fromisoformat(iso_text)
+        if parsed.tzinfo is None:
+            parsed = parsed.replace(tzinfo=timezone)
+        return parsed.astimezone(dt.timezone.utc)
+    except Exception:
+        return None
+
+
+def build_lexicographic_descriptors(scoring_cfg, prefer_lower):
+    out = []
+    for item in scoring_cfg.get("lexicographic_fields", []):
+        if isinstance(item, str):
+            field_name = item.strip()
+            order = "asc" if prefer_lower else "desc"
+        else:
+            field_name = str(item.get("field", "")).strip()
+            order = str(item.get("order", "")).strip().lower()
+            if not order:
+                order = "asc" if prefer_lower else "desc"
+        out.append({"field": field_name, "order": order})
+    return out
+
+
+def parse_scored_records(records, field_map, record_mapping_cfg, scoring_cfg):
+    if not scoring_cfg.get("enabled", False):
+        return []
+
+    strategy = str(scoring_cfg.get("strategy", "weighted_average")).strip()
+    prefer_lower = bool(scoring_cfg.get("prefer_lower", False))
+
+    timezone = parse_timezone(record_mapping_cfg.get("created_time_timezone", "UTC"))
+    time_formats = [str(x) for x in record_mapping_cfg.get("created_time_formats", [])]
+
+    weighted_fields = scoring_cfg.get("weighted_fields", []) if strategy == "weighted_average" else []
+    lex_descriptors = build_lexicographic_descriptors(scoring_cfg, prefer_lower) if strategy == "lexicographic" else []
+
+    needed_fields = set()
+    for item in weighted_fields:
+        needed_fields.add(str(item.get("field", "")).strip())
+    for item in lex_descriptors:
+        needed_fields.add(str(item.get("field", "")).strip())
+    for item in scoring_cfg.get("tie_breakers", []):
+        needed_fields.add(str(item.get("field", "")).strip())
+    needed_fields.discard("domain")
+    needed_fields.discard("created_at")
+
+    out = []
+    for record in records:
+        domain = normalize_domain(resolve_field(record, "domain", field_map))
+        if not domain:
+            continue
+
+        created_raw = resolve_field(record, "created_at", field_map)
+        created_at = parse_created_time(created_raw, time_formats, timezone)
+
+        field_values = {}
+        for field_name in needed_fields:
+            field_values[field_name] = resolve_field(record, field_name, field_map)
+
+        score_value = None
+        scores = []
+        lex_values = []
+
+        if strategy == "weighted_average":
+            total = 0.0
+            total_weight = 0.0
+            missing = False
+            for item in weighted_fields:
+                field_name = str(item.get("field", "")).strip()
+                weight = float(item.get("weight"))
+                raw_v = resolve_field(record, field_name, field_map)
+                val = to_float_or_none(raw_v)
+                scores.append(val)
+                if val is None:
+                    missing = True
+                    continue
+                field_lower = item.get("prefer_lower", prefer_lower)
+                if str(item.get("order", "")).lower() == "desc":
+                    field_lower = False
+                elif str(item.get("order", "")).lower() == "asc":
+                    field_lower = True
+                effective_val = val if (bool(field_lower) == prefer_lower) else -val
+                total += effective_val * weight
+                total_weight += weight
+
+            if not missing and total_weight > 0:
+                score_value = total / total_weight
+
+        if strategy == "lexicographic":
+            for item in lex_descriptors:
+                field_name = item["field"]
+                order = item["order"]
+                raw_v = resolve_field(record, field_name, field_map)
+                num_v = to_float_or_none(raw_v)
+                v = num_v if num_v is not None else raw_v
+                lex_values.append({"field": field_name, "value": v, "order": order})
+                scores.append(v)
+
+        out.append(
+            {
+                "domain": domain,
+                "created_at": created_at,
+                "created_raw": created_raw,
+                "scores": scores,
+                "score_value": score_value,
+                "lex_values": lex_values,
+                "field_values": field_values,
+                "raw_record": record,
+            }
+        )
+    return out
+
+
+def cmp_scalar(a, b, order):
+    a_none = a is None
+    b_none = b is None
+    if a_none and b_none:
+        return 0
+    if a_none:
+        return 1
+    if b_none:
+        return -1
+
+    if isinstance(a, dt.datetime):
+        a = a.timestamp()
+    if isinstance(b, dt.datetime):
+        b = b.timestamp()
+
+    a_num = to_float_or_none(a)
+    b_num = to_float_or_none(b)
+    if a_num is not None and b_num is not None:
+        if a_num < b_num:
+            base = -1
+        elif a_num > b_num:
+            base = 1
+        else:
+            base = 0
+    else:
+        a_s = str(a).lower()
+        b_s = str(b).lower()
+        if a_s < b_s:
+            base = -1
+        elif a_s > b_s:
+            base = 1
+        else:
+            base = 0
+
+    return base if order == "asc" else -base
+
+
+def get_sort_field_value(record, field_name):
+    if field_name == "domain":
+        return record.get("domain")
+    if field_name == "created_at":
+        return record.get("created_at")
+    return record.get("field_values", {}).get(field_name)
+
+
+def rank_scored_records(records, scoring_cfg):
+    if not records:
+        return []
+
+    within_hours = float(scoring_cfg.get("within_hours", 0))
+    strategy = str(scoring_cfg.get("strategy", "weighted_average")).strip()
+    prefer_lower = bool(scoring_cfg.get("prefer_lower", False))
+    tie_breakers = scoring_cfg.get("tie_breakers", [])
+
+    if within_hours > 0:
+        now = dt.datetime.now(dt.timezone.utc)
+        cutoff = now - dt.timedelta(hours=within_hours)
+        recent = [r for r in records if r.get("created_at") is not None and r["created_at"] >= cutoff]
+        candidates = recent if recent else records
+    else:
+        candidates = records
+
+    default_lex_order = "asc" if prefer_lower else "desc"
+
+    def compare(a, b):
+        if strategy == "weighted_average":
+            order = "asc" if prefer_lower else "desc"
+            c = cmp_scalar(a.get("score_value"), b.get("score_value"), order)
+            if c != 0:
+                return c
+        elif strategy == "lexicographic":
+            a_lex = a.get("lex_values", [])
+            b_lex = b.get("lex_values", [])
+            n = max(len(a_lex), len(b_lex))
+            for i in range(n):
+                av = a_lex[i]["value"] if i < len(a_lex) else None
+                bv = b_lex[i]["value"] if i < len(b_lex) else None
+                order = default_lex_order
+                if i < len(a_lex) and a_lex[i].get("order"):
+                    order = a_lex[i]["order"]
+                c = cmp_scalar(av, bv, order)
+                if c != 0:
+                    return c
+
+        for item in tie_breakers:
+            field_name = str(item.get("field", "")).strip()
+            order = str(item.get("order", "asc")).strip().lower()
+            av = get_sort_field_value(a, field_name)
+            bv = get_sort_field_value(b, field_name)
+            c = cmp_scalar(av, bv, order)
+            if c != 0:
+                return c
+
+        return cmp_scalar(a.get("domain"), b.get("domain"), "asc")
+
+    return sorted(candidates, key=functools.cmp_to_key(compare))
+
+
+def choose_top_candidate_domains(filtered_domains, top_n, ranked_scored):
+    if ranked_scored:
+        domains_by_score = [x["domain"] for x in ranked_scored]
+        return domains_by_score[:top_n]
+    return filtered_domains[:top_n]
+
+
+def choose_domain(filtered_domains, top_n, ranked_scored, last_good="", switch_margin_pct=0.0):
+    if ranked_scored:
+        top_domains = choose_top_candidate_domains(filtered_domains, top_n, ranked_scored)
+        if top_domains:
+            best_domain = top_domains[0]
+            if switch_margin_pct > 0 and last_good and best_domain != last_good:
+                last_cand = next((r for r in ranked_scored if r.get("domain") == last_good), None)
+                best_cand = next((r for r in ranked_scored if r.get("domain") == best_domain), None)
+                if last_cand and best_cand:
+                    best_score = to_float_or_none(best_cand.get("score_value"))
+                    last_score = to_float_or_none(last_cand.get("score_value"))
+                    if best_score is not None and last_score is not None and last_score != 0:
+                        improvement_pct = abs((last_score - best_score) / last_score) * 100.0
+                        if improvement_pct < switch_margin_pct:
+                            top_reordered = [last_good] + [d for d in top_domains if d != last_good]
+                            return last_good, top_reordered[:top_n]
+            return best_domain, top_domains
+
+    if filtered_domains:
+        return filtered_domains[0], filtered_domains[:top_n]
+    return None, []
+
+
+def text_or_blank(value):
+    if value is None:
+        return ""
+    return str(value).strip()
+
+
+def set_if_nonempty_text(obj, key, value):
+    text = text_or_blank(value)
+    if text:
+        obj[key] = text
+
+
+def base_top_candidate(domain, source_type):
+    candidate = {"domain": domain, "source_type": source_type}
+    if IPV4_RE.match(domain):
+        candidate["ip"] = domain
+    return candidate
+
+
+def maybe_resolve_field(record, field_name, field_map):
+    if not isinstance(record, dict):
+        return None
+    if field_name not in field_map:
+        return None
+    return resolve_field(record, field_name, field_map)
+
+
+def build_cfst_candidate(row):
+    domain = row.get("domain", "")
+    candidate = base_top_candidate(domain=domain, source_type="cfst_local")
+    set_if_nonempty_text(candidate, "ip", row.get("ip") or domain)
+    set_if_nonempty_text(candidate, "loss_rate", row.get("loss_rate"))
+    set_if_nonempty_text(candidate, "avg_latency", row.get("avg_latency"))
+    set_if_nonempty_text(candidate, "download_speed", row.get("download_speed"))
+    set_if_nonempty_text(candidate, "region", row.get("region"))
+    return candidate
+
+
+def build_api_candidate(domain, record, field_map, scored_record=None):
+    candidate = base_top_candidate(domain=domain, source_type="api")
+
+    if record:
+        set_if_nonempty_text(candidate, "created_raw", maybe_resolve_field(record, "created_at", field_map))
+        set_if_nonempty_text(candidate, "avg_latency", maybe_resolve_field(record, "avg_latency", field_map))
+        set_if_nonempty_text(candidate, "loss_rate", maybe_resolve_field(record, "avg_pkg_lost_rate", field_map))
+        set_if_nonempty_text(candidate, "download_speed", maybe_resolve_field(record, "download_speed", field_map))
+        location_country = text_or_blank(maybe_resolve_field(record, "location_country", field_map))
+        location_city = text_or_blank(maybe_resolve_field(record, "location_city", field_map))
+        region = "/".join([x for x in [location_country, location_city] if x])
+        set_if_nonempty_text(candidate, "region", region)
+
+    if scored_record:
+        if scored_record.get("score_value") is not None:
+            candidate["score_value"] = scored_record.get("score_value")
+        scores = list(scored_record.get("scores", []))
+        if scores:
+            candidate["scores"] = scores
+        if "created_raw" not in candidate:
+            set_if_nonempty_text(candidate, "created_raw", scored_record.get("created_raw"))
+
+    return candidate
+
+
+def build_top_candidates(source_type, candidate_domains, cfst_rows=None, records=None, field_map=None, ranked_scored=None):
+    if source_type == "cfst_local":
+        row_map = {}
+        for row in cfst_rows or []:
+            domain = row.get("domain", "")
+            if domain and domain not in row_map:
+                row_map[domain] = row
+        return [build_cfst_candidate(row_map.get(d, {"domain": d, "ip": d})) for d in candidate_domains]
+
+    ranked_map = {}
+    for item in ranked_scored or []:
+        domain = item.get("domain", "")
+        if domain and domain not in ranked_map:
+            ranked_map[domain] = item
+
+    record_map = {}
+    for record in records or []:
+        domain = normalize_domain(maybe_resolve_field(record, "domain", field_map or {}))
+        if domain and domain not in record_map:
+            record_map[domain] = record
+
+    out = []
+    for domain in candidate_domains:
+        scored_record = ranked_map.get(domain)
+        record = None
+        if scored_record:
+            record = scored_record.get("raw_record")
+        if record is None:
+            record = record_map.get(domain)
+        out.append(build_api_candidate(domain, record, field_map or {}, scored_record=scored_record))
+    return out

scripts/rotator/sources.py

@@ -0,0 +1,151 @@
+import csv
+import json
+import os
+import re
+import subprocess
+import urllib.request
+
+from .config import build_url, resolve_path
+from .filter import DOMAIN_RE, IPV4_RE, normalize_domain, flatten_values, get_by_json_path, get_values_by_path
+
+
+def fetch_api_json(cfg):
+    api = cfg["api"]
+    url = build_url(api["url"], api.get("params", {}))
+    method = api.get("method", "GET").upper()
+    headers = api.get("headers", {})
+    timeout = int(api.get("timeout_sec", 10))
+    body_obj = api.get("body")
+    body = None
+    if body_obj is not None:
+        body = json.dumps(body_obj).encode("utf-8")
+        headers = {**headers, "Content-Type": "application/json"}
+
+    req = urllib.request.Request(url=url, data=body, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=timeout) as resp:
+        raw = resp.read().decode("utf-8", errors="replace")
+    return json.loads(raw)
+
+
+def parse_domains(payload, parser_cfg):
+    domains = []
+
+    for p in parser_cfg.get("field_paths", []):
+        values = get_values_by_path(payload, p)
+        domains.extend(flatten_values(values))
+
+    for p in parser_cfg.get("json_paths", []):
+        v = get_by_json_path(payload, p)
+        if v is not None:
+            domains.extend(flatten_values(v))
+
+    if not domains:
+        regex_s = parser_cfg.get("regex", r"[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}")
+        text = json.dumps(payload, ensure_ascii=True)
+        domains.extend(re.findall(regex_s, text))
+
+    clean = []
+    seen = set()
+    for d in domains:
+        d = str(d).strip().lower().rstrip(".")
+        if (DOMAIN_RE.match(d) or IPV4_RE.match(d)) and d not in seen:
+            seen.add(d)
+            clean.append(d)
+    return clean
+
+
+def extract_records(payload, record_mapping):
+    records_path = str(record_mapping.get("records_path", "")).strip()
+    raw = get_values_by_path(payload, records_path)
+    return [x for x in raw if isinstance(x, dict)]
+
+
+def load_cfst_rows(cfg, config_path_abs):
+    cfst_cfg = cfg.get("cfst_local", {})
+    config_dir = os.path.dirname(config_path_abs)
+
+    work_dir = resolve_path(config_dir, cfst_cfg.get("work_dir", "./cfst"))
+    binary_path = resolve_path(work_dir, cfst_cfg.get("binary", "./cfst"))
+    result_file = resolve_path(work_dir, cfst_cfg.get("result_file", "result.csv"))
+    encoding = str(cfst_cfg.get("encoding", "utf-8")).strip() or "utf-8"
+    skip_run = bool(cfst_cfg.get("skip_run", False))
+    timeout_sec = int(cfst_cfg.get("run_timeout_sec", 600))
+
+    run_args = cfst_cfg.get("run_args", ["-o", os.path.basename(result_file)])
+    if not isinstance(run_args, list):
+        raise ValueError("cfst_local.run_args must be an array")
+    command = [binary_path] + [str(x) for x in run_args]
+
+    if not skip_run:
+        completed = subprocess.run(
+            command,
+            cwd=work_dir,
+            check=False,
+            capture_output=True,
+            text=True,
+            encoding=encoding,
+            errors="replace",
+            timeout=timeout_sec,
+        )
+        if completed.returncode != 0:
+            stderr = (completed.stderr or "").strip()
+            stdout = (completed.stdout or "").strip()
+            details = stderr or stdout or f"exit code {completed.returncode}"
+            raise RuntimeError(f"cfst run failed: {details}")
+
+    if not os.path.exists(result_file):
+        raise RuntimeError(f"cfst result file not found: {result_file}")
+
+    with open(result_file, "r", encoding=encoding, errors="replace", newline="") as f:
+        reader = csv.reader(f)
+        rows = [row for row in reader if any(str(col).strip() for col in row)]
+
+    header_rows = int(cfst_cfg.get("header_rows", 1))
+    if len(rows) <= header_rows:
+        raise RuntimeError("cfst result has no data rows")
+
+    columns_cfg = cfst_cfg.get("columns", {})
+    if not isinstance(columns_cfg, dict):
+        raise ValueError("cfst_local.columns must be an object")
+
+    def col_index(name, default_index):
+        raw = columns_cfg.get(name, default_index)
+        try:
+            idx = int(raw)
+        except Exception as exc:
+            raise ValueError(f"cfst_local.columns.{name} must be an integer") from exc
+        if idx < 0:
+            raise ValueError(f"cfst_local.columns.{name} must be >= 0")
+        return idx
+
+    ip_idx = col_index("ip", 0)
+    sent_idx = col_index("sent", 1)
+    received_idx = col_index("received", 2)
+    loss_idx = col_index("loss_rate", 3)
+    latency_idx = col_index("avg_latency", 4)
+    speed_idx = col_index("download_speed", 5)
+    region_idx = col_index("region", 6)
+
+    out = []
+    for row in rows[header_rows:]:
+        if ip_idx >= len(row):
+            continue
+        domain = normalize_domain(row[ip_idx])
+        if not domain:
+            continue
+        out.append(
+            {
+                "domain": domain,
+                "ip": domain,
+                "sent": row[sent_idx].strip() if sent_idx < len(row) else "",
+                "received": row[received_idx].strip() if received_idx < len(row) else "",
+                "loss_rate": row[loss_idx].strip() if loss_idx < len(row) else "",
+                "avg_latency": row[latency_idx].strip() if latency_idx < len(row) else "",
+                "download_speed": row[speed_idx].strip() if speed_idx < len(row) else "",
+                "region": row[region_idx].strip() if region_idx < len(row) else "",
+            }
+        )
+
+    if not out:
+        raise RuntimeError("cfst result parsed to zero valid rows")
+    return out

scripts/rotator/storage.py

@@ -0,0 +1,66 @@
+import datetime as dt
+import json
+import os
+import tempfile
+
+
+def utc_now_iso():
+    """Return current UTC time in ISO 8601 format (e.g. 2026-10-13T12:00:00Z)."""
+    return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
+
+
+def read_json_file(path, default=None):
+    """Read and parse JSON from file, returning `default` if file missing or corrupt."""
+    if default is None:
+        default = {}
+    if not os.path.exists(path):
+        return default
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            return json.load(f)
+    except (ValueError, json.JSONDecodeError):
+        return default
+
+
+def _atomic_write(path, payload_bytes):
+    """Write `payload_bytes` to `path` atomically.
+
+    Strategy: write to a sibling temp file, fsync, then os.replace.
+    Concurrent readers either see the previous version or the new
+    one — never a truncated/empty file. Required for state.json,
+    which is the sole fallback source on the next run.
+    """
+    parent = os.path.dirname(path) or "."
+    os.makedirs(parent, exist_ok=True)
+    fd, tmp_path = tempfile.mkstemp(
+        prefix=".{}.".format(os.path.basename(path)),
+        suffix=".tmp",
+        dir=parent,
+    )
+    try:
+        with os.fdopen(fd, "wb") as f:
+            f.write(payload_bytes)
+            f.flush()
+            try:
+                os.fsync(f.fileno())
+            except OSError:
+                # fsync may fail on some filesystems (e.g. tmpfs without backing store)
+                pass
+        os.replace(tmp_path, path)
+    except Exception:
+        try:
+            os.unlink(tmp_path)
+        except OSError:
+            pass
+        raise
+
+
+def write_json_file(path, data):
+    """Atomically write data as formatted JSON."""
+    payload = json.dumps(data, ensure_ascii=True, indent=2)
+    _atomic_write(path, payload.encode("utf-8"))
+
+
+def write_text_file(path, data):
+    """Atomically write text string to file."""
+    _atomic_write(path, data.encode("utf-8"))

scripts/run_update_and_commit.sh

@@ -21,7 +21,17 @@ log_err() {
 
 SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 DEFAULT_APP_DIR="$(cd "${SCRIPT_DIR}/.." && pwd)"
-APP_DIR="$(git -C "$DEFAULT_APP_DIR" rev-parse --show-toplevel 2>/dev/null || printf '%s' "$DEFAULT_APP_DIR")"
+APP_DIR="${APP_DIR:-$(git -C "$DEFAULT_APP_DIR" rev-parse --show-toplevel 2>/dev/null || printf '%s' "$DEFAULT_APP_DIR")}"
+
+PYTHON_BIN="${PYTHON_BIN:-}"
+if [[ -z "$PYTHON_BIN" ]]; then
+  if [[ -x "/usr/bin/python3" ]]; then
+    PYTHON_BIN="/usr/bin/python3"
+  else
+    PYTHON_BIN="$(command -v python3 || echo "python3")"
+  fi
+fi
+
 force_commit="${GIT_FORCE_COMMIT:-0}"
 if [[ "${1:-}" == "--force-commit" ]]; then
   force_commit="1"
@@ -86,6 +96,7 @@ http_user="${GIT_HTTP_USERNAME:-git}"
 http_token="${GIT_HTTP_TOKEN:-}"
 http_token_file="${GIT_HTTP_TOKEN_FILE:-}"
 credential_helper="${GIT_CREDENTIAL_HELPER:-}"
+max_state_commits="${GIT_MAX_STATE_COMMITS:-50}"
 ts="$(date -u +"%Y-%m-%dT%H:%M:%SZ")"
 
 if [[ -z "$http_token" ]] && [[ -n "$http_token_file" ]] && [[ -r "$http_token_file" ]]; then
@@ -111,9 +122,12 @@ git_auth() {
   fi
 }
 
-output_settings_json="$(/usr/bin/python3 "${APP_DIR}/scripts/domain_updater.py" --config "$CONFIG_PATH" --print-output-settings)"
-mapfile -t output_settings < <(
-  printf '%s' "$output_settings_json" | /usr/bin/python3 -c '
+output_settings_json="$("$PYTHON_BIN" "${APP_DIR}/scripts/domain_updater.py" --config "$CONFIG_PATH" --print-output-settings)"
+output_settings=()
+while IFS= read -r line; do
+  output_settings+=("$line")
+done < <(
+  printf '%s' "$output_settings_json" | "$PYTHON_BIN" -c '
 import json, sys
 settings = json.load(sys.stdin)
 for key in ["runtime_dir", "selected_text_path", "selected_json_path", "state_path", "vars_path"]:
@@ -128,7 +142,7 @@ STATE_FILE="${output_settings[3]}"
 VARS_FILE="${output_settings[4]}"
 
 repo_relpath() {
-  /usr/bin/python3 - "$APP_DIR" "$1" <<'PY'
+  "$PYTHON_BIN" - "$APP_DIR" "$1" <<'PY'
 import os
 import sys
 
@@ -147,7 +161,7 @@ PY
 
 json_field() {
   local key="$1"
-  /usr/bin/python3 -c '
+  "$PYTHON_BIN" -c '
 import json
 import sys
 
@@ -165,7 +179,7 @@ else:
 tmp_result_json="$(mktemp "${TMPDIR:-/tmp}/vmess_result.XXXXXX")"
 trap 'rm -f "$tmp_result_json"' EXIT INT TERM
 
-/usr/bin/python3 "${APP_DIR}/scripts/domain_updater.py" --config "$CONFIG_PATH" --result-json-file "$tmp_result_json"
+"$PYTHON_BIN" "${APP_DIR}/scripts/domain_updater.py" --config "$CONFIG_PATH" --result-json-file "$tmp_result_json"
 if [[ -s "$tmp_result_json" ]]; then
   updater_status="$(json_field status < "$tmp_result_json" 2>/dev/null || echo "")"
 else
@@ -336,7 +350,9 @@ for src in "${tracked_src_files[@]}"; do
   dst="$work_dir/$rel"
   mkdir -p "$(dirname "$dst")"
   if [[ -f "$src" ]]; then
-    cp "$src" "$dst"
+    if [[ "$src" != "$dst" ]]; then
+      cp "$src" "$dst"
+    fi
   else
     rm -f "$dst"
   fi
@@ -373,10 +389,34 @@ if [[ "$force_commit" == "1" ]]; then
   commit_message="manual: value ${after}, updated at ${ts}"
 fi
 
-git -C "$work_dir" \
-  -c user.name="$commit_name" \
-  -c user.email="$commit_email" \
-  commit "${commit_extra_args[@]}" -m "$commit_message"
+commit_cmd=(git -C "$work_dir" -c user.name="$commit_name" -c user.email="$commit_email" commit)
+if [[ "${#commit_extra_args[@]}" -gt 0 ]]; then
+  commit_cmd+=("${commit_extra_args[@]}")
+fi
+commit_cmd+=(-m "$commit_message")
+"${commit_cmd[@]}"
+
+history_trimmed="0"
+if [[ "$max_state_commits" -gt 0 ]]; then
+  commit_count="$(git -C "$work_dir" rev-list --count HEAD 2>/dev/null || echo "1")"
+  if [[ "$commit_count" -gt "$max_state_commits" ]]; then
+    log "[vmess-domain-rotator] runtime history reached ${commit_count} commits, trimming older history (retaining last ${max_state_commits})"
+    git -C "$work_dir" branch -f "${runtime_branch}-untrimmed" HEAD >/dev/null 2>&1 || true
+    base_commit="$(git -C "$work_dir" rev-parse "HEAD~$((max_state_commits - 1))" 2>/dev/null || true)"
+    if [[ -n "$base_commit" ]]; then
+      tree_sha="$(git -C "$work_dir" rev-parse "${base_commit}^{tree}" 2>/dev/null || true)"
+      if [[ -n "$tree_sha" ]]; then
+        root_commit="$(git -C "$work_dir" commit-tree "$tree_sha" -m "chore(prune): baseline snapshot before rolling rotation" 2>/dev/null || true)"
+        if [[ -n "$root_commit" ]] && git -C "$work_dir" rebase --onto "$root_commit" "$base_commit" "$runtime_branch" >/dev/null 2>&1; then
+          history_trimmed="1"
+          log "[vmess-domain-rotator] history successfully pruned to ${max_state_commits} commits"
+        else
+          git -C "$work_dir" reset --hard "${runtime_branch}-untrimmed" >/dev/null 2>&1 || true
+        fi
+      fi
+    fi
+  fi
+fi
 
 if [[ "$push_enabled" != "1" ]]; then
   log "[vmess-domain-rotator] git push disabled by GIT_PUSH_ENABLED=${push_enabled}"
@@ -387,19 +427,41 @@ elif [[ -z "$push_remote" ]]; then
   fi
   log "[vmess-domain-rotator] no remote found, skip git push"
 else
+  push_opts=()
+  if [[ "$history_trimmed" == "1" ]]; then
+    push_opts+=(--force-with-lease)
+  fi
+  push_opts+=("$push_remote" "$runtime_branch:$runtime_branch")
+
   push_ok="0"
   if git -C "$work_dir" rev-parse --abbrev-ref --symbolic-full-name "@{u}" >/dev/null 2>&1; then
-    if git_auth "$work_dir" push "$push_remote" "$runtime_branch:$runtime_branch"; then
+    if git_auth "$work_dir" push "${push_opts[@]}"; then
       push_ok="1"
       log "[vmess-domain-rotator] pushed to ${push_remote}/${runtime_branch}"
+    elif [[ "$history_trimmed" == "1" ]]; then
+      log "[vmess-domain-rotator] remote rejected force-with-lease for history trimming, restoring full history and retrying normal push"
+      git -C "$work_dir" reset --hard "${runtime_branch}-untrimmed" >/dev/null 2>&1 || true
+      if git_auth "$work_dir" push "$push_remote" "$runtime_branch:$runtime_branch"; then
+        push_ok="1"
+        log "[vmess-domain-rotator] pushed to ${push_remote}/${runtime_branch} (fallback to append-only)"
+      fi
     fi
   else
-    if git_auth "$work_dir" push -u "$push_remote" "$runtime_branch:$runtime_branch"; then
+    if git_auth "$work_dir" push -u "${push_opts[@]}"; then
       push_ok="1"
       log "[vmess-domain-rotator] pushed to ${push_remote}/${runtime_branch} (set upstream)"
+    elif [[ "$history_trimmed" == "1" ]]; then
+      log "[vmess-domain-rotator] remote rejected force-with-lease for history trimming, restoring full history and retrying normal push"
+      git -C "$work_dir" reset --hard "${runtime_branch}-untrimmed" >/dev/null 2>&1 || true
+      if git_auth "$work_dir" push -u "$push_remote" "$runtime_branch:$runtime_branch"; then
+        push_ok="1"
+        log "[vmess-domain-rotator] pushed to ${push_remote}/${runtime_branch} (set upstream fallback)"
+      fi
     fi
   fi
 
+  git -C "$work_dir" branch -D "${runtime_branch}-untrimmed" >/dev/null 2>&1 || true
+
   if [[ "$push_ok" != "1" ]]; then
     log "[vmess-domain-rotator] git push failed"
     log "[vmess-domain-rotator] hint: configure non-interactive auth (credential.helper store, SSH deploy key, or GIT_HTTP_USERNAME/GIT_HTTP_TOKEN)"

tests/test_domain_updater.py

@@ -268,6 +268,39 @@ class TestDomainUpdaterP0P1Bugs(unittest.TestCase):
         proc = subprocess.run(cmd, capture_output=True, text=True)
         self.assertEqual(proc.returncode, 0)
 
+    def test_modular_rotator_package_architecture_and_facade_integrity(self):
+        """[方案B 架构重构验证] 验证 rotator 模块化子包解耦与 domain_updater.py 外观门面契约完整性"""
+        import rotator
+        from rotator import config, filter, sources, scoring, storage, notify, main
+
+        # 验证各职责模块的核心导出符号存在且可调用
+        self.assertTrue(callable(config.validate_config))
+        self.assertTrue(callable(config.build_output_settings))
+        self.assertTrue(callable(filter.apply_filter))
+        self.assertTrue(callable(sources.fetch_api_json))
+        self.assertTrue(callable(scoring.rank_scored_records))
+        self.assertTrue(callable(storage.write_json_file))
+        self.assertTrue(callable(notify.run_notify))
+        self.assertTrue(callable(main.run_updater))
+
+        # 验证顶层 domain_updater 外观门面已完整映射所有对外契约符号
+        required_facade_symbols = [
+            "validate_config",
+            "apply_filter",
+            "run_notify",
+            "utc_now_iso",
+            "write_text_file",
+            "write_json_file",
+            "read_json_file",
+            "rank_scored_records",
+            "choose_domain",
+            "DOMAIN_RE",
+            "IPV4_RE",
+            "main",
+        ]
+        for sym in required_facade_symbols:
+            self.assertTrue(hasattr(du, sym), f"Facade 必须导出符号: {sym}")
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_router_scripts.py

@@ -37,6 +37,71 @@ class TestRouterScriptsBugsAndContracts(unittest.TestCase):
         self.assertIn('CFST_DISABLE_DOWNLOAD', content)
         self.assertIn('-dd', content)
 
+    def test_build_router_package_ssot_sync(self):
+        """[方案B SSOT驱动验证] 验证 build_router_package.sh --sync-from-config 自动将 JSON 测速与输出参数投影至路由器配置"""
+        import json
+        import os
+        import subprocess
+        import tarfile
+        import tempfile
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            tmppath = pathlib.Path(tmpdir)
+            base_conf = tmppath / "base_router.conf"
+            base_conf.write_text("""CFST_DISPLAY_COUNT="10"
+CFST_PORT="443"
+CFST_DISABLE_DOWNLOAD="0"
+TOP_N="3"
+VALUE_TEXT_FILE="current_ip.txt"
+""")
+
+            json_conf = tmppath / "config.local.json"
+            json_conf.write_text(json.dumps({
+                "selection": {"top_n": 5},
+                "output": {"selected_value_file": "ssot_custom_ip.txt"},
+                "cfst_local": {
+                    "run_args": ["-p", "25", "-f", "my_ips.txt", "-tp", "8443", "-dd"]
+                }
+            }))
+
+            out_tar = tmppath / "test_pkg.tar.gz"
+            script_path = PROJECT_ROOT / "build_router_package.sh"
+
+            env = os.environ.copy()
+            env["ROUTER_CONF"] = str(base_conf)
+            env["OUT_TAR"] = str(out_tar)
+            env["TMP_DIR"] = str(tmppath / "build_tmp")
+
+            proc = subprocess.run(
+                [
+                    "sh",
+                    str(script_path),
+                    "--sync-from-config",
+                    str(json_conf),
+                ],
+                env=env,
+                capture_output=True,
+                text=True,
+                cwd=str(PROJECT_ROOT),
+            )
+
+            self.assertEqual(proc.returncode, 0, f"打包失败: {proc.stderr}\nSTDOUT: {proc.stdout}")
+            self.assertTrue(out_tar.exists(), "生成的 tar.gz 必须存在")
+
+            # 解压并校验打包中的 config_router.conf
+            with tarfile.open(out_tar, "r:gz") as tar:
+                conf_member = tar.extractfile("vmess/config_router.conf")
+                self.assertIsNotNone(conf_member)
+                conf_content = conf_member.read().decode("utf-8")
+
+            # 验证 SSOT 同步生效
+            self.assertIn('TOP_N="5"', conf_content)
+            self.assertIn('VALUE_TEXT_FILE="ssot_custom_ip.txt"', conf_content)
+            self.assertIn('CFST_DISPLAY_COUNT="25"', conf_content)
+            self.assertIn('CFST_IP_FILE="my_ips.txt"', conf_content)
+            self.assertIn('CFST_PORT="8443"', conf_content)
+            self.assertIn('CFST_DISABLE_DOWNLOAD="1"', conf_content)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_shell_lifecycle.py

@@ -1,7 +1,9 @@
+import json
 import os
 import pathlib
 import shutil
 import subprocess
+import sys
 import tempfile
 import unittest
 
@@ -100,6 +102,90 @@ class TestShellLifecycleP0P1Bugs(unittest.TestCase):
         self.assertIn('is_fallback="1"', script_content)
         self.assertIn('"$is_fallback" == "1"', script_content)
 
+    def test_mode_a_git_history_rolling_prune(self):
+        """[方案B 模式A验证] 验证 run_update_and_commit.sh 在达到阈值时自动修剪孤儿分支历史至 max_state_commits"""
+        repo_dir = self.work_dir / "state_repo"
+        repo_dir.mkdir()
+        # 初始化 git repo
+        subprocess.run(["git", "init", "-b", "main"], cwd=str(repo_dir), check=True, capture_output=True)
+        subprocess.run(["git", "config", "user.name", "Test Runner"], cwd=str(repo_dir), check=True)
+        subprocess.run(["git", "config", "user.email", "test@example.com"], cwd=str(repo_dir), check=True)
+
+        # 模拟生成 5 个初始历史 commit
+        target_file = repo_dir / "current.txt"
+        for i in range(1, 6):
+            target_file.write_text(f"domain{i}.example.com\n")
+            subprocess.run(["git", "add", "."], cwd=str(repo_dir), check=True)
+            subprocess.run(["git", "commit", "-m", f"update {i}"], cwd=str(repo_dir), check=True, capture_output=True)
+
+        initial_count = int(subprocess.check_output(["git", "rev-list", "--count", "HEAD"], cwd=str(repo_dir)).decode().strip())
+        self.assertEqual(initial_count, 5)
+
+        # 准备真实 updater 所需的测试配置与数据源
+        csv_file = self.work_dir / "result.csv"
+        csv_file.write_text("IP 地址,已用时间,下载速度 (MB/s)\nnew-best.example.com,20 ms,12.50\n")
+
+        test_config = {
+            "source": {"type": "cfst_local"},
+            "cfst_local": {
+                "work_dir": str(self.work_dir),
+                "binary": "echo",
+                "result_file": str(csv_file),
+                "skip_run": True,
+                "run_args": [],
+            },
+            "output": {
+                "runtime_dir": str(repo_dir),
+                "selected_value_file": "current.txt",
+                "selected_value_json": "current.json",
+                "selected_value_json_key": "domain",
+                "state_file": "state.json",
+                "state_last_good_key": "last_good",
+                "export_vars_file": "vars.json",
+                "substore_value_key": "DOMAIN",
+            },
+        }
+        config_path = self.work_dir / "test_config.json"
+        config_path.write_text(json.dumps(test_config))
+
+        script_path = PROJECT_ROOT / "scripts" / "run_update_and_commit.sh"
+        env = os.environ.copy()
+        env["APP_DIR"] = str(repo_dir)
+        env["GIT_RUNTIME_BRANCH"] = "main"
+        env["GIT_PUSH_ENABLED"] = "0"
+        env["GIT_MAX_STATE_COMMITS"] = "3"  # 设置阈值为 3
+        env["PYTHON_BIN"] = sys.executable
+
+        # 必须带上 scripts 目录（含 domain_updater.py 与 rotator 子包）在 APP_DIR
+        app_scripts_dir = repo_dir / "scripts"
+        shutil.copytree(PROJECT_ROOT / "scripts", app_scripts_dir)
+
+        # 运行 run_update_and_commit.sh
+        proc = subprocess.run(
+            [
+                "bash",
+                str(script_path),
+                "--force-commit",
+                str(config_path),
+            ],
+            env=env,
+            capture_output=True,
+            text=True,
+            cwd=str(self.work_dir),
+        )
+
+        combined_output = f"{proc.stdout}\n{proc.stderr}"
+        self.assertIn("trimming older history (retaining last 3)", combined_output)
+        self.assertIn("history successfully pruned to 3 commits", combined_output)
+
+        # 检查 commit 总数是否被精确修剪至 3
+        pruned_count = int(subprocess.check_output(["git", "rev-list", "--count", "HEAD"], cwd=str(repo_dir)).decode().strip())
+        self.assertEqual(pruned_count, 3, "历史 Commit 数量必须严格限制在 max_state_commits = 3")
+
+        # 检查最新内容是否完整保存
+        self.assertEqual(target_file.read_text().strip(), "new-best.example.com")
+
 
 if __name__ == "__main__":
     unittest.main()
+