feat: 数据视图API(会话/详情/推送日志/删除,租户scope强制)
查询语义自 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 <noreply@anthropic.com>
This commit is contained in:
@@ -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/ 前缀加中间件豁免(平台后台不依赖租户域名解析);
|
||||
|
||||
@@ -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}
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user