354 lines
12 KiB
Python
354 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from app.conf_models import ConfResolvedModule, ConfRuntimePlan
|
|
from app.models import ClientConfig, SourceConfig, SourceSnapshot
|
|
from app.services.rules import load_rule_payload
|
|
|
|
|
|
def _source_auto_group_name(source: Any) -> str:
|
|
display_name = source["display_name"] if isinstance(source, dict) else source.display_name
|
|
return f"{display_name} 自动"
|
|
|
|
|
|
def conf_source_to_source_config(source: dict[str, Any] | Any, *, base_path: Path) -> SourceConfig:
|
|
source_type = source["source_type"] if isinstance(source, dict) else source.source_type
|
|
value = source["value"] if isinstance(source, dict) else source.value
|
|
options = source["options"] if isinstance(source, dict) else source.options
|
|
cache_ttl = source["cache_ttl"] if isinstance(source, dict) else source.cache_ttl
|
|
key = source["key"] if isinstance(source, dict) else source.key
|
|
|
|
if source_type not in {"url", "file"}:
|
|
raise ValueError(f"Unsupported conf source type for runtime route: {source_type}")
|
|
|
|
resolved_value = value
|
|
if source_type == "file":
|
|
candidate = Path(value)
|
|
if not candidate.is_absolute():
|
|
candidate = (base_path.resolve().parent / candidate).resolve()
|
|
resolved_value = str(candidate)
|
|
|
|
kind = str(options.get("kind", "auto")).strip() or "auto"
|
|
display_name = str(options.get("display_name", key)).strip() or key
|
|
prefix = str(options.get("prefix", "")).strip()
|
|
suffix = str(options.get("suffix", "")).strip()
|
|
include_regex = str(options.get("include_regex", "")).strip() or None
|
|
exclude_regex = str(options.get("exclude_regex", "")).strip() or None
|
|
headers: dict[str, str] = {}
|
|
for option_key, option_value in options.items():
|
|
if option_key.startswith("header."):
|
|
header_name = option_key.split(".", 1)[1].strip()
|
|
if header_name:
|
|
headers[header_name] = option_value
|
|
|
|
return SourceConfig(
|
|
enabled=True,
|
|
kind=kind,
|
|
url=resolved_value,
|
|
display_name=display_name,
|
|
headers=headers,
|
|
include_regex=include_regex,
|
|
exclude_regex=exclude_regex,
|
|
prefix=prefix,
|
|
suffix=suffix,
|
|
cache_ttl_seconds=cache_ttl,
|
|
)
|
|
|
|
|
|
def _module_rule_provider_entry(
|
|
module: ConfResolvedModule,
|
|
*,
|
|
client: ClientConfig,
|
|
base_url: str,
|
|
public_path: str,
|
|
profile_key: str,
|
|
) -> dict[str, Any]:
|
|
suffix = Path(module.path).suffix.lower()
|
|
return {
|
|
"behavior": "classical",
|
|
"format": "yaml" if suffix in {".yaml", ".yml"} else "text",
|
|
"url": f"{base_url}/{public_path}/conf/rules/{profile_key}/{module.key}.yaml",
|
|
"interval": client.rule_interval,
|
|
}
|
|
|
|
|
|
def _is_inline_module(module: ConfResolvedModule) -> bool:
|
|
return Path(module.path).suffix.lower() not in {".yaml", ".yml"}
|
|
|
|
|
|
def build_conf_rule_provider_entries(
|
|
plan: ConfRuntimePlan,
|
|
*,
|
|
client: ClientConfig,
|
|
base_url: str,
|
|
public_path: str,
|
|
) -> dict[str, dict[str, Any]]:
|
|
return {
|
|
module.key: _module_rule_provider_entry(module, client=client, base_url=base_url, public_path=public_path, profile_key=plan.profile_key)
|
|
for module in plan.selected_modules
|
|
if not _is_inline_module(module)
|
|
}
|
|
|
|
|
|
def _render_inline_rule(payload_line: str, policy: str) -> str:
|
|
rendered = payload_line.strip()
|
|
if not rendered:
|
|
return ""
|
|
if rendered.endswith(",no-resolve"):
|
|
parts = [part.strip() for part in rendered.split(",")]
|
|
parts.insert(len(parts) - 1, policy)
|
|
return ",".join(parts)
|
|
return f"{rendered},{policy}"
|
|
|
|
|
|
def build_conf_rule_set_references(plan: ConfRuntimePlan, *, base_path: Path) -> list[str]:
|
|
config_dir = base_path.resolve().parent
|
|
lines: list[str] = []
|
|
final_lines: list[str] = []
|
|
for rule in plan.prepend_rules:
|
|
if rule.startswith("MATCH,"):
|
|
final_lines.append(rule)
|
|
else:
|
|
lines.append(rule)
|
|
for module in plan.selected_modules:
|
|
if _is_inline_module(module):
|
|
module_path = (config_dir / module.path).resolve()
|
|
for payload_line in load_rule_payload(module_path):
|
|
rendered = _render_inline_rule(payload_line, module.policy)
|
|
if rendered:
|
|
lines.append(rendered)
|
|
else:
|
|
lines.append(f"RULE-SET,{module.key},{module.policy}")
|
|
for builtin in plan.selected_builtins:
|
|
if builtin.rule_line.startswith("MATCH,"):
|
|
final_lines.append(builtin.rule_line)
|
|
else:
|
|
lines.append(builtin.rule_line)
|
|
for rule in plan.append_rules:
|
|
if rule.startswith("MATCH,"):
|
|
final_lines.append(rule)
|
|
else:
|
|
lines.append(rule)
|
|
return [*lines, *final_lines]
|
|
|
|
|
|
def _build_group_for_thin(
|
|
group: Any,
|
|
*,
|
|
client_type: str,
|
|
client: ClientConfig,
|
|
selected_source_names: list[str],
|
|
source_auto_names: list[str],
|
|
selectors: dict[str, str],
|
|
) -> dict[str, Any]:
|
|
built: dict[str, Any] = {
|
|
"name": group.name,
|
|
"type": group.group_type,
|
|
}
|
|
if group.selector_key:
|
|
built["filter"] = selectors[group.selector_key]
|
|
if client_type == "mihomo":
|
|
built["use"] = selected_source_names
|
|
else:
|
|
built["include-all"] = True
|
|
else:
|
|
proxies: list[str] = []
|
|
include_all = False
|
|
for member in group.members:
|
|
if member == "@source_auto_groups":
|
|
proxies.extend(source_auto_names)
|
|
elif member == "@all_proxies":
|
|
include_all = True
|
|
else:
|
|
proxies.append(member)
|
|
built["proxies"] = proxies
|
|
if include_all:
|
|
if client_type == "mihomo":
|
|
built["include-all-providers"] = True
|
|
else:
|
|
built["include-all"] = True
|
|
if group.group_type != "select":
|
|
built["url"] = group.url or str(client.test_url)
|
|
built["interval"] = group.interval or client.test_interval
|
|
if group.tolerance is not None:
|
|
built["tolerance"] = group.tolerance
|
|
return built
|
|
|
|
|
|
def _build_group_for_bundle(
|
|
group: Any,
|
|
*,
|
|
client: ClientConfig,
|
|
selectors: dict[str, str],
|
|
all_proxy_names: list[str],
|
|
source_auto_names: list[str],
|
|
) -> dict[str, Any]:
|
|
built: dict[str, Any] = {
|
|
"name": group.name,
|
|
"type": group.group_type,
|
|
}
|
|
if group.selector_key:
|
|
pattern = selectors[group.selector_key]
|
|
built["proxies"] = [name for name in all_proxy_names if re.search(pattern, name)] or [client.direct_policy]
|
|
else:
|
|
proxies: list[str] = []
|
|
for member in group.members:
|
|
if member == "@source_auto_groups":
|
|
proxies.extend(source_auto_names)
|
|
continue
|
|
if member == "@all_proxies":
|
|
proxies.extend(all_proxy_names)
|
|
continue
|
|
else:
|
|
proxies.append(member)
|
|
built["proxies"] = proxies or [client.direct_policy]
|
|
if group.group_type != "select":
|
|
built["url"] = group.url or str(client.test_url)
|
|
built["interval"] = group.interval or client.test_interval
|
|
if group.tolerance is not None:
|
|
built["tolerance"] = group.tolerance
|
|
return built
|
|
|
|
|
|
def build_conf_thin_profile(
|
|
*,
|
|
client_type: str,
|
|
client: ClientConfig,
|
|
plan: ConfRuntimePlan,
|
|
base_path: Path,
|
|
base_url: str,
|
|
public_path: str,
|
|
) -> dict[str, Any]:
|
|
profile: dict[str, Any] = {
|
|
"mode": plan.mode or client.mode,
|
|
"ipv6": client.ipv6 if plan.ipv6 is None else plan.ipv6,
|
|
"proxy-providers": {},
|
|
}
|
|
log_level = plan.log_level or client.log_level
|
|
if log_level:
|
|
profile["log-level"] = log_level
|
|
if client_type == "mihomo":
|
|
if client.mixed_port is not None:
|
|
profile["mixed-port"] = client.mixed_port
|
|
if client.socks_port is not None:
|
|
profile["socks-port"] = client.socks_port
|
|
profile["allow-lan"] = client.allow_lan if plan.allow_lan is None else plan.allow_lan
|
|
|
|
selected_source_names = [item.key for item in plan.selected_sources]
|
|
source_auto_names = [_source_auto_group_name(item) for item in plan.selected_sources]
|
|
for name in selected_source_names:
|
|
source = next(item for item in plan.selected_sources if item.key == name)
|
|
group_name = _source_auto_group_name(source)
|
|
if client_type == "mihomo":
|
|
profile["proxy-providers"][name] = {
|
|
"type": "http",
|
|
"url": f"{base_url}/{public_path}/conf/providers/{plan.profile_key}/{name}.yaml",
|
|
"path": f"./providers/{plan.profile_key}-{name}.yaml",
|
|
"interval": client.provider_interval,
|
|
"health-check": {
|
|
"enable": True,
|
|
"url": str(client.test_url),
|
|
"interval": client.test_interval,
|
|
},
|
|
}
|
|
else:
|
|
profile["proxy-providers"][name] = {
|
|
"url": f"{base_url}/{public_path}/conf/providers/{plan.profile_key}/{name}.yaml",
|
|
"interval": client.provider_interval,
|
|
}
|
|
profile.setdefault("proxy-groups", []).append(
|
|
{
|
|
"name": group_name,
|
|
"type": "url-test",
|
|
"url": str(client.test_url),
|
|
"interval": client.test_interval,
|
|
"use": [name],
|
|
}
|
|
)
|
|
|
|
profile["proxy-groups"].extend(
|
|
[
|
|
_build_group_for_thin(
|
|
group,
|
|
client_type=client_type,
|
|
client=client,
|
|
selected_source_names=selected_source_names,
|
|
source_auto_names=source_auto_names,
|
|
selectors=plan.selectors,
|
|
)
|
|
for group in plan.groups
|
|
]
|
|
)
|
|
profile["rule-providers"] = build_conf_rule_provider_entries(plan, client=client, base_url=base_url, public_path=public_path)
|
|
profile["rules"] = build_conf_rule_set_references(plan, base_path=base_path)
|
|
return profile
|
|
|
|
|
|
def build_conf_bundle_profile(*, client_type: str, client: ClientConfig, plan: ConfRuntimePlan, snapshots: list[SourceSnapshot], rules: list[str]) -> dict[str, Any]:
|
|
profile: dict[str, Any] = {
|
|
"mode": plan.mode or client.mode,
|
|
"ipv6": client.ipv6 if plan.ipv6 is None else plan.ipv6,
|
|
}
|
|
log_level = plan.log_level or client.log_level
|
|
if log_level:
|
|
profile["log-level"] = log_level
|
|
if client_type == "mihomo":
|
|
if client.mixed_port is not None:
|
|
profile["mixed-port"] = client.mixed_port
|
|
if client.socks_port is not None:
|
|
profile["socks-port"] = client.socks_port
|
|
profile["allow-lan"] = client.allow_lan if plan.allow_lan is None else plan.allow_lan
|
|
|
|
proxies: list[dict[str, Any]] = []
|
|
all_proxy_names: list[str] = []
|
|
source_proxy_names: dict[str, list[str]] = {}
|
|
seen: set[str] = set()
|
|
for snapshot in snapshots:
|
|
current_source_names: list[str] = []
|
|
for proxy in snapshot.document.proxies:
|
|
candidate = dict(proxy)
|
|
name = str(candidate.get("name", "")).strip()
|
|
if not name:
|
|
continue
|
|
original = name
|
|
index = 2
|
|
while name in seen:
|
|
name = f"{original} #{index}"
|
|
index += 1
|
|
seen.add(name)
|
|
candidate["name"] = name
|
|
proxies.append(candidate)
|
|
all_proxy_names.append(name)
|
|
current_source_names.append(name)
|
|
source_proxy_names[snapshot.name] = current_source_names
|
|
|
|
profile["proxies"] = proxies
|
|
profile["proxy-groups"] = []
|
|
for source in plan.selected_sources:
|
|
profile["proxy-groups"].append(
|
|
{
|
|
"name": _source_auto_group_name(source),
|
|
"type": "url-test",
|
|
"url": str(client.test_url),
|
|
"interval": client.test_interval,
|
|
"proxies": source_proxy_names.get(source.key) or [client.direct_policy],
|
|
}
|
|
)
|
|
profile["proxy-groups"].extend(
|
|
[
|
|
_build_group_for_bundle(
|
|
group,
|
|
client=client,
|
|
selectors=plan.selectors,
|
|
all_proxy_names=all_proxy_names,
|
|
source_auto_names=[_source_auto_group_name(item) for item in plan.selected_sources],
|
|
)
|
|
for group in plan.groups
|
|
]
|
|
)
|
|
profile["rules"] = rules
|
|
return profile
|