7 Commits

Author SHA1 Message Date
riglen
b29e245ff5 gitignore 2026-07-29 13:54:04 +08:00
riglen
3feab7d5d8 gitignore 2026-07-29 13:53:01 +08:00
riglen
9bb7858d42 配置 2026-05-28 17:11:57 +08:00
riglen
28c49ce299 配置 2026-04-22 10:25:01 +08:00
riglen
86cabce537 配置 2026-04-21 20:10:04 +08:00
riglen
d5cfae22ea 重构 2026-04-21 16:56:06 +08:00
riglen
05e0355e14 step1 2026-04-21 15:47:09 +08:00
23 changed files with 2682 additions and 3 deletions

View File

@@ -4,6 +4,10 @@ HOST=0.0.0.0
PORT=18080
LOG_LEVEL=info
# 上游订阅抓取时使用的 User-Agent。
# 如果机场对默认 UA 返回 406可改成你客户端实际使用的 UA例如 Clash Party 抓包得到的值。
UPSTREAM_USER_AGENT=sub-provider/0.2
# 对外访问前缀,尽量改成足够长的随机字符串
PUBLIC_PATH=change-me-random-hash-path

2
.gitignore vendored
View File

@@ -207,3 +207,5 @@ cython_debug/
marimo/_static/
marimo/_lsp/
__marimo__/
output/
data/

View File

@@ -40,6 +40,11 @@ sub-provider/
rules.py
subscriptions.py
config/
conf/
base.conf
profiles/
default.conf
lite.conf
sources.yaml
rules/
reject.yaml
@@ -69,6 +74,7 @@ cp .env.example .env
- `PUBLIC_PATH` 改成足够长的随机字符串
- `PUBLIC_BASE_URL` 建议填写你反代后的最终访问地址,例如 `https://sub.example.com`
- 如果上游订阅地址对默认 UA 返回 `406`,可在 `.env` 里设置 `UPSTREAM_USER_AGENT`,填成你客户端实际使用的 UA例如 Clash Party 抓包出来的值
- `AIRPORT_A_URL` / `AIRPORT_B_URL` / `AIRPORT_C_URL` 都可以直接填订阅地址,项目会自动判断是 YAML 还是 URI 订阅
- 也可以直接填本地文件路径,例如 `/app/data/sources/airport-b.txt``file:///app/data/sources/airport-b.txt`
- 允许把其中一个留空;留空时这个机场会自动跳过
@@ -90,10 +96,14 @@ docker compose up -d --build
`https://YOUR_DOMAIN/<PUBLIC_PATH>/clients/mihomo.yaml?sources=airport-a,airport-b,airport-c`
- Stash 薄壳入口:
`https://YOUR_DOMAIN/<PUBLIC_PATH>/clients/stash.yaml?sources=airport-a,airport-b,airport-c`
- Conf 薄壳入口:
`https://YOUR_DOMAIN/<PUBLIC_PATH>/conf/clients/mihomo/default.yaml`
- Mihomo/OpenClash bundle
`https://YOUR_DOMAIN/<PUBLIC_PATH>/bundle/mihomo.yaml?sources=airport-a,airport-b,airport-c`
- Stash bundle
`https://YOUR_DOMAIN/<PUBLIC_PATH>/bundle/stash.yaml?sources=airport-a,airport-b,airport-c`
- Conf bundle
`https://YOUR_DOMAIN/<PUBLIC_PATH>/conf/bundle/mihomo/default.yaml`
---
@@ -152,6 +162,77 @@ HEAD /<PUBLIC_PATH>/bundle/stash.yaml?sources=airport-a,airport-b,airport-c
- 上游订阅原始内容也会按 TTL 落盘缓存到 `data/fetch-cache/`,默认 900 秒
- 如果你想固定使用本地文件订阅,可以把文件放到 `data/sources/`,再把 `AIRPORT_*_URL` 指向 `/app/data/sources/xxx.txt`
### 5. Conf 配置入口
```text
GET /<PUBLIC_PATH>/conf/clients/{client_type}/{profile_key}.yaml
GET /<PUBLIC_PATH>/conf/bundle/{client_type}/{profile_key}.yaml
GET /<PUBLIC_PATH>/conf/providers/{profile_key}/{name}.yaml
GET /<PUBLIC_PATH>/conf/rules/{profile_key}/{name}.yaml
```
特点:
-`sources.yaml` 路由保持不变conf 路由独立存在
- 新链路读取 `config/conf/base.conf``config/conf/profiles/*.conf`
- 可以并行对比旧输出和 conf 输出,不影响日常使用
- conf 路由按请求懒加载配置conf 写坏时不会影响旧接口启动
---
## Conf 配置说明
当前已落地一套 ACL4SSR-like 的 `.conf` 配置链路:
- `config/conf/base.conf`
负责 `source``selector``group``module``builtin` 等全局定义
- `config/conf/profiles/*.conf`
负责 `sources``include_modules``include_builtins`
以及 `override_policy``prepend_rule``append_rule`
这条链路目前的目标不是替换旧接口,而是:
- 保持旧 YAML 配置和旧路由继续可用
- 用新 conf 路由做并行验证
- 逐步把输出收敛到和旧链路一致
### 示例
`base.conf` 里一条规则模块:
```ini
module = apple,../rules/acl4ssr/Apple.yaml,🍎 苹果服务,80,true
```
`default.conf` 里启用模块:
```ini
include_modules = apple,openai
include_builtins = geoip_cn,final
```
---
## 对比脚本
项目内置了一个本地对比脚本:
```bash
python scripts/compare_conf_outputs.py
```
它会:
- 启动本地 `TestClient`
- 同时请求旧路由和 conf 路由
- 输出 `proxy-providers``rule-providers``proxy-groups``rules` 的差异摘要
如果要指定样例源文件:
```bash
python scripts/compare_conf_outputs.py --sample-source C:\path\to\sample.yaml
```
---
## 默认策略结构
@@ -191,3 +272,15 @@ HEAD /<PUBLIC_PATH>/bundle/stash.yaml?sources=airport-a,airport-b,airport-c
- `sources` 填多个机场源:**只取第一个**机场源的 `Subscription-Userinfo`
这样 Stash、Clash Party 这类客户端读取配置订阅头时,行为是稳定可预期的。
---
## 上游 `406` 处理
上游订阅抓取默认使用 `sub-provider/0.2` 作为 `User-Agent`。如果机场对这个 UA 返回 `406 Not Acceptable`,可以直接在 `.env` 里覆盖:
```env
UPSTREAM_USER_AGENT=Clash Party/<你的版本号或抓包值>
```
代码里也兼容旧变量名 `DEFAULT_USER_AGENT`,但建议后续统一用 `UPSTREAM_USER_AGENT`

147
app/conf_models.py Normal file
View File

@@ -0,0 +1,147 @@
from __future__ import annotations
from pathlib import Path
from typing import Literal
from pydantic import BaseModel, Field
class ConfSource(BaseModel):
key: str
source_type: Literal["url", "file", "inline", "base64"] = "url"
value: str
enabled: bool = True
cache_ttl: int | None = None
options: dict[str, str] = Field(default_factory=dict)
line_no: int
class ConfSelector(BaseModel):
key: str
regex: str
line_no: int
class ConfGroup(BaseModel):
name: str
group_type: str
tokens: list[str] = Field(default_factory=list)
raw: str
line_no: int
class ConfModule(BaseModel):
key: str
path: str
policy: str
order: int
enabled: bool = True
line_no: int
def resolved_path(self, config_dir: Path) -> Path:
return (config_dir / self.path).resolve()
class ConfBuiltin(BaseModel):
key: str
builtin_type: Literal["GEOIP", "FINAL"]
value: str = ""
policy: str
order: int
enabled: bool = True
line_no: int
class ConfBaseConfig(BaseModel):
listen: str | None = None
output_dir: str | None = None
cache_dir: str | None = None
mode: str | None = None
allow_lan: bool | None = None
log_level: str | None = None
ipv6: bool | None = None
append_userinfo_header: bool | None = None
userinfo_source_policy: str | None = None
sources: list[ConfSource] = Field(default_factory=list)
selectors: list[ConfSelector] = Field(default_factory=list)
groups: list[ConfGroup] = Field(default_factory=list)
modules: list[ConfModule] = Field(default_factory=list)
builtins: list[ConfBuiltin] = Field(default_factory=list)
class ConfPolicyOverride(BaseModel):
module_key: str
policy: str
line_no: int
class ConfInlineRule(BaseModel):
value: str
line_no: int
class ConfProfileConfig(BaseModel):
name: str | None = None
enabled: bool = True
sources: list[str] = Field(default_factory=list)
include_modules: list[str] = Field(default_factory=list)
exclude_modules: list[str] = Field(default_factory=list)
include_builtins: list[str] = Field(default_factory=list)
override_policies: list[ConfPolicyOverride] = Field(default_factory=list)
prepend_rules: list[ConfInlineRule] = Field(default_factory=list)
append_rules: list[ConfInlineRule] = Field(default_factory=list)
class ConfResolvedSource(BaseModel):
key: str
display_name: str
source_type: str
value: str
enabled: bool = True
cache_ttl: int | None = None
options: dict[str, str] = Field(default_factory=dict)
class ConfResolvedModule(BaseModel):
key: str
path: str
policy: str
order: int
enabled: bool = True
class ConfResolvedBuiltin(BaseModel):
key: str
rule_line: str
policy: str
order: int
enabled: bool = True
class ConfResolvedGroup(BaseModel):
name: str
group_type: str
members: list[str] = Field(default_factory=list)
selector_key: str | None = None
url: str | None = None
interval: int | None = None
tolerance: int | None = None
raw: str
class ConfRuntimePlan(BaseModel):
profile_key: str
profile_name: str | None = None
selected_sources: list[ConfResolvedSource] = Field(default_factory=list)
selected_modules: list[ConfResolvedModule] = Field(default_factory=list)
selected_builtins: list[ConfResolvedBuiltin] = Field(default_factory=list)
selectors: dict[str, str] = Field(default_factory=dict)
groups: list[ConfResolvedGroup] = Field(default_factory=list)
prepend_rules: list[str] = Field(default_factory=list)
append_rules: list[str] = Field(default_factory=list)
mode: str | None = None
allow_lan: bool | None = None
log_level: str | None = None
ipv6: bool | None = None
append_userinfo_header: bool | None = None
userinfo_source_policy: str | None = None

View File

@@ -1,9 +1,10 @@
from __future__ import annotations
import os
from functools import lru_cache
from pathlib import Path
from pydantic import Field
from pydantic import AliasChoices, Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -25,9 +26,14 @@ class Settings(BaseSettings):
cache_ttl_seconds: int = 900
bundle_cache_ttl_seconds: int = 600
max_proxy_name_length: int = 80
default_user_agent: str = "sub-provider/0.2"
default_user_agent: str = Field(
default="sub-provider/0.2",
validation_alias=AliasChoices("UPSTREAM_USER_AGENT", "DEFAULT_USER_AGENT"),
)
sources_file: Path = CONFIG_DIR / "sources.yaml"
conf_base_file: Path = CONFIG_DIR / "conf" / "base.conf"
conf_profiles_dir: Path = CONFIG_DIR / "conf" / "profiles"
rules_dir: Path = CONFIG_DIR / "rules"
bundle_cache_dir: Path = ROOT_DIR / "output" / "bundle-cache"
fetch_cache_dir: Path = DATA_DIR / "fetch-cache"
@@ -38,6 +44,16 @@ class Settings(BaseSettings):
extra="ignore",
)
@field_validator("default_user_agent", mode="before")
@classmethod
def _coalesce_default_user_agent(cls, value: str) -> str:
if isinstance(value, str) and value.strip():
return value
legacy = os.getenv("DEFAULT_USER_AGENT", "").strip()
if legacy:
return legacy
return "sub-provider/0.2"
@lru_cache(maxsize=1)
def get_settings() -> Settings:

View File

@@ -8,6 +8,10 @@ from fastapi.responses import Response
from app.config import get_settings
from app.models import RuleConfig, SourceConfig, SourceSnapshot
from app.services.bundle_cache import build_bundle_cache_key, load_bundle_cache, save_bundle_cache
from app.services.conf_config_store import load_conf_store
from app.services.conf_loader import ConfConfigError
from app.services.conf_profiles import build_conf_bundle_profile, build_conf_thin_profile, conf_source_to_source_config
from app.services.conf_runtime import build_conf_rule_lines, resolve_conf_runtime_plan
from app.services.loader import load_app_config
from app.services.profiles import build_bundle_profile, build_thin_profile, dump_yaml
from app.services.rules import load_rule_text
@@ -69,6 +73,37 @@ def _rule_path(rule: RuleConfig):
return path
def _load_conf_profile_runtime(profile_key: str):
try:
base_config, profiles = load_conf_store()
except ConfConfigError as exc:
raise HTTPException(status_code=500, detail=f"invalid conf config: {exc}") from exc
profile = profiles.get(profile_key)
if profile is None:
raise HTTPException(status_code=404, detail="conf profile not found")
try:
plan = resolve_conf_runtime_plan(
profile_key=profile_key,
base_config=base_config,
profile_config=profile,
base_path=settings.conf_base_file,
)
except ConfConfigError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return base_config, profile, plan
def _conf_source_items(plan) -> list[tuple[str, SourceConfig]]:
items: list[tuple[str, SourceConfig]] = []
for source in plan.selected_sources:
try:
items.append((source.key, conf_source_to_source_config(source, base_path=settings.conf_base_file)))
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return items
async def _build_quota_headers(source_items: list[tuple[str, SourceConfig]]) -> dict[str, str]:
headers: dict[str, str] = {}
quota = await get_first_quota(source_items)
@@ -161,6 +196,58 @@ async def client_profile(client_type: str, request: Request, sources: str | None
return _yaml_response(content, request, headers=headers, filename=f"{client_type}.yaml")
@app.api_route(PUBLIC_PREFIX + "/conf/rules/{profile_key}/{name}.yaml", methods=["GET", "HEAD"])
async def conf_rule_file(profile_key: str, name: str, request: Request) -> Response:
_, _, plan = _load_conf_profile_runtime(profile_key)
module = next((item for item in plan.selected_modules if item.key == name), None)
if module is None:
raise HTTPException(status_code=404, detail="conf rule not found")
path = (settings.conf_base_file.resolve().parent / module.path).resolve()
if not path.is_file():
raise HTTPException(status_code=404, detail="conf rule file missing")
content = load_rule_text(path)
return _yaml_response(content, request, filename=f"{profile_key}-{name}.yaml")
@app.api_route(PUBLIC_PREFIX + "/conf/providers/{profile_key}/{name}.yaml", methods=["GET", "HEAD"])
async def conf_provider(profile_key: str, name: str, request: Request) -> Response:
_, _, plan = _load_conf_profile_runtime(profile_key)
source_items = _conf_source_items(plan)
source = next((item for item in source_items if item[0] == name), None)
if source is None:
raise HTTPException(status_code=404, detail="conf provider not found")
try:
document = await build_provider_document(source[0], source[1])
except Exception as exc: # noqa: BLE001
logger.exception("conf provider failed: profile=%s source=%s", profile_key, name)
raise HTTPException(status_code=502, detail=f"failed to build conf provider: {exc}") from exc
content = dump_provider_yaml(document)
headers = await _build_quota_headers([source])
return _yaml_response(content, request, headers=headers, filename=f"{profile_key}-{name}.yaml")
@app.api_route(PUBLIC_PREFIX + "/conf/clients/{client_type}/{profile_key}.yaml", methods=["GET", "HEAD"])
async def conf_client_profile(client_type: str, profile_key: str, request: Request) -> Response:
client = app_config.clients.get(client_type)
if client is None:
raise HTTPException(status_code=404, detail="client config not found")
_, _, plan = _load_conf_profile_runtime(profile_key)
source_items = _conf_source_items(plan)
content = dump_yaml(
build_conf_thin_profile(
client_type=client_type,
client=client,
plan=plan,
base_path=settings.conf_base_file,
base_url=_base_url(request),
public_path=(app_config.public_path or settings.public_path).strip("/"),
)
)
headers = {"profile-update-interval": str(client.provider_interval)}
headers.update(await _build_quota_headers(source_items))
return _yaml_response(content, request, headers=headers, filename=f"{client_type}-{profile_key}.yaml")
@app.api_route(PUBLIC_PREFIX + "/bundle/{client_type}.yaml", methods=["GET", "HEAD"])
async def bundle_profile(
client_type: str,
@@ -215,3 +302,58 @@ async def bundle_profile(
headers={key: value for key, value in headers.items() if key != "X-Sub-Provider-Bundle-Cache"},
)
return _yaml_response(content, request, headers=headers, filename=f"bundle-{client_type}.yaml")
@app.api_route(PUBLIC_PREFIX + "/conf/bundle/{client_type}/{profile_key}.yaml", methods=["GET", "HEAD"])
async def conf_bundle_profile(client_type: str, profile_key: str, request: Request, force_refresh: bool = Query(default=False)) -> Response:
client = app_config.clients.get(client_type)
if client is None:
raise HTTPException(status_code=404, detail="client config not found")
_, _, plan = _load_conf_profile_runtime(profile_key)
source_items = _conf_source_items(plan)
cache_key = build_bundle_cache_key(
client_type=f"conf-{client_type}-{profile_key}",
source_names=[name for name, _ in source_items],
)
if not force_refresh:
cached = load_bundle_cache(
cache_dir=settings.bundle_cache_dir,
cache_key=cache_key,
ttl_seconds=settings.bundle_cache_ttl_seconds,
)
if cached is not None:
headers = {
"profile-update-interval": str(client.provider_interval),
"X-Sub-Provider-Bundle-Cache": "HIT",
}
headers.update(cached.headers)
return _yaml_response(content=cached.content, request=request, headers=headers, filename=f"bundle-{client_type}-{profile_key}.yaml")
try:
snapshots = await build_source_snapshots(source_items)
rules = build_conf_rule_lines(plan, base_path=settings.conf_base_file)
except Exception as exc: # noqa: BLE001
logger.exception("conf_bundle_profile failed: client=%s profile=%s", client_type, profile_key)
raise HTTPException(status_code=502, detail=f"failed to build conf bundle: {exc}") from exc
content = dump_yaml(
build_conf_bundle_profile(
client_type=client_type,
client=client,
plan=plan,
snapshots=snapshots,
rules=rules,
)
)
headers = {
"profile-update-interval": str(client.provider_interval),
"X-Sub-Provider-Bundle-Cache": "BYPASS" if force_refresh else "MISS",
}
headers.update(_quota_headers_from_snapshots(snapshots))
save_bundle_cache(
cache_dir=settings.bundle_cache_dir,
cache_key=cache_key,
content=content,
headers={key: value for key, value in headers.items() if key != "X-Sub-Provider-Bundle-Cache"},
)
return _yaml_response(content, request, headers=headers, filename=f"bundle-{client_type}-{profile_key}.yaml")

View File

@@ -0,0 +1,12 @@
from __future__ import annotations
from app.config import get_settings
from app.conf_models import ConfBaseConfig, ConfProfileConfig
from app.services.conf_loader import load_conf_base, load_conf_profiles
def load_conf_store() -> tuple[ConfBaseConfig, dict[str, ConfProfileConfig]]:
settings = get_settings()
base = load_conf_base(settings.conf_base_file)
profiles = load_conf_profiles(settings.conf_profiles_dir, base_config=base)
return base, profiles

360
app/services/conf_loader.py Normal file
View File

@@ -0,0 +1,360 @@
from __future__ import annotations
import csv
import os
import re
from pathlib import Path
from typing import Any
from app.conf_models import (
ConfBaseConfig,
ConfBuiltin,
ConfGroup,
ConfInlineRule,
ConfModule,
ConfPolicyOverride,
ConfProfileConfig,
ConfSelector,
ConfSource,
)
_ENV_PATTERN = re.compile(r"\$\{([A-Z0-9_]+)\}")
_SCALAR_KEYS = {
"listen",
"output_dir",
"cache_dir",
"mode",
"allow_lan",
"log_level",
"ipv6",
"append_userinfo_header",
"userinfo_source_policy",
}
_BOOL_KEYS = {"allow_lan", "ipv6", "append_userinfo_header"}
_MULTI_KEYS = {"source", "selector", "group", "module", "builtin"}
_SOURCE_TYPES = {"url", "file", "inline", "base64"}
_PROFILE_SCALAR_KEYS = {
"name",
"enabled",
"sources",
"include_modules",
"exclude_modules",
"include_builtins",
}
_PROFILE_MULTI_KEYS = {"override_policy", "prepend_rule", "append_rule"}
class ConfConfigError(ValueError):
def __init__(self, path: Path, line_no: int, message: str) -> None:
self.path = path
self.line_no = line_no
self.message = message
super().__init__(f"{path}:{line_no}: {message}")
def _expand_env(value: str) -> str:
return _ENV_PATTERN.sub(lambda match: os.getenv(match.group(1), ""), value)
def _parse_bool(value: str, *, path: Path, line_no: int, key: str) -> bool:
normalized = value.strip().lower()
mapping = {
"true": True,
"yes": True,
"1": True,
"false": False,
"no": False,
"0": False,
}
if normalized not in mapping:
raise ConfConfigError(path, line_no, f"invalid boolean for {key}: {value}")
return mapping[normalized]
def _parse_csv_fields(value: str, *, path: Path, line_no: int) -> list[str]:
try:
row = next(csv.reader([value], skipinitialspace=True))
except Exception as exc: # noqa: BLE001
raise ConfConfigError(path, line_no, f"invalid csv payload: {exc}") from exc
return [field.strip() for field in row]
def _split_key_value(raw_line: str, *, path: Path, line_no: int) -> tuple[str, str]:
if "=" not in raw_line:
raise ConfConfigError(path, line_no, "expected key = value")
key, value = [part.strip() for part in raw_line.split("=", 1)]
if not key:
raise ConfConfigError(path, line_no, "key cannot be empty")
return key, value
def _iter_conf_lines(path: Path) -> list[tuple[int, str]]:
records: list[tuple[int, str]] = []
for line_no, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
line = raw_line.strip()
if not line or line.startswith("#") or line.startswith(";"):
continue
records.append((line_no, raw_line))
return records
def _ensure_named_records(records: list[Any], *, key: str, path: Path) -> None:
seen: dict[str, int] = {}
for record in records:
name = getattr(record, key)
line_no = getattr(record, "line_no")
previous_line = seen.get(name)
if previous_line is not None:
raise ConfConfigError(path, line_no, f"duplicate {record.__class__.__name__} key '{name}', first defined at line {previous_line}")
seen[name] = line_no
def _parse_source(value: str, *, path: Path, line_no: int) -> ConfSource:
fields = _parse_csv_fields(_expand_env(value), path=path, line_no=line_no)
if len(fields) < 3:
raise ConfConfigError(path, line_no, "source requires at least 3 fields: key,type,value")
key, source_type, source_value, *extras = fields
if not key:
raise ConfConfigError(path, line_no, "source key cannot be empty")
if source_type not in _SOURCE_TYPES:
raise ConfConfigError(path, line_no, f"unsupported source type: {source_type}")
options: dict[str, str] = {}
enabled = True
cache_ttl: int | None = None
for item in extras:
if "=" not in item:
raise ConfConfigError(path, line_no, f"source option must be k=v: {item}")
option_key, option_value = [part.strip() for part in item.split("=", 1)]
if not option_key:
raise ConfConfigError(path, line_no, "source option key cannot be empty")
option_value = _expand_env(option_value)
if option_key == "enabled":
enabled = _parse_bool(option_value, path=path, line_no=line_no, key="source.enabled")
elif option_key == "cache_ttl":
try:
cache_ttl = int(option_value)
except ValueError as exc:
raise ConfConfigError(path, line_no, f"invalid integer for source.cache_ttl: {option_value}") from exc
else:
options[option_key] = option_value
return ConfSource(
key=key,
source_type=source_type,
value=source_value,
enabled=enabled,
cache_ttl=cache_ttl,
options=options,
line_no=line_no,
)
def _parse_selector(value: str, *, path: Path, line_no: int) -> ConfSelector:
fields = _parse_csv_fields(value, path=path, line_no=line_no)
if len(fields) != 2:
raise ConfConfigError(path, line_no, "selector requires exactly 2 fields: key,regex")
key, regex = fields
if not key or not regex:
raise ConfConfigError(path, line_no, "selector key and regex cannot be empty")
return ConfSelector(key=key, regex=regex, line_no=line_no)
def _parse_group(value: str, *, path: Path, line_no: int) -> ConfGroup:
segments = [segment.strip() for segment in value.split("`")]
if len(segments) < 2:
raise ConfConfigError(path, line_no, "group requires at least name`type")
name, group_type, *tokens = segments
if not name or not group_type:
raise ConfConfigError(path, line_no, "group name and type cannot be empty")
return ConfGroup(name=name, group_type=group_type, tokens=tokens, raw=value.strip(), line_no=line_no)
def _parse_module(value: str, *, path: Path, line_no: int, config_dir: Path) -> ConfModule:
fields = _parse_csv_fields(value, path=path, line_no=line_no)
if len(fields) != 5:
raise ConfConfigError(path, line_no, "module requires exactly 5 fields: key,path,policy,order,enabled")
key, module_path, policy, order_text, enabled_text = fields
if not key or not module_path or not policy:
raise ConfConfigError(path, line_no, "module key, path, and policy cannot be empty")
try:
order = int(order_text)
except ValueError as exc:
raise ConfConfigError(path, line_no, f"invalid integer for module.order: {order_text}") from exc
enabled = _parse_bool(enabled_text, path=path, line_no=line_no, key="module.enabled")
resolved_path = (config_dir / module_path).resolve()
if not resolved_path.is_file():
raise ConfConfigError(path, line_no, f"module path does not exist: {module_path}")
return ConfModule(key=key, path=module_path, policy=policy, order=order, enabled=enabled, line_no=line_no)
def _parse_builtin(value: str, *, path: Path, line_no: int) -> ConfBuiltin:
fields = _parse_csv_fields(value, path=path, line_no=line_no)
if len(fields) != 6:
raise ConfConfigError(path, line_no, "builtin requires exactly 6 fields: key,type,value,policy,order,enabled")
key, builtin_type, builtin_value, policy, order_text, enabled_text = fields
if builtin_type not in {"GEOIP", "FINAL"}:
raise ConfConfigError(path, line_no, f"unsupported builtin type: {builtin_type}")
if not key or not policy:
raise ConfConfigError(path, line_no, "builtin key and policy cannot be empty")
if builtin_type == "GEOIP" and not builtin_value:
raise ConfConfigError(path, line_no, "builtin GEOIP requires a value")
try:
order = int(order_text)
except ValueError as exc:
raise ConfConfigError(path, line_no, f"invalid integer for builtin.order: {order_text}") from exc
enabled = _parse_bool(enabled_text, path=path, line_no=line_no, key="builtin.enabled")
return ConfBuiltin(
key=key,
builtin_type=builtin_type,
value=builtin_value,
policy=policy,
order=order,
enabled=enabled,
line_no=line_no,
)
def _parse_string_list(value: str, *, path: Path, line_no: int) -> list[str]:
fields = _parse_csv_fields(_expand_env(value), path=path, line_no=line_no)
return [field for field in fields if field]
def _parse_override_policy(value: str, *, path: Path, line_no: int) -> ConfPolicyOverride:
fields = _parse_csv_fields(value, path=path, line_no=line_no)
if len(fields) != 2:
raise ConfConfigError(path, line_no, "override_policy requires exactly 2 fields: module_key,new_policy")
module_key, policy = fields
if not module_key or not policy:
raise ConfConfigError(path, line_no, "override_policy module key and policy cannot be empty")
return ConfPolicyOverride(module_key=module_key, policy=policy, line_no=line_no)
def _parse_inline_rule(value: str, *, path: Path, line_no: int, key: str) -> ConfInlineRule:
expanded = _expand_env(value).strip()
if not expanded:
raise ConfConfigError(path, line_no, f"{key} cannot be empty")
return ConfInlineRule(value=expanded, line_no=line_no)
def _validate_profile_references(
profile: ConfProfileConfig,
base: ConfBaseConfig,
*,
path: Path,
field_lines: dict[str, int],
) -> None:
source_keys = {item.key for item in base.sources}
module_keys = {item.key for item in base.modules}
builtin_keys = {item.key for item in base.builtins}
for source in profile.sources:
if source not in source_keys:
raise ConfConfigError(path, field_lines.get("sources", 1), f"profile references unknown source: {source}")
for module in profile.include_modules:
if module not in module_keys:
raise ConfConfigError(
path,
field_lines.get("include_modules", 1),
f"profile references unknown module in include_modules: {module}",
)
for module in profile.exclude_modules:
if module not in module_keys:
raise ConfConfigError(
path,
field_lines.get("exclude_modules", 1),
f"profile references unknown module in exclude_modules: {module}",
)
for builtin in profile.include_builtins:
if builtin not in builtin_keys:
raise ConfConfigError(
path,
field_lines.get("include_builtins", 1),
f"profile references unknown builtin in include_builtins: {builtin}",
)
for override in profile.override_policies:
if override.module_key not in module_keys:
raise ConfConfigError(path, override.line_no, f"override_policy references unknown module: {override.module_key}")
def load_conf_base(path: Path) -> ConfBaseConfig:
config = ConfBaseConfig()
scalar_values: dict[str, Any] = {}
config_dir = path.resolve().parent
for line_no, raw_line in _iter_conf_lines(path):
key, value = _split_key_value(raw_line, path=path, line_no=line_no)
if key in _SCALAR_KEYS:
if key in scalar_values:
raise ConfConfigError(path, line_no, f"duplicate scalar key: {key}")
expanded_value = _expand_env(value)
if key in _BOOL_KEYS:
scalar_values[key] = _parse_bool(expanded_value, path=path, line_no=line_no, key=key)
else:
scalar_values[key] = expanded_value
continue
if key not in _MULTI_KEYS:
raise ConfConfigError(path, line_no, f"unknown directive: {key}")
if key == "source":
config.sources.append(_parse_source(value, path=path, line_no=line_no))
elif key == "selector":
config.selectors.append(_parse_selector(value, path=path, line_no=line_no))
elif key == "group":
config.groups.append(_parse_group(value, path=path, line_no=line_no))
elif key == "module":
config.modules.append(_parse_module(value, path=path, line_no=line_no, config_dir=config_dir))
elif key == "builtin":
config.builtins.append(_parse_builtin(value, path=path, line_no=line_no))
for key, parsed_value in scalar_values.items():
setattr(config, key, parsed_value)
_ensure_named_records(config.sources, key="key", path=path)
_ensure_named_records(config.selectors, key="key", path=path)
_ensure_named_records(config.groups, key="name", path=path)
_ensure_named_records(config.modules, key="key", path=path)
_ensure_named_records(config.builtins, key="key", path=path)
return config
def load_conf_profile(path: Path, *, base_config: ConfBaseConfig) -> ConfProfileConfig:
profile = ConfProfileConfig()
scalar_values: dict[str, Any] = {}
field_lines: dict[str, int] = {}
for line_no, raw_line in _iter_conf_lines(path):
key, value = _split_key_value(raw_line, path=path, line_no=line_no)
if key in _PROFILE_SCALAR_KEYS:
if key in scalar_values:
raise ConfConfigError(path, line_no, f"duplicate scalar key: {key}")
field_lines[key] = line_no
if key == "enabled":
scalar_values[key] = _parse_bool(_expand_env(value), path=path, line_no=line_no, key=key)
elif key in {"sources", "include_modules", "exclude_modules", "include_builtins"}:
scalar_values[key] = _parse_string_list(value, path=path, line_no=line_no)
else:
scalar_values[key] = _expand_env(value)
continue
if key not in _PROFILE_MULTI_KEYS:
raise ConfConfigError(path, line_no, f"unknown directive: {key}")
if key == "override_policy":
profile.override_policies.append(_parse_override_policy(value, path=path, line_no=line_no))
elif key == "prepend_rule":
profile.prepend_rules.append(_parse_inline_rule(value, path=path, line_no=line_no, key=key))
elif key == "append_rule":
profile.append_rules.append(_parse_inline_rule(value, path=path, line_no=line_no, key=key))
for key, parsed_value in scalar_values.items():
setattr(profile, key, parsed_value)
_validate_profile_references(profile, base_config, path=path, field_lines=field_lines)
return profile
def load_conf_profiles(directory: Path, *, base_config: ConfBaseConfig) -> dict[str, ConfProfileConfig]:
profiles: dict[str, ConfProfileConfig] = {}
for path in sorted(directory.glob("*.conf")):
profile = load_conf_profile(path, base_config=base_config)
profile_key = path.stem
if profile_key in profiles:
raise ConfConfigError(path, 1, f"duplicate profile file key: {profile_key}")
profiles[profile_key] = profile
return profiles

View File

@@ -0,0 +1,353 @@
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

View File

@@ -0,0 +1,252 @@
from __future__ import annotations
from pathlib import Path
from app.conf_models import (
ConfBaseConfig,
ConfGroup,
ConfProfileConfig,
ConfResolvedBuiltin,
ConfResolvedGroup,
ConfResolvedModule,
ConfResolvedSource,
ConfRuntimePlan,
)
from app.services.conf_loader import ConfConfigError
from app.services.rules import load_rule_payload
_GROUP_TYPES = {"select", "url-test", "fallback", "load-balance"}
_CIDR_CHARS = set("0123456789abcdefABCDEF:./")
_SELECT_SPECIAL_MEMBERS = {"@source_auto_groups", "@all_proxies"}
def _parse_group_member(token: str) -> str:
if token.startswith("[]"):
return token[2:]
return token
def _parse_group(group: ConfGroup, *, selectors: dict[str, str], known_group_names: set[str], path: Path) -> ConfResolvedGroup:
if group.group_type not in _GROUP_TYPES:
raise ConfConfigError(path, group.line_no, f"unsupported group type: {group.group_type}")
if group.group_type == "select":
selector_key: str | None = None
members = [_parse_group_member(token) for token in group.tokens if token.strip()]
if not members:
raise ConfConfigError(path, group.line_no, "select group must define at least one member")
for token in group.tokens:
stripped = token.strip()
if stripped.startswith("@"):
if stripped in _SELECT_SPECIAL_MEMBERS:
continue
selector_key = stripped[1:]
if selector_key not in selectors:
raise ConfConfigError(path, group.line_no, f"group references unknown selector: {selector_key}")
continue
if stripped.startswith("[]"):
target = stripped[2:]
if target not in {"DIRECT", "REJECT"} and target not in known_group_names:
raise ConfConfigError(path, group.line_no, f"group references unknown group: {target}")
return ConfResolvedGroup(
name=group.name,
group_type=group.group_type,
members=[] if selector_key else members,
selector_key=selector_key,
raw=group.raw,
)
if not group.tokens:
raise ConfConfigError(path, group.line_no, f"{group.group_type} group requires selector/member arguments")
first = group.tokens[0].strip()
selector_key: str | None = None
members: list[str] = []
if first.startswith("@"):
selector_key = first[1:]
if selector_key not in selectors:
raise ConfConfigError(path, group.line_no, f"group references unknown selector: {selector_key}")
else:
members.append(_parse_group_member(first))
if first.startswith("[]"):
target = first[2:]
if target not in {"DIRECT", "REJECT"} and target not in known_group_names:
raise ConfConfigError(path, group.line_no, f"group references unknown group: {target}")
url = group.tokens[1].strip() if len(group.tokens) > 1 and group.tokens[1].strip() else None
interval: int | None = None
tolerance: int | None = None
if len(group.tokens) > 2 and group.tokens[2].strip():
try:
interval = int(group.tokens[2].split(",", 1)[0].strip())
except ValueError as exc:
raise ConfConfigError(path, group.line_no, f"invalid group interval: {group.tokens[2]}") from exc
if len(group.tokens) > 3 and group.tokens[3].strip():
tolerance_token = group.tokens[3].split(",")[-1].strip()
if tolerance_token:
try:
tolerance = int(tolerance_token)
except ValueError as exc:
raise ConfConfigError(path, group.line_no, f"invalid group tolerance: {group.tokens[3]}") from exc
return ConfResolvedGroup(
name=group.name,
group_type=group.group_type,
members=members,
selector_key=selector_key,
url=url,
interval=interval,
tolerance=tolerance,
raw=group.raw,
)
def _resolve_builtin_rule_line(builtin_type: str, value: str, policy: str) -> str:
if builtin_type == "GEOIP":
return f"GEOIP,{value},{policy}"
return f"MATCH,{policy}"
def _render_rule_line(payload_line: str, policy: str) -> str:
rendered = payload_line.strip()
if not rendered:
return ""
if "," not in rendered:
if "/" in rendered and set(rendered) <= _CIDR_CHARS:
prefix = "IP-CIDR6" if ":" in rendered else "IP-CIDR"
rendered = f"{prefix},{rendered}"
else:
rendered = f"DOMAIN-SUFFIX,{rendered}"
parts = [part.strip() for part in rendered.split(",")]
if parts and parts[-1] == "no-resolve":
parts.insert(len(parts) - 1, policy)
return ",".join(parts)
return f"{rendered},{policy}"
def build_conf_rule_lines(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:
module_path = (config_dir / module.path).resolve()
if not module_path.is_file():
raise FileNotFoundError(f"Rule file missing: {module.path}")
for payload_line in load_rule_payload(module_path):
rendered = _render_rule_line(payload_line, module.policy)
if rendered:
if rendered.startswith("MATCH,"):
final_lines.append(rendered)
else:
lines.append(rendered)
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 resolve_conf_runtime_plan(
*,
profile_key: str,
base_config: ConfBaseConfig,
profile_config: ConfProfileConfig,
base_path: Path,
) -> ConfRuntimePlan:
if not profile_config.enabled:
raise ConfConfigError(base_path, 1, f"profile is disabled: {profile_key}")
selectors = {item.key: item.regex for item in base_config.selectors}
source_map = {item.key: item for item in base_config.sources}
module_map = {item.key: item for item in base_config.modules}
builtin_map = {item.key: item for item in base_config.builtins}
override_map = {item.module_key: item.policy for item in profile_config.override_policies}
selected_source_keys = profile_config.sources or [item.key for item in base_config.sources if item.enabled]
selected_sources = [
ConfResolvedSource(
key=source_map[key].key,
display_name=str(source_map[key].options.get("display_name", source_map[key].key)).strip() or source_map[key].key,
source_type=source_map[key].source_type,
value=source_map[key].value,
enabled=source_map[key].enabled,
cache_ttl=source_map[key].cache_ttl,
options=dict(source_map[key].options),
)
for key in selected_source_keys
if source_map[key].enabled
]
excluded_modules = set(profile_config.exclude_modules)
include_keys = profile_config.include_modules or [item.key for item in base_config.modules if item.enabled]
selected_modules: list[ConfResolvedModule] = []
for key in include_keys:
module = module_map[key]
if not module.enabled or key in excluded_modules:
continue
selected_modules.append(
ConfResolvedModule(
key=module.key,
path=module.path,
policy=override_map.get(module.key, module.policy),
order=module.order,
enabled=module.enabled,
)
)
selected_modules.sort(key=lambda item: (item.order, item.key))
builtin_keys = profile_config.include_builtins or [item.key for item in base_config.builtins if item.enabled]
selected_builtins: list[ConfResolvedBuiltin] = []
for key in builtin_keys:
builtin = builtin_map[key]
if not builtin.enabled:
continue
selected_builtins.append(
ConfResolvedBuiltin(
key=builtin.key,
rule_line=_resolve_builtin_rule_line(builtin.builtin_type, builtin.value, builtin.policy),
policy=builtin.policy,
order=builtin.order,
enabled=builtin.enabled,
)
)
selected_builtins.sort(key=lambda item: (item.order, item.key))
known_group_names = {group.name for group in base_config.groups}
resolved_groups = [
_parse_group(group, selectors=selectors, known_group_names=known_group_names, path=base_path)
for group in base_config.groups
]
return ConfRuntimePlan(
profile_key=profile_key,
profile_name=profile_config.name,
selected_sources=selected_sources,
selected_modules=selected_modules,
selected_builtins=selected_builtins,
selectors=selectors,
groups=resolved_groups,
prepend_rules=[item.value for item in profile_config.prepend_rules],
append_rules=[item.value for item in profile_config.append_rules],
mode=base_config.mode,
allow_lan=base_config.allow_lan,
log_level=base_config.log_level,
ipv6=base_config.ipv6,
append_userinfo_header=base_config.append_userinfo_header,
userinfo_source_policy=base_config.userinfo_source_policy,
)

82
config/conf/base.conf Normal file
View File

@@ -0,0 +1,82 @@
listen = 0.0.0.0:3000
output_dir = ./output
cache_dir = ./data/cache
mode = rule
allow_lan = true
log_level = info
ipv6 = true
append_userinfo_header = true
userinfo_source_policy = first_enabled_source
source = airport-a,url,${AIRPORT_A_URL},enabled=true,cache_ttl=1800,display_name=A,prefix=[A]
source = airport-b,url,${AIRPORT_B_URL},enabled=true,cache_ttl=1800,display_name=B,prefix=[B]
source = airport-c,url,${AIRPORT_C_URL},enabled=true,cache_ttl=1800,display_name=C,prefix=[C]
selector = all,.*
selector = hk,(?i)(港|hk|hong kong|hongkong)
selector = tw,(?i)(台|新北|彰化|tw|taiwan)
selector = sg,(?i)(新加坡|坡|狮城|sg|singapore)
selector = jp,(?i)(日本|东京|大阪|埼玉|jp|japan)
selector = us,(?i)(美|us|united states)
selector = kr,(?i)(kr|korea|kor|首尔|韩|韓)
selector = netflix,(?i)(nf|奈飞|解锁|netflix|media)
group = ♻️ 自动选择`url-test`@all`https://www.gstatic.com/generate_204`300,,50
group = ☁️ 机场选择`select`[]♻️ 自动选择`@source_auto_groups`[]DIRECT
group = 🚀 手动切换`select`@all_proxies`[]DIRECT
group = 🚀 节点选择`select`[]☁️ 机场选择`[]♻️ 自动选择`[]🇭🇰 香港节点`[]🇨🇳 台湾节点`[]🇸🇬 狮城节点`[]🇯🇵 日本节点`[]🇺🇲 美国节点`[]🇰🇷 韩国节点`[]🎥 奈飞节点`[]🚀 手动切换`[]DIRECT
group = 💬 Ai平台`select`[]🚀 节点选择`[]♻️ 自动选择`[]🇺🇲 美国节点`[]🇸🇬 狮城节点`[]🇯🇵 日本节点`[]🇭🇰 香港节点`[]🇨🇳 台湾节点`[]🇰🇷 韩国节点`[]🚀 手动切换`[]DIRECT
group = 📢 谷歌`select`[]🚀 节点选择`[]DIRECT`[]🇺🇲 美国节点`[]🇭🇰 香港节点`[]🇨🇳 台湾节点`[]🇸🇬 狮城节点`[]🇯🇵 日本节点`[]🇰🇷 韩国节点`[]🚀 手动切换
group = 🎮 PT平台`select`[]🚀 节点选择`[]DIRECT`[]🇭🇰 香港节点`[]🇨🇳 台湾节点`[]🇸🇬 狮城节点`[]🇯🇵 日本节点`[]🇺🇲 美国节点`[]🇰🇷 韩国节点`[]🚀 手动切换
group = 📲 电报消息`select`[]🚀 节点选择`[]♻️ 自动选择`[]🇸🇬 狮城节点`[]🇭🇰 香港节点`[]🇨🇳 台湾节点`[]🇯🇵 日本节点`[]🇺🇲 美国节点`[]🇰🇷 韩国节点`[]🚀 手动切换`[]DIRECT
group = 📹 油管视频`select`[]🚀 节点选择`[]♻️ 自动选择`[]🇸🇬 狮城节点`[]🇭🇰 香港节点`[]🇨🇳 台湾节点`[]🇯🇵 日本节点`[]🇺🇲 美国节点`[]🇰🇷 韩国节点`[]🚀 手动切换`[]DIRECT
group = 🎥 奈飞视频`select`[]🎥 奈飞节点`[]🚀 节点选择`[]♻️ 自动选择`[]🇸🇬 狮城节点`[]🇭🇰 香港节点`[]🇨🇳 台湾节点`[]🇯🇵 日本节点`[]🇺🇲 美国节点`[]🇰🇷 韩国节点`[]🚀 手动切换`[]DIRECT
group = 🌍 国外媒体`select`[]🚀 节点选择`[]♻️ 自动选择`[]🇭🇰 香港节点`[]🇨🇳 台湾节点`[]🇸🇬 狮城节点`[]🇯🇵 日本节点`[]🇺🇲 美国节点`[]🇰🇷 韩国节点`[]🚀 手动切换`[]DIRECT
group = Ⓜ️ 微软Bing`select`[]DIRECT`[]🚀 节点选择`[]🇺🇲 美国节点`[]🇭🇰 香港节点`[]🇨🇳 台湾节点`[]🇸🇬 狮城节点`[]🇯🇵 日本节点`[]🇰🇷 韩国节点`[]🚀 手动切换
group = Ⓜ️ 微软云盘`select`[]DIRECT`[]🚀 节点选择`[]🇺🇲 美国节点`[]🇭🇰 香港节点`[]🇨🇳 台湾节点`[]🇸🇬 狮城节点`[]🇯🇵 日本节点`[]🇰🇷 韩国节点`[]🚀 手动切换
group = Ⓜ️ 微软服务`select`[]DIRECT`[]🚀 节点选择`[]🇺🇲 美国节点`[]🇭🇰 香港节点`[]🇨🇳 台湾节点`[]🇸🇬 狮城节点`[]🇯🇵 日本节点`[]🇰🇷 韩国节点`[]🚀 手动切换
group = 🍎 苹果服务`select`[]DIRECT`[]🚀 节点选择`[]🇺🇲 美国节点`[]🇭🇰 香港节点`[]🇨🇳 台湾节点`[]🇸🇬 狮城节点`[]🇯🇵 日本节点`[]🇰🇷 韩国节点`[]🚀 手动切换
group = 🎮 游戏平台`select`[]🚀 节点选择`[]DIRECT`[]🇺🇲 美国节点`[]🇭🇰 香港节点`[]🇨🇳 台湾节点`[]🇸🇬 狮城节点`[]🇯🇵 日本节点`[]🇰🇷 韩国节点`[]🚀 手动切换
group = 🎯 全球直连`select`[]DIRECT`[]🚀 节点选择`[]♻️ 自动选择
group = 🛑 广告拦截`select`[]REJECT`[]DIRECT
group = 🍃 应用净化`select`[]REJECT`[]DIRECT
group = 🐟 漏网之鱼`select`[]DIRECT`[]🚀 节点选择`[]♻️ 自动选择`[]🇭🇰 香港节点`[]🇨🇳 台湾节点`[]🇸🇬 狮城节点`[]🇯🇵 日本节点`[]🇺🇲 美国节点`[]🇰🇷 韩国节点`[]🎥 奈飞节点`[]🚀 手动切换
group = 🇭🇰 香港节点`url-test`@hk`https://www.gstatic.com/generate_204`300,,50
group = 🇨🇳 台湾节点`url-test`@tw`https://www.gstatic.com/generate_204`300,,50
group = 🇸🇬 狮城节点`url-test`@sg`https://www.gstatic.com/generate_204`300,,50
group = 🇯🇵 日本节点`url-test`@jp`https://www.gstatic.com/generate_204`300,,50
group = 🇺🇲 美国节点`url-test`@us`https://www.gstatic.com/generate_204`300,,150
group = 🇰🇷 韩国节点`url-test`@kr`https://www.gstatic.com/generate_204`300,,50
group = 🎥 奈飞节点`select`@netflix
module = custom-proxy,../rules/modules/custom-proxy.list,🚀 节点选择,1,true
module = local-network,../rules/acl4ssr/LocalAreaNetwork.yaml,🎯 全球直连,10,true
module = unban,../rules/acl4ssr/UnBan.yaml,🎯 全球直连,20,true
module = reject,../rules/acl4ssr/BanAD.yaml,🛑 广告拦截,40,true
module = app-purify,../rules/acl4ssr/BanProgramAD.yaml,🍃 应用净化,50,true
module = google,../rules/acl4ssr/Google.yaml,📢 谷歌,60,true
module = google-cn,../rules/acl4ssr/GoogleCN.yaml,🎯 全球直连,61,true
module = steam-cn,../rules/acl4ssr/SteamCN.yaml,🎯 全球直连,62,true
module = microsoft-bing,../rules/acl4ssr/Bing.yaml,Ⓜ️ 微软Bing,70,true
module = microsoft-onedrive,../rules/acl4ssr/OneDrive.yaml,Ⓜ️ 微软云盘,71,true
module = microsoft,../rules/acl4ssr/Microsoft.yaml,Ⓜ️ 微软服务,72,true
module = apple,../rules/acl4ssr/Apple.yaml,🍎 苹果服务,80,true
module = telegram,../rules/acl4ssr/Telegram.yaml,📲 电报消息,90,true
module = ai,../rules/acl4ssr/AI.yaml,💬 Ai平台,99,true
module = openai,../rules/acl4ssr/OpenAi.yaml,💬 Ai平台,100,true
module = youtube,../rules/acl4ssr/YouTube.yaml,📹 油管视频,110,true
module = netflix,../rules/acl4ssr/Netflix.yaml,🎥 奈飞视频,111,true
module = proxy-media,../rules/acl4ssr/ProxyMedia.yaml,🌍 国外媒体,112,true
module = games-epic,../rules/acl4ssr/Epic.yaml,🎮 游戏平台,120,true
module = games-origin,../rules/acl4ssr/Origin.yaml,🎮 游戏平台,121,true
module = games-sony,../rules/acl4ssr/Sony.yaml,🎮 游戏平台,122,true
module = games-steam,../rules/acl4ssr/Steam.yaml,🎮 游戏平台,123,true
module = games-nintendo,../rules/acl4ssr/Nintendo.yaml,🎮 游戏平台,124,true
module = pt,../rules/acl4ssr/PrivateTracker.yaml,🎮 PT平台,125,true
module = cn-domain,../rules/acl4ssr/ChinaDomain.yaml,🎯 全球直连,130,true
module = cn-company-ip,../rules/acl4ssr/ChinaCompanyIp.yaml,🎯 全球直连,131,true
module = download,../rules/acl4ssr/Download.yaml,🎯 全球直连,132,true
module = proxy-gfw,../rules/acl4ssr/ProxyGFWlistLite.yaml,🚀 节点选择,140,true
builtin = geoip_cn,GEOIP,CN,🎯 全球直连,9000,true
builtin = final,FINAL,,🚀 节点选择,9999,true

View File

@@ -0,0 +1,6 @@
name = 仅 A
enabled = true
sources = airport-a
exclude_modules =
include_builtins = geoip_cn,final

View File

@@ -0,0 +1,6 @@
name = 三机场全开
enabled = true
sources = airport-a,airport-b,airport-c
exclude_modules =
include_builtins = geoip_cn,final

View File

@@ -0,0 +1,6 @@
name = 仅 B
enabled = true
sources = airport-b
exclude_modules =
include_builtins = geoip_cn,final

View File

@@ -0,0 +1,6 @@
name = 仅 C
enabled = true
sources = airport-c
exclude_modules =
include_builtins = geoip_cn,final

View File

@@ -0,0 +1,6 @@
name = 默认完整配置
enabled = true
sources = airport-a,airport-b
exclude_modules =
include_builtins = geoip_cn,final

View File

@@ -0,0 +1,9 @@
name = 精简配置
enabled = true
sources = airport-a
include_modules = openai
exclude_modules =
include_builtins = final
prepend_rule = DOMAIN-SUFFIX,lite.example.com,DIRECT

View File

@@ -0,0 +1,104 @@
payload:
- DOMAIN-KEYWORD,facebook
- DOMAIN-KEYWORD,github
- DOMAIN-KEYWORD,instagram
- DOMAIN-KEYWORD,twitter
- DOMAIN-KEYWORD,whatsapp
- DOMAIN-SUFFIX,bbc.co
- DOMAIN-SUFFIX,bbc.co.uk
- DOMAIN-SUFFIX,bbc.com
- DOMAIN-SUFFIX,bbc.in
- DOMAIN-SUFFIX,bbcchinese.com
- DOMAIN-SUFFIX,bbci.co.uk
- DOMAIN-SUFFIX,bloomberg.cn
- DOMAIN-SUFFIX,bloomberg.com
- DOMAIN-SUFFIX,bloomberg.de
- DOMAIN-SUFFIX,bloombergview.com
- DOMAIN-SUFFIX,cdninstagram.com
- DOMAIN-SUFFIX,cloudflare-dns.com
- DOMAIN-SUFFIX,cloudflare-ipfs.com
- DOMAIN-SUFFIX,cloudflarestatus.com
- DOMAIN-SUFFIX,discord.co
- DOMAIN-SUFFIX,discord.com
- DOMAIN-SUFFIX,discord.gg
- DOMAIN-SUFFIX,discord.media
- DOMAIN-SUFFIX,discordapp.com
- DOMAIN-SUFFIX,discordapp.net
- DOMAIN-SUFFIX,economist.com
- DOMAIN-SUFFIX,facebook.br
- DOMAIN-SUFFIX,facebook.com
- DOMAIN-SUFFIX,facebook.de
- DOMAIN-SUFFIX,facebook.design
- DOMAIN-SUFFIX,facebook.hu
- DOMAIN-SUFFIX,facebook.in
- DOMAIN-SUFFIX,facebook.net
- DOMAIN-SUFFIX,facebook.nl
- DOMAIN-SUFFIX,facebook.se
- DOMAIN-SUFFIX,facebookmail.com
- DOMAIN-SUFFIX,fbcdn.com
- DOMAIN-SUFFIX,fbcdn.net
- DOMAIN-SUFFIX,github.blog
- DOMAIN-SUFFIX,github.com
- DOMAIN-SUFFIX,github.io
- DOMAIN-SUFFIX,githubapp.com
- DOMAIN-SUFFIX,githubassets.com
- DOMAIN-SUFFIX,githubcopilot.com
- DOMAIN-SUFFIX,githubmemory.com
- DOMAIN-SUFFIX,githubusercontent.com
- DOMAIN-SUFFIX,gitlab.com
- DOMAIN-SUFFIX,gitlab.io
- DOMAIN-SUFFIX,gitlab.net
- DOMAIN-SUFFIX,instagram.com
- DOMAIN-SUFFIX,jsdelivr.net
- DOMAIN-SUFFIX,medium.com
- DOMAIN-SUFFIX,netlify.app
- DOMAIN-SUFFIX,nytimes.com
- DOMAIN-SUFFIX,nytimes.map.fastly.net
- DOMAIN-SUFFIX,oraclecloud.com
- DOMAIN-SUFFIX,pages.dev
- DOMAIN-SUFFIX,pincong.rocks
- DOMAIN-SUFFIX,rawgithub.com
- DOMAIN-SUFFIX,redd.it
- DOMAIN-SUFFIX,reddit.com
- DOMAIN-SUFFIX,reddithelp.com
- DOMAIN-SUFFIX,redditlist.com
- DOMAIN-SUFFIX,redditmedia.com
- DOMAIN-SUFFIX,redditspace.com
- DOMAIN-SUFFIX,redditstatic.com
- DOMAIN-SUFFIX,reuters.com
- DOMAIN-SUFFIX,reutersmedia.net
- DOMAIN-SUFFIX,signal.org
- DOMAIN-SUFFIX,sourceforge.net
- DOMAIN-SUFFIX,static-economist.com
- DOMAIN-SUFFIX,substack.com
- DOMAIN-SUFFIX,thefacebook.com
- DOMAIN-SUFFIX,threads.com
- DOMAIN-SUFFIX,threads.net
- DOMAIN-SUFFIX,twitter.com
- DOMAIN-SUFFIX,twitter.jp
- DOMAIN-SUFFIX,twitter4j.org
- DOMAIN-SUFFIX,twittercounter.com
- DOMAIN-SUFFIX,twitterfeed.com
- DOMAIN-SUFFIX,twittergadget.com
- DOMAIN-SUFFIX,twitterkr.com
- DOMAIN-SUFFIX,twittermail.com
- DOMAIN-SUFFIX,twitterrific.com
- DOMAIN-SUFFIX,twittertim.es
- DOMAIN-SUFFIX,v2ex.com
- DOMAIN-SUFFIX,vercel.app
- DOMAIN-SUFFIX,whatsapp.com
- DOMAIN-SUFFIX,whatsapp.net
- DOMAIN-SUFFIX,wikibooks.org
- DOMAIN-SUFFIX,wikimedia.org
- DOMAIN-SUFFIX,wikinews.org
- DOMAIN-SUFFIX,wikipedia.com
- DOMAIN-SUFFIX,wikipedia.org
- DOMAIN-SUFFIX,wikiquote.org
- DOMAIN-SUFFIX,wikisource.org
- DOMAIN-SUFFIX,wikiversity.org
- DOMAIN-SUFFIX,wikivoyage.org
- DOMAIN-SUFFIX,wiktionary.org
- DOMAIN-SUFFIX,workers.dev
- DOMAIN-SUFFIX,wsj.com
- DOMAIN-SUFFIX,wsj.net
- DOMAIN-SUFFIX,wsjhk.com

View File

@@ -0,0 +1,8 @@
DOMAIN-KEYWORD,cloudflare
DOMAIN-KEYWORD,hetzner
DOMAIN-KEYWORD,hdkylin
DOMAIN-KEYWORD,steamusercontent
DOMAIN-KEYWORD,steamcontent
DOMAIN-KEYWORD,nintendo
DOMAIN-KEYWORD,vscode
DOMAIN-KEYWORD,btschool

View File

@@ -435,7 +435,7 @@ rules:
interval: 86400
proxy-gfw:
file: acl4ssr/ProxyGFWlist.yaml
file: acl4ssr/ProxyGFWlistLite.yaml
behavior: classical
format: yaml
policy: "{{ main_policy }}"

809
docs/ini改造方案.md Normal file
View File

@@ -0,0 +1,809 @@
可以。下面这份就是一版 **“类 ACL4SSR 风格 INI/CONF 配置整体说明”**,你可以直接丢给 Codex。
它刻意 **不引入 YAML**,只保留高密度、行导向、声明式的写法。你现在贴的配置本身已经在用 `[custom]``ruleset=``custom_proxy_group=``enable_rule_generator=true``overwrite_original_rules=true` 这套思路subconverter 现有 ACL4SSR 相关配置和外部配置 gist 里也能看到同类结构,甚至有把公共规则和公共分组拆出去再导入的写法。 ([GitHub][1])
---
# sub-provider 类 ACL4SSR 风格 INI 配置说明(草案)
## 1. 目标
将 sub-provider 的配置设计为一种 **ACL4SSR-like 的高密度 INI/CONF DSL**,用于:
* 管理订阅源
* 定义节点选择器
* 定义代理组
* 注册规则模块
* 选择输出 profile
* 生成完整 Clash/Mihomo YAML
这里的重点不是兼容标准 INI 解析器,而是:
> **保留 INI 的可读感 + ACL4SSR 的高密度声明风格 + 方便 Python 自定义解析**
---
## 2. 设计原则
### 2.1 保留 ACL4SSR 的使用习惯
沿用这些核心心智模型:
* `ruleset = 规则模块 -> 目标策略组`
* `group = 策略组定义`
* `selector = 节点筛选器`
* `source = 订阅源`
* `profile = 输出方案`
### 2.2 高密度、行导向
每一条配置尽量一行表达一个对象,方便:
* 快速编辑
* 批量复制
* diff
* 注释开关
* 手工维护
### 2.3 规则正文与配置分离
* `*.list` 文件只保存规则主体
* 目标策略组由配置指定
* 生成器负责最终拼接
### 2.4 不追求严格标准 INI
原因:
* 标准 INI 不擅长重复 key
* 不擅长表达列表对象
* 不擅长表达高密度分组语法
因此这里定义的是:
> **INI-like / CONF-like 自定义配置格式**
---
## 3. 文件组织
推荐结构:
```text
config/
base.conf
profiles/
default.conf
lite.conf
media.conf
rules/
modules/
Riglen.list
LocalAreaNetwork.list
UnBan.list
BanAD.list
BanProgramAD.list
Google.list
GoogleCN.list
SteamCN.list
Bing.list
OneDrive.list
Microsoft.list
Apple.list
Telegram.list
AI.list
OpenAi.list
YouTube.list
Netflix.list
ProxyMedia.list
ProxyGFWlist.list
Pt.list
ChinaDomain.list
ChinaCompanyIp.list
Download.list
```
### 说明
* `base.conf`:全局配置、源、选择器、组定义、模块注册、内置规则
* `profiles/*.conf`:不同输出方案
* `rules/modules/*.list`:规则正文
---
## 4. 基础语法
## 4.1 注释
支持两种注释:
```ini
; 这是注释
# 这也是注释
```
## 4.2 空行
空行忽略。
## 4.3 键值形式
基本格式:
```ini
key = value
```
左右空格允许存在,解析时应 trim。
## 4.4 重复 key
允许重复 key。
例如:
```ini
source = ...
source = ...
selector = ...
selector = ...
group = ...
group = ...
module = ...
module = ...
```
解析器应将这类 key 视为“多条记录”,而不是覆盖。
## 4.5 大小写
建议:
* 指令名小写
* 值区分大小写
* 策略组名、规则内容、正则表达式保持原样
---
## 5. 支持的指令
本方案定义以下核心指令:
* `listen`
* `output_dir`
* `cache_dir`
* `mode`
* `allow_lan`
* `log_level`
* `ipv6`
* `append_userinfo_header`
* `userinfo_source_policy`
以及对象型指令:
* `source =`
* `selector =`
* `group =`
* `module =`
* `builtin =`
profile 内支持:
* `name =`
* `enabled =`
* `sources =`
* `include_modules =`
* `exclude_modules =`
* `include_builtins =`
* `override_policy =`
* `prepend_rule =`
* `append_rule =`
---
## 6. base.conf 说明
## 6.1 全局标量配置
### 示例
```ini
listen = 0.0.0.0:3000
output_dir = ./output
cache_dir = ./data/cache
mode = rule
allow_lan = true
log_level = info
ipv6 = false
append_userinfo_header = true
userinfo_source_policy = first_enabled_source
```
### 含义
* `listen`:监听地址
* `output_dir`:生成结果输出目录
* `cache_dir`:缓存目录
* `mode`Clash 模式,通常为 `rule`
* `allow_lan`:是否允许局域网访问
* `log_level`:日志级别
* `ipv6`:是否开启 IPv6
* `append_userinfo_header`:是否透传订阅流量头
* `userinfo_source_policy`:多源时如何选取订阅信息头
---
## 6.2 source 指令
### 目标
定义订阅输入源。
### 格式
```ini
source = key,type,value,enabled=true,cache_ttl=1800
```
### 示例
```ini
source = airport_main,url,https://example.com/sub?token=xxxx,enabled=true,cache_ttl=1800
source = local_links,file,./data/local_links.txt,enabled=false
```
### 字段说明
* 第 1 列:`key`
* 第 2 列:`type`
* 第 3 列:`value`
* 后续:可选命名参数
### 支持类型
* `url`:远程订阅链接
* `file`:本地文件
* `inline`:内联文本
* `base64`base64 内容
### 建议解析规则
前三列固定,后续按 `k=v` 解析。
---
## 6.3 selector 指令
### 目标
定义节点筛选器,供代理组复用。
### 格式
```ini
selector = key,regex
```
### 示例
```ini
selector = all,.*
selector = hk,(港|HK|hk|Hong Kong|HongKong|hongkong)
selector = jp,(日本|东京|大阪|JP|Japan)
selector = us,(美|洛杉矶|西雅图|US|United States)
selector = netflix,(NF|奈飞|解锁|Netflix|NETFLIX|Media)
```
### 含义
* `key`:选择器名
* `regex`:匹配节点名的正则表达式
### 约定
后续在 `group` 指令中通过 `@selector_key` 引用。
---
## 6.4 group 指令
这是整个 DSL 的核心。
它保留 ACL4SSR 风格的高密度分组写法。你的示例里已经大量使用 `custom_proxy_group=` 这种反引号分隔格式。 现有 subconverter 相关 ACL4SSR 配置里也是同类写法。([GitHub][1])
### 格式
```ini
group = 组名`类型`参数1`参数2`参数3...
```
### 成员约定
* `[]组名`:引用已有组
* `[]DIRECT`:内置 DIRECT
* `[]REJECT`:内置 REJECT
* `@selector_key`:引用 selector 动态筛节点
* `.*`:兼容保留写法,可映射为全部节点
### 示例
```ini
group = 🚀 节点选择`select`[]♻️ 自动选择`[]🇭🇰 香港节点`[]🇯🇵 日本节点`[]🚀 手动切换`[]DIRECT
group = 🚀 手动切换`select`@all
group = ♻️ 自动选择`url-test`@all`http://www.gstatic.com/generate_204`300,,50
group = 🇭🇰 香港节点`url-test`@hk`http://www.gstatic.com/generate_204`300,,50
group = 🇯🇵 日本节点`url-test`@jp`http://www.gstatic.com/generate_204`300,,50
group = 🎥 奈飞节点`select`@netflix
group = 🐟 漏网之鱼`select`[]DIRECT`[]🚀 节点选择`[]♻️ 自动选择
```
### 解析规则建议
#### `select`
```ini
group = 组名`select`成员1`成员2`成员3...
```
#### `url-test`
```ini
group = 组名`url-test`@selector`测试URL`间隔,,容差
```
#### `fallback`
```ini
group = 组名`fallback`@selector`测试URL`间隔,,容差
```
#### `load-balance`
```ini
group = 组名`load-balance`@selector`测试URL`间隔,,容差
```
### 设计建议
不要再继续沿用 `custom_proxy_group=` 这个旧名字。
内部 DSL 里直接统一为:
```ini
group = ...
```
这样更短,也更像你自己的项目语法。
---
## 6.5 module 指令
### 目标
注册规则模块。
### 格式
```ini
module = key,path,policy,order,enabled
```
### 示例
```ini
module = riglen,./rules/modules/Riglen.list,🚀 节点选择,10,true
module = lan,./rules/modules/LocalAreaNetwork.list,🎯 全球直连,20,true
module = banad,./rules/modules/BanAD.list,🛑 广告拦截,40,true
module = apple,./rules/modules/Apple.list,🍎 苹果服务,80,true
module = ai,./rules/modules/AI.list,💬 Ai平台,100,true
module = openai,./rules/modules/OpenAi.list,💬 Ai平台,101,true
```
### 字段说明
* `key`:模块唯一标识
* `path`:规则文件路径
* `policy`:默认策略组
* `order`:排序
* `enabled`:默认是否启用
### 规则文件约定
规则文件只保存规则主体,不含目标策略组。
例如 `OpenAi.list`
```ini
DOMAIN-SUFFIX,openai.com
DOMAIN-SUFFIX,chatgpt.com
DOMAIN-SUFFIX,oaistatic.com
```
生成时补成:
```ini
DOMAIN-SUFFIX,openai.com,💬 Ai平台
```
---
## 6.6 builtin 指令
### 目标
定义不来自 `.list` 文件的内置规则。
### 格式
```ini
builtin = key,type,value,policy,order,enabled
```
### 示例
```ini
builtin = geoip_cn,GEOIP,CN,🎯 全球直连,9000,true
builtin = final,FINAL,,🐟 漏网之鱼,9999,true
```
### 映射规则
* `GEOIP + CN + 🎯 全球直连`
-> `GEOIP,CN,🎯 全球直连`
* `FINAL + 🐟 漏网之鱼`
-> `MATCH,🐟 漏网之鱼`
### 说明
这里保留了 ACL4SSR 配置里 `[]GEOIP,CN``[]FINAL` 这种内置规则思想,只是换成更适合你项目的数据化声明。你的现有配置里已经有这两类用法。 ([GitHub][1])
---
## 7. profile 配置说明
每个 profile 一个文件,放在 `config/profiles/` 下。
---
## 7.1 基本字段
### 示例
```ini
name = 默认完整配置
enabled = true
sources = airport_main
include_modules = riglen,lan,unban,banad,apple,telegram,ai,openai
exclude_modules =
include_builtins = geoip_cn,final
```
### 含义
* `name`profile 名称
* `enabled`:是否启用
* `sources`:本 profile 使用哪些订阅源
* `include_modules`:启用哪些模块
* `exclude_modules`:从启用列表中排除哪些模块
* `include_builtins`:启用哪些内置规则
---
## 7.2 override_policy 指令
### 目标
覆盖模块默认策略组。
### 格式
```ini
override_policy = module_key,new_policy
```
### 示例
```ini
override_policy = apple,🚀 节点选择
override_policy = ai,💬 Ai平台
```
### 含义
例如某模块默认是 `🍎 苹果服务`,某 profile 想改成 `🚀 节点选择`,就在 profile 里覆盖。
---
## 7.3 prepend_rule / append_rule 指令
### 目标
给特定 profile 注入自定义规则。
### 示例
```ini
prepend_rule = DOMAIN-SUFFIX,internal.example.com,DIRECT
append_rule = DOMAIN-SUFFIX,test.example.com,🚀 节点选择
```
### 含义
* `prepend_rule`:插在 rules 最前
* `append_rule`:插在 builtin 之前或之后,由实现约定
### 建议
`MATCH` / `FINAL` 仍应最后输出。
---
## 8. 解析器行为规范
## 8.1 base.conf
解析器应支持:
* 普通标量键值
* 多条 `source`
* 多条 `selector`
* 多条 `group`
* 多条 `module`
* 多条 `builtin`
## 8.2 profile.conf
解析器应支持:
* 普通标量键值
* 多条 `override_policy`
* 多条 `prepend_rule`
* 多条 `append_rule`
## 8.3 字段 trim
建议默认:
* 去掉左右空白
* 不改动组名内部空格
* 不改动正则原文
## 8.4 布尔值
支持:
* `true/false`
* `yes/no`
* `1/0`
内部统一成布尔值。
## 8.5 列表字段
例如:
```ini
sources = airport_main,local_links
include_modules = apple,ai,openai
```
按逗号切分trim 后转数组。
## 8.6 module/source 的 CSV 解析
建议用 Python `csv` 模块解析逗号分隔,而不是简单 `split(",")`,避免后续扩展时出问题。
## 8.7 group 的反引号解析
建议:
1. 去掉前缀 `group =`
2. 按反引号 `` ` `` 分段
3. 第一段为组名
4. 第二段为类型
5. 后续段按组类型解释
---
## 9. 生成流程规范
生成器处理顺序建议如下:
### 第 1 步:加载 base.conf
构建:
* 源表
* selector 表
* group 定义表
* module 注册表
* builtin 注册表
### 第 2 步:加载指定 profile.conf
得到:
* 使用哪些源
* 启用哪些模块
* 启用哪些 builtin
* policy override
* prepend/append 规则
### 第 3 步:拉取并解析节点
* 拉 URL 订阅
* 读本地文件
* 统一解析为节点对象
* 去重 / 重命名 / 过滤
### 第 4 步:按 selector 构建节点集合
例如:
* `hk` -> 匹配香港节点
* `us` -> 匹配美国节点
* `all` -> 全部节点
### 第 5 步:构建 proxy-groups
解析 `group = ...` 指令,生成最终组定义。
### 第 6 步:构建 rules
1. 按模块顺序加载 `.list`
2. 对每行规则补上目标 policy
3. 应用 `override_policy`
4. 加入 `prepend_rule`
5. 加入 builtin
6. 加入 `append_rule`
7. 确保 `FINAL/MATCH` 最后
### 第 7 步:渲染完整 YAML
这里是生成器内部输出,不需要在配置层暴露 YAML 结构给用户。
---
## 10. 错误处理建议
以下情况应报错并指出行号:
* 未知指令
* `group` 语法不合法
* `module` 路径不存在
* profile 引用了不存在的 `source`
* profile 引用了不存在的 `module`
* `override_policy` 指向未知模块
* `@selector_key` 未定义
* builtin 类型不支持
以下情况可仅 warning
* selector 匹配不到节点
* group 最终成员为空
* 模块文件为空
* 规则重复
---
## 11. 兼容性策略建议
## 11.1 兼容旧 ACL4SSR 心智模型
虽然这是你自己的 DSL但建议
* 注释继续支持 `;`
* 保留 `ruleset -> policy` 的思想
* 保留高密度 `group` 行语法
* 保留 `GEOIP` / `FINAL` 的 builtin 概念
## 11.2 不必兼容旧关键字原样
内部不建议继续沿用:
* `ruleset=`
* `custom_proxy_group=`
建议统一改成更短的:
* `module =`
* `group =`
这样语义更清晰:
* `module` 是规则模块注册
* `group` 是代理组定义
---
## 12. 最小示例
## 12.1 base.conf
```ini
listen = 0.0.0.0:3000
mode = rule
allow_lan = true
log_level = info
source = airport_main,url,https://example.com/sub?token=xxxx,enabled=true,cache_ttl=1800
selector = all,.*
selector = hk,(港|HK|hk|Hong Kong|HongKong|hongkong)
selector = us,(美|US|United States)
group = 🚀 节点选择`select`[]♻️ 自动选择`[]🇭🇰 香港节点`[]🇺🇲 美国节点`[]DIRECT
group = ♻️ 自动选择`url-test`@all`http://www.gstatic.com/generate_204`300,,50
group = 🇭🇰 香港节点`url-test`@hk`http://www.gstatic.com/generate_204`300,,50
group = 🇺🇲 美国节点`url-test`@us`http://www.gstatic.com/generate_204`300,,150
group = 🐟 漏网之鱼`select`[]DIRECT`[]🚀 节点选择
module = apple,./rules/modules/Apple.list,🍎 苹果服务,80,true
module = openai,./rules/modules/OpenAi.list,💬 Ai平台,100,true
builtin = geoip_cn,GEOIP,CN,🎯 全球直连,9000,true
builtin = final,FINAL,,🐟 漏网之鱼,9999,true
```
## 12.2 profiles/default.conf
```ini
name = 默认完整配置
enabled = true
sources = airport_main
include_modules = apple,openai
exclude_modules =
include_builtins = geoip_cn,final
override_policy = apple,🚀 节点选择
prepend_rule = DOMAIN-SUFFIX,internal.example.com,DIRECT
```
---
## 13. 给 Codex 的实现要求
你可以直接把下面这段交给 Codex
```md
请为 sub-provider 设计并实现一套类 ACL4SSR 风格的 INI/CONF 配置系统,不要引入 YAML 配置层。
目标:
- 使用高密度、行导向、声明式配置
- 保留 ACL4SSR 的使用习惯,但不要求 100% 兼容其原始关键字
- 规则正文继续使用 rules/modules/*.list
- 配置文件负责声明:
- source
- selector
- group
- module
- builtin
- profile
要求:
1. base.conf 负责全局配置、source、selector、group、module、builtin。
2. profiles/*.conf 负责具体输出 profile。
3. group 使用反引号分隔的 ACL4SSR-like 语法。
4. module 指向规则文件路径和默认策略组。
5. builtin 用于表达 GEOIP、FINAL 这类内置规则。
6. 生成器读取这些 conf 文件后,生成完整 Clash/Mihomo YAML。
7. 不要接数据库,不要做面板。
8. 解析器要能给出清晰的行号错误提示。
9. profile 支持 override_policy、prepend_rule、append_rule。
10. 先实现最小可用版本,再逐步扩展。
```
---
## 14. 最终结论
你现在最适合的路线,就是:
> **规则正文继续放 `.list`**
> **配置层改成类 ACL4SSR 风格的 `.conf`**
> **Python 写一个轻量 parser + generator**
> **最终输出完整 YAML**
这条路最顺,也最符合你现在的使用习惯。
[1]: https://github.com/tindy2013/subconverter/blob/master/base/config/ACL4SSR_NoMicrosoft.ini "subconverter/base/config/ACL4SSR_NoMicrosoft.ini at master · tindy2013/subconverter · GitHub"

View 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()

61
tests/test_conf_parity.py Normal file
View File

@@ -0,0 +1,61 @@
from __future__ import annotations
import os
import sys
import unittest
from pathlib import Path
from tempfile import TemporaryDirectory
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",
)
class ConfParityTest(unittest.TestCase):
def test_compare_script_core_matches(self) -> None:
with TemporaryDirectory() as tmp:
sample = Path(tmp) / "sample.yaml"
_write_sample_source(sample)
os.environ["AIRPORT_A_URL"] = str(sample)
os.environ["AIRPORT_B_URL"] = str(sample)
os.environ["AIRPORT_C_URL"] = str(sample)
from scripts.compare_conf_outputs import compare_outputs
result = compare_outputs("mihomo", "default", "airport-a,airport-b", sample_source=sample)
self.assertTrue(result["thin"]["proxy_provider_match"])
self.assertTrue(result["thin"]["rule_provider_match"])
self.assertTrue(result["thin"]["group_name_match"])
self.assertTrue(result["bundle"]["group_name_match"])
self.assertTrue(result["bundle"]["proxy_count_match"])
self.assertTrue(result["bundle"]["rules_count_match"])
self.assertEqual(result["thin"]["legacy"]["rules_head"], result["thin"]["conf"]["rules_head"])
self.assertEqual(result["bundle"]["legacy"]["rules_tail"], result["bundle"]["conf"]["rules_tail"])
if __name__ == "__main__":
unittest.main()