Files
sub-provider/app/models.py
riglen 09a9faa1be opt
2026-03-31 16:39:23 +08:00

113 lines
3.3 KiB
Python

from __future__ import annotations
from typing import Any, Literal
from pydantic import BaseModel, Field, HttpUrl
class SourceConfig(BaseModel):
enabled: bool = True
kind: Literal["clash_yaml"] = "clash_yaml"
url: str
display_name: str | None = None
headers: dict[str, str] = Field(default_factory=dict)
include_regex: str | None = None
exclude_regex: str | None = None
prefix: str = ""
suffix: str = ""
cache_ttl_seconds: int | None = None
class RuleConfig(BaseModel):
file: str | None = None
behavior: Literal["domain", "ipcidr", "classical"] = "domain"
format: Literal["yaml", "text", "mrs"] = "yaml"
interval: int = 86400
policy: str
no_resolve: bool = False
payload: list[str] = Field(default_factory=list)
class RegionConfig(BaseModel):
name: str
filter: str
tolerance: int = 50
class ProxyGroupConfig(BaseModel):
name: str
type: Literal["select", "url-test"] = "select"
proxies: list[str] = Field(default_factory=list)
filter: str | None = None
tolerance: int | None = None
url: HttpUrl | None = None
interval: int | None = None
class ClientConfig(BaseModel):
title: str
provider_interval: int = 21600
rule_interval: int = 86400
test_url: HttpUrl = "https://www.gstatic.com/generate_204"
test_interval: int = 300
main_policy: str = "🚀 节点选择"
source_policy: str = "☁️ 机场选择"
mixed_auto_policy: str = "♻️ 自动选择"
manual_policy: str = "🚀 手动切换"
direct_policy: str = "DIRECT"
mode: str = "rule"
allow_lan: bool = True
ipv6: bool = True
mixed_port: int | None = 7890
socks_port: int | None = 7891
log_level: str | None = "info"
class AppConfig(BaseModel):
public_path: str | None = None
sources: dict[str, SourceConfig]
rules: dict[str, RuleConfig] = Field(default_factory=dict)
clients: dict[str, ClientConfig] = Field(default_factory=dict)
regions: dict[str, RegionConfig] = Field(default_factory=dict)
selector_groups: list[ProxyGroupConfig] = Field(default_factory=list)
policy_groups: list[ProxyGroupConfig] = Field(default_factory=list)
class FetchResult(BaseModel):
text: str
headers: dict[str, str] = Field(default_factory=dict)
class ProviderDocument(BaseModel):
proxies: list[dict[str, Any]]
class SubscriptionUserInfo(BaseModel):
upload: int | None = None
download: int | None = None
total: int | None = None
expire: int | None = None
def to_header_value(self) -> str:
parts: list[str] = []
if self.upload is not None:
parts.append(f"upload={self.upload}")
if self.download is not None:
parts.append(f"download={self.download}")
if self.total is not None:
parts.append(f"total={self.total}")
if self.expire is not None:
parts.append(f"expire={self.expire}")
return "; ".join(parts)
def is_empty(self) -> bool:
return self.upload is None and self.download is None and self.total is None and self.expire is None
class SourceSnapshot(BaseModel):
name: str
display_name: str
document: ProviderDocument
headers: dict[str, str] = Field(default_factory=dict)
quota: SubscriptionUserInfo | None = None