Files

118 lines
4.1 KiB
Python

"""Conditional LLMWiki provider configuration.
DeerFlow owns only the connection to WeKnora. Database, Redis, document
parsing, storage and model configuration stay entirely in WeKnora's own
deployment files. An effective non-empty API URL is the sole provider switch.
"""
from __future__ import annotations
from dataclasses import dataclass
from urllib.parse import urlsplit, urlunsplit
from pydantic import BaseModel, ConfigDict, Field, field_validator
def normalize_service_url(value: str | None) -> str:
"""Normalize an optional HTTP(S) service URL and reject embedded secrets."""
raw = str(value or "").strip()
if not raw:
return ""
parsed = urlsplit(raw)
if parsed.scheme not in {"http", "https"} or not parsed.hostname:
raise ValueError("WeKnora addresses must be absolute http(s) URLs")
if parsed.username is not None or parsed.password is not None:
raise ValueError("WeKnora addresses must not contain credentials")
if parsed.query or parsed.fragment:
raise ValueError("WeKnora addresses must not contain query strings or fragments")
path = parsed.path.rstrip("/")
return urlunsplit((parsed.scheme, parsed.netloc, path, "", ""))
class WeKnoraConnectionConfig(BaseModel):
"""The complete DeerFlow-owned connection surface for WeKnora."""
api_base_url: str = Field(default="", description="WeKnora app API base URL; non-empty enables WeKnora mode")
web_base_url: str = Field(default="", description="Optional internal WeKnora administration URL")
admin_email: str = Field(
default="",
description=(
"Optional WeKnora administrator email used only by DeerFlow's "
"server-side detail-page iframe proxy. Prefer $ENV references."
),
)
admin_password: str = Field(
default="",
repr=False,
description=(
"Optional WeKnora administrator password used only by DeerFlow's "
"server-side detail-page iframe proxy. Prefer $ENV references."
),
)
model_config = ConfigDict(extra="ignore")
@field_validator("api_base_url", "web_base_url")
@classmethod
def _validate_url(cls, value: str) -> str:
return normalize_service_url(value)
class LlmWikiConfig(BaseModel):
"""Top-level ``config.yaml`` section for conditional WeKnora mode."""
weknora: WeKnoraConnectionConfig = Field(default_factory=WeKnoraConnectionConfig)
model_config = ConfigDict(extra="ignore")
class LlmWikiRuntimeOverride(BaseModel):
"""Administrator-written URL override stored in system settings.
``enabled=False`` means "use config.yaml". When enabled, an empty API URL
deliberately forces legacy mode even if the source file has an address.
Secrets are never stored here.
"""
enabled: bool = False
api_base_url: str = ""
web_base_url: str = ""
model_config = ConfigDict(extra="ignore")
@field_validator("api_base_url", "web_base_url")
@classmethod
def _validate_url(cls, value: str) -> str:
return normalize_service_url(value)
@dataclass(frozen=True)
class ResolvedLlmWikiRuntime:
provider: str
api_base_url: str
web_base_url: str
admin_email: str
admin_password: str
override_enabled: bool
@property
def weknora_enabled(self) -> bool:
return self.provider == "weknora"
def resolve_llmwiki_runtime(
config: LlmWikiConfig,
override: LlmWikiRuntimeOverride | None = None,
) -> ResolvedLlmWikiRuntime:
"""Resolve the deterministic provider from file config plus URL override."""
use_override = bool(override and override.enabled)
api_base_url = override.api_base_url if use_override and override else config.weknora.api_base_url
web_base_url = override.web_base_url if use_override and override else config.weknora.web_base_url
return ResolvedLlmWikiRuntime(
provider="weknora" if api_base_url else "legacy",
api_base_url=api_base_url,
web_base_url=web_base_url,
admin_email=config.weknora.admin_email,
admin_password=config.weknora.admin_password,
override_enabled=use_override,
)