"""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 copy 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 from app.platform.defaults import DEFAULT_CONFIG @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 为空)") # 站点停用过滤(终审 I-1,与 DB 源同一语义):status 键缺省 = active # (既有 tenant.json 均无该键,兼容);停用站点不进 LoadedTenant—— # 域名随之不进 registry 索引(该域名 404)、site_by_slug 取不到。 sites = [s for s in sites if not isinstance(s, dict) or s.get("status", "active") == "active"] # 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 同构嵌套)。 已知边界(终审 I-1 已接线):site.status != "active" 的站点**不进** LoadedTenant(_load_one 过滤)——停用站点的域名不进 registry 索引 (该域名 404),site_by_slug 取不到;全停用输出空 sites(零站点校验 只对库表形状)。 """ 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"(或其不是对象)") # 站点停用过滤(终审 I-1):status != "active" 的站点不进 # LoadedTenant——其域名不进 registry 域名索引(该域名 404)、 # site_by_slug 取不到。与"租户停用 → TenantDisabledError → 503 # 中性页"的语义差:站点停用 = 这张门面下线,404 合理(不是服务暂停)。 # 全部站点停用 = 输出空 sites(零站点校验只对库表形状——行数为零 # 才是数据坏;全停用是合法运营状态,域名全 404)。 sites = [{"slug": s.slug, "name": s.name, "status": s.status, "domains": list(domains_by_site.get(s.id, [])), "brand": dict(s.brand_overlay or {})} for s in sites_rows if s.status == "active"] return LoadedTenant( slug=tenant.slug, name=tenant.name, status=tenant.status, sites=sites, snapshot=snapshot, seed_dir=None) # ---- 出厂档案快照(Task 5 创建向导/Task 6 重置出厂的公共路径) ---------------------- # 自 tenant_routes 迁入(Task 6 微审注记:配置 API 的 reset 也要用,挪公共位置 # 供两个路由模块消费——此处已是文件源的地盘,零新增依赖)。 # 出厂档案目录:与 app.main.SEED_DIR 同一锚定规则(仓库根/seed_profiles), # 但不 import app.main——main 模块级挂路由,反向导入制造循环依赖。 SEED_DIR = pathlib.Path(__file__).resolve().parents[2] / "seed_profiles" # blank 身份组清理清单(brand 的这几个键置出厂默认;其余 brand 字段保留) _IDENTITY_BRAND_KEYS = ("title", "whatsapp", "email", "privacy_url") def _blankify(snapshot: dict) -> dict: """出厂档案快照 → blank 形态:身份组置出厂默认,其余(含 wall/prompts) 原样保留。深拷贝后改——纯函数纪律,绝不动入参。""" out = copy.deepcopy(snapshot) brand = dict(out.get("brand") or {}) for key in _IDENTITY_BRAND_KEYS: brand[key] = DEFAULT_CONFIG["brand"][key] out["brand"] = brand push = dict(out.get("push") or {}) push["prefix"] = DEFAULT_CONFIG["push"]["prefix"] out["push"] = push engine = dict(out.get("engine") or {}) engine["context_seed_keys"] = dict( DEFAULT_CONFIG["engine"]["context_seed_keys"]) out["engine"] = engine return out def _profile_snapshot(profile: str) -> dict: """profile → 首版快照:auto/jiancai 读对应出厂档案原样;blank 读 jiancai 档案再去身份组(wall/prompts 保留——见 Task 5 模块 docstring)。 重置出厂(spec §5.6"代码库当前模板,随平台升级演进")与创建向导共用 本函数:每次调用现读 seed 目录,不缓存。未知 profile → LookupError (read_tenant 的文件源契约,调用方映射 400)。""" base_slug = "jiancai" if profile == "blank" else profile snapshot = FileTenantSource(SEED_DIR).read_tenant(base_slug).snapshot return _blankify(snapshot) if profile == "blank" else snapshot