This commit is contained in:
riglen
2026-04-09 11:11:49 +08:00
parent 0dbb84d308
commit 4e97dbc369
3 changed files with 41 additions and 4 deletions

View File

@@ -1,5 +1,7 @@
from __future__ import annotations from __future__ import annotations
import logging
from fastapi import FastAPI, HTTPException, Query, Request from fastapi import FastAPI, HTTPException, Query, Request
from fastapi.responses import Response from fastapi.responses import Response
@@ -18,6 +20,7 @@ from app.services.subscriptions import (
) )
settings = get_settings() settings = get_settings()
logger = logging.getLogger(__name__)
app = FastAPI(title=settings.app_name) app = FastAPI(title=settings.app_name)
app_config = load_app_config(settings.sources_file) app_config = load_app_config(settings.sources_file)
PUBLIC_PREFIX = "/" + (app_config.public_path or settings.public_path).strip("/") PUBLIC_PREFIX = "/" + (app_config.public_path or settings.public_path).strip("/")
@@ -37,6 +40,7 @@ def _base_url(request: Request) -> str:
def _resolve_sources(sources: str | None) -> list[tuple[str, SourceConfig]]: def _resolve_sources(sources: str | None) -> list[tuple[str, SourceConfig]]:
enabled = [(name, src) for name, src in app_config.sources.items() if src.enabled and str(src.url).strip()] enabled = [(name, src) for name, src in app_config.sources.items() if src.enabled and str(src.url).strip()]
if not sources: if not sources:
logger.info("resolve_sources default: selected=%s", [name for name, _ in enabled])
return enabled return enabled
names = [item.strip() for item in sources.split(",") if item.strip()] names = [item.strip() for item in sources.split(",") if item.strip()]
@@ -52,6 +56,7 @@ def _resolve_sources(sources: str | None) -> list[tuple[str, SourceConfig]]:
seen.add(name) seen.add(name)
if not selected: if not selected:
raise HTTPException(status_code=400, detail="no sources selected") raise HTTPException(status_code=400, detail="no sources selected")
logger.info("resolve_sources explicit: requested=%s selected=%s", sources, [name for name, _ in selected])
return selected return selected
@@ -100,6 +105,7 @@ async def merged_provider(request: Request, sources: str | None = Query(default=
try: try:
document = await build_merged_provider_document(source_items) document = await build_merged_provider_document(source_items)
except Exception as exc: # noqa: BLE001 except Exception as exc: # noqa: BLE001
logger.exception("merged_provider failed: sources=%s", [name for name, _ in source_items])
raise HTTPException(status_code=502, detail=f"failed to build merged provider: {exc}") from exc raise HTTPException(status_code=502, detail=f"failed to build merged provider: {exc}") from exc
content = dump_provider_yaml(document) content = dump_provider_yaml(document)
@@ -116,6 +122,7 @@ async def provider(name: str, request: Request) -> Response:
try: try:
document = await build_provider_document(name, source) document = await build_provider_document(name, source)
except Exception as exc: # noqa: BLE001 except Exception as exc: # noqa: BLE001
logger.exception("provider failed: source=%s", name)
raise HTTPException(status_code=502, detail=f"failed to build provider: {exc}") from exc raise HTTPException(status_code=502, detail=f"failed to build provider: {exc}") from exc
content = dump_provider_yaml(document) content = dump_provider_yaml(document)
@@ -174,6 +181,7 @@ async def bundle_profile(
ttl_seconds=settings.bundle_cache_ttl_seconds, ttl_seconds=settings.bundle_cache_ttl_seconds,
) )
if cached is not None: if cached is not None:
logger.info("bundle cache hit: client=%s sources=%s", client_type, [name for name, _ in source_items])
headers = { headers = {
"profile-update-interval": str(client.provider_interval), "profile-update-interval": str(client.provider_interval),
"X-Sub-Provider-Bundle-Cache": "HIT", "X-Sub-Provider-Bundle-Cache": "HIT",
@@ -184,6 +192,7 @@ async def bundle_profile(
try: try:
snapshots = await build_source_snapshots(source_items) snapshots = await build_source_snapshots(source_items)
except Exception as exc: # noqa: BLE001 except Exception as exc: # noqa: BLE001
logger.exception("bundle_profile failed: client=%s sources=%s", client_type, [name for name, _ in source_items])
raise HTTPException(status_code=502, detail=f"failed to build bundle: {exc}") from exc raise HTTPException(status_code=502, detail=f"failed to build bundle: {exc}") from exc
content = dump_yaml( content = dump_yaml(

View File

@@ -2,6 +2,7 @@ from __future__ import annotations
import base64 import base64
import json import json
import logging
import re import re
from typing import Any, Iterable from typing import Any, Iterable
from urllib.parse import parse_qs, unquote, urlparse from urllib.parse import parse_qs, unquote, urlparse
@@ -16,6 +17,7 @@ from app.services.fetch_cache import build_fetch_cache_key, load_fetch_cache, sa
from app.services.headers import parse_subscription_userinfo from app.services.headers import parse_subscription_userinfo
logger = logging.getLogger(__name__)
_fetch_cache: TTLCache[FetchResult] = TTLCache() _fetch_cache: TTLCache[FetchResult] = TTLCache()
_provider_cache: TTLCache[ProviderDocument] = TTLCache() _provider_cache: TTLCache[ProviderDocument] = TTLCache()
_snapshot_cache: TTLCache[SourceSnapshot] = TTLCache() _snapshot_cache: TTLCache[SourceSnapshot] = TTLCache()
@@ -26,6 +28,7 @@ async def fetch_source(name: str, source: SourceConfig) -> FetchResult:
ttl = source.cache_ttl_seconds or settings.cache_ttl_seconds ttl = source.cache_ttl_seconds or settings.cache_ttl_seconds
cached = _fetch_cache.get(name) cached = _fetch_cache.get(name)
if cached is not None: if cached is not None:
logger.info("fetch_source memory cache hit: source=%s ttl=%s", name, ttl)
return cached return cached
cache_key = build_fetch_cache_key(name=name, url=source.url) cache_key = build_fetch_cache_key(name=name, url=source.url)
@@ -35,17 +38,24 @@ async def fetch_source(name: str, source: SourceConfig) -> FetchResult:
ttl_seconds=ttl, ttl_seconds=ttl,
) )
if disk_cached is not None: if disk_cached is not None:
logger.info("fetch_source disk cache hit: source=%s ttl=%s path=%s", name, ttl, settings.fetch_cache_dir)
result = FetchResult(text=disk_cached.text, headers=disk_cached.headers) result = FetchResult(text=disk_cached.text, headers=disk_cached.headers)
_fetch_cache.set(name, result, ttl) _fetch_cache.set(name, result, ttl)
return result return result
headers = {"User-Agent": settings.default_user_agent} headers = {"User-Agent": settings.default_user_agent}
headers.update(source.headers) headers.update(source.headers)
logger.info("fetch_source upstream request: source=%s url=%s ttl=%s", name, source.url, ttl)
async with httpx.AsyncClient(timeout=settings.request_timeout_seconds, follow_redirects=True) as client: async with httpx.AsyncClient(timeout=settings.request_timeout_seconds, follow_redirects=True) as client:
try:
response = await client.get(source.url, headers=headers) response = await client.get(source.url, headers=headers)
response.raise_for_status() response.raise_for_status()
except Exception:
logger.exception("fetch_source upstream request failed: source=%s url=%s", name, source.url)
raise
result = FetchResult(text=response.text, headers=dict(response.headers)) result = FetchResult(text=response.text, headers=dict(response.headers))
logger.info("fetch_source upstream success: source=%s status=%s bytes=%s", name, response.status_code, len(result.text))
_fetch_cache.set(name, result, ttl) _fetch_cache.set(name, result, ttl)
save_fetch_cache( save_fetch_cache(
cache_dir=settings.fetch_cache_dir, cache_dir=settings.fetch_cache_dir,
@@ -66,8 +76,9 @@ async def build_provider_document(name: str, source: SourceConfig) -> ProviderDo
fetched = await fetch_source(name, source) fetched = await fetch_source(name, source)
proxies = parse_source_proxies(fetched.text, source.kind) proxies = parse_source_proxies(fetched.text, source.kind, source_name=name)
proxies = transform_proxies(proxies, source, settings.max_proxy_name_length) proxies = transform_proxies(proxies, source, settings.max_proxy_name_length)
logger.info("build_provider_document success: source=%s proxies=%s", name, len(proxies))
document = ProviderDocument(proxies=proxies) document = ProviderDocument(proxies=proxies)
_provider_cache.set(cache_key, document, ttl) _provider_cache.set(cache_key, document, ttl)
@@ -98,6 +109,7 @@ async def build_source_snapshot(name: str, source: SourceConfig) -> SourceSnapsh
async def build_source_snapshots(source_items: Iterable[tuple[str, SourceConfig]]) -> list[SourceSnapshot]: async def build_source_snapshots(source_items: Iterable[tuple[str, SourceConfig]]) -> list[SourceSnapshot]:
snapshots: list[SourceSnapshot] = [] snapshots: list[SourceSnapshot] = []
for name, source in source_items: for name, source in source_items:
logger.info("build_source_snapshots start: source=%s", name)
snapshots.append(await build_source_snapshot(name, source)) snapshots.append(await build_source_snapshot(name, source))
return snapshots return snapshots
@@ -152,7 +164,7 @@ def parse_clash_yaml_proxies(text: str) -> list[dict[str, Any]]:
return normalized return normalized
def parse_source_proxies(text: str, source_kind: str) -> list[dict[str, Any]]: def parse_source_proxies(text: str, source_kind: str, source_name: str = "-") -> list[dict[str, Any]]:
parsers: dict[str, list] = { parsers: dict[str, list] = {
"auto": [parse_clash_yaml_proxies, parse_base64_uri_proxies, parse_uri_text_proxies], "auto": [parse_clash_yaml_proxies, parse_base64_uri_proxies, parse_uri_text_proxies],
"clash_yaml": [parse_clash_yaml_proxies], "clash_yaml": [parse_clash_yaml_proxies],
@@ -164,10 +176,25 @@ def parse_source_proxies(text: str, source_kind: str) -> list[dict[str, Any]]:
try: try:
proxies = parser(text) proxies = parser(text)
if proxies: if proxies:
logger.info(
"parse_source_proxies success: source=%s kind=%s parser=%s proxies=%s",
source_name,
source_kind,
parser.__name__,
len(proxies),
)
return proxies return proxies
except Exception as exc: # noqa: BLE001 except Exception as exc: # noqa: BLE001
logger.warning(
"parse_source_proxies parser failed: source=%s kind=%s parser=%s error=%s",
source_name,
source_kind,
parser.__name__,
exc,
)
errors.append(f"{parser.__name__}: {exc}") errors.append(f"{parser.__name__}: {exc}")
detail = "; ".join(errors) if errors else f"unsupported source kind: {source_kind}" detail = "; ".join(errors) if errors else f"unsupported source kind: {source_kind}"
logger.error("parse_source_proxies failed: source=%s kind=%s detail=%s", source_name, source_kind, detail)
raise ValueError(f"Failed to parse upstream subscription: {detail}") raise ValueError(f"Failed to parse upstream subscription: {detail}")

View File

@@ -9,4 +9,5 @@ services:
- .env - .env
volumes: volumes:
- ./config:/app/config:ro - ./config:/app/config:ro
- ./data:/app/data
- ./output:/app/output - ./output:/app/output