浅念
ec55bc38ab
查询语义自 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>
351 lines
15 KiB
Python
351 lines
15 KiB
Python
"""数据视图 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}
|