diff --git a/app/engine/models.py b/app/engine/models.py index 1ef4aa2..ade615e 100644 --- a/app/engine/models.py +++ b/app/engine/models.py @@ -2,7 +2,7 @@ import enum import uuid from datetime import datetime, timezone -from sqlalchemy import String, Text, Boolean, Integer, ForeignKey, JSON, Enum as SAEnum +from sqlalchemy import String, Text, Boolean, Integer, ForeignKey, JSON, Enum as SAEnum, text from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship @@ -159,3 +159,27 @@ class LeadPushLog(Base): default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc), ) + + +async def ensure_schema(engine) -> None: + """create_all + 旧库加列迁移(自 app/main.py lifespan/_migrate 提取,逻辑逐字等价)。 + + jc 上线库(conversations 无 page_token / evaluations 无 correction_count) + 经此升级;新库 create_all 已含全部列,加列分支空跑。TenantRegistry 懒建 + 每租户库时复用(Phase 1);生产装配见 app/main.py lifespan。 + """ + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + result = await conn.execute(text("PRAGMA table_info(conversations)")) + columns = {row[1] for row in result} + if "page_token" not in columns: + await conn.execute(text( + "ALTER TABLE conversations ADD COLUMN page_token VARCHAR(64)" + )) + + result = await conn.execute(text("PRAGMA table_info(evaluations)")) + columns = {row[1] for row in result} + if "correction_count" not in columns: + await conn.execute(text( + "ALTER TABLE evaluations ADD COLUMN correction_count INTEGER DEFAULT 0" + )) diff --git a/app/main.py b/app/main.py index 1589b10..44ba3aa 100644 --- a/app/main.py +++ b/app/main.py @@ -2,11 +2,10 @@ from contextlib import asynccontextmanager from fastapi import FastAPI from fastapi.staticfiles import StaticFiles -from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from app.engine.config import settings, ensure_dirs, STATIC_DIR -from app.engine.models import Base +from app.engine.models import ensure_schema # 后台推送任务的独立连接写结果可能撞上 SMTP 窗口内持有的写锁,默认 5s @@ -16,28 +15,10 @@ engine = create_async_engine(settings.DATABASE_URL, echo=settings.APP_DEBUG, async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) -async def _migrate(conn): - result = await conn.execute(text("PRAGMA table_info(conversations)")) - columns = {row[1] for row in result} - if "page_token" not in columns: - await conn.execute(text( - "ALTER TABLE conversations ADD COLUMN page_token VARCHAR(64)" - )) - - result = await conn.execute(text("PRAGMA table_info(evaluations)")) - columns = {row[1] for row in result} - if "correction_count" not in columns: - await conn.execute(text( - "ALTER TABLE evaluations ADD COLUMN correction_count INTEGER DEFAULT 0" - )) - - @asynccontextmanager async def lifespan(app: FastAPI): ensure_dirs() - async with engine.begin() as conn: - await conn.run_sync(Base.metadata.create_all) - await _migrate(conn) + await ensure_schema(engine) # 询盘推送(设计 8.2): 清扫崩溃遗留的 stale pending → failed # 函数内导入: lead_push_service 顶部 from app.main import async_session, # 顶层互相导入会循环 diff --git a/app/platform/registry.py b/app/platform/registry.py new file mode 100644 index 0000000..93aaed7 --- /dev/null +++ b/app/platform/registry.py @@ -0,0 +1,236 @@ +"""TenantRegistry——进程内租户注册表(Phase 1 文件加载器)。 + +数据源:seed_profiles//tenant.json(站点/品牌/引擎配置)+ 同目录 +wall.json(并入 snapshot["wall"],compile_wall 编译期显性炸)。Phase 2 换 +platform.db 时本模块的对外接口不变(spec §5.3:后台保存→原子替换条目, +进行中的会话下一条消息即用新配置)。 + +关键行为: +- resolve_site(host):host 小写去端口 → 域名索引 → (TenantRuntime, Site); + 未知 Host → UnknownTenantError(spec §4:404,不做兜底租户)。 + resolve(host) 只是它的便捷形式——双 API 一层实现,防两套解析漂移。 +- 懒建:首次 resolve 才建 engine(WAL/pool=2)+ session_factory + ensure_schema, + per-slug asyncio.Lock 串行,双检防并发双建;建库失败不留半开 engine。 +- registry_reload(slug):重读磁盘→校验(坏 JSON/墙/域名冲突全在替换前抛, + 旧条目不动)→原子替换 _by_slug 条目并重建域名索引→dispose 旧 engine + (Windows 下不关=库文件句柄泄漏)。 + +测试/teardown 纪律:凡建过 engine 的 Registry 用完必须 await dispose_all() +(Windows 文件锁,tmp 目录删不掉)。 +""" +import asyncio +import json +import pathlib +from dataclasses import dataclass + +from sqlalchemy import event +from sqlalchemy.ext.asyncio import ( + AsyncEngine, + AsyncSession, + async_sessionmaker, + create_async_engine, +) + +from app.engine.models import ensure_schema +from app.engine.wall_schema import CompiledWall, compile_wall +from app.platform.tenant_config import Site, TenantConfig + + +class UnknownTenantError(LookupError): + """Host/域名无法映射到任何租户(调用方语义:聊天/公开端点 404)。""" + + +def make_tenant_engine(db_path: pathlib.Path) -> AsyncEngine: + """每租户 SQLite engine:pool=2 无溢出(单租户低并发),忙等 30s + (对齐 jc 现状:后台推送连接可能撞 SMTP 窗口写锁),打开即 WAL + (spec §6.1 硬纪律:读写不互斥,优于靠 timeout 硬扛)。""" + engine = create_async_engine( + f"sqlite+aiosqlite:///{db_path.as_posix()}", + pool_size=2, max_overflow=0, + connect_args={"timeout": 30}) + + @event.listens_for(engine.sync_engine, "connect") + def _wal(dbapi_conn, _): # dbapi_conn = aiosqlite 同步适配层 + dbapi_conn.execute("PRAGMA journal_mode=WAL") + + return engine + + +def _normalize_host(host: str) -> str: + """Host 头归一:去空白/小写/去端口/去 FQDN 尾点(真实请求 Host:8018、 + 大写都可能出现);IPv6 字面量 [::1]:8018 取方括号内部分。""" + host = (host or "").strip().lower().rstrip(".") + if host.startswith("["): + return host.split("]", 1)[0].lstrip("[") + return host.split(":", 1)[0] + + +@dataclass +class TenantRuntime: + """单租户运行时。config/db_path/compiled_wall 加载期就绪;engine/ + session_factory 由 Registry 首次 resolve 懒建;manager 由装配线注入 + (Task 5b),prompts 由 Task 6 注入——本任务不含。""" + + config: TenantConfig + db_path: pathlib.Path + compiled_wall: CompiledWall + engine: AsyncEngine | None = None + session_factory: async_sessionmaker | None = None + + def site_by_slug(self, slug: str) -> Site | None: + """按 slug 取站点(空域名站点如 es 靠嵌入代码 data-site 接入)。""" + for site in self.config.sites: + if site.slug == slug: + return site + return None + + +class TenantRegistry: + """tenant slug → TenantRuntime;域名 → (tenant_slug, site_slug) 索引。 + reload 重建索引而非增量维护——租户数是个位数,整建最简且无漂移。""" + + def __init__(self): + self._seed_dir: pathlib.Path | None = None + self._data_dir: pathlib.Path | None = None + self._by_slug: dict[str, TenantRuntime] = {} + self._domain_index: dict[str, tuple[str, str]] = {} + self._locks: dict[str, asyncio.Lock] = {} + + # ---- 加载 ----------------------------------------------------------- + + @classmethod + def load_from_dir(cls, seed_dir, data_dir) -> "TenantRegistry": + """seed_profiles//tenant.json(+同目录 wall.json)→ Registry。 + + 加载期显性失败(不流入运行期):目录无任何 tenant.json(seed_dir + 路径拼错→静默空注册表→全量 404,必须炸)、slug 与目录名不一致、 + 零站点、wall 编译错误、域名重复绑定。 + """ + reg = cls() + reg._seed_dir = pathlib.Path(seed_dir) + reg._data_dir = pathlib.Path(data_dir) + tenant_jsons = sorted(reg._seed_dir.glob("*/tenant.json")) + if not tenant_jsons: + raise ValueError(f"{reg._seed_dir} 下没有任何 /tenant.json") + for tenant_json in tenant_jsons: + rt = reg._load_tenant(tenant_json.parent) + if rt.config.slug in reg._by_slug: + raise ValueError(f"租户 slug 重复: {rt.config.slug}") + reg._by_slug[rt.config.slug] = rt + reg._domain_index = reg._build_domain_index(reg._by_slug) + return reg + + def _load_tenant(self, tenant_dir: pathlib.Path) -> TenantRuntime: + slug = tenant_dir.name + tenant_json = json.loads( + (tenant_dir / "tenant.json").read_text(encoding="utf-8")) + if tenant_json.get("slug") != slug: + raise ValueError( + f"{tenant_dir / 'tenant.json'}: 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(引擎消费,单一编译) + wall_cfg = json.loads( + (tenant_dir / "wall.json").read_text(encoding="utf-8")) + snapshot = dict(tenant_json.get("config") or {}) + snapshot["wall"] = wall_cfg + config = TenantConfig.from_snapshot( + slug, tenant_json.get("name", slug), snapshot, + sites=sites, status=tenant_json.get("status", "active")) + return TenantRuntime( + config=config, + db_path=self._data_dir / f"{slug}.db", + compiled_wall=compile_wall(wall_cfg)) + + def _build_domain_index( + self, by_slug: dict[str, TenantRuntime] + ) -> dict[str, tuple[str, str]]: + index: dict[str, tuple[str, str]] = {} + for slug, rt in by_slug.items(): + for site in rt.config.sites: + for domain in site.domains: + key = _normalize_host(domain) + if not key: + continue + if index.get(key, (slug, site.slug)) != (slug, site.slug): + raise ValueError( + f"域名 {key} 同时绑定 {index[key][0]} 与 " + f"{slug} 的站点(域名不可重复绑定)") + index[key] = (slug, site.slug) + return index + + # ---- 解析(对外 API)-------------------------------------------------- + + async def resolve_site(self, host: str) -> tuple[TenantRuntime, Site]: + """host → (TenantRuntime, Site)。首次 resolve 懒建该租户 engine。""" + mapped = self._domain_index.get(_normalize_host(host)) + if mapped is None: + raise UnknownTenantError(f"未知 Host: {host!r}") + tenant_slug, site_slug = mapped + rt = self._by_slug[tenant_slug] + await self._ensure_ready(rt) + site = rt.site_by_slug(site_slug) + if site is None: # 防御:索引与配置同源重建,正常不可达 + raise UnknownTenantError( + f"站点 {site_slug!r} 不存在于租户 {tenant_slug!r}") + return rt, site + + async def resolve(self, host: str) -> TenantRuntime: + """resolve_site 的便捷形式(双 API 一层实现)。""" + return (await self.resolve_site(host))[0] + + # ---- 懒建 / 热加载 / 释放 --------------------------------------------- + + async def _ensure_ready(self, rt: TenantRuntime) -> TenantRuntime: + """懒建 engine + session_factory + ensure_schema,per-slug Lock 串行。 + + 双检:抢到锁后重看 engine——等锁期间前一个协程可能已建好。 + 建库失败 dispose 半开 engine 并保持未就绪,下次 resolve 重试。 + """ + if rt.engine is not None: + return rt + lock = self._locks.setdefault(rt.config.slug, asyncio.Lock()) + async with lock: + if rt.engine is not None: + return rt + self._data_dir.mkdir(parents=True, exist_ok=True) + engine = make_tenant_engine(rt.db_path) + try: + await ensure_schema(engine) + except Exception: + await engine.dispose() + raise + rt.engine = engine + rt.session_factory = async_sessionmaker( + engine, class_=AsyncSession, expire_on_commit=False) + return rt + + async def registry_reload(self, slug: str) -> None: + """重读磁盘,原子替换条目;旧 runtime 若已建 engine,dispose 掉。 + + 先全量校验(重读 tenant.json/wall.json + 候选域名索引)再替换—— + 坏配置抛异常时旧条目原样服务,不留半更新状态。 + """ + if slug not in self._by_slug: + raise UnknownTenantError(f"租户不存在: {slug!r}") + old = self._by_slug[slug] + rt = self._load_tenant(self._seed_dir / slug) + candidate = dict(self._by_slug) + candidate[slug] = rt + new_index = self._build_domain_index(candidate) + self._by_slug[slug] = rt # 原子替换(单事件循环内无 await) + self._domain_index = new_index + if old.engine is not None: + await old.engine.dispose() + old.engine = old.session_factory = None + + async def dispose_all(self) -> None: + """关闭全部 engine 并复位为未就绪(测试 teardown 用——Windows 文件锁; + 复位后若再 resolve 会重新懒建,可安全复用)。""" + for rt in self._by_slug.values(): + if rt.engine is not None: + await rt.engine.dispose() + rt.engine = rt.session_factory = None diff --git a/seed_profiles/auto/tenant.json b/seed_profiles/auto/tenant.json new file mode 100644 index 0000000..9d34699 --- /dev/null +++ b/seed_profiles/auto/tenant.json @@ -0,0 +1,28 @@ +{ + "slug": "auto", + "name": "HESU Auto 汽车", + "status": "active", + "sites": [ + { + "slug": "main", + "domains": ["chat.agent.hnxinyouhe.com"] + } + ], + "config": { + "brand": { + "title": "HESU Support", + "color": "#f08a31", + "whatsapp": "+8619037179250", + "email": "info@borderlesscar.com" + }, + "push": { + "prefix": "ai-kefu" + }, + "engine": { + "context_seed_keys": { + "vehicle_name": "app_vehicle_name", + "vehicle_id": "app_vehicle_id" + } + } + } +} diff --git a/seed_profiles/jiancai/tenant.json b/seed_profiles/jiancai/tenant.json new file mode 100644 index 0000000..c1982f9 --- /dev/null +++ b/seed_profiles/jiancai/tenant.json @@ -0,0 +1,37 @@ +{ + "slug": "jiancai", + "name": "HESUPIPE 建材", + "status": "active", + "sites": [ + { + "slug": "main", + "domains": ["chat2.agent.hnxinyouhe.com", "localhost.jc"] + }, + { + "slug": "es", + "domains": [], + "brand": { + "title": "HESUPIPE Soporte", + "welcome": "¡Hola! ¿En qué podemos ayudarle?" + } + } + ], + "config": { + "brand": { + "title": "HESUPIPE Support", + "color": "#066fc0", + "whatsapp": "+8615937108317", + "email": "hesu@hesuu.com", + "privacy_url": "https://hesupipe.com/privacy-policy" + }, + "push": { + "prefix": "ai-kefu-jc" + }, + "engine": { + "context_seed_keys": { + "product_name": "app_product_name", + "product_id": "app_product_id" + } + } + } +} diff --git a/tests/test_registry.py b/tests/test_registry.py new file mode 100644 index 0000000..01a5ebd --- /dev/null +++ b/tests/test_registry.py @@ -0,0 +1,107 @@ +"""TenantRegistry 单元契约(Task 5a 文件加载器)。 + +seed_profiles//tenant.json + 同目录 wall.json → Registry; +Host(小写/去端口)→站点→租户;首次 resolve 懒建 engine(WAL/pool=2)+ +ensure_schema;reload 原子换条目并 dispose 旧 engine。 + +Windows 纪律:每个建过 engine 的用例结尾 await reg.dispose_all()—— +连接不关,文件锁着,tmp_path 清理 PermissionError。 +""" +import asyncio + +import pytest + +from app.platform.registry import ( + TenantRegistry, + UnknownTenantError, + make_tenant_engine, +) + +SEED_DIR = "seed_profiles" + + +async def test_resolve_by_host(tmp_path): + reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path) + rt = await reg.resolve("chat2.agent.hnxinyouhe.com") + assert rt.config.slug == "jiancai" + assert rt.config.data["push"]["prefix"] == "ai-kefu-jc" + await reg.dispose_all() + + +async def test_unknown_host_404_semantics(tmp_path): + reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path) + with pytest.raises(UnknownTenantError): + await reg.resolve("nope.example.com") + await reg.dispose_all() + + +async def test_tenant_db_isolated_per_slug(tmp_path): + reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path) + a = await reg.resolve("chat2.agent.hnxinyouhe.com") + b = await reg.resolve("chat.agent.hnxinyouhe.com") + assert a.db_path != b.db_path and a.db_path.name == "jiancai.db" + await reg.dispose_all() + + +async def test_reload_swaps_config_atomically(tmp_path): + reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path) + rt = await reg.resolve("chat2.agent.hnxinyouhe.com") + await reg.registry_reload("jiancai") # 重新读盘 + assert await reg.resolve("chat2.agent.hnxinyouhe.com") is not rt + assert rt.engine is None # 旧 runtime 的 engine 必须已 dispose(Windows 文件锁) + await reg.dispose_all() + + +async def test_site_brand_overlay_and_inheritance(tmp_path): + reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path) + rt, site = await reg.resolve_site("chat2.agent.hnxinyouhe.com") + assert site.slug == "main" + es = rt.site_by_slug("es") # 空域名站点按 slug 取 + assert es.brand["title"] == "HESUPIPE Soporte" # 覆盖生效 + assert es.brand["color"] == rt.config.data["brand"]["color"] # 未覆盖项继承租户默认 + await reg.dispose_all() + + +async def test_host_normalization_case_and_port(tmp_path): + """真实 Host 头带端口/可能大写(spec §4:Host 识别必须先归一)。""" + reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path) + rt = await reg.resolve("CHAT2.Agent.hNxinyouHe.com:8018") + assert rt.config.slug == "jiancai" + with pytest.raises(UnknownTenantError): + await reg.resolve("NOPE.example.com:8018") + await reg.dispose_all() + + +async def test_concurrent_first_resolve_builds_engine_once(tmp_path, monkeypatch): + """两个协程同时首 resolve 同一租户:engine 只建一次(asyncio.Lock 串行)。""" + import app.platform.registry as registry_mod + + calls = [] + real_make = registry_mod.make_tenant_engine + + def counting_make(db_path): + calls.append(db_path) + return real_make(db_path) + + monkeypatch.setattr(registry_mod, "make_tenant_engine", counting_make) + + reg = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path) + rts = await asyncio.gather( + *[reg.resolve("chat2.agent.hnxinyouhe.com") for _ in range(5)]) + assert len(calls) == 1 # 无锁则各协程各建一个,前 4 个泄漏 + assert all(rt is rts[0] for rt in rts) # 同一 runtime 实例 + assert rts[0].engine is not None and rts[0].session_factory is not None + await reg.dispose_all() + + +async def test_make_tenant_engine_enables_wal(tmp_path): + """spec §6.1 硬纪律:每租户库打开即 PRAGMA journal_mode=WAL(读写不互斥)。""" + from sqlalchemy import text + + engine = make_tenant_engine(tmp_path / "wal_check.db") + try: + async with engine.connect() as conn: + mode = (await conn.execute(text("PRAGMA journal_mode"))).scalar() + assert str(mode).lower() == "wal" + finally: + await engine.dispose()