app/services/request_log_store.py

"""请求日志存储：完整记录每个业务请求的入参/出参/耗时/错误。

借鉴 ds2api 的设计：
- 完整记录请求标头、请求体、响应体、响应标头、耗时、错误码
- 容量轮转：保留最近 N 条（默认 500，可配 0/100/200/500/1000）
- 列表查询返回摘要（不带大 body），详情查询返回完整
- 敏感信息脱敏（authorization / api_key / x-xtc-admin-key 等）

记录范围：/v1/* 和 /admin/api/* 路径
不记录：/health /docs /openapi.json /redoc /admin（HTML 页面）静态资源
"""
from __future__ import annotations

import json
import re
import time
from typing import Any, Optional

from ..database import get_conn


# ===== 常量 =====

# 敏感标头/key 脱敏
_SENSITIVE_HEADERS = {
    "authorization", "x-xtc-admin-key", "x-xtc-access-key", "x-xtc-access-token",
    "cookie", "x-api-key", "x-hf-token", "hf-token",
}
_SENSITIVE_BODY_KEYS = {
    "api_key", "apikey", "api_keys", "apikey", "key", "token", "access_token",
    "access_key", "admin_key", "password", "secret", "authorization",
}
# body 最大记录长度（超过截断）
_MAX_BODY_LEN = 64 * 1024  # 64KB
# 标头最大记录长度
_MAX_HEADERS_LEN = 16 * 1024  # 16KB


# ===== 脱敏 =====

def _redact_headers(headers: dict) -> dict:
    """脱敏敏感标头：保留前 4 位 + ***。"""
    out = {}
    for k, v in headers.items():
        kl = k.lower()
        if kl in _SENSITIVE_HEADERS:
            sv = str(v)
            if len(sv) <= 8:
                out[k] = "***"
            else:
                out[k] = sv[:4] + "***" + sv[-2:]
        else:
            out[k] = v
    return out


def _redact_body(body: Any) -> Any:
    """递归脱敏 body 中的敏感字段。"""
    if isinstance(body, dict):
        out = {}
        for k, v in body.items():
            if k.lower() in _SENSITIVE_BODY_KEYS:
                out[k] = "***REDACTED***"
            else:
                out[k] = _redact_body(v)
        return out
    if isinstance(body, list):
        return [_redact_body(item) for item in body]
    return body


def _truncate(s: Optional[str], max_len: int) -> Optional[str]:
    if s is None:
        return None
    if len(s) <= max_len:
        return s
    return s[:max_len] + f"\n...[truncated, total {len(s)} bytes]"


# 设备 ID 长度上限（防止异常长字符串刷屏/撑爆索引）
_MAX_DEVICE_ID_LEN = 128


def _normalize_device_id(raw: Any) -> str:
    """归一化设备 ID：去空白、截断、空值返回空字符串（确保入库）。"""
    if raw is None:
        return ""
    s = str(raw).strip()
    if not s:
        return ""
    if len(s) > _MAX_DEVICE_ID_LEN:
        s = s[:_MAX_DEVICE_ID_LEN]
    return s


# ===== 容量配置 =====

# 内存缓存：limit 只在管理员改设置时变化，缓存后避免每次 insert/list 都 SELECT kv。
_limit_cache: Optional[int] = None


def get_limit() -> int:
    """读取容量上限。0 = 禁用记录。命中缓存直接返回。"""
    global _limit_cache
    if _limit_cache is not None:
        return _limit_cache
    try:
        with get_conn() as conn:
            row = conn.execute(
                "SELECT value FROM kv WHERE key = 'request_log_limit'"
            ).fetchone()
            _limit_cache = int(row["value"]) if row else 500
    except Exception:
        return 500
    return _limit_cache


def set_limit(limit: int) -> int:
    """设置容量上限，并立即触发轮转。"""
    global _limit_cache
    limit = max(0, min(5000, int(limit)))
    now = int(time.time())
    with get_conn() as conn:
        conn.execute(
            "INSERT INTO kv(key, value, updated_at) VALUES(?,?,?) "
            "ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at",
            ("request_log_limit", str(limit), now),
        )
    _limit_cache = limit
    if limit > 0:
        _rotate(limit)
    return limit


def _rotate(limit: int) -> int:
    """删除超限的旧记录（事件循环线程调用）。返回删除条数。"""
    with get_conn() as conn:
        return _do_rotate(conn, limit)


def _do_rotate(conn, limit: int) -> int:
    """在给定连接上执行轮转。"""
    row = conn.execute(
        "SELECT id FROM request_log ORDER BY id DESC LIMIT 1 OFFSET ?",
        (limit,),
    ).fetchone()
    if not row:
        return 0
    threshold = int(row["id"])
    cur = conn.execute("DELETE FROM request_log WHERE id < ?", (threshold,))
    return cur.rowcount


# ===== 写入 =====

# 后台写线程的插入计数器，用于阈值触发轮转（单线程写，无需加锁）。
_insert_count = 0


def insert(record: dict) -> Optional[int]:
    """插入一条请求日志（异步：转交后台写线程，立即返回 None）。

    limit=0 时不插入。脱敏/截断在事件循环上完成（开销小），INSERT + 轮转
    在后台写线程执行，避免阻塞事件循环。
    """
    global _insert_count
    limit = get_limit()
    if limit <= 0:
        return None
    now = int(time.time())
    args = (
        now,
        record.get("method", ""),
        record.get("path", ""),
        record.get("query"),
        record.get("status_code"),
        record.get("elapsed_ms"),
        record.get("access_key"),
        record.get("client_ip"),
        record.get("user_agent"),
        record.get("provider"),
        record.get("model"),
        1 if record.get("stream") else 0,
        1 if record.get("ok") else 0,
        record.get("error_code"),
        record.get("error_message"),
        _truncate(record.get("request_headers"), _MAX_HEADERS_LEN),
        _truncate(record.get("request_body"), _MAX_BODY_LEN),
        _truncate(record.get("response_body"), _MAX_BODY_LEN),
        _truncate(record.get("response_headers"), _MAX_HEADERS_LEN),
        _normalize_device_id(record.get("device_id")),
    )
    sql = (
        """INSERT INTO request_log(
            ts, method, path, query, status_code, elapsed_ms,
            access_key, client_ip, user_agent, provider, model, stream,
            ok, error_code, error_message,
            request_headers, request_body, response_body, response_headers, device_id
        ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"""
    )
    rotate_limit = limit

    def _task(c):
        global _insert_count
        c.execute(sql, args)
        _insert_count += 1
        # 每 50 次写入触发一次轮转（阈值触发，避免每次都查）
        if _insert_count % 50 == 0:
            _do_rotate(c, rotate_limit)

    try:
        from ..db_writer import enqueue, is_started
        if is_started():
            enqueue(_task)
            return None
    except Exception:
        pass
    # 回退：写线程未启动（如启动前/测试），在事件循环上内联写入
    with get_conn() as conn:
        conn.execute(sql, args)
    return None


# ===== 查询 =====

def _category_filter(category: Optional[str]) -> tuple[str, list]:
    """返回 (SQL 片段, 参数) 用于按请求类别过滤。

    - ``business``（默认）：只看业务请求 ``/v1/*``（对话/伪流式/OpenAI 兼容等）
    - ``admin``：只看管理面板访问 ``/admin/api/*``
    - ``all`` 或其它：不过滤
    """
    cat = str(category or "business").lower()
    if cat == "admin":
        return " AND path LIKE '/admin/api/%'", []
    if cat == "all":
        return "", []
    # 默认 business：排除管理面板访问，只保留 /v1/*
    return " AND path LIKE '/v1/%'", []


def list_records(
    *,
    limit: int = 50,
    offset: int = 0,
    path: Optional[str] = None,
    method: Optional[str] = None,
    status: Optional[str] = None,
    access_key: Optional[str] = None,
    ok: Optional[bool] = None,
    hours: Optional[int] = None,
    keyword: Optional[str] = None,
    category: Optional[str] = None,
    device_id: Optional[str] = None,
) -> dict:
    """列表查询：返回摘要（不含 request_body / response_body / headers）。"""
    sql = "SELECT id, ts, method, path, query, status_code, elapsed_ms, access_key, client_ip, user_agent, provider, model, stream, ok, error_code, error_message, device_id FROM request_log WHERE 1=1"
    args: list = []
    cat_sql, cat_args = _category_filter(category)
    if cat_sql:
        sql += cat_sql
        args += cat_args
    if path:
        sql += " AND path LIKE ?"
        args.append(f"%{path}%")
    if method:
        sql += " AND method = ?"
        args.append(method.upper())
    if status:
        # 支持 2xx/3xx/4xx/5xx 通配
        s = str(status).strip()
        if s.endswith("xx"):
            prefix = s[0]
            sql += " AND CAST(status_code/100 AS INTEGER) = ?"
            args.append(int(prefix))
        else:
            sql += " AND status_code = ?"
            args.append(int(s))
    if access_key:
        sql += " AND access_key = ?"
        args.append(access_key)
    if ok is not None:
        sql += " AND ok = ?"
        args.append(1 if ok else 0)
    if hours:
        since = int(time.time()) - hours * 3600
        sql += " AND ts >= ?"
        args.append(since)
    if keyword:
        sql += " AND (path LIKE ? OR error_message LIKE ? OR error_code LIKE ?)"
        kw = f"%{keyword}%"
        args += [kw, kw, kw]
    if device_id:
        sql += " AND device_id = ?"
        args.append(device_id)
    # 总数
    count_sql = f"SELECT COUNT(*) AS c FROM ({sql})"
    sql += " ORDER BY ts DESC, id DESC LIMIT ? OFFSET ?"
    args += [limit, offset]
    with get_conn() as conn:
        total = int(conn.execute(count_sql, args[:-2]).fetchone()["c"])
        rows = conn.execute(sql, args).fetchall()
    items = [_row_to_summary(r) for r in rows]
    return {"items": items, "count": len(items), "total": total, "limit": limit, "offset": offset}


def get_record(rid: int) -> Optional[dict]:
    """详情查询：返回完整记录（含 body / headers）。"""
    with get_conn() as conn:
        row = conn.execute(
            "SELECT * FROM request_log WHERE id = ?",
            (rid,),
        ).fetchone()
    if not row:
        return None
    return _row_to_detail(row)


def clear_all() -> int:
    """清空全部请求日志。返回删除条数。"""
    with get_conn() as conn:
        cur = conn.execute("DELETE FROM request_log")
        return cur.rowcount


def clear_device_records(device_id: str) -> int:
    """清空某设备的全部请求日志。返回删除条数。"""
    if not device_id:
        return 0
    with get_conn() as conn:
        cur = conn.execute(
            "DELETE FROM request_log WHERE device_id = ?", (device_id,)
        )
        return cur.rowcount


def delete_record(rid: int) -> bool:
    with get_conn() as conn:
        cur = conn.execute("DELETE FROM request_log WHERE id = ?", (rid,))
        return cur.rowcount > 0


# ===== 行转换 =====

def _row_to_summary(row) -> dict:
    """摘要：不含大字段。"""
    return {
        "id": int(row["id"]),
        "ts": int(row["ts"]),
        "method": row["method"],
        "path": row["path"],
        "query": row["query"],
        "status_code": row["status_code"],
        "elapsed_ms": int(row["elapsed_ms"]) if row["elapsed_ms"] is not None else None,
        "access_key": row["access_key"],
        "client_ip": row["client_ip"],
        "user_agent": row["user_agent"],
        "provider": row["provider"],
        "model": row["model"],
        "stream": bool(row["stream"]),
        "ok": bool(row["ok"]),
        "error_code": row["error_code"],
        "error_message": row["error_message"],
        "device_id": row["device_id"],
    }


def _row_to_detail(row) -> dict:
    """详情：含全部字段。"""
    d = _row_to_summary(row)
    # 解析 JSON 字段
    for k in ("request_headers", "request_body", "response_body", "response_headers"):
        v = row[k]
        if v is None:
            d[k] = None
            continue
        # 尝试 JSON 解析
        try:
            d[k] = json.loads(v)
        except Exception:
            d[k] = v
    return d


# ===== 统计 =====

def stats(hours: int = 24, category: Optional[str] = None) -> dict:
    """请求日志统计。``category`` 语义同 :func:`list_records`。"""
    since = int(time.time()) - hours * 3600
    cat_sql, cat_args = _category_filter(category)
    base_cond = " WHERE ts >= ?"
    base_cond += cat_sql
    with get_conn() as conn:
        total = conn.execute(
            "SELECT COUNT(*) AS c FROM request_log" + base_cond,
            [since] + cat_args,
        ).fetchone()
        by_status = conn.execute(
            "SELECT CAST(status_code/100 AS INTEGER) AS bucket, COUNT(*) AS c "
            "FROM request_log" + base_cond + " GROUP BY bucket ORDER BY bucket",
            [since] + cat_args,
        ).fetchall()
        by_path = conn.execute(
            "SELECT path, COUNT(*) AS c, AVG(elapsed_ms) AS avg_ms, "
            "SUM(CASE WHEN ok=1 THEN 1 ELSE 0 END) AS ok_c "
            "FROM request_log" + base_cond + " GROUP BY path ORDER BY c DESC LIMIT 20",
            [since] + cat_args,
        ).fetchall()
        by_error = conn.execute(
            "SELECT error_code, COUNT(*) AS c FROM request_log "
            + base_cond + " AND ok = 0 AND error_code IS NOT NULL "
            "GROUP BY error_code ORDER BY c DESC LIMIT 20",
            [since] + cat_args,
        ).fetchall()
        slow = conn.execute(
            "SELECT id, ts, method, path, elapsed_ms, status_code, ok "
            "FROM request_log" + base_cond + " AND elapsed_ms IS NOT NULL "
            "ORDER BY elapsed_ms DESC LIMIT 10",
            [since] + cat_args,
        ).fetchall()
    return {
        "hours": hours,
        "total": int(total["c"]) if total else 0,
        "by_status": {str(r["bucket"]) + "xx": int(r["c"]) for r in by_status},
        "by_path": [
            {
                "path": r["path"],
                "count": int(r["c"]),
                "avg_ms": round(float(r["avg_ms"]), 1) if r["avg_ms"] else 0,
                "ok": int(r["ok_c"]),
            }
            for r in by_path
        ],
        "by_error": [{"code": r["error_code"], "count": int(r["c"])} for r in by_error],
        "slowest": [
            {
                "id": int(r["id"]),
                "ts": int(r["ts"]),
                "method": r["method"],
                "path": r["path"],
                "elapsed_ms": int(r["elapsed_ms"]),
                "status_code": r["status_code"],
                "ok": bool(r["ok"]),
            }
            for r in slow
        ],
    }


# ===== 设备视图 =====

def list_devices(
    *,
    hours: Optional[int] = None,
    limit: int = 100,
    offset: int = 0,
    keyword: Optional[str] = None,
) -> dict:
    """按 device_id 聚合：列出有记录的设备及其统计。

    返回每个设备的：总请求数、成功数、失败数、首次/最近活跃时间、
    最近使用的 provider/model/client_ip/user_agent。
    """
    cond = " WHERE device_id != ''"
    args: list = []
    if hours:
        since = int(time.time()) - hours * 3600
        cond += " AND ts >= ?"
        args.append(since)
    if keyword:
        cond += " AND device_id LIKE ?"
        args.append(f"%{keyword}%")
    # 外层 SELECT 做 count + 最近值聚合
    sql = (
        "SELECT device_id, COUNT(*) AS total, "
        "SUM(CASE WHEN ok=1 THEN 1 ELSE 0 END) AS ok_c, "
        "SUM(CASE WHEN ok=0 THEN 1 ELSE 0 END) AS fail_c, "
        "MIN(ts) AS first_seen, MAX(ts) AS last_seen "
        "FROM request_log" + cond + " GROUP BY device_id "
        "ORDER BY last_seen DESC LIMIT ? OFFSET ?"
    )
    args += [limit, offset]
    count_sql = (
        "SELECT COUNT(*) AS c FROM (SELECT 1 FROM request_log" + cond
        + " GROUP BY device_id)"
    )
    with get_conn() as conn:
        total = int(conn.execute(count_sql, args[:-2]).fetchone()["c"])
        rows = conn.execute(sql, args).fetchall()
        items = []
        for r in rows:
            did = r["device_id"]
            # 取该设备最近一条请求的 provider/model/ip/ua
            last = conn.execute(
                "SELECT provider, model, client_ip, user_agent, path, status_code, ts "
                "FROM request_log WHERE device_id = ? "
                "ORDER BY ts DESC, id DESC LIMIT 1",
                (did,),
            ).fetchone()
            items.append({
                "device_id": did,
                "total": int(r["total"]),
                "ok": int(r["ok_c"]),
                "fail": int(r["fail_c"]),
                "first_seen": int(r["first_seen"]) if r["first_seen"] else None,
                "last_seen": int(r["last_seen"]) if r["last_seen"] else None,
                "last_provider": last["provider"] if last else None,
                "last_model": last["model"] if last else None,
                "last_client_ip": last["client_ip"] if last else None,
                "last_user_agent": last["user_agent"] if last else None,
                "last_path": last["path"] if last else None,
                "last_status": last["status_code"] if last else None,
            })
    return {"items": items, "count": len(items), "total": total, "limit": limit, "offset": offset}


def list_device_records(
    *,
    device_id: str,
    limit: int = 50,
    offset: int = 0,
    hours: Optional[int] = None,
    keyword: Optional[str] = None,
    scope: Optional[str] = None,
) -> dict:
    """某设备的最近请求记录（简化视图：含 request_body 用于提取文本）。

    keyword: 关键词，在 request_body / response_body 上做 LIKE 搜索。
    scope:   筛选范围，'req' 仅请求、'resp' 仅响应、'both' 或 None 两者都搜。
    """
    did = _normalize_device_id(device_id)
    if not did:
        return {"items": [], "count": 0, "total": 0, "limit": limit, "offset": offset}
    sql = (
        "SELECT id, ts, method, path, query, status_code, elapsed_ms, "
        "access_key, client_ip, provider, model, stream, ok, "
        "error_code, error_message, request_body, response_body "
        "FROM request_log WHERE device_id = ?"
    )
    args: list = [did]
    if hours:
        since = int(time.time()) - hours * 3600
        sql += " AND ts >= ?"
        args.append(since)
    # 文本筛选
    kw = (keyword or "").strip()
    if kw:
        s = (scope or "both").strip().lower()
        like = f"%{kw}%"
        if s == "req":
            sql += " AND request_body LIKE ?"
            args.append(like)
        elif s == "resp":
            sql += " AND response_body LIKE ?"
            args.append(like)
        else:
            # both / 默认：任一命中即可
            sql += " AND (request_body LIKE ? OR response_body LIKE ?)"
            args += [like, like]
    count_sql = f"SELECT COUNT(*) AS c FROM ({sql})"
    sql += " ORDER BY ts DESC, id DESC LIMIT ? OFFSET ?"
    args += [limit, offset]
    with get_conn() as conn:
        total = int(conn.execute(count_sql, args[:-2]).fetchone()["c"])
        rows = conn.execute(sql, args).fetchall()
    items = [_row_to_device_record(r) for r in rows]
    return {"items": items, "count": len(items), "total": total, "limit": limit, "offset": offset}


def _row_to_device_record(row) -> dict:
    """设备视图专用记录：含请求/响应摘要 + 完整文本 + 原始 body（供下载）。"""
    # request_body 是 JSON 字符串（已在中间件层脱敏 json.dumps）
    rb_raw = row["request_body"]
    rb_parsed = None
    if rb_raw:
        try:
            rb_parsed = json.loads(rb_raw)
        except Exception:
            rb_parsed = rb_raw
    text_summary = _extract_text_summary(rb_parsed)
    text_full = _extract_text_full(rb_parsed)
    # response_body：流式取摘要+完整，非流式取完整
    resp_raw = row["response_body"]
    resp_parsed = None
    if resp_raw:
        try:
            resp_parsed = json.loads(resp_raw)
        except Exception:
            resp_parsed = resp_raw
    resp_summary = _extract_response_summary(resp_parsed)
    resp_full = _extract_response_full(resp_parsed)
    return {
        "id": int(row["id"]),
        "ts": int(row["ts"]),
        "method": row["method"],
        "path": row["path"],
        "status_code": row["status_code"],
        "elapsed_ms": int(row["elapsed_ms"]) if row["elapsed_ms"] is not None else None,
        "access_key": row["access_key"],
        "client_ip": row["client_ip"],
        "provider": row["provider"],
        "model": row["model"],
        "stream": bool(row["stream"]),
        "ok": bool(row["ok"]),
        "error_code": row["error_code"],
        "error_message": row["error_message"],
        "text_summary": text_summary,
        "text_full": text_full,
        "response_summary": resp_summary,
        "response_full": resp_full,
        "request_raw": rb_raw or "",
        "response_raw": resp_raw or "",
    }


def _extract_text_summary(body: Any) -> str:
    """从请求体提取用户输入文本（短摘要，用于卡片预览，≤200 字符）。"""
    full = _extract_text_full(body)
    if len(full) <= 200:
        return full
    return full[:200] + " …"


def _extract_text_full(body: Any) -> str:
    """从请求体提取完整用户输入文本（不截断）。

    兼容 OpenAI messages 形态、XTC input 简化形态、多模态 content 数组。
    """
    if body is None:
        return ""
    if isinstance(body, str):
        return body
    if not isinstance(body, dict):
        return str(body)
    # 1) 简化形态：input 字段
    inp = body.get("input")
    if isinstance(inp, str) and inp.strip():
        return inp.strip()
    # 2) messages 形态
    msgs = body.get("messages")
    if isinstance(msgs, list) and msgs:
        parts: list[str] = []
        for m in msgs:
            if not isinstance(m, dict):
                continue
            role = str(m.get("role") or "").strip()
            content = m.get("content")
            text = ""
            if isinstance(content, str):
                text = content
            elif isinstance(content, list):
                # OpenAI 多模态：[{type:text,text:...},{type:image_url,...}]
                chunks = []
                for c in content:
                    if isinstance(c, dict):
                        if c.get("type") == "text" and c.get("text"):
                            chunks.append(str(c["text"]))
                        elif c.get("type") == "image_url":
                            chunks.append("[图片]")
                text = " ".join(chunks)
            if text:
                prefix = f"[{role}] " if role else ""
                parts.append(prefix + text.strip())
        return "\n".join(parts)
    return ""


def _extract_response_summary(resp: Any) -> str:
    """从响应体提取简要文本（≤200 字符，用于卡片预览）。"""
    full = _extract_response_full(resp)
    if len(full) <= 200:
        return full
    return full[:200] + " …"


def _extract_response_full(resp: Any) -> str:
    """从响应体提取完整文本（不截断）。

    - 流式：合并所有 chunk 的 delta 文本
    - 非流式：取 choices[0].message.content
    - 错误：拼错误信息
    """
    if resp is None:
        return ""
    if isinstance(resp, str):
        return resp
    if isinstance(resp, dict):
        # 流式摘要
        if resp.get("type") == "streaming":
            chunks = resp.get("sampled_chunks") or []
            total = resp.get("total_chunks")
            bytes_ = resp.get("total_bytes")
            # 合并所有 chunk 的 delta 文本
            text_parts: list[str] = []
            for c in chunks:
                if not isinstance(c, str):
                    continue
                # SSE data: {...}
                for line in c.splitlines():
                    line = line.strip()
                    if line.startswith("data:"):
                        payload = line[5:].strip()
                        if not payload or payload == "[DONE]":
                            continue
                        try:
                            obj = json.loads(payload)
                        except Exception:
                            continue
                        txt = _extract_delta_text(obj)
                        if txt:
                            text_parts.append(txt)
            joined = "".join(text_parts)
            header = f"[流式 {total} chunks / {bytes_} B]"
            return (header + "\n" + joined) if joined else header
        # OpenAI 非流式响应
        choices = resp.get("choices")
        if isinstance(choices, list) and choices:
            first = choices[0]
            if isinstance(first, dict):
                msg = first.get("message") or {}
                content = msg.get("content") if isinstance(msg, dict) else None
                if isinstance(content, str) and content:
                    return content
                if isinstance(content, list):
                    texts = []
                    for c in content:
                        if isinstance(c, dict) and c.get("type") == "text":
                            texts.append(str(c.get("text") or ""))
                    return "".join(texts)
        # 错误响应
        if resp.get("error"):
            err = resp["error"]
            if isinstance(err, dict):
                return f"[错误] {err.get('code','')} {err.get('message','')}"
            return f"[错误] {err}"
    return str(resp)


def _extract_delta_text(obj: Any) -> str:
    """从单个 OpenAI 流式 chunk 中提取增量文本。"""
    if not isinstance(obj, dict):
        return ""
    choices = obj.get("choices")
    if not isinstance(choices, list) or not choices:
        return ""
    first = choices[0]
    if not isinstance(first, dict):
        return ""
    delta = first.get("delta") or {}
    if isinstance(delta, dict):
        content = delta.get("content")
        if isinstance(content, str):
            return content
    return ""