重构
This commit is contained in:
riglen
195
scripts/compare_conf_outputs.py
Normal file
195
scripts/compare_conf_outputs.py
Normal file
@@ -0,0 +1,195 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user