BossAi/api.py

from dotenv import load_dotenv
load_dotenv()  # 从 .env 加载 KIMI_API_KEY 等配置，密钥不进代码仓库

import os
import time
import json
import logging
from fastapi import FastAPI, HTTPException, UploadFile, File, Request, Response, Header
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from bossai import (
    SessionManager, assess, action_summary, extract_text, KimiClient,
    SCENE_PROMPTS, EMOTION_MODE_PREFIX,
    build_dynamic_prompt, build_emotion_prefix,
    extract_profile_signals, build_profile_context,
    UserProfile, UserProfileStore, UserStore,
    KnowledgeStore, SettingsStore, ActionSummaryStore,
    ExpertStore, ConsultationStore, RatingStore,
    DocumentGenerator, RegulationUpdateManager, FactFinder,
    ChatStore, redact,
)

# ===== 日志脱敏 =====
class RedactingFilter(logging.Filter):
    """日志过滤器:脱敏手机号、身份证号、银行卡号。"""
    import re as _re
    _PHONE_PAT = _re.compile(r'1[3-9]\d{9}')
    _IDCARD_PAT = _re.compile(r'\b\d{17}[0-9Xx]\b')
    _BANK_PAT = _re.compile(r'\b\d{15,19}\b')

    def filter(self, record: logging.LogRecord) -> bool:
        msg = str(record.getMessage())
        msg = self._PHONE_PAT.sub('[手机号]', msg)
        msg = self._IDCARD_PAT.sub('[证件号]', msg)
        msg = self._BANK_PAT.sub('[账号]', msg)
        record.msg = msg
        record.args = ()
        return True

logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger("bossai")
logger.addFilter(RedactingFilter())

# ===== 可观测性指标 =====
class Metrics:
    """轻量内存指标,无需 Prometheus 依赖。"""
    def __init__(self):
        self.request_count = 0
        self.request_latency_sum = 0.0
        self.model_calls = {"cloud": 0, "local": 0}
        self.destroy_count = 0
        self.error_count = 0

    def record_request(self, duration: float):
        self.request_count += 1
        self.request_latency_sum += duration

    def record_model(self, location: str):
        self.model_calls[location] = self.model_calls.get(location, 0) + 1

    def record_destroy(self):
        self.destroy_count += 1

    def record_error(self):
        self.error_count += 1

    def snapshot(self) -> dict:
        avg = self.request_latency_sum / self.request_count if self.request_count else 0
        return {
            "request_count": self.request_count,
            "avg_latency_ms": round(avg * 1000, 1),
            "model_calls": self.model_calls,
            "destroy_count": self.destroy_count,
            "error_count": self.error_count,
            "destroy_stats": sessions.destroy_stats(),
        }


app = FastAPI(title="左右(BossAi)", version="1.1.0")
sessions = SessionManager()
knowledge = KnowledgeStore()
settings_store = SettingsStore()
summary_store = ActionSummaryStore()
regulation_mgr = RegulationUpdateManager()
fact_finder = FactFinder(max_rounds=2, max_questions_per_round=3)
profile_store = UserProfileStore()
user_store = UserStore()
expert_store = ExpertStore()
consultation_store = ConsultationStore()
rating_store = RatingStore()
chat_store = ChatStore()
metrics = Metrics()
try:
    kimi = KimiClient()
except RuntimeError:
    kimi = None


# ===== 请求计时 + 日志脱敏中间件 =====
@app.middleware("http")
async def metrics_middleware(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    metrics.record_request(duration)
    # 日志脱敏:只记录路径和状态,不记录 body
    logger.info(f"{request.method} {request.url.path} {response.status_code} {duration:.3f}s")
    if response.status_code >= 500:
        metrics.record_error()
    return response


# ===== 简易限流(内存令牌桶,每 IP 每秒 10 请求) =====
_rate_buckets: dict[str, list[float]] = {}
RATE_LIMIT = int(os.getenv("RATE_LIMIT", "20"))  # 请求/秒,生产建议 10
RATE_WINDOW = 1.0


@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    if request.url.path.startswith("/health") or request.url.path.startswith("/metrics"):
        return await call_next(request)
    client_ip = request.client.host if request.client else "unknown"
    now = time.time()
    bucket = _rate_buckets.setdefault(client_ip, [])
    # 清理过期
    bucket[:] = [t for t in bucket if now - t < RATE_WINDOW]
    if len(bucket) >= RATE_LIMIT:
        return Response(content='{"detail":"rate limit exceeded"}', status_code=429,
                        media_type="application/json", headers={"Retry-After": "1"})
    bucket.append(now)
    return await call_next(request)


class SessionRequest(BaseModel):
    mode: str = "normal"


# ===== 用户认证 =====

class RegisterRequest(BaseModel):
    """注册请求。"""
    phone: str
    password: str
    name: str = ""


class LoginRequest(BaseModel):
    """登录请求。"""
    phone: str
    password: str


def get_current_phone(authorization: str | None = Header(default=None)) -> str | None:
    """从 Authorization header 解析 JWT token,返回手机号。

    用于需要用户身份的端点(ask/ask-stream/profile)。
    未登录时返回 None(端点自行处理)。
    """
    if not authorization or not authorization.startswith("Bearer "):
        return None
    token = authorization[7:]
    return user_store.verify_token(token)


@app.post("/auth/register")
def register(req: RegisterRequest):
    """手机号+密码注册,返回 JWT token。"""
    try:
        result = user_store.register(req.phone, req.password, req.name)
        return result
    except ValueError as e:
        raise HTTPException(400, str(e))


@app.post("/auth/login")
def login(req: LoginRequest):
    """手机号+密码登录,返回 JWT token。支持用户、专家、管理员三种角色。"""
    # 先查用户表
    try:
        return user_store.login(req.phone, req.password)
    except ValueError:
        pass
    # 再查专家表
    import bcrypt as _bcrypt
    with user_store._conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT id, phone, password, name FROM experts WHERE phone = %s AND status = 'active'",
                (req.phone,),
            )
            expert_row = cur.fetchone()
            # 查管理员表
            cur.execute(
                "SELECT id, phone, password, name FROM admins WHERE phone = %s",
                (req.phone,),
            )
            admin_row = cur.fetchone()
    if expert_row and _bcrypt.checkpw(req.password.encode(), expert_row[2].encode()):
        token = user_store._make_token(expert_row[1])
        return {"token": token, "phone": expert_row[1], "name": expert_row[3] or "", "role": "expert", "expert_id": expert_row[0]}
    if admin_row and _bcrypt.checkpw(req.password.encode(), admin_row[2].encode()):
        token = user_store._make_token(admin_row[1])
        return {"token": token, "phone": admin_row[1], "name": admin_row[3] or "", "role": "admin", "admin_id": admin_row[0]}
    raise HTTPException(401, "手机号或密码错误")


class FactAnswer(BaseModel):
    """用户对追问的回答。"""
    dim_key: str
    dim_desc: str
    answer: str


class AskRequest(BaseModel):
    question: str
    scene: str = "legal"
    emotion_mode: str | None = None
    """用户补充的事实回答(多轮追问后传入)"""
    fact_answers: list[FactAnswer] | None = None
    """手机号(从 JWT token 解析,用于用户画像)"""
    phone: str | None = None


class KnowledgeAddRequest(BaseModel):
    title: str
    content: str
    source: str = "manual"
    doc_type: str = "law"
    region: str | None = None
    effective_date: str | None = None
    keywords: str | None = None


class SettingsUpdateRequest(BaseModel):
    settings: dict[str, str]


class TicketCreateRequest(BaseModel):
    question: str
    risk_level: str
    conclusion: str | None = None
    session_id: str | None = None
    contact: str | None = None
    priority: str = "normal"


class TicketUpdateRequest(BaseModel):
    status: str
    assignee: str | None = None
    resolution: str | None = None


class ExpertRuleUpdateRequest(BaseModel):
    expert_type: str | None = None
    sla_hours: int | None = None
    auto_assign: str | None = None


class SummarySaveRequest(BaseModel):
    session_id: str | None = None
    summary: dict


@app.get("/health")
def health():
    return {"status": "ok", "service": "左右(BossAi)", "version": "1.1.0", "db": "connected"}


class SampleRequest(BaseModel):
    """生成示例问题请求。"""
    scene: str = "legal"
    topic: str | None = None
    existing: list[str] | None = None  # 已展示的示例,避免重复


@app.post("/samples")
def generate_samples(req: SampleRequest):
    """用 AI 生成场景示例问题,供前端"换一批"使用。

    AI 不可用时降级到内置示例。
    """
    # 内置示例(降级用)
    builtin = {
        "legal": ["员工工伤后能否开除?", "收到税务稽查通知怎么办?", "客户拖欠货款 3 个月如何催收?", "合同违约金怎么约定?"],
        "decision": ["是否应该接一个付款周期 90 天的大单?", "现在适合扩张开第二家店吗?", "该不该降价 15% 抢竞品的客户?", "合伙人要退股,怎么谈条件?"],
        "life": ["怎么和青春期的孩子沟通不想他接班的事?", "父母年纪大了,养老方案怎么选?", "最近睡眠很差,有什么改善建议?", "体检报告有几项异常,需要进一步检查吗?"],
        "emotion": ["最近压力很大,睡不着觉", "合伙人背叛了,我很愤怒", "明天的谈判我拖延了一周没准备", "感觉撑不下去了"],
    }

    if not kimi:
        return {"samples": builtin.get(req.scene, builtin["legal"])}

    scene_config = {
        "legal": {"desc": "法律/法务", "focus": "劳动用工、合同审查、债权债务、股东纠纷、税务风险、诉讼仲裁等法律问题"},
        "decision": {"desc": "经营决策", "focus": "扩张收缩、定价策略、招人裁员、融资合伙、项目承接、业务转型等经营决策"},
        "life": {"desc": "生活助理", "focus": "家庭沟通、子女教育、父母养老、睡眠压力、出差日程、健康习惯等个人生活问题,不是企业经营问题"},
        "emotion": {"desc": "情绪疏导", "focus": "压力、焦虑、愤怒、拖延、疲惫等情绪状态,老板可能倾诉的心理感受"},
    }
    config = scene_config.get(req.scene, scene_config["legal"])

    existing = req.existing or []
    prompt = (
        f"你是{config['desc']}领域的专业顾问。请生成 4 个该领域的典型问题,"
        f"用户(中小企业老板)可能会问的问题。\n\n"
        f"问题方向:{config['focus']}\n\n"
    )
    if req.topic:
        prompt += f"具体话题:{req.topic}\n\n"
    if existing:
        prompt += "已展示过的问题(不要重复):\n"
        for s in existing:
            prompt += f"- {s}\n"
        prompt += "\n"
    prompt += (
        "要求:\n"
        "1. 问题真实、具体、贴近中小企业老板的实际场景\n"
        "2. 每个问题 15-30 字\n"
        "3. 不要重复已展示过的问题\n"
        "4. 只返回 JSON 数组,如 [\"问题1\", \"问题2\", \"问题3\", \"问题4\"]\n"
    )

    try:
        answer = kimi.chat(
            [{"role": "system", "content": "你是问题生成助手,只返回 JSON 数组。"},
             {"role": "user", "content": prompt}],
            temperature=0.8,
            timeout=15,
        )
        # 解析 JSON
        answer = answer.strip()
        # 去除 markdown 代码块标记
        if answer.startswith("```"):
            answer = answer.split("\n", 1)[1] if "\n" in answer else answer[3:]
        if answer.endswith("```"):
            answer = answer[:-3]
        answer = answer.strip()
        # 修复中文引号(常见 LLM 输出问题)
        answer = answer.replace("\u201c", '"').replace("\u201d", '"')
        start = answer.find("[")
        end = answer.rfind("]")
        if start == -1 or end == -1:
            return {"samples": builtin.get(req.scene, builtin["legal"])}
        samples = json.loads(answer[start:end + 1])
        # 过滤非字符串和空值
        samples = [s for s in samples if isinstance(s, str) and s.strip()][:4]
        if len(samples) < 2:
            return {"samples": builtin.get(req.scene, builtin["legal"])}
        return {"samples": samples}
    except Exception as e:
        import logging
        logging.warning(f"/samples AI 生成失败: {type(e).__name__}: {e}")
        return {"samples": builtin.get(req.scene, builtin["legal"])}


@app.get("/metrics")
def get_metrics():
    """可观测性指标:请求数/延迟/模型调用量/销毁统计。"""
    return metrics.snapshot()


@app.get("/", include_in_schema=False)
def home():
    from fastapi.responses import HTMLResponse
    return HTMLResponse(open("web.html", encoding="utf-8").read())


@app.post("/sessions")
def create_session(req: SessionRequest):
    s = sessions.create(req.mode)
    return {"session_id": s.session_id, "mode": s.mode, "expires_at": s.expires_at}


@app.post("/sessions/{sid}/ask")
def ask(sid: str, req: AskRequest, authorization: str | None = Header(default=None)):
    # 从 token 解析手机号(用于用户画像)
    req.phone = get_current_phone(authorization)
    try:
        sessions.append(sid, "user", req.question)
    except KeyError as e:
        raise HTTPException(410, str(e))
    # 获取会话模式(用于决定是否落库)
    session = sessions.sessions.get(sid)
    session_mode = session.mode if session else "normal"
    card = assess(req.question, kimi_client=kimi)
    answer = None
    rag_context = []
    inference_location = "local"
    model_name = "local-triage"
    fallback_used = False
    if kimi:
        prompt = build_dynamic_prompt(req.scene, req.question, kimi_client=kimi)
        # 注入用户画像
        if req.phone:
            profile = profile_store.get(req.phone)
            profile_ctx = build_profile_context(profile)
            if profile_ctx:
                prompt = prompt + "\n\n" + profile_ctx
        if req.scene == "emotion" and req.emotion_mode:
            prompt = build_emotion_prefix(req.emotion_mode, req.question, kimi_client=kimi) + prompt
        # RAG:按场景检索对应分类知识上下文
        ctx = knowledge.build_context(req.question, limit=3, scene=req.scene)
        if ctx:
            prompt = prompt + "\n\n" + ctx
            rag_context = knowledge.search(req.question, limit=3,
                                           doc_types=knowledge.SCENE_DOC_TYPES.get(req.scene))
        # 带降级的云端调用
        result = kimi.chat_with_fallback(
            [{"role": "system", "content": prompt}, {"role": "user", "content": req.question}],
            local_fallback=card.conclusion,
        )
        answer = result["answer"]
        inference_location = result["inference_location"]
        model_name = result["model"]
        fallback_used = result["fallback_used"]
        metrics.record_model(inference_location)
    assistant_content = answer or card.conclusion
    sessions.append(sid, "assistant", assistant_content)

    # 落库:普通/保密模式保存消息+风险卡+元数据,阅后即焚不落库
    if req.phone and session_mode != "ephemeral":
        try:
            # 用户消息
            chat_store.save_message(
                session_id=sid, phone=req.phone, mode=session_mode, role="user",
                content=req.question, scene=req.scene,
            )
            # AI 回复(含风险卡和元数据)
            chat_store.save_message(
                session_id=sid, phone=req.phone, mode=session_mode, role="assistant",
                content=assistant_content, scene=req.scene,
                risk_level=card.level, risk_conclusion=card.conclusion,
                risk_actions=card.actions, risk_evidence=card.evidence,
                risk_escalation=card.escalation,
                inference_location=inference_location, model_name=model_name,
                fallback_used=fallback_used,
                rag_context=rag_context,
            )
        except Exception as e:
            logger.warning(f"chat_store save failed: {e}")

    # 更新用户画像(提取抽象信号,非对话内容)
    if req.phone:
        try:
            signals = extract_profile_signals(req.question, req.scene, kimi_client=kimi)
            profile_store.update(req.phone, signals, scene=req.scene)
        except Exception:
            pass

    return {
        "risk": card.__dict__,
        "answer": answer,
        "action_summary": action_summary(card),
        "model": model_name,
        "scene": req.scene,
        "rag_context": rag_context,
        "inference_location": inference_location,
        "fallback_used": fallback_used,
    }


@app.post("/sessions/{sid}/ask-analyze")
def ask_analyze(sid: str, req: AskRequest, authorization: str | None = Header(default=None)):
    req.phone = get_current_phone(authorization)
    """分析用户问题,返回追问问题或标记事实已充分。

    前端流程:
    1. 用户提问 → 调用 ask-analyze → 返回追问问题(如有)
    2. 用户回答追问 → 再次调用 ask-analyze(带 fact_answers)→ 返回更多追问或 need_analysis=true
    3. need_analysis=true 后 → 调用 ask-stream(带 fact_answers)→ 流式返回逐层分析结论

    Returns:
        {
            "need_ask": bool,      # 是否需要追问
            "need_analysis": bool,  # 是否可以进入分析阶段
            "round": int,           # 当前追问轮次
            "questions": [...],     # 追问问题(need_ask=true 时)
            "question_type": str,   # 问题类型
            "risk_level": str,      # 初步风险等级
        }
    """
    try:
        sessions._active(sid)
    except KeyError as e:
        raise HTTPException(410, str(e))

    # 初步风险分诊(AI 语义分诊 + 规则兜底)
    card = assess(req.question, kimi_client=kimi)

    # 事实追问分析(AI 生成追问 + 规则兜底)
    answers = [a.model_dump() for a in (req.fact_answers or [])]
    prompt_or_questions, meta = fact_finder.build_analysis_prompt(
        req.question, req.scene, answers, kimi_client=kimi,
    )

    # 收集用户画像中的已知事实(带入内容)
    profile_facts: list[dict] = []
    if req.phone:
        try:
            profile = profile_store.get(req.phone)
            if profile.industry:
                profile_facts.append({"dim_key": "industry", "dim_desc": "行业", "answer": profile.industry})
            if profile.company_size:
                profile_facts.append({"dim_key": "company_size", "dim_desc": "企业规模", "answer": profile.company_size})
            if profile.role:
                profile_facts.append({"dim_key": "role", "dim_desc": "角色", "answer": profile.role})
            if profile.risk_tolerance:
                profile_facts.append({"dim_key": "risk_tolerance", "dim_desc": "风险偏好", "answer": profile.risk_tolerance})
            if profile.decision_style:
                profile_facts.append({"dim_key": "decision_style", "dim_desc": "决策风格", "answer": profile.decision_style})
            if profile.key_facts:
                for k, v in profile.key_facts.items():
                    profile_facts.append({"dim_key": f"fact_{k}", "dim_desc": k, "answer": str(v)})
        except Exception:
            pass

    if meta["need_ask"]:
        return {
            "need_ask": True,
            "need_analysis": False,
            "round": meta["round"],
            "questions": meta["questions"],
            "question_type": meta["analysis"]["question_type"],
            "risk_level": card.level,
            "source": meta.get("source", "rule"),
            "profile_facts": profile_facts,
        }
    else:
        return {
            "need_ask": False,
            "need_analysis": True,
            "round": len(answers) // 3 + 1,
            "fact_summary": meta["fact_summary"],
            "question_type": meta["analysis"]["question_type"],
            "risk_level": card.level,
            "profile_facts": profile_facts,
        }


@app.post("/sessions/{sid}/ask-stream")
def ask_stream(sid: str, req: AskRequest, authorization: str | None = Header(default=None)):
    req.phone = get_current_phone(authorization)
    """流式提问:SSE 返回风险卡 + 逐 token 答复。

    事件格式:
    - event: risk\ndata: {风险卡 JSON}\n\n
    - event: delta\ndata: {\"content\": \"token\"}\n\n
    - event: done\ndata: {完整响应 JSON}\n\n
    - event: error\ndata: {\"detail\": \"错误信息\"}\n\n
    """
    try:
        sessions.append(sid, "user", req.question)
    except KeyError as e:
        raise HTTPException(410, str(e))

    # 获取会话模式(用于决定是否落库)
    session = sessions.sessions.get(sid)
    session_mode = session.mode if session else "normal"
    card = assess(req.question, kimi_client=kimi)
    inference_location = "local"
    model_name = "local-triage"
    fallback_used = False
    rag_context = []

    # 构建 prompt(与非流式一致)
    # 构建 prompt(动态定制 + 规则兜底)
    fact_summary = None
    if req.fact_answers:
        answers = [a.model_dump() for a in req.fact_answers]
        analysis = fact_finder.analyze(req.question, req.scene)
        fact_summary = fact_finder.summarize_facts(req.question, analysis, answers)

    prompt = build_dynamic_prompt(req.scene, req.question, kimi_client=kimi,
                                   fact_summary=fact_summary)
    # 注入用户画像
    if req.phone:
        profile = profile_store.get(req.phone)
        profile_ctx = build_profile_context(profile)
        if profile_ctx:
            prompt = prompt + "\n\n" + profile_ctx
    if req.scene == "emotion" and req.emotion_mode:
        prompt = build_emotion_prefix(req.emotion_mode, req.question, kimi_client=kimi) + prompt
    # RAG:按场景检索对应分类知识上下文
    ctx = knowledge.build_context(req.question, limit=3, scene=req.scene)
    if ctx:
        prompt = prompt + "\n\n" + ctx
        rag_context = knowledge.search(req.question, limit=3,
                                       doc_types=knowledge.SCENE_DOC_TYPES.get(req.scene))

    # 注入事实摘要(多轮追问后)
    if fact_summary:
        prompt = prompt + "\n\n" + fact_summary

    def event_stream():
        """SSE 事件生成器。"""
        nonlocal inference_location, model_name, fallback_used
        full_answer = []
        try:
            # 1. 先发风险卡
            yield f"event: risk\ndata: {json.dumps(card.__dict__, ensure_ascii=False)}\n\n"

            # 2. 流式答复
            if kimi:
                messages = [
                    {"role": "system", "content": prompt},
                    {"role": "user", "content": req.question},
                ]
                try:
                    for token in kimi.chat_stream(messages):
                        full_answer.append(token)
                        yield f"event: delta\ndata: {json.dumps({'content': token}, ensure_ascii=False)}\n\n"
                    inference_location = "cloud"
                    model_name = kimi.model
                except Exception as e:
                    # 云端失败,降级到本地分诊结论
                    logging.warning("kimi stream failed, fallback: %s", e)
                    fallback_text = card.conclusion
                    full_answer.append(fallback_text)
                    yield f"event: delta\ndata: {json.dumps({'content': fallback_text}, ensure_ascii=False)}\n\n"
                    inference_location = "local"
                    model_name = "local-triage"
                    fallback_used = True
            else:
                # 无 Kimi 配置,用本地分诊结论
                fallback_text = card.conclusion
                full_answer.append(fallback_text)
                yield f"event: delta\ndata: {json.dumps({'content': fallback_text}, ensure_ascii=False)}\n\n"

            answer = "".join(full_answer)
            sessions.append(sid, "assistant", answer)
            metrics.record_model(inference_location)

            # 落库:普通/保密模式保存消息+风险卡+元数据,阅后即焚不落库
            if req.phone and session_mode != "ephemeral":
                try:
                    chat_store.save_message(
                        session_id=sid, phone=req.phone, mode=session_mode, role="user",
                        content=req.question, scene=req.scene,
                    )
                    chat_store.save_message(
                        session_id=sid, phone=req.phone, mode=session_mode, role="assistant",
                        content=answer, scene=req.scene,
                        risk_level=card.level, risk_conclusion=card.conclusion,
                        risk_actions=card.actions, risk_evidence=card.evidence,
                        risk_escalation=card.escalation,
                        inference_location=inference_location, model_name=model_name,
                        fallback_used=fallback_used,
                    )
                except Exception as e:
                    logging.warning(f"chat_store save failed: {e}")

            # 3. 完成事件(含元数据)
            done_data = {
                "answer": answer,
                "action_summary": action_summary(card),
                "model": model_name,
                "scene": req.scene,
                "rag_context": rag_context,
                "inference_location": inference_location,
                "fallback_used": fallback_used,
                "fact_summary": fact_summary,
            }

            # 更新用户画像(流式完成后,提取抽象信号)
            if req.phone:
                try:
                    signals = extract_profile_signals(
                        req.question, req.scene, kimi_client=kimi,
                        fact_answers=[a.model_dump() for a in req.fact_answers] if req.fact_answers else None,
                    )
                    profile_store.update(req.phone, signals, scene=req.scene)
                except Exception:
                    pass

            yield f"event: done\ndata: {json.dumps(done_data, ensure_ascii=False)}\n\n"

        except Exception as e:
            yield f"event: error\ndata: {json.dumps({'detail': str(e)}, ensure_ascii=False)}\n\n"

    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",  # nginx 不缓冲
        },
    )


@app.post("/sessions/{sid}/destroy")
def destroy(sid: str):
    if not sessions.emergency_stop(sid):
        raise HTTPException(404, "session unavailable")
    metrics.record_destroy()
    status = sessions.destroy_status(sid)
    return {"destroyed": True, "session_id": sid, "destroy_log": status["destroy_log"] if status else []}


@app.get("/sessions/{sid}/destroy-status")
def destroy_status(sid: str):
    """查询会话销毁状态和日志(不含正文)。"""
    status = sessions.destroy_status(sid)
    if not status:
        raise HTTPException(404, "session not found")
    return status


@app.post("/sessions/{sid}/file")
async def inspect_file(sid: str, file: UploadFile = File(...)):
    try:
        sessions._active(sid)
    except KeyError as e:
        raise HTTPException(410, str(e))
    data = await file.read()
    if len(data) > 10 * 1024 * 1024:
        raise HTTPException(413, "file too large")
    text = extract_text(file.filename or "", data)
    # 文件名脱敏后记录日志
    logger.info(f"file uploaded: {redact(file.filename or '')} bytes={len(data)} extracted={len(text)}")
    return {
        "filename": file.filename,
        "bytes": len(data),
        "text_preview": text[:500],
        "text_length": len(text),
        "message": "文件已进入临时会话区" if text else "文件已接收,暂不支持该格式解析",
    }


# ===== 脱敏行动摘要 =====

@app.post("/summaries/preview")
def preview_summary(req: SummarySaveRequest):
    """保存前预览:展示将保留的脱敏字段。"""
    return {"savable_fields": summary_store.SAVABLE_FIELDS, "preview": summary_store.preview(req.summary)}


@app.post("/summaries")
def save_summary(req: SummarySaveRequest):
    """保存脱敏行动摘要(仅白名单字段,二次脱敏)。"""
    return summary_store.save(req.session_id, req.summary)


@app.get("/summaries")
def list_summaries(limit: int = 20):
    """列出已保存的脱敏摘要。"""
    return {"results": summary_store.list(limit=limit)}


# ===== 法规知识库 RAG =====

@app.get("/knowledge")
def list_knowledge(doc_type: str | None = None, limit: int = 50):
    """列出知识库文档,支持按类型筛选。"""
    return {"results": knowledge.list_all(doc_type=doc_type, limit=limit)}


@app.post("/knowledge")
def add_knowledge(req: KnowledgeAddRequest):
    """新增法规/案例/模板文档。"""
    return knowledge.add(
        title=req.title, content=req.content, source=req.source,
        doc_type=req.doc_type, region=req.region,
        effective_date=req.effective_date, keywords=req.keywords,
    )


@app.get("/knowledge/search")
def search_knowledge(q: str, doc_type: str | None = None, limit: int = 5):
    """RAG 检索:关键词匹配 + TF-IDF 排序。"""
    return {"results": knowledge.search(q, limit=limit, doc_type=doc_type)}


@app.get("/knowledge/vector-search")
def vector_search_knowledge(q: str, limit: int = 5):
    """向量检索:TF-IDF 余弦相似度。"""
    return {"results": knowledge.vector_search(q, limit=limit)}


@app.delete("/knowledge/{doc_id}")
def delete_knowledge(doc_id: int):
    """删除知识库文档。"""
    if not knowledge.delete(doc_id):
        raise HTTPException(404, "document not found")
    return {"deleted": True, "id": doc_id}


# ===== 用户画像 =====

@app.get("/profile/me")
def get_my_profile(authorization: str | None = Header(default=None)):
    """获取当前登录用户的画像。"""
    phone = get_current_phone(authorization)
    if not phone:
        raise HTTPException(401, "未登录或登录已过期")
    profile = profile_store.get(phone)
    return {
        "phone": profile.phone,
        "industry": profile.industry,
        "company_size": profile.company_size,
        "role": profile.role,
        "common_scenes": profile.common_scenes,
        "common_topics": profile.common_topics,
        "risk_tolerance": profile.risk_tolerance,
        "communication_style": profile.communication_style,
        "decision_style": profile.decision_style,
        "key_facts": profile.key_facts,
        "interaction_count": profile.interaction_count,
        "last_active": profile.last_active,
    }


@app.put("/profile/me")
def update_my_profile(body: dict, authorization: str | None = Header(default=None)):
    """更新当前登录用户的画像(用户确认事实后调用)。

    Body:
        {
            "industry": "...",
            "company_size": "...",
            "role": "...",
            "risk_tolerance": "...",
            "decision_style": "...",
            "key_facts": {"事实名": "事实值", ...}
        }
    """
    phone = get_current_phone(authorization)
    if not phone:
        raise HTTPException(401, "未登录或登录已过期")
    # 构造 signals,只更新非空字段
    signals: dict = {}
    for key in ("industry", "company_size", "role", "risk_tolerance",
                 "communication_style", "decision_style"):
        val = body.get(key)
        if val:
            signals[key] = val
    if body.get("key_facts") and isinstance(body["key_facts"], dict):
        signals["key_facts"] = body["key_facts"]
    if not signals:
        return {"updated": False}
    signals["_source"] = "manual"  # 标记为用户手动确认
    profile = profile_store.update(phone, signals)
    return {
        "updated": True,
        "industry": profile.industry,
        "company_size": profile.company_size,
        "role": profile.role,
        "risk_tolerance": profile.risk_tolerance,
        "decision_style": profile.decision_style,
        "key_facts": profile.key_facts,
    }


@app.get("/profile/me/history")
def get_my_profile_history(authorization: str | None = Header(default=None)):
    """获取当前登录用户的画像历史版本。"""
    phone = get_current_phone(authorization)
    if not phone:
        raise HTTPException(401, "未登录或登录已过期")
    return {"results": profile_store.list_history(phone)}


# ===== 用户设置持久化 =====

@app.get("/settings")
def get_settings():
    """读取全部用户设置。"""
    return {"settings": settings_store.get_all()}


@app.put("/settings")
def update_settings(req: SettingsUpdateRequest):
    """批量更新用户设置。"""
    updated = settings_store.set_many(req.settings)
    return {"updated": True, "settings": settings_store.get_all()}


# ===== 专家库(用户端) =====

class ConsultationRequest(BaseModel):
    """一键咨询:将 AI 对话话题发给专家。"""
    expert_id: int
    topic: str                                    # 咨询话题
    context: str = ""                             # 对话上下文(脱敏摘要)
    question: str = ""                            # 给专家的具体问题


class RatingRequest(BaseModel):
    """服务评价。"""
    consultation_id: int
    rating: int                                   # 1-5 星
    comment: str = ""


@app.get("/experts")
def list_experts(category: str | None = None):
    """列出专家,支持按分类筛选(legal/decision/emotion/life)。"""
    return {"results": expert_store.list_experts(category=category, status="active")}


@app.get("/experts/{expert_id}")
def get_expert(expert_id: int):
    """获取专家详情 + 评价列表。"""
    expert = expert_store.get_expert(expert_id)
    if not expert:
        raise HTTPException(404, "专家不存在")
    ratings = rating_store.list_by_expert(expert_id, limit=10)
    expert["ratings"] = ratings
    return expert


@app.post("/experts/{expert_id}/select")
def select_expert(expert_id: int, authorization: str | None = Header(default=None)):
    """用户选定专家,发起服务请求(待专家确认)。"""
    phone = get_current_phone(authorization)
    if not phone:
        raise HTTPException(401, "未登录或登录已过期")
    expert = expert_store.get_expert(expert_id)
    if not expert:
        raise HTTPException(404, "专家不存在")
    if expert["status"] != "active":
        raise HTTPException(400, "该专家已下架")
    try:
        result = expert_store.create_service(phone, expert_id, expert["category"])
        return result
    except ValueError as e:
        raise HTTPException(400, str(e))


@app.get("/my/services")
def my_services(authorization: str | None = Header(default=None)):
    """查看我的专家服务列表。"""
    phone = get_current_phone(authorization)
    if not phone:
        raise HTTPException(401, "未登录或登录已过期")
    return {"results": expert_store.list_user_services(phone)}


@app.get("/my/consultations")
def my_consultations(status: str | None = None,
                     authorization: str | None = Header(default=None)):
    """查看我的咨询列表。"""
    phone = get_current_phone(authorization)
    if not phone:
        raise HTTPException(401, "未登录或登录已过期")
    return {"results": consultation_store.list_user_consultations(phone, status=status)}


@app.post("/consultations")
def create_consultation(req: ConsultationRequest,
                        authorization: str | None = Header(default=None)):
    """一键咨询:将 AI 对话话题发给专家。"""
    phone = get_current_phone(authorization)
    if not phone:
        raise HTTPException(401, "未登录或登录已过期")
    expert = expert_store.get_expert(req.expert_id)
    if not expert:
        raise HTTPException(404, "专家不存在")
    # 检查是否有 active 服务关系
    services = expert_store.list_user_services(phone)
    active_service = next(
        (s for s in services if s["expert_id"] == req.expert_id and s["status"] == "active"),
        None,
    )
    if not active_service:
        raise HTTPException(400, "请先选定该专家并等待专家确认")
    # 脱敏上下文
    safe_context = redact(req.context) if req.context else ""
    safe_question = redact(req.question) if req.question else ""
    return consultation_store.create(
        service_id=active_service["service_id"],
        user_phone=phone, expert_id=req.expert_id,
        category=expert["category"], topic=req.topic,
        context=safe_context, question=safe_question,
    )


@app.post("/consultations/{consultation_id}/rate")
def rate_consultation(consultation_id: int, req: RatingRequest,
                      authorization: str | None = Header(default=None)):
    """评价专家服务。"""
    phone = get_current_phone(authorization)
    if not phone:
        raise HTTPException(401, "未登录或登录已过期")
    consultation = consultation_store.get(consultation_id)
    if not consultation:
        raise HTTPException(404, "咨询不存在")
    if consultation["user_phone"] != phone:
        raise HTTPException(403, "无权评价他人的咨询")
    if consultation["status"] != "replied":
        raise HTTPException(400, "专家尚未回复,暂无法评价")
    try:
        return rating_store.create(
            consultation_id=consultation_id, user_phone=phone,
            expert_id=consultation["expert_id"],
            rating=req.rating, comment=req.comment,
        )
    except ValueError as e:
        raise HTTPException(400, str(e))


# ===== 专家工作台 =====

@app.get("/expert/workbench")
def expert_workbench(authorization: str | None = Header(default=None)):
    """专家工作台:查看自己的服务关系和待回复咨询。"""
    phone = get_current_phone(authorization)
    if not phone:
        raise HTTPException(401, "未登录或登录已过期")
    # 通过手机号查找专家
    experts = expert_store.list_experts()
    expert = next((e for e in experts if e["phone"] == phone), None)
    if not expert:
        raise HTTPException(403, "您不是注册专家")
    services = expert_store.list_expert_services(expert["id"])
    consultations = consultation_store.list_expert_consultations(expert["id"])
    return {"expert": expert, "services": services, "consultations": consultations}


@app.post("/expert/services/{service_id}/accept")
def accept_service(service_id: int, authorization: str | None = Header(default=None)):
    """专家确认服务关系(签协议)。"""
    phone = get_current_phone(authorization)
    if not phone:
        raise HTTPException(401, "未登录或登录已过期")
    experts = expert_store.list_experts()
    expert = next((e for e in experts if e["phone"] == phone), None)
    if not expert:
        raise HTTPException(403, "您不是注册专家")
    if not expert_store.accept_service(service_id, expert["id"]):
        raise HTTPException(400, "确认失败(服务不存在或已确认)")
    return {"accepted": True}


@app.post("/my/services/{service_id}/cancel")
def cancel_service_user(service_id: int, authorization: str | None = Header(default=None)):
    """用户取消服务关系(无需专家确认)。"""
    phone = get_current_phone(authorization)
    if not phone:
        raise HTTPException(401, "未登录或登录已过期")
    if not expert_store.close_service(service_id, user_phone=phone):
        raise HTTPException(400, "取消失败(服务不存在或已结束)")
    return {"cancelled": True}


@app.post("/expert/services/{service_id}/cancel")
def cancel_service_expert(service_id: int, authorization: str | None = Header(default=None)):
    """专家取消服务关系(无需用户确认)。"""
    phone = get_current_phone(authorization)
    if not phone:
        raise HTTPException(401, "未登录或登录已过期")
    experts = expert_store.list_experts()
    expert = next((e for e in experts if e["phone"] == phone), None)
    if not expert:
        raise HTTPException(403, "您不是注册专家")
    if not expert_store.close_service(service_id, expert_id=expert["id"]):
        raise HTTPException(400, "取消失败(服务不存在或已结束)")
    return {"cancelled": True}


@app.post("/expert/consultations/{consultation_id}/reply")
def reply_consultation(consultation_id: int, body: dict,
                       authorization: str | None = Header(default=None)):
    """专家回复咨询。"""
    phone = get_current_phone(authorization)
    if not phone:
        raise HTTPException(401, "未登录或登录已过期")
    experts = expert_store.list_experts()
    expert = next((e for e in experts if e["phone"] == phone), None)
    if not expert:
        raise HTTPException(403, "您不是注册专家")
    reply_text = body.get("reply", "").strip()
    if not reply_text:
        raise HTTPException(400, "回复内容不能为空")
    if not consultation_store.reply(consultation_id, expert["id"], reply_text):
        raise HTTPException(400, "回复失败(咨询不存在或已回复)")
    return {"replied": True}


# ===== 对话历史 =====

class HistoryVerifyRequest(BaseModel):
    password: str


@app.get("/history/sessions")
def list_history_sessions(
    mode: str | None = None,
    authorization: str | None = Header(default=None),
):
    """列出用户的对话历史会话(按时间倒序)。

    普通模式和保密模式的历史会话都会列出(仅摘要,不含正文)。
    阅后即焚不落库,不会出现。

    Args:
        mode: 可选模式过滤(normal/confidential)
    """
    phone = get_current_phone(authorization)
    if not phone:
        raise HTTPException(401, "未登录")
    return {"results": chat_store.list_sessions(phone, mode=mode)}


@app.get("/history/sessions/{session_id}")
def get_history_messages(
    session_id: str,
    authorization: str | None = Header(default=None),
):
    """获取会话的对话历史。

    - 普通模式:直接返回明文消息。
    - 保密模式:返回加密标记,需调用 /history/sessions/{sid}/verify 验证密码后获取明文。

    Args:
        session_id: 会话 ID
    """
    phone = get_current_phone(authorization)
    if not phone:
        raise HTTPException(401, "未登录")
    # 先取消息判断模式
    messages = chat_store.get_messages(session_id, phone, decrypt=False)
    if not messages:
        raise HTTPException(404, "会话不存在或无历史")
    mode = messages[0].get("mode", "normal")
    if mode == "confidential":
        # 保密模式:不直接返回明文,前端需先验证密码
        return {
            "session_id": session_id,
            "mode": "confidential",
            "need_verify": True,
            "message_count": len(messages),
        }
    # 普通模式:直接返回明文
    return {
        "session_id": session_id,
        "mode": "normal",
        "need_verify": False,
        "messages": messages,
    }


@app.post("/history/sessions/{session_id}/verify")
def verify_and_get_history(
    session_id: str,
    req: HistoryVerifyRequest,
    authorization: str | None = Header(default=None),
):
    """保密模式:验证登录密码后返回解密的历史消息。

    二次验证用登录密码(bcrypt 校验),验证通过后解密返回明文。
    """
    phone = get_current_phone(authorization)
    if not phone:
        raise HTTPException(401, "未登录")
    # 二次验证登录密码
    with user_store._conn() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT password FROM users WHERE phone = %s", (phone,))
            row = cur.fetchone()
    if not row:
        raise HTTPException(401, "用户不存在")
    import bcrypt
    if not bcrypt.checkpw(req.password.encode(), row[0].encode()):
        raise HTTPException(403, "密码验证失败")
    # 验证通过,返回解密后的消息
    messages = chat_store.get_messages(session_id, phone, decrypt=True)
    if not messages:
        raise HTTPException(404, "会话不存在或无历史")
    return {
        "session_id": session_id,
        "mode": "confidential",
        "need_verify": False,
        "verified": True,
        "messages": messages,
    }


@app.delete("/history/sessions/{session_id}")
def delete_history_session(
    session_id: str,
    authorization: str | None = Header(default=None),
):
    """删除会话的所有历史消息(用户主动清除历史)。"""
    phone = get_current_phone(authorization)
    if not phone:
        raise HTTPException(401, "未登录")
    deleted = chat_store.delete_session(session_id, phone)
    if deleted == 0:
        raise HTTPException(404, "会话不存在或无历史")
    return {"deleted": True, "session_id": session_id, "count": deleted}


# ===== 管理后台 =====

@app.get("/admin/experts")
def admin_list_experts(status: str | None = None):
    """管理端:列出所有专家(含下架)。"""
    return {"results": expert_store.list_experts(status=status)}


@app.patch("/admin/experts/{expert_id}/status")
def admin_update_expert_status(expert_id: int, body: dict):
    """管理端:上架/下架专家。"""
    status = body.get("status", "")
    if status not in ("active", "inactive"):
        raise HTTPException(400, "状态应为 active 或 inactive")
    if not expert_store.update_status(expert_id, status):
        raise HTTPException(404, "专家不存在")
    return {"updated": True, "expert_id": expert_id, "status": status}


class ExpertUpdateRequest(BaseModel):
    """编辑专家信息请求(所有字段可选)。"""
    name: str | None = None
    category: str | None = None
    title: str | None = None
    bio: str | None = None
    expertise: str | None = None
    avatar_url: str | None = None
    status: str | None = None


@app.patch("/admin/experts/{expert_id}")
def admin_update_expert(expert_id: int, req: ExpertUpdateRequest):
    """管理端:编辑专家信息(姓名/分类/职称/简介/专长/头像/状态)。"""
    try:
        updated = expert_store.update_expert(
            expert_id,
            name=req.name, category=req.category, title=req.title,
            bio=req.bio, expertise=req.expertise,
            avatar_url=req.avatar_url, status=req.status,
        )
    except ValueError as e:
        raise HTTPException(400, str(e))
    if not updated:
        raise HTTPException(404, "专家不存在")
    return {"updated": True, "expert": updated}


@app.post("/admin/experts/{expert_id}/avatar")
async def admin_upload_expert_avatar(expert_id: int, file: UploadFile = File(...)):
    """管理端:上传专家头像,保存到 uploads/avatars/,返回可访问 URL。"""
    # 校验专家存在
    if not expert_store.get_expert(expert_id):
        raise HTTPException(404, "专家不存在")
    data = await file.read()
    if len(data) > 5 * 1024 * 1024:
        raise HTTPException(413, "图片不能超过 5MB")
    # 校验文件类型
    allowed = {"image/jpeg", "image/png", "image/webp"}
    if file.content_type not in allowed:
        raise HTTPException(400, f"仅支持 {', '.join(allowed)} 格式")
    # 保存到 uploads/avatars/
    upload_dir = os.path.join(os.path.dirname(__file__), "uploads", "avatars")
    os.makedirs(upload_dir, exist_ok=True)
    ext = {"image/jpeg": ".jpg", "image/png": ".png", "image/webp": ".webp"}[file.content_type]
    filename = f"expert_{expert_id}{ext}"
    filepath = os.path.join(upload_dir, filename)
    with open(filepath, "wb") as f:
        f.write(data)
    # 返回可访问 URL(经 Vite 代理或 nginx 反代)
    avatar_url = f"/uploads/avatars/{filename}"
    # 同步更新数据库
    expert_store.update_expert(expert_id, avatar_url=avatar_url)
    logger.info(f"expert {expert_id} avatar uploaded: {avatar_url}")
    return {"avatar_url": avatar_url}


@app.get("/admin/users")
def admin_list_users(q: str | None = None, limit: int = 50):
    """管理端:列出所有注册用户,支持搜索,含对话统计和状态。"""
    with user_store._conn() as conn:
        with conn.cursor() as cur:
            if q:
                cur.execute(
                    "SELECT id, phone, name, status, created_at FROM users "
                    "WHERE phone LIKE %s OR name LIKE %s "
                    "ORDER BY created_at DESC LIMIT %s",
                    (f"%{q}%", f"%{q}%", limit),
                )
            else:
                cur.execute(
                    "SELECT id, phone, name, status, created_at FROM users "
                    "ORDER BY created_at DESC LIMIT %s",
                    (limit,),
                )
            rows = cur.fetchall()
            # 批量查询每个用户的对话数和最后活跃时间
            results = []
            for r in rows:
                uid, phone, name, status, created_at = r
                cur.execute(
                    "SELECT COUNT(*), MAX(created_at), "
                    "(array_agg(meta_data->>'scene' ORDER BY created_at DESC) FILTER (WHERE role = 'assistant'))[1] "
                    "FROM chat_messages WHERE phone = %s",
                    (phone,),
                )
                stat = cur.fetchone()
                msg_count = stat[0] or 0
                last_active = stat[1]
                last_scene = stat[2]
                results.append({
                    "id": uid,
                    "phone": phone[:3] + "****" + phone[-4:],
                    "name": name,
                    "status": status,
                    "created_at": str(created_at),
                    "message_count": msg_count,
                    "last_active": str(last_active) if last_active else None,
                    "last_scene": last_scene,
                })
    return {"results": results}


@app.post("/admin/users/{user_id}/reset-password")
def admin_reset_password(user_id: int, body: dict):
    """管理端:重置用户密码。"""
    new_password = body.get("password", "")
    if not new_password or len(new_password) < 6:
        raise HTTPException(400, "密码至少 6 位")
    if not user_store.reset_password(user_id, new_password):
        raise HTTPException(404, "用户不存在")
    return {"reset": True}


@app.post("/admin/users/{user_id}/status")
def admin_update_user_status(user_id: int, body: dict):
    """管理端:激活/停用用户。"""
    status = body.get("status", "")
    if status not in ("active", "inactive"):
        raise HTTPException(400, "状态应为 active 或 inactive")
    if not user_store.update_status(user_id, status):
        raise HTTPException(404, "用户不存在")
    return {"status": status}


@app.get("/admin/stats")
def admin_stats():
    """管理端:平台统计。"""
    with user_store._conn() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT count(*) FROM users")
            user_count = cur.fetchone()[0]
            cur.execute("SELECT count(*) FROM experts WHERE status = 'active'")
            expert_active = cur.fetchone()[0]
            cur.execute("SELECT count(*) FROM experts")
            expert_total = cur.fetchone()[0]
            cur.execute("SELECT count(*) FROM expert_services WHERE status = 'active'")
            service_active = cur.fetchone()[0]
            cur.execute("SELECT count(*) FROM expert_consultations")
            consultation_total = cur.fetchone()[0]
            cur.execute("SELECT count(*) FROM expert_consultations WHERE status = 'pending'")
            consultation_pending = cur.fetchone()[0]
    return {
        "users": user_count,
        "experts_active": expert_active,
        "experts_total": expert_total,
        "services_active": service_active,
        "consultations_total": consultation_total,
        "consultations_pending": consultation_pending,
    }


@app.get("/admin/ai-usage")
def admin_ai_usage():
    """管理端:用户 AI 使用情况统计(不含具体对话内容)。

    返回维度:
    - 总提问数、活跃用户数
    - 按场景分布(legal/decision/life/emotion)
    - 按隐私模式分布(normal/confidential,ephemeral 不落库)
    - 按风险等级分布(green/amber/orange/red)
    - 按推理位置分布(cloud/local)
    - 近 7 天每日提问趋势
    - 按用户统计(脱敏手机号,提问数、最后活跃时间、常用场景)
    """
    with user_store._conn() as conn:
        with conn.cursor() as cur:
            # 总提问数(用户消息)
            cur.execute("SELECT count(*) FROM chat_messages WHERE role = 'user'")
            total_questions = cur.fetchone()[0]

            # 活跃用户数(有提问记录的不同手机号)
            cur.execute("SELECT count(DISTINCT phone) FROM chat_messages WHERE role = 'user'")
            active_users = cur.fetchone()[0]

            # 按场景分布
            cur.execute("""
                SELECT meta_data->>'scene' AS scene, count(*) AS cnt
                FROM chat_messages WHERE role = 'user'
                GROUP BY scene ORDER BY cnt DESC
            """)
            by_scene = [{"scene": r[0] or "unknown", "count": r[1]} for r in cur.fetchall()]

            # 按隐私模式分布
            cur.execute("""
                SELECT mode, count(*) AS cnt
                FROM chat_messages WHERE role = 'user'
                GROUP BY mode ORDER BY cnt DESC
            """)
            by_mode = [{"mode": r[0], "count": r[1]} for r in cur.fetchall()]

            # 按风险等级分布(AI 回复中的风险卡)
            cur.execute("""
                SELECT risk_data->>'level' AS level, count(*) AS cnt
                FROM chat_messages WHERE role = 'assistant'
                GROUP BY level ORDER BY cnt DESC
            """)
            by_risk = [{"level": r[0] or "unknown", "count": r[1]} for r in cur.fetchall()]

            # 按推理位置分布
            cur.execute("""
                SELECT meta_data->>'inference_location' AS loc, count(*) AS cnt
                FROM chat_messages WHERE role = 'assistant'
                GROUP BY loc ORDER BY cnt DESC
            """)
            by_inference = [{"location": r[0] or "unknown", "count": r[1]} for r in cur.fetchall()]

            # 近 7 天每日提问趋势
            cur.execute("""
                SELECT date_trunc('day', created_at)::date AS d, count(*) AS cnt
                FROM chat_messages WHERE role = 'user'
                  AND created_at >= now() - interval '7 days'
                GROUP BY d ORDER BY d
            """)
            daily_trend = [{"date": r[0].isoformat(), "count": r[1]} for r in cur.fetchall()]

            # 按用户统计(脱敏手机号)
            cur.execute("""
                SELECT phone,
                       count(*) FILTER (WHERE role = 'user') AS questions,
                       count(*) FILTER (WHERE role = 'assistant') AS replies,
                       max(created_at) AS last_active,
                       (array_agg(meta_data->>'scene' ORDER BY created_at DESC)
                        FILTER (WHERE role = 'user'))[1] AS top_scene
                FROM chat_messages
                GROUP BY phone
                ORDER BY questions DESC
                LIMIT 50
            """)
            by_user = []
            for r in cur.fetchall():
                phone = r[0]
                # 脱敏:138****1234
                masked = phone[:3] + "****" + phone[-4:] if len(phone) >= 8 else phone
                by_user.append({
                    "phone_masked": masked,
                    "questions": r[1],
                    "replies": r[2],
                    "last_active": r[3].isoformat() if r[3] else None,
                    "top_scene": r[4],
                })

    return {
        "total_questions": total_questions,
        "active_users": active_users,
        "by_scene": by_scene,
        "by_mode": by_mode,
        "by_risk": by_risk,
        "by_inference": by_inference,
        "daily_trend": daily_trend,
        "by_user": by_user,
    }


@app.get("/admin/expert-activity")
def admin_expert_activity():
    """管理端:用户与专家沟通情况统计(不含具体咨询内容)。

    返回维度:
    - 总咨询数、已回复数、待回复数、已关闭数
    - 按分类分布(legal/decision/emotion/life)
    - 按状态分布(pending/replied/closed)
    - 按专家分布(专家名、咨询数、已回复数、待回复数、平均评分)
    - 近 7 天每日咨询趋势
    - 按用户分布(脱敏手机号,咨询数)
    """
    with user_store._conn() as conn:
        with conn.cursor() as cur:
            # 按状态分布
            cur.execute("""
                SELECT status, count(*) AS cnt
                FROM expert_consultations
                GROUP BY status ORDER BY cnt DESC
            """)
            by_status = [{"status": r[0], "count": r[1]} for r in cur.fetchall()]

            # 按分类分布
            cur.execute("""
                SELECT category, count(*) AS cnt
                FROM expert_consultations
                GROUP BY category ORDER BY cnt DESC
            """)
            by_category = [{"category": r[0], "count": r[1]} for r in cur.fetchall()]

            # 按专家分布(含平均评分)
            cur.execute("""
                SELECT e.name, e.category, e.title,
                       count(c.id) AS total,
                       count(c.id) FILTER (WHERE c.status = 'replied') AS replied,
                       count(c.id) FILTER (WHERE c.status = 'pending') AS pending,
                       COALESCE(avg(r.rating), 0) AS avg_rating,
                       count(r.id) AS rating_count
                FROM experts e
                LEFT JOIN expert_consultations c ON c.expert_id = e.id
                LEFT JOIN expert_ratings r ON r.consultation_id = c.id
                GROUP BY e.id, e.name, e.category, e.title
                ORDER BY total DESC
            """)
            by_expert = [{
                "name": r[0], "category": r[1], "title": r[2],
                "total": r[3], "replied": r[4], "pending": r[5],
                "avg_rating": round(float(r[6]), 1), "rating_count": r[7],
            } for r in cur.fetchall()]

            # 近 7 天每日咨询趋势
            cur.execute("""
                SELECT date_trunc('day', created_at)::date AS d, count(*) AS cnt
                FROM expert_consultations
                WHERE created_at >= now() - interval '7 days'
                GROUP BY d ORDER BY d
            """)
            daily_trend = [{"date": r[0].isoformat(), "count": r[1]} for r in cur.fetchall()]

            # 按用户分布(脱敏)
            cur.execute("""
                SELECT user_phone, count(*) AS cnt, max(created_at) AS last_active
                FROM expert_consultations
                GROUP BY user_phone
                ORDER BY cnt DESC
                LIMIT 50
            """)
            by_user = []
            for r in cur.fetchall():
                phone = r[0]
                masked = phone[:3] + "****" + phone[-4:] if len(phone) >= 8 else phone
                by_user.append({
                    "phone_masked": masked,
                    "consultations": r[1],
                    "last_active": r[2].isoformat() if r[2] else None,
                })

    total = sum(s["count"] for s in by_status)
    replied = next((s["count"] for s in by_status if s["status"] == "replied"), 0)
    pending = next((s["count"] for s in by_status if s["status"] == "pending"), 0)

    return {
        "total": total,
        "replied": replied,
        "pending": pending,
        "by_status": by_status,
        "by_category": by_category,
        "by_expert": by_expert,
        "daily_trend": daily_trend,
        "by_user": by_user,
    }


# ===== 文件生成 =====

class DocumentGenerateRequest(BaseModel):
    template: str  # demand_letter/labor_notice/termination_notice/evidence_checklist/general_notice
    fields: dict
    session_id: str | None = None  # 阅后即焚会话禁止导出


@app.get("/documents/templates")
def list_templates():
    """列出可用文书模板。"""
    return {"templates": DocumentGenerator.TEMPLATES}


@app.post("/documents/generate")
def generate_document(req: DocumentGenerateRequest):
    """生成法律文书 DOCX。

    阅后即焚(ephemeral)会话禁止导出。
    """
    # 检查会话模式
    ephemeral = False
    if req.session_id:
        session = sessions.sessions.get(req.session_id)
        if session and session.mode == "ephemeral" and not session.destroyed:
            ephemeral = True
    if ephemeral:
        raise HTTPException(403, "阅后即焚模式下禁止导出文件")

    gen = DocumentGenerator(ephemeral_mode=ephemeral, kimi_client=kimi)
    try:
        data = gen.generate(req.template, req.fields)
    except ValueError as e:
        raise HTTPException(400, str(e))

    from fastapi.responses import Response
    # 文件名用 ASCII(中文文件名在 header 中会编码失败)
    filename = f"{req.template}_{int(time.time())}.docx"
    return Response(
        content=data,
        media_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
    )


# ===== 法规更新管理 =====

class RegulationUpdateRequest(BaseModel):
    doc_id: int
    content: str
    title: str | None = None
    effective_date: str | None = None
    keywords: str | None = None
    snapshot_reason: str = "法规更新"


class RegulationRollbackRequest(BaseModel):
    version_num: int


@app.get("/regulations/check-updates")
def check_regulation_updates():
    """扫描知识库,返回需要更新的法规列表。"""
    return {"results": regulation_mgr.check_updates()}


@app.post("/regulations/update")
def update_regulation(req: RegulationUpdateRequest):
    """更新法规文档,自动创建更新前快照。"""
    try:
        result = regulation_mgr.update_document(
            doc_id=req.doc_id, content=req.content, title=req.title,
            effective_date=req.effective_date, keywords=req.keywords,
            snapshot_reason=req.snapshot_reason,
        )
        return result
    except ValueError as e:
        raise HTTPException(404, str(e))


@app.get("/regulations/{doc_id}/versions")
def list_regulation_versions(doc_id: int):
    """列出法规文档的所有版本快照。"""
    return {"results": regulation_mgr.list_versions(doc_id)}


@app.post("/regulations/{doc_id}/rollback")
def rollback_regulation(doc_id: int, req: RegulationRollbackRequest):
    """回滚法规文档到指定版本。"""
    try:
        return regulation_mgr.rollback(doc_id, req.version_num)
    except ValueError as e:
        raise HTTPException(404, str(e))


# ===== 静态文件托管:上传的专家头像等 =====
from fastapi.staticfiles import StaticFiles
_upload_dir = os.path.join(os.path.dirname(__file__), "uploads")
os.makedirs(_upload_dir, exist_ok=True)
app.mount("/uploads", StaticFiles(directory=_upload_dir), name="uploads")