122 lines
4.0 KiB
Python
122 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
from app.models import AppConfig, SourceSnapshot
|
|
from app.services.policy_group_builder import build_bundle_groups, build_thin_groups
|
|
from app.services.profile_resolver import resolve_profile
|
|
from app.services.proxy_pipeline import build_bundle_proxy_inventory
|
|
from app.services.rule_resolver import build_rule_provider_entries, iter_resolved_rule_lines
|
|
|
|
|
|
def dump_yaml(data: dict[str, Any]) -> str:
|
|
return yaml.safe_dump(data, allow_unicode=True, sort_keys=False, default_flow_style=False)
|
|
|
|
|
|
def _build_profile_header(*, resolved_profile) -> dict[str, Any]:
|
|
client = resolved_profile.client
|
|
profile: dict[str, Any] = {"mode": client.mode, "ipv6": client.ipv6}
|
|
if client.log_level:
|
|
profile["log-level"] = client.log_level
|
|
if resolved_profile.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
|
|
return profile
|
|
|
|
|
|
def _build_proxy_providers(
|
|
*,
|
|
resolved_profile,
|
|
base_url: str,
|
|
public_path: str,
|
|
) -> dict[str, dict[str, Any]]:
|
|
proxy_providers: dict[str, dict[str, Any]] = {}
|
|
client = resolved_profile.client
|
|
for name in resolved_profile.selected_sources:
|
|
if resolved_profile.client_type == "mihomo":
|
|
proxy_providers[name] = {
|
|
"type": "http",
|
|
"url": f"{base_url}/{public_path}/providers/{name}.yaml",
|
|
"path": f"./providers/{name}.yaml",
|
|
"interval": client.provider_interval,
|
|
"health-check": {
|
|
"enable": True,
|
|
"url": str(client.test_url),
|
|
"interval": client.test_interval,
|
|
},
|
|
}
|
|
else:
|
|
proxy_providers[name] = {
|
|
"url": f"{base_url}/{public_path}/providers/{name}.yaml",
|
|
"interval": client.provider_interval,
|
|
}
|
|
return proxy_providers
|
|
|
|
|
|
def build_thin_profile(
|
|
*,
|
|
client_type: str,
|
|
app_config: AppConfig,
|
|
selected_source_names: list[str],
|
|
base_url: str,
|
|
public_path: str,
|
|
) -> dict[str, Any]:
|
|
resolved_profile = resolve_profile(
|
|
app_config=app_config,
|
|
client_type=client_type,
|
|
selected_source_names=selected_source_names,
|
|
)
|
|
profile = _build_profile_header(resolved_profile=resolved_profile)
|
|
profile["proxy-providers"] = _build_proxy_providers(
|
|
resolved_profile=resolved_profile,
|
|
base_url=base_url,
|
|
public_path=public_path,
|
|
)
|
|
profile["proxy-groups"] = build_thin_groups(
|
|
resolved_profile=resolved_profile,
|
|
selected_source_names=selected_source_names,
|
|
)
|
|
profile["rule-providers"] = build_rule_provider_entries(
|
|
resolved_profile=resolved_profile,
|
|
base_url=base_url,
|
|
public_path=public_path,
|
|
)
|
|
profile["rules"] = iter_resolved_rule_lines(
|
|
resolved_profile=resolved_profile,
|
|
include_rule_set_references=True,
|
|
inline_file_payloads=False,
|
|
)
|
|
return profile
|
|
|
|
|
|
def build_bundle_profile(
|
|
*,
|
|
client_type: str,
|
|
app_config: AppConfig,
|
|
snapshots: list[SourceSnapshot],
|
|
) -> dict[str, Any]:
|
|
resolved_profile = resolve_profile(
|
|
app_config=app_config,
|
|
client_type=client_type,
|
|
selected_source_names=[snapshot.name for snapshot in snapshots],
|
|
)
|
|
profile = _build_profile_header(resolved_profile=resolved_profile)
|
|
proxy_nodes, source_proxy_names = build_bundle_proxy_inventory(snapshots)
|
|
profile["proxies"] = [node.to_proxy_dict() for node in proxy_nodes]
|
|
profile["proxy-groups"] = build_bundle_groups(
|
|
resolved_profile=resolved_profile,
|
|
snapshots=snapshots,
|
|
source_proxy_names=source_proxy_names,
|
|
)
|
|
profile["rules"] = iter_resolved_rule_lines(
|
|
resolved_profile=resolved_profile,
|
|
include_rule_set_references=False,
|
|
inline_file_payloads=True,
|
|
)
|
|
return profile
|