From ec55bc38ab4992a3357445dd7f6e35688f6168df Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Tue, 13 Oct 2026 11:00:39 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=95=B0=E6=8D=AE=E8=A7=86=E5=9B=BEAPI?= =?UTF-8?q?=EF=BC=88=E4=BC=9A=E8=AF=9D/=E8=AF=A6=E6=83=85/=E6=8E=A8?= =?UTF-8?q?=E9=80=81=E6=97=A5=E5=BF=97/=E5=88=A0=E9=99=A4=EF=BC=8C?= =?UTF-8?q?=E7=A7=9F=E6=88=B7scope=E5=BC=BA=E5=88=B6=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 查询语义自 jc admin.py 移植,路由多租户化(/api/platform/tenants/{slug}/...): - runtime 按 slug 取: registry.runtimes() 匹配 + ensure_ready(resolve 需域名, 平台后台只有 slug);未懒建租户首次查询触发建库不炸 - 会话列表: ≥1 消息的会话、created_at 倒序、intent/site(client_info.site)/ 日期区间 from-to(边界含,to 日期-only 含全天)筛选,分页 page/page_size≤100, 每行含意向/联系方式/email_sent/IP 归国(ip_service 复用);total 对筛选后 集合计数(修 jc 旧缺陷: 其会话列表 total 不随 intent 变) - 详情: 元数据+client_info+画像 chips(raw_info 去 _ 前缀键)+消息全量+意向块 - 删除: jc 原序逐表显式删,补齐 lead_push_logs(兑现 FK 声明的 CASCADE, 免孤儿行);不存在 404 - push-logs: load_only 排除 request_payload 大列,payload_size(func.length 服务端算)作省略标记;status/conversation_id 筛选 - 鉴权 _tenant_guard(slug): 他租户 403/未认证 401 有测试钉住;非法筛选值 显性 400(jc 静默忽略) Co-Authored-By: Claude Code --- app/main.py | 5 + app/platform/routes/data_routes.py | 350 +++++++++++++++++++++ tests/test_data_views.py | 477 +++++++++++++++++++++++++++++ 3 files changed, 832 insertions(+) create mode 100644 app/platform/routes/data_routes.py create mode 100644 tests/test_data_views.py diff --git a/app/main.py b/app/main.py index 4e2b4ae..65a449f 100644 --- a/app/main.py +++ b/app/main.py @@ -85,6 +85,7 @@ from app.platform.routes.config_routes import ( router as platform_config_router, settings_router as platform_settings_router, ) +from app.platform.routes.data_routes import router as platform_data_router from app.platform.routes.tenant_routes import router as platform_tenant_router from app.platform.routes.upload_routes import router as platform_upload_router app.include_router(chat_router) @@ -101,6 +102,10 @@ app.include_router(platform_settings_router) # 头像上传(Task 7):multipart 魔数校验 + uuid 重命名 + 配置版本管线 # (require_tenant);/api/public/avatar 的三级解析链消费上传产物。 app.include_router(platform_upload_router) +# 数据视图(Task 8): 会话列表(筛选/分页)/详情/删除 + 推送日志——跑在租户 +# runtime 的 session_factory 上(slug → runtimes() 匹配 + ensure_ready 懒建), +# 鉴权 _tenant_guard;查询语义自 jc admin.py 移植。 +app.include_router(platform_data_router) # 平台认证路由(Task 2):本任务即挂——已知租户 Host(如 jiancai)下 # /api/platform/auth/* 经 Host 中间件直达,TDD 循环闭环。Task 3 将为 # /api/platform/ 前缀加中间件豁免(平台后台不依赖租户域名解析); diff --git a/app/platform/routes/data_routes.py b/app/platform/routes/data_routes.py new file mode 100644 index 0000000..1f4ecfb --- /dev/null +++ b/app/platform/routes/data_routes.py @@ -0,0 +1,350 @@ +"""数据视图 API(Phase 2 Task 8): /api/platform/tenants/{slug}/conversations* +与 /push-logs——查询语义自 jc 旧单租户 admin.py 移植,路由形态多租户化。 + +移植要点(相对 jc 的差异均为多租户化所需,行为语义保持): +- 数据跑在该租户 runtime 的 session_factory 上(各自租户库,天然隔离); + runtime 按 slug 经 registry.runtimes() 匹配 + ensure_ready——resolve 系列 + 需要域名,而平台后台只有 slug;runtimes() 是加载期快照,ensure_ready 内部 + 的现势检查保证 reload 竞态下也不会产出孤儿 engine。未懒建租户(创建后 + 从未被域名 resolve)在此触发建库(ensure_schema 幂等),首次查询不炸。 +- 鉴权 _tenant_guard(slug)(super 或本租户管理员;他租户 403,依赖先于 + 处理器跑,404 检查在其后——与 tenant_routes 同序)。 +- 列表: 只收至少一条消息的会话(jc 语义,消息计数子查询 INNER JOIN)、 + created_at 倒序、intent 筛选(evaluations 左联 + where)、分页 + page/page_size(≤100)。新增 site 筛选(client_info["site"],WS init 与 + REST ?site= 的会话站点标)与日期区间 from/to(ISO 日期或日期时间; + 日期-only 的 to 视为含当日全天——边界含)。total 对筛选后的集合计数 + (jc 会话列表的 total 不随 intent 筛选变,系旧缺陷;jc push-logs 的 + total 就是筛选后计数,此处对齐后者)。 +- 意向/推送状态非法值显性 400(jc 静默忽略 bogus 筛选,用户以为滤了 + 实则全量——改为显性拒绝)。 +- 每行 IP 归国复用 ip_service.resolve_country(中文国名)。 +- 详情: 会话元数据 + client_info 原样 + 画像 chips(raw_info 去 _ 前缀 + 键,_cc/_cc_asked 内部态不上桌——与 lead_push_service 的展示过滤同款) + + 消息全量(空内容滤除,jc 语义)+ 意向块。 +- 删除: jc 原序 evaluation → profile → message → conversation 显式逐表 + (SQLite FK 强制默认关,ondelete=CASCADE 不会自动兑现),并补齐 + lead_push_logs(jc 删除时该表会后遗孤儿行指向已删会话;FK 声明本就是 + CASCADE,此处显式兑现声明语义)。不存在 → 404(jc 恒 {"ok": true}, + 平台侧区分"删过了"与"删成功")。 +- push-logs: load_only 排除 request_payload 大列(含完整对话,一页几十条 + 大 JSON 的 I/O 与反序列化都省掉);响应带 payload_size(func.length + 服务端算,内容不出库)作省略标记。status/conversation_id 筛选与分页 + 同 jc。 +""" +import re +from datetime import datetime, timezone + +from fastapi import APIRouter, Depends, HTTPException, Query, Request +from sqlalchemy import delete, func, select +from sqlalchemy.orm import load_only + +from app.engine.models import ( + Conversation, + CustomerProfile, + Evaluation, + IntentLevel, + LeadPushLog, + Message, + MessageRole, +) +from app.engine.services.ip_service import resolve_country +from app.platform.registry import TenantRuntime +from app.platform.routes.tenant_routes import _tenant_guard + +router = APIRouter(prefix="/api/platform/tenants", tags=["platform-data"]) + +PAGE_SIZE_MAX = 100 +INTENTS = ("interested", "not_interested", "pending") +PUSH_STATUSES = ("pending", "success", "failed", "skipped") + +_DATE_ONLY = re.compile(r"^\d{4}-\d{2}-\d{2}$") + + +# ---- 助手 ------------------------------------------------------------------------- + + +async def _runtime_by_slug(request: Request, slug: str) -> TenantRuntime: + """slug → 就绪 runtime:registry.runtimes() 快照里匹配 config.slug, + 再 ensure_ready(懒建 engine + ensure_schema;未懒建租户在此触发建库)。 + + 不走 resolve(需要域名,平台后台只有 slug)。ensure_ready 的现势检查 + (条目被 reload 换掉则按新条目重走)与 per-slug 锁串行在此同样生效, + 无孤儿 engine 窗口。registry 未加载该 slug → 404(DB 源下租户经创建 + API/启动 load_all 进 registry,创建即可用)。""" + registry = request.app.state.registry + rt = next((r for r in registry.runtimes() if r.config.slug == slug), None) + if rt is None: + raise HTTPException(status_code=404, detail=f"租户不存在: {slug!r}") + return await registry.ensure_ready(rt) + + +def _page_params(page: int, page_size: int) -> tuple[int, int]: + """分页参数校验: page ≥ 1,1 ≤ page_size ≤ 100(非法显性 400)。""" + if page < 1: + raise HTTPException(status_code=400, detail="page 须为正整数") + if not 1 <= page_size <= PAGE_SIZE_MAX: + raise HTTPException( + status_code=400, + detail=f"page_size 须为 1-{PAGE_SIZE_MAX}(实际 {page_size})") + return page, page_size + + +def _parse_dt(raw: str, param: str, *, end_of_day: bool = False) -> datetime: + """ISO 日期/日期时间 → naive UTC(与库内存储同一对比空间)。 + + 日期-only 的 to 视为含当日全天(23:59:59.999999)——边界含;带时区的 + 输入归一 UTC 后去 tzinfo。无法解析 → 400。""" + try: + dt = datetime.fromisoformat(raw) + except (ValueError, TypeError) as exc: + raise HTTPException( + status_code=400, + detail=f"{param}: 无法解析的 ISO 日期/日期时间: {raw!r}") from exc + if end_of_day and _DATE_ONLY.fullmatch(raw.strip()): + dt = dt.replace(hour=23, minute=59, second=59, microsecond=999999) + if dt.tzinfo is not None: + dt = dt.astimezone(timezone.utc).replace(tzinfo=None) + return dt + + +def _iso(value: datetime | None) -> str | None: + return value.isoformat() if value else None + + +def _enum_value(value): + return value.value if value is not None else None + + +# ---- 会话: 列表 / 详情 / 删除 ------------------------------------------------------- + + +@router.get("/{slug}/conversations") +async def list_conversations( + slug: str, request: Request, + intent: str | None = None, + site: str | None = None, + from_dt: str | None = Query(default=None, alias="from"), + to_dt: str | None = Query(default=None, alias="to"), + page: int = 1, page_size: int = 20, + user=Depends(_tenant_guard)): + """会话列表(jc admin 查询语义移植): 只有至少一条消息的会话、created_at + 倒序、每行含意向/联系方式/email_sent/IP 归国。筛选 intent/site/日期 + 区间(边界含),分页 page/page_size(≤100)。""" + page, size = _page_params(page, page_size) + if intent is not None and intent not in INTENTS: + raise HTTPException( + status_code=400, + detail=f"intent 只允许 {INTENTS}(实际 {intent!r})") + bound_from = _parse_dt(from_dt, "from") if from_dt else None + bound_to = _parse_dt(to_dt, "to", end_of_day=True) if to_dt else None + + rt = await _runtime_by_slug(request, slug) + async with rt.session_factory() as db: + # 消息计数 + 首条用户消息(预览)子查询——INNER JOIN 即"至少一条消息" + msg_subq = ( + select( + Message.conversation_id, + func.count().label("mc"), + func.min(Message.content).filter( + Message.role == MessageRole.user).label("first_user_msg"), + ).group_by(Message.conversation_id).subquery()) + stmt = ( + select(Conversation, msg_subq.c.mc, + msg_subq.c.first_user_msg, Evaluation) + .join(msg_subq, msg_subq.c.conversation_id == Conversation.id) + .outerjoin(Evaluation, + Evaluation.conversation_id == Conversation.id) + .order_by(Conversation.created_at.desc())) + if intent: + stmt = stmt.where(Evaluation.intent_level == IntentLevel(intent)) + if site: + stmt = stmt.where( + Conversation.client_info["site"].as_string() == site) + if bound_from is not None: + stmt = stmt.where(Conversation.created_at >= bound_from) + if bound_to is not None: + stmt = stmt.where(Conversation.created_at <= bound_to) + + total = (await db.execute( + select(func.count()).select_from(stmt.subquery()))).scalar() or 0 + rows = (await db.execute( + stmt.offset((page - 1) * size).limit(size))).all() + + items = [] + for conv, msg_count, first_user_msg, evaluation in rows: + client_info = conv.client_info or {} + ip = client_info.get("ip", "") + items.append({ + "id": conv.id, + "session_id": conv.session_id, + "status": _enum_value(conv.status), + "source": _enum_value(conv.source), + "site": client_info.get("site"), + "ip": ip, + "device": client_info.get("user_agent", ""), + "country": resolve_country(ip), + "intent_level": _enum_value( + evaluation.intent_level if evaluation else None), + "reasoning": evaluation.reasoning if evaluation else None, + "contact_info": evaluation.contact_info if evaluation else None, + "email_sent": evaluation.email_sent if evaluation else False, + "message_count": msg_count, + "preview": (first_user_msg or "")[:50], + "created_at": _iso(conv.created_at), + }) + return {"total": total, "page": page, "page_size": size, "items": items} + + +@router.get("/{slug}/conversations/{conv_id}") +async def conversation_detail( + slug: str, conv_id: str, request: Request, + user=Depends(_tenant_guard)): + """会话详情: 元数据 + client_info 原样 + 画像 chips(raw_info 去 _ + 前缀键)与 summary + 意向块 + 消息全量(时间升序,空内容滤除)。""" + rt = await _runtime_by_slug(request, slug) + async with rt.session_factory() as db: + conv = (await db.execute( + select(Conversation).where(Conversation.id == conv_id) + )).scalar_one_or_none() + if conv is None: + raise HTTPException( + status_code=404, detail=f"会话不存在: {conv_id!r}") + messages = (await db.execute( + select(Message) + .where(Message.conversation_id == conv_id, + Message.content != "") + .order_by(Message.created_at))).scalars().all() + profile = (await db.execute( + select(CustomerProfile) + .where(CustomerProfile.conversation_id == conv_id) + )).scalar_one_or_none() + evaluation = (await db.execute( + select(Evaluation) + .where(Evaluation.conversation_id == conv_id) + )).scalar_one_or_none() + + client_info = conv.client_info or {} + ip = client_info.get("ip", "") + raw_info = (profile.raw_info if profile else None) or {} + return { + "id": conv.id, + "session_id": conv.session_id, + "status": _enum_value(conv.status), + "source": _enum_value(conv.source), + "client_info": client_info, + "site": client_info.get("site"), + "ip": ip, + "device": client_info.get("user_agent", ""), + "country": resolve_country(ip), + "created_at": _iso(conv.created_at), + "updated_at": _iso(conv.updated_at), + "evaluation": None if evaluation is None else { + "intent_level": _enum_value(evaluation.intent_level), + "reasoning": evaluation.reasoning, + "contact_info": evaluation.contact_info, + "email_sent": evaluation.email_sent, + "correction_count": evaluation.correction_count, + "evaluated_at": _iso(evaluation.evaluated_at), + }, + "profile": None if profile is None else { + "chips": {k: v for k, v in raw_info.items() + if not str(k).startswith("_")}, + "summary": profile.summary, + }, + "messages": [ + {"role": _enum_value(m.role), "content": m.content, + "time": _iso(m.created_at)} + for m in messages if m.content and m.content.strip()], + } + + +@router.delete("/{slug}/conversations/{conv_id}") +async def delete_conversation( + slug: str, conv_id: str, request: Request, + user=Depends(_tenant_guard)): + """删除会话: jc 原序逐表显式删(evaluation → profile → [补齐 + lead_push_logs] → message → conversation)——SQLite FK 强制默认关, + ondelete=CASCADE 不自动兑现,显式删才不留孤儿。不存在 → 404。""" + rt = await _runtime_by_slug(request, slug) + async with rt.session_factory() as db: + found = (await db.execute( + select(Conversation.id).where(Conversation.id == conv_id) + )).first() + if found is None: + raise HTTPException( + status_code=404, detail=f"会话不存在: {conv_id!r}") + await db.execute( + delete(Evaluation).where(Evaluation.conversation_id == conv_id)) + await db.execute( + delete(CustomerProfile) + .where(CustomerProfile.conversation_id == conv_id)) + await db.execute( + delete(LeadPushLog) + .where(LeadPushLog.conversation_id == conv_id)) + await db.execute( + delete(Message).where(Message.conversation_id == conv_id)) + await db.execute( + delete(Conversation).where(Conversation.id == conv_id)) + await db.commit() + return {"ok": True} + + +# ---- 推送日志 --------------------------------------------------------------------- + + +@router.get("/{slug}/push-logs") +async def list_push_logs( + slug: str, request: Request, + status: str | None = None, + conversation_id: str | None = None, + page: int = 1, page_size: int = 20, + user=Depends(_tenant_guard)): + """询盘推送日志列表(jc admin 语义移植): load_only 排除 request_payload + 大列(含完整对话)——内容不出库,响应带 payload_size(func.length + 服务端算)作省略标记,排障看库。status/conversation_id 筛选。""" + page, size = _page_params(page, page_size) + if status is not None and status not in PUSH_STATUSES: + raise HTTPException( + status_code=400, + detail=f"status 只允许 {PUSH_STATUSES}(实际 {status!r})") + + rt = await _runtime_by_slug(request, slug) + async with rt.session_factory() as db: + stmt = ( + select(LeadPushLog, + func.length(LeadPushLog.request_payload) + .label("payload_size")) + .options(load_only( + LeadPushLog.id, LeadPushLog.evaluation_id, + LeadPushLog.conversation_id, LeadPushLog.push_type, + LeadPushLog.status, LeadPushLog.attempts, + LeadPushLog.outcome, LeadPushLog.response_status, + LeadPushLog.error_message, LeadPushLog.created_at, + LeadPushLog.updated_at)) + .order_by(LeadPushLog.created_at.desc())) + if status: + stmt = stmt.where(LeadPushLog.status == status) + if conversation_id: + stmt = stmt.where(LeadPushLog.conversation_id == conversation_id) + + total = (await db.execute( + select(func.count()).select_from(stmt.subquery()))).scalar() or 0 + rows = (await db.execute( + stmt.offset((page - 1) * size).limit(size))).all() + + items = [{ + "id": log.id, + "evaluation_id": log.evaluation_id, + "conversation_id": log.conversation_id, + "push_type": log.push_type, + "status": log.status, + "attempts": log.attempts, + "outcome": log.outcome, + "response_status": log.response_status, + "error_message": log.error_message, + "payload_size": payload_size, + "created_at": _iso(log.created_at), + "updated_at": _iso(log.updated_at), + } for log, payload_size in rows] + return {"total": total, "page": page, "page_size": size, "items": items} diff --git a/tests/test_data_views.py b/tests/test_data_views.py new file mode 100644 index 0000000..c982ea7 --- /dev/null +++ b/tests/test_data_views.py @@ -0,0 +1,477 @@ +"""数据视图 API(Phase 2 Task 8): 会话列表(筛选/分页)/详情/删除 + 推送日志。 + +环境: db_source_app/super_client(DB 源闭环,租户全部经创建 API 进入 +registry)。查询语义自 jc 旧单租户 admin.py 移植(仅至少一条消息的会话进 +列表/创建时间倒序/意向筛选/load_only 免大字段),路由形态多租户化。 + +种子纪律: 租户库五表(conversations/messages/customer_profiles/ +evaluations/lead_push_logs)直接 ORM 插——独立 NullPool engine 直连同一 +db 文件,一次 asyncio.run 内建表(create_all,与 registry ensure_schema +同构幂等)+插数+dispose,绝不碰 registry 的 pool=2 常驻连接(跨循环 +纪律,同 test_tenant_admin_api 的 _db_fetch 模式)。 + +懒建触发: 经创建 API 进 registry 的租户从未被域名 resolve(engine 仍 +None),数据视图首次查询必须经 ensure_ready 建库不炸(空列表 200,库文件 +落地)——test_lazy_tenant_first_query_builds_db 钉住。 +""" +import asyncio +from datetime import datetime, timezone + +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine +from sqlalchemy.pool import NullPool + +from app.engine.models import ( + Base, + Conversation, + ConversationSource, + ConversationStatus, + CustomerProfile, + Evaluation, + IntentLevel, + LeadPushLog, + Message, + MessageRole, +) + +TENANTS = "/api/platform/tenants" +LOGIN = "/api/platform/auth/login" +CHANGE_PW = "/api/platform/auth/change-password" +UTC = timezone.utc +# 推送日志大字段唯一标记: 列表响应里绝不出现(payload lazy 钉子) +PAYLOAD_MARKER = "UNISERVICE-PAYLOAD-9f1c2d" + + +# ---- 助手(同 test_tenant_admin_api 的模式) ---------------------------------------- + + +def _auth(client): + return {"Authorization": f"Bearer {client.platform_token}"} + + +def _bearer(token): + return {"Authorization": f"Bearer {token}"} + + +def _create(client, slug="demo", admin=None, password=None, domains=None): + return client.post(TENANTS, headers=_auth(client), json={ + "name": f"{slug} 租户", + "slug": slug, + "profile": "jiancai", + "admin_username": admin or f"{slug}-admin", + "admin_password": password or f"{slug}-pass-1", + "domains": domains or [], + }) + + +def _tenant_admin_token(client, username, old_password, new_password="tn-pass-1"): + r = client.post(LOGIN, json={"username": username, "password": old_password}) + assert r.status_code == 200 + token = r.json()["token"] + r = client.post(CHANGE_PW, headers=_bearer(token), json={ + "old_password": old_password, "new_password": new_password}) + assert r.status_code == 200 + return token + + +# ---- 租户库直连(NullPool 跨循环纪律) ---------------------------------------------- + + +def _tenant_engine(root, slug): + tenants = root / "tenants" + tenants.mkdir(parents=True, exist_ok=True) + return create_async_engine( + f"sqlite+aiosqlite:///{(tenants / f'{slug}.db').as_posix()}", + poolclass=NullPool) + + +def _seed(root, slug, populate): + """直连租户库造数: create_all + populate(session) + commit,跑完 dispose。""" + + async def run(): + engine = _tenant_engine(root, slug) + try: + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + async with async_sessionmaker(engine, expire_on_commit=False)() as session: + await populate(session) + await session.commit() + finally: + await engine.dispose() + + asyncio.run(run()) + + +def _count(root, slug, table): + """直连租户库数行(删除级联断言用)。""" + + async def run(): + engine = _tenant_engine(root, slug) + try: + async with AsyncSession(engine) as session: + return (await session.execute( + select(func.count()).select_from(table))).scalar() + finally: + await engine.dispose() + + return asyncio.run(run()) + + +# ---- 种子工厂 --------------------------------------------------------------------- + + +def _conv(cid, at, *, site="main", ip="127.0.0.1", lang=None, + status=ConversationStatus.completed, + source=ConversationSource.widget): + client_info = {"ip": ip, "user_agent": f"UA-{cid}"} + if site: + client_info["site"] = site + if lang: + client_info["lang"] = lang + return Conversation( + id=cid, session_id=f"sess-{cid}", status=status, source=source, + client_info=client_info, created_at=at, updated_at=at) + + +def _msg(cid, role, content, at): + return Message(conversation_id=cid, role=role, content=content, created_at=at) + + +# ---- 核心 4 项 -------------------------------------------------------------------- + + +def test_conversations_list_with_filters(super_client, tmp_path): + _create(super_client, slug="demo") + + async def populate(session): + at1 = datetime(2026, 1, 5, 12, 0, tzinfo=UTC) + at2 = datetime(2026, 1, 6, 9, 0, tzinfo=UTC) + at4 = datetime(2026, 2, 1, 8, 0, tzinfo=UTC) + # c1: es 站/interested/2 条消息/已留资已发邮件 + session.add(_conv("c1", at1, site="es", ip="8.8.8.8", lang="en")) + session.add(_msg("c1", MessageRole.user, "您好，我想买切割机", at1)) + session.add(_msg("c1", MessageRole.assistant, "您好！请问要切什么材料？", at1)) + session.add(Evaluation( + conversation_id="c1", intent_level=IntentLevel.interested, + reasoning="明确采购意向", contact_info={"email": "buy@example.com"}, + email_sent=True, evaluated_at=at1)) + # c2: main 站/not_interested/1 条消息 + session.add(_conv("c2", at2, site="main", ip="127.0.0.1")) + session.add(_msg("c2", MessageRole.user, "just browsing", at2)) + session.add(Evaluation( + conversation_id="c2", intent_level=IntentLevel.not_interested, + contact_info={}, evaluated_at=at2)) + # c3: 零消息——jc 语义: 不进列表(哪怕 intent 命中) + session.add(_conv("c3", at2, site="es")) + session.add(Evaluation( + conversation_id="c3", intent_level=IntentLevel.interested, + evaluated_at=at2)) + # c4: 2 月的会话(日期区间排除用) + session.add(_conv("c4", at4, site="main")) + session.add(_msg("c4", MessageRole.user, "february lead", at4)) + session.add(Evaluation( + conversation_id="c4", intent_level=IntentLevel.pending, + evaluated_at=at4)) + + _seed(tmp_path, "demo", populate) + + def _list(query=""): + r = super_client.get(f"{TENANTS}/demo/conversations{query}", + headers=_auth(super_client)) + assert r.status_code == 200, r.text + return r.json() + + # 默认: 只有带消息的会话,创建时间倒序 + body = _list() + assert body["total"] == 3 + assert [i["id"] for i in body["items"]] == ["c4", "c2", "c1"] + row = next(i for i in body["items"] if i["id"] == "c1") + assert row["session_id"] == "sess-c1" + assert row["status"] == "completed" + assert row["source"] == "widget" + assert row["site"] == "es" + assert row["ip"] == "8.8.8.8" + assert row["country"] == "美国" # ip_service 归国复用 + assert row["intent_level"] == "interested" + assert row["reasoning"] == "明确采购意向" + assert row["contact_info"]["email"] == "buy@example.com" + assert row["email_sent"] is True + assert row["message_count"] == 2 + assert row["preview"] == "您好，我想买切割机" + assert row["created_at"].startswith("2026-01-05") + row2 = next(i for i in body["items"] if i["id"] == "c2") + assert row2["country"] == "本地" # 127.0.0.1 + assert row2["email_sent"] is False + + # intent 筛选(c3 同意向但零消息,不得混入) + body = _list("?intent=interested") + assert body["total"] == 1 + assert [i["id"] for i in body["items"]] == ["c1"] + + # site 筛选(client_info.site) + body = _list("?site=es") + assert body["total"] == 1 + assert [i["id"] for i in body["items"]] == ["c1"] + + # 日期区间,边界含: from=精确时刻 → c1(等于边界)在内 + body = _list("?from=2026-01-05T12:00:00") + assert body["total"] == 3 + # to=日期(整天含) → 只 c1(12:00 ≤ 当日 23:59:59.999999),c2 次日出局 + body = _list("?to=2026-01-05") + assert [i["id"] for i in body["items"]] == ["c1"] + # 闭区间 [01-06, 01-06] → 只 c2 + body = _list("?from=2026-01-06&to=2026-01-06") + assert [i["id"] for i in body["items"]] == ["c2"] + # 跨月区间排除 c4 + body = _list("?from=2026-01-01&to=2026-01-31") + assert [i["id"] for i in body["items"]] == ["c2", "c1"] + + # 分页 + body = _list("?page_size=2&page=1") + assert body["total"] == 3 + assert [i["id"] for i in body["items"]] == ["c4", "c2"] + body = _list("?page_size=2&page=2") + assert [i["id"] for i in body["items"]] == ["c1"] + body = _list("?page_size=2&page=3") + assert body["items"] == [] + + # 上限与非法参数 + assert super_client.get( + f"{TENANTS}/demo/conversations?page_size=101", + headers=_auth(super_client)).status_code == 400 + assert super_client.get( + f"{TENANTS}/demo/conversations?page=0", + headers=_auth(super_client)).status_code == 400 + assert super_client.get( + f"{TENANTS}/demo/conversations?intent=bogus", + headers=_auth(super_client)).status_code == 400 + assert super_client.get( + f"{TENANTS}/demo/conversations?from=not-a-date", + headers=_auth(super_client)).status_code == 400 + + +def test_conversation_detail_with_messages(super_client, tmp_path): + _create(super_client, slug="demo") + + async def populate(session): + at = datetime(2026, 1, 5, 12, 0, tzinfo=UTC) + session.add(_conv("c1", at, site="es", ip="8.8.8.8", lang="en")) + session.add(_msg("c1", MessageRole.user, "您好，我想买切割机", at)) + session.add(_msg("c1", MessageRole.assistant, + "您好！请问要切什么材料？", at)) + # 空内容消息: jc 详情语义不收 + session.add(_msg("c1", MessageRole.assistant, "", at)) + session.add(CustomerProfile( + conversation_id="c1", + raw_info={"姓名": "John", "产品": "切割机", + "_cc": {"whatsapp": "+8613800000000"}}, + summary="尼日利亚买家,切割机")) + session.add(Evaluation( + conversation_id="c1", intent_level=IntentLevel.interested, + reasoning="明确采购意向", contact_info={"email": "buy@example.com"}, + email_sent=True, correction_count=1, evaluated_at=at)) + + _seed(tmp_path, "demo", populate) + + r = super_client.get(f"{TENANTS}/demo/conversations/c1", + headers=_auth(super_client)) + assert r.status_code == 200, r.text + d = r.json() + # 会话元数据 + client_info 原样 + assert d["id"] == "c1" + assert d["session_id"] == "sess-c1" + assert d["status"] == "completed" + assert d["source"] == "widget" + assert d["client_info"] == {"ip": "8.8.8.8", "user_agent": "UA-c1", + "site": "es", "lang": "en"} + assert d["site"] == "es" + assert d["country"] == "美国" + assert d["created_at"].startswith("2026-01-05") + assert d["updated_at"].startswith("2026-01-05") + # 画像 chips: raw_info 去 _ 前缀键(_cc 内部态不上桌) + assert d["profile"]["chips"] == {"姓名": "John", "产品": "切割机"} + assert d["profile"]["summary"] == "尼日利亚买家,切割机" + # 意向块 + assert d["evaluation"]["intent_level"] == "interested" + assert d["evaluation"]["email_sent"] is True + assert d["evaluation"]["correction_count"] == 1 + # 消息全量(按时间升序,空内容被滤) + assert [(m["role"], m["content"]) for m in d["messages"]] == [ + ("user", "您好，我想买切割机"), + ("assistant", "您好！请问要切什么材料？")] + assert all(m["time"] for m in d["messages"]) + + # 未知会话 / 未知租户 + assert super_client.get(f"{TENANTS}/demo/conversations/nope", + headers=_auth(super_client)).status_code == 404 + assert super_client.get(f"{TENANTS}/ghost/conversations/c1", + headers=_auth(super_client)).status_code == 404 + + +def test_push_logs_payload_lazy(super_client, tmp_path): + _create(super_client, slug="demo") + + payload = {"customer": {"name": "John"}, + "messages": [f"msg-{i}" for i in range(40)], + "marker": PAYLOAD_MARKER} + + async def populate(session): + at1 = datetime(2026, 1, 5, 12, 0, tzinfo=UTC) + at2 = datetime(2026, 1, 6, 9, 0, tzinfo=UTC) + session.add(_conv("c1", at1)) + session.add(_msg("c1", MessageRole.user, "您好", at1)) + session.add(CustomerProfile(conversation_id="c1", raw_info={})) + ev_ok = Evaluation( + conversation_id="c1", intent_level=IntentLevel.interested, + contact_info={"email": "buy@example.com"}, evaluated_at=at1) + session.add(ev_ok) + await session.flush() + session.add(LeadPushLog( + evaluation_id=ev_ok.id, conversation_id="c1", push_type="lead", + status="success", attempts=1, request_payload=payload, + response_status=200, outcome="accepted", created_at=at1, + updated_at=at1)) + session.add(LeadPushLog( + evaluation_id=ev_ok.id, conversation_id="c1", push_type="correction", + status="failed", attempts=2, request_payload=payload, + response_status=500, error_message="upstream 500", created_at=at2, + updated_at=at2)) + + _seed(tmp_path, "demo", populate) + + r = super_client.get(f"{TENANTS}/demo/push-logs", headers=_auth(super_client)) + assert r.status_code == 200, r.text + body = r.json() + assert body["total"] == 2 + # 大字段 lazy: 响应绝不含 payload 内容与键名,但有长度标记 + assert PAYLOAD_MARKER not in r.text + assert "request_payload" not in r.text + by_status = {i["status"]: i for i in body["items"]} + assert set(by_status) == {"success", "failed"} + ok = by_status["success"] + assert ok["conversation_id"] == "c1" + assert ok["push_type"] == "lead" + assert ok["outcome"] == "accepted" + assert ok["response_status"] == 200 + assert ok["attempts"] == 1 + assert isinstance(ok["payload_size"], int) and ok["payload_size"] > 0 + assert by_status["failed"]["error_message"] == "upstream 500" + + # status 筛选 / conversation_id 筛选 / 非法值 + r = super_client.get(f"{TENANTS}/demo/push-logs?status=failed", + headers=_auth(super_client)) + assert r.status_code == 200 + assert [i["status"] for i in r.json()["items"]] == ["failed"] + r = super_client.get(f"{TENANTS}/demo/push-logs?conversation_id=c2", + headers=_auth(super_client)) + assert r.status_code == 200 + assert r.json()["total"] == 0 + assert super_client.get(f"{TENANTS}/demo/push-logs?status=bogus", + headers=_auth(super_client)).status_code == 400 + # 分页 + r = super_client.get(f"{TENANTS}/demo/push-logs?page_size=1&page=2", + headers=_auth(super_client)) + assert r.status_code == 200 + assert len(r.json()["items"]) == 1 + + +def test_tenant_scope_isolation(super_client, tmp_path): + _create(super_client, slug="jca", admin="jca-admin", + password="jca-pass-1") + _create(super_client, slug="jcb", admin="jcb-admin", + password="jcb-pass-1") + + async def populate(session): + at = datetime(2026, 1, 5, 12, 0, tzinfo=UTC) + session.add(_conv("b1", at)) + session.add(_msg("b1", MessageRole.user, "hello", at)) + + _seed(tmp_path, "jcb", populate) + + token = _tenant_admin_token(super_client, "jca-admin", "jca-pass-1") + # jca 管理员看 jcb 的数据 → 403(列表/详情/删除/推送日志全挡) + assert super_client.get( + f"{TENANTS}/jcb/conversations", + headers=_bearer(token)).status_code == 403 + assert super_client.get( + f"{TENANTS}/jcb/conversations/b1", + headers=_bearer(token)).status_code == 403 + assert super_client.delete( + f"{TENANTS}/jcb/conversations/b1", + headers=_bearer(token)).status_code == 403 + assert super_client.get( + f"{TENANTS}/jcb/push-logs", + headers=_bearer(token)).status_code == 403 + # 自己的 → 200(空库) + r = super_client.get(f"{TENANTS}/jca/conversations", + headers=_bearer(token)) + assert r.status_code == 200 + assert r.json()["total"] == 0 + # 超管看 jcb → 200 且数据在 + r = super_client.get(f"{TENANTS}/jcb/conversations", + headers=_auth(super_client)) + assert r.status_code == 200 + assert [i["id"] for i in r.json()["items"]] == ["b1"] + # 未认证 → 401 + assert super_client.get( + f"{TENANTS}/jcb/conversations").status_code == 401 + + +# ---- 删除 / 懒建 ------------------------------------------------------------------ + + +def test_delete_conversation_cascades(super_client, tmp_path): + _create(super_client, slug="demo") + + async def populate(session): + at = datetime(2026, 1, 5, 12, 0, tzinfo=UTC) + session.add(_conv("c1", at)) + session.add(_msg("c1", MessageRole.user, "您好", at)) + session.add(_msg("c1", MessageRole.assistant, "你好", at)) + session.add(CustomerProfile(conversation_id="c1", raw_info={"姓名": "J"})) + ev = Evaluation(conversation_id="c1", + intent_level=IntentLevel.interested, + contact_info={"email": "x@y.com"}, evaluated_at=at) + session.add(ev) + await session.flush() + session.add(LeadPushLog( + evaluation_id=ev.id, conversation_id="c1", push_type="lead", + status="success", request_payload={"m": 1}, created_at=at, + updated_at=at)) + + _seed(tmp_path, "demo", populate) + + r = super_client.delete(f"{TENANTS}/demo/conversations/c1", + headers=_auth(super_client)) + assert r.status_code == 200, r.text + assert r.json() == {"ok": True} + # jc 删除语义(级联消息/画像/意向) + 推送日志补齐: 五表全空 + for table in (Conversation, Message, CustomerProfile, Evaluation, + LeadPushLog): + assert _count(tmp_path, "demo", table) == 0, table.__tablename__ + # 详情随之 404;重复删 404 + assert super_client.get(f"{TENANTS}/demo/conversations/c1", + headers=_auth(super_client)).status_code == 404 + assert super_client.delete(f"{TENANTS}/demo/conversations/c1", + headers=_auth(super_client)).status_code == 404 + + +def test_lazy_tenant_first_query_builds_db(super_client, db_source_app, tmp_path): + """未懒建租户(创建后从未被域名 resolve,engine 仍 None)首次查询数据 + 视图: ensure_ready 建库建表,空列表 200,不炸。""" + _create(super_client, slug="fresh") + registry = db_source_app.state.registry + rt = registry._by_slug["fresh"] + assert rt.engine is None # 前提: 确未懒建 + + r = super_client.get(f"{TENANTS}/fresh/conversations", + headers=_auth(super_client)) + assert r.status_code == 200, r.text + assert r.json() == {"total": 0, "page": 1, "page_size": 20, "items": []} + assert rt.engine is not None # 首次查询触发懒建 + assert (tmp_path / "tenants" / "fresh.db").exists() + # 推送日志同样触发(复用已建 engine,再查不炸) + r = super_client.get(f"{TENANTS}/fresh/push-logs", + headers=_auth(super_client)) + assert r.status_code == 200 + assert r.json()["total"] == 0