B

app/services/subscriptions.py

@@ -4,13 +4,14 @@ import base64
 import json
 import logging
 import re
+from pathlib import Path
 from typing import Any, Iterable
 from urllib.parse import parse_qs, unquote, urlparse
 
 import httpx
 import yaml
 
-from app.config import get_settings
+from app.config import ROOT_DIR, get_settings
 from app.models import FetchResult, ProviderDocument, SourceConfig, SourceSnapshot
 from app.services.cache import TTLCache
 from app.services.fetch_cache import build_fetch_cache_key, load_fetch_cache, save_fetch_cache
@@ -31,6 +32,17 @@ async def fetch_source(name: str, source: SourceConfig) -> FetchResult:
         logger.info("fetch_source memory cache hit: source=%s ttl=%s", name, ttl)
         return cached
 
+    local_path = resolve_local_source_path(source.url)
+    if local_path is not None:
+        logger.info("fetch_source local file: source=%s path=%s", name, local_path)
+        try:
+            result = FetchResult(text=local_path.read_text(encoding="utf-8"), headers={})
+        except Exception:
+            logger.exception("fetch_source local file read failed: source=%s path=%s", name, local_path)
+            raise
+        _fetch_cache.set(name, result, ttl)
+        return result
+
     cache_key = build_fetch_cache_key(name=name, url=source.url)
     disk_cached = load_fetch_cache(
         cache_dir=settings.fetch_cache_dir,
@@ -66,6 +78,34 @@ async def fetch_source(name: str, source: SourceConfig) -> FetchResult:
         return result
 
 
+def resolve_local_source_path(raw: str) -> Path | None:
+    candidate = raw.strip()
+    if not candidate:
+        return None
+
+    parsed = urlparse(candidate)
+    if parsed.scheme in {"http", "https"}:
+        return None
+
+    if parsed.scheme == "file":
+        path_text = unquote(parsed.path or "")
+        if parsed.netloc:
+            path_text = f"//{parsed.netloc}{path_text}"
+        if re.match(r"^/[A-Za-z]:", path_text):
+            path_text = path_text[1:]
+        path = Path(path_text)
+    else:
+        path = Path(candidate)
+
+    if not path.is_absolute():
+        path = ROOT_DIR / path
+
+    resolved = path.resolve()
+    if not resolved.is_file():
+        raise FileNotFoundError(f"Local source file not found: {resolved}")
+    return resolved
+
+
 async def build_provider_document(name: str, source: SourceConfig) -> ProviderDocument:
     settings = get_settings()
     ttl = source.cache_ttl_seconds or settings.cache_ttl_seconds