from __future__ import annotations import argparse import json import os import sys from pathlib import Path from tempfile import TemporaryDirectory from typing import Any import yaml from fastapi.testclient import TestClient ROOT_DIR = Path(__file__).resolve().parent.parent if str(ROOT_DIR) not in sys.path: sys.path.insert(0, str(ROOT_DIR)) def _write_sample_source(path: Path) -> None: path.write_text( "\n".join( [ "proxies:", " - name: hk-demo", " type: ss", " server: 1.1.1.1", " port: 443", " cipher: aes-128-gcm", " password: demo-pass", " - name: us-demo", " type: ss", " server: 2.2.2.2", " port: 443", " cipher: aes-128-gcm", " password: demo-pass", ] ), encoding="utf-8", ) def _load_yaml(text: str) -> dict[str, Any]: data = yaml.safe_load(text) if not isinstance(data, dict): raise ValueError("response is not a yaml mapping") return data def _list_names(items: list[dict[str, Any]] | None) -> list[str]: names: list[str] = [] for item in items or []: if not isinstance(item, dict): continue name = item.get("name") if isinstance(name, str) and name.strip(): names.append(name) return names def _mapping_keys(mapping: dict[str, Any] | None) -> list[str]: if not isinstance(mapping, dict): return [] return sorted(mapping.keys()) def _group_map(groups: list[dict[str, Any]] | None) -> dict[str, dict[str, Any]]: result: dict[str, dict[str, Any]] = {} for group in groups or []: if not isinstance(group, dict): continue name = group.get("name") if isinstance(name, str) and name.strip(): result[name] = group return result def _extract_summary(data: dict[str, Any]) -> dict[str, Any]: return { "proxy_provider_keys": _mapping_keys(data.get("proxy-providers")), "rule_provider_keys": _mapping_keys(data.get("rule-providers")), "proxy_group_names": _list_names(data.get("proxy-groups")), "proxy_names": _list_names(data.get("proxies")), "rules_count": len(data.get("rules") or []), "rules_head": list(data.get("rules") or [])[:5], "rules_tail": list(data.get("rules") or [])[-5:], } def _compare_group_members(legacy_groups: list[dict[str, Any]], conf_groups: list[dict[str, Any]]) -> list[dict[str, Any]]: mismatches: list[dict[str, Any]] = [] legacy_map = _group_map(legacy_groups) conf_map = _group_map(conf_groups) for name in sorted(set(legacy_map) & set(conf_map)): legacy = legacy_map[name] conf = conf_map[name] legacy_members = legacy.get("proxies") or legacy.get("use") or [] conf_members = conf.get("proxies") or conf.get("use") or [] if legacy_members != conf_members: mismatches.append( { "group": name, "legacy_members": legacy_members, "conf_members": conf_members, } ) return mismatches def compare_outputs(client_type: str, profile_key: str, sources: str, *, sample_source: Path | None) -> dict[str, Any]: if sample_source is not None: os.environ["AIRPORT_A_URL"] = str(sample_source) os.environ["AIRPORT_B_URL"] = str(sample_source) os.environ["AIRPORT_C_URL"] = str(sample_source) from app.main import PUBLIC_PREFIX, app client = TestClient(app) legacy_thin = client.get(f"{PUBLIC_PREFIX}/clients/{client_type}.yaml?sources={sources}") legacy_bundle = client.get(f"{PUBLIC_PREFIX}/bundle/{client_type}.yaml?sources={sources}&force_refresh=true") conf_thin = client.get(f"{PUBLIC_PREFIX}/conf/clients/{client_type}/{profile_key}.yaml") conf_bundle = client.get(f"{PUBLIC_PREFIX}/conf/bundle/{client_type}/{profile_key}.yaml?force_refresh=true") responses = { "legacy_thin": legacy_thin, "legacy_bundle": legacy_bundle, "conf_thin": conf_thin, "conf_bundle": conf_bundle, } for name, response in responses.items(): if response.status_code != 200: raise RuntimeError(f"{name} returned {response.status_code}: {response.text}") legacy_thin_yaml = _load_yaml(legacy_thin.text) legacy_bundle_yaml = _load_yaml(legacy_bundle.text) conf_thin_yaml = _load_yaml(conf_thin.text) conf_bundle_yaml = _load_yaml(conf_bundle.text) result = { "inputs": { "client_type": client_type, "profile_key": profile_key, "sources": sources, }, "thin": { "legacy": _extract_summary(legacy_thin_yaml), "conf": _extract_summary(conf_thin_yaml), "group_member_mismatches": _compare_group_members( legacy_thin_yaml.get("proxy-groups") or [], conf_thin_yaml.get("proxy-groups") or [], ), }, "bundle": { "legacy": _extract_summary(legacy_bundle_yaml), "conf": _extract_summary(conf_bundle_yaml), "group_member_mismatches": _compare_group_members( legacy_bundle_yaml.get("proxy-groups") or [], conf_bundle_yaml.get("proxy-groups") or [], ), }, } result["thin"]["proxy_provider_match"] = result["thin"]["legacy"]["proxy_provider_keys"] == result["thin"]["conf"]["proxy_provider_keys"] result["thin"]["rule_provider_match"] = result["thin"]["legacy"]["rule_provider_keys"] == result["thin"]["conf"]["rule_provider_keys"] result["thin"]["group_name_match"] = result["thin"]["legacy"]["proxy_group_names"] == result["thin"]["conf"]["proxy_group_names"] result["bundle"]["group_name_match"] = result["bundle"]["legacy"]["proxy_group_names"] == result["bundle"]["conf"]["proxy_group_names"] result["bundle"]["proxy_count_match"] = len(result["bundle"]["legacy"]["proxy_names"]) == len(result["bundle"]["conf"]["proxy_names"]) result["bundle"]["rules_count_match"] = result["bundle"]["legacy"]["rules_count"] == result["bundle"]["conf"]["rules_count"] return result def main() -> None: parser = argparse.ArgumentParser(description="Compare legacy YAML routes with conf routes") parser.add_argument("--client-type", default="mihomo") parser.add_argument("--profile", default="default") parser.add_argument("--sources", default="airport-a,airport-b") parser.add_argument("--sample-source", default="") args = parser.parse_args() if args.sample_source: sample_source = Path(args.sample_source).resolve() else: sample_source = None if sample_source is None: with TemporaryDirectory() as tmp: generated = Path(tmp) / "sample-proxies.yaml" _write_sample_source(generated) result = compare_outputs(args.client_type, args.profile, args.sources, sample_source=generated) else: result = compare_outputs(args.client_type, args.profile, args.sources, sample_source=sample_source) print(json.dumps(result, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()