"""TenantSource——租户数据源抽象(Phase 2 Task 4)。 registry 只认本模块的 TenantSource 协议(async load_all / load_tenant), 不再直连文件系统:测试/出厂档案走 FileTenantSource,生产(Task 11 起装配) 走 DbTenantSource(platform.db)。两源输出同构 LoadedTenant: - sites: raw 站点列表,形态 = 文件源 tenant.json 的 sites 数组 ({slug, name, domains, brand})——DB 源把 sites.brand_overlay 列映射进 "brand" 键(TenantConfig.from_snapshot 的站点覆盖合并链消费同一形态, 未覆盖项继承租户级 brand); - snapshot: 租户配置快照 dict = 配置组(brand/push/ai/...) ⊕ wall(墙 词表)⊕ prompts(提示词四件套文本)——两源键形态一致,registry 侧 单一消费路径(compile_wall / PromptProvider.from_config)。 文件源行为增量(本任务,等价迁移):prompts 四件套并入 snapshot["prompts"] ——旧路径 PromptProvider.from_dir 直读盘、不进快照;对齐后 DB 源(prompts 嵌 config_json)与文件源输出形态一致,registry 不再感知源差异。 错误契约(registry 依赖,加载期显性失败不流入运行期): - 未知 slug → LookupError(registry_reload 映射为 UnknownTenantError); - DB 源数据坏(缺版本/坏 JSON/非对象/零站点/缺 wall·prompts 组)→ ValueError 带租户名(与版本号); - 文件源数据坏:坏 JSON/slug 与目录不一致/零站点 → ValueError 带路径, wall/prompts 文件缺失 → FileNotFoundError 带路径(read_text 原样)。 """ import json import pathlib from dataclasses import dataclass from typing import Protocol from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession from app.engine.services.ai_service import PROMPTS_KEYS from app.platform.db import Site, SiteDomain, Tenant, TenantConfigVersion @dataclass(frozen=True) class LoadedTenant: """单租户的源侧加载产物(两源同构)。 snapshot 含 wall/prompts 键(见模块 docstring);sites 为 raw dict 列表(TenantConfig.from_snapshot 直接消费);seed_dir 文件源专属 (租户目录,DB 源 None——avatar 解析链对 None 回退静态兜底)。 """ slug: str name: str status: str sites: list[dict] snapshot: dict seed_dir: pathlib.Path | None = None class TenantSource(Protocol): """registry 的数据源接口(纯读,无缓存——每次现读)。""" async def load_all(self) -> list[LoadedTenant]: """全量加载(启动/整表重读)。空源 → ValueError(路径拼错静默 空注册表 = 全量 404,必须炸)。""" ... async def load_tenant(self, slug: str) -> LoadedTenant: """单租户重读(registry_reload 用)。未知 slug → LookupError。""" ... def _read_json(path: pathlib.Path): """读 JSON 文件;读不了/解析不了都带路径抛 ValueError——哪份文件 坏了一目了然(裸 FileNotFoundError/JSONDecodeError 不含路径上下文)。 (自 registry 原样迁入,报错文案逐字等价。)""" try: text = path.read_text(encoding="utf-8") except OSError as exc: raise ValueError(f"{path}: 无法读取: {exc}") from exc try: return json.loads(text) except ValueError as exc: # JSONDecodeError 等 raise ValueError(f"{path}: 无效 JSON: {exc}") from exc class FileTenantSource: """文件源:seed_profiles//(测试/出厂档案)。 行为自 registry._load_tenant 逐字迁入(含校验与报错文案);唯一增量 = prompts 四件套读入 snapshot["prompts"](两源形态对齐)。读盘是纯 同步快操作——同步本体 read_all/read_tenant 供 load_from_dir 便捷构造 同步调用;async load_all/load_tenant 为 TenantSource 协议适配 (registry 的统一 async 消费面,直接返回本体结果)。 """ def __init__(self, seed_dir): self.seed_dir = pathlib.Path(seed_dir) # ---- 同步本体 --------------------------------------------------------- def read_all(self) -> list[LoadedTenant]: tenant_jsons = sorted(self.seed_dir.glob("*/tenant.json")) if not tenant_jsons: raise ValueError(f"{self.seed_dir} 下没有任何 /tenant.json") return [self.read_tenant(p.parent.name) for p in tenant_jsons] def read_tenant(self, slug: str) -> LoadedTenant: tenant_json_path = self.seed_dir / slug / "tenant.json" if not tenant_json_path.is_file(): raise LookupError(f"文件源没有租户 {slug!r}(缺 {tenant_json_path})") tenant_json = _read_json(tenant_json_path) if tenant_json.get("slug") != slug: raise ValueError( f"{tenant_json_path}: slug 字段 " f"{tenant_json.get('slug')!r} 与目录名 {slug!r} 不一致") sites = tenant_json.get("sites") or [] if not sites: raise ValueError(f"{slug}: 每租户至少一个站点(sites 为空)") # wall 并入 snapshot["wall"]:data["wall"] 保留原始词表(后台编辑/ # 回滚用),编译产物挂 runtime.compiled_wall(引擎消费,单一编译) snapshot = dict(tenant_json.get("config") or {}) snapshot["wall"] = _read_json(self.seed_dir / slug / "wall.json") prompts_dir = self.seed_dir / slug / "prompts" snapshot["prompts"] = { name: (prompts_dir / f"{name}.md").read_text(encoding="utf-8") for name in PROMPTS_KEYS } return LoadedTenant( slug=slug, name=tenant_json.get("name", slug), status=tenant_json.get("status", "active"), sites=list(sites), snapshot=snapshot, seed_dir=self.seed_dir / slug) # ---- 协议适配(registry 的 async 消费面)-------------------------------- async def load_all(self) -> list[LoadedTenant]: return self.read_all() async def load_tenant(self, slug: str) -> LoadedTenant: return self.read_tenant(slug) class DbTenantSource: """platform.db 源(生产,Task 11 起装配;本任务先交付能力)。 tenants JOIN sites(brand_overlay → 站点 brand 覆盖层)/site_domains + 当前生效版本(MAX(version),db.py 纪律:历史不可变,当前版取最大) 的 config_json 解析。版本快照由 Task 5 创建向导写入(config 组 ⊕ wall ⊕ prompts 同构嵌套)。 已知边界:site.status 列暂不参与映射(全量输出)——Task 5 站点 CRUD/停用接线时定语义。 """ def __init__(self, engine: AsyncEngine): self._engine = engine async def load_all(self) -> list[LoadedTenant]: async with AsyncSession(self._engine) as session: tenants = (await session.execute( select(Tenant).order_by(Tenant.slug))).scalars().all() if not tenants: raise ValueError( "platform.db 没有任何租户(tenants 表为空)——" "库路径拼错会静默空注册表(全量 404),必须炸") return [await self._load_one(session, t) for t in tenants] async def load_tenant(self, slug: str) -> LoadedTenant: async with AsyncSession(self._engine) as session: tenant = (await session.execute( select(Tenant).where(Tenant.slug == slug))).scalar_one_or_none() if tenant is None: raise LookupError(f"platform.db 中不存在租户: {slug!r}") return await self._load_one(session, tenant) async def _load_one( self, session: AsyncSession, tenant: Tenant) -> LoadedTenant: sites_rows = (await session.execute( select(Site).where(Site.tenant_id == tenant.id) .order_by(Site.id))).scalars().all() if not sites_rows: raise ValueError(f"{tenant.slug}: 每租户至少一个站点(sites 为空)") domains = (await session.execute( select(SiteDomain) .where(SiteDomain.site_id.in_([s.id for s in sites_rows])) .order_by(SiteDomain.id))).scalars().all() domains_by_site: dict[int, list[str]] = {} for d in domains: domains_by_site.setdefault(d.site_id, []).append(d.domain) version = (await session.execute( select(TenantConfigVersion) .where(TenantConfigVersion.tenant_id == tenant.id) .order_by(TenantConfigVersion.version.desc()) .limit(1))).scalar_one_or_none() if version is None: raise ValueError( f"租户 {tenant.name!r}({tenant.slug}) 没有任何配置版本" f"(tenant_config_versions 无行)——先在后台保存配置出版本 1") try: snapshot = json.loads(version.config_json) except ValueError as exc: raise ValueError( f"租户 {tenant.name!r}({tenant.slug}) 版本 " f"{version.version} 的 config_json 无效 JSON: {exc}") from exc if not isinstance(snapshot, dict): raise ValueError( f"租户 {tenant.name!r}({tenant.slug}) 版本 " f"{version.version} 的 config_json 不是 JSON 对象" f"(实际 {type(snapshot).__name__})") for key in ("wall", "prompts"): if not isinstance(snapshot.get(key), dict): raise ValueError( f"租户 {tenant.name!r}({tenant.slug}) 版本 " f"{version.version} 的 config_json 缺少 {key!r} 配置组" f"(或其不是对象)") sites = [{"slug": s.slug, "name": s.name, "domains": list(domains_by_site.get(s.id, [])), "brand": dict(s.brand_overlay or {})} for s in sites_rows] return LoadedTenant( slug=tenant.slug, name=tenant.name, status=tenant.status, sites=sites, snapshot=snapshot, seed_dir=None)