qihuaiflow/backend/python/app/api/chat_media.py

# Auto-extracted from chat.py during Phase 1 refactor (Strangler Fig).
# Source of truth: scripts/phase1_extract.py. Do not hand-merge back.

from fastapi import APIRouter, Depends, HTTPException, Query, Response
from sqlalchemy.orm import Session
from sqlalchemy import and_, desc, asc, func
from pydantic import BaseModel, Field
from typing import Optional
import uuid
from datetime import datetime, timezone, timedelta
import re
import json
import asyncio
import math
import httpx
import os
import logging
from ..database import get_db, SessionLocal
from ..models import (



    User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog,



    KnowledgeBase, KnowledgeDocument, KnowledgeDocumentChunk, EvolutionSkill,



    EvolutionTrajectory, MarketApp, AppPurchase, AppInstallation,



    MCPServer,



    gen_uuid



)
from ..auth import get_current_user
from ..platform_knowledge import get_platform_knowledge
from ..mcp_client import list_mcp_tools, call_mcp_tool
from ..middleware.log_service import LogService
from ..capability_registry import resolve_auto_model
from app.api.chat_utils import (_html_escape, _proxy_for_url, cst_now, logger)
from app.core.observability import span, get_trace_id

def detect_media_intent(text: str) -> str | None:



    """根据用户输入判断是否为 图像/视频/语音 生成请求，返回对应能力或 None。







    用于 Auto 模式：用户未指定具体模型时，依据意图自动匹配平台各类能力的默认模型。



    疑问句（含「吗/呢/怎么/能否」或结尾问号）不触发，避免「你能生成视频吗」被误判。



    """



    if not text:



        return None



    t = text.lower().strip()



    # 疑问句不触发媒体生成



    if t.endswith(("？", "?")) or any(q in t for q in ("吗", "呢", "怎么", "如何", "能否", "是否可以", "可以吗", "会不会", "why", "how")):



        return None



    video_kw = ("视频", "短片", "影片", "vide", "动画视频", "一段视频", "做个视频", "生成视频",



                "视频生成", "来段视频", "拍个视频", "做段视频", "视频剪辑", "做视频",



                "生成一个视频", "制作一个视频", "帮我做视频", "制作视频",



                "生成短片", "做短片", "制作短片", "一段短片")



    image_kw = ("图片", "图像", "插画", "海报", "照片", "头像", "壁纸", "漫画", "表情包", "logo", "图标",



                "imag", "画一张", "画张", "画一只", "画只", "画个", "画幅",



                "来张", "来一张", "来幅", "发张图", "发个图",



                "生成图片", "生成一张", "生成一幅", "生成图", "图像生成",



                "做一张", "做张图", "做张", "出张图", "出一张图",



                "帮我画", "给我画", "请画", "绘制", "作画", "ai绘画", "ai画图",



                "画个图", "画张图", "画一幅图", "一张图", "一幅画", "生成一幅",



                "画画", "画一下", "画出来", "生成个图", "生成一张图", "生成个图片",



                "做图片", "做个图", "制图", "整张图", "整张图出来", "搞张图",



                "帮我生成图片", "帮我生成图像", "给我生成图片", "给我生成图像",



                "生成一张图片", "生成一个图片", "制作一张图", "制作一张图片",



                "来一张图片", "画一张图片", "画一幅", "画一副")



    audio_kw = ("语音", "配音", "朗读", "播报", "念一段", "有声", "tts", "音频", "生成语音",



                "读一下", "念一下", "语音合成")



    if any(k in t for k in video_kw):



        return "video"



    if any(k in t for k in image_kw):



        return "image"



    if any(k in t for k in audio_kw):



        return "audio"



    return None

def detect_ambiguous_creative_intent(text: str):



    """检测「模糊创作意图」：用户使用了创作类动词(制作/创作/做/生成/画/写/设计…)，



    但未明确产出形式(文案/图片/视频/其他)。此时应与用户确认方向，而不是直接调用大模型



    （某些模型会对这类开放式创作请求返回 "high risk" 拒答）。







    返回 None 表示无需确认；否则返回 {"options": [...]}。



    """



    if not text:



        return None



    t = text.strip()



    tl = t.lower()



    # 已有明确媒体意图(图像/视频/语音)则无需确认



    if detect_media_intent(t):



        return None



    # 疑问句不触发



    if tl.endswith(("？", "?")) or any(q in tl for q in ("吗", "呢", "怎么", "如何", "能否", "是否可以", "可以吗", "会不会", "why", "how", "什么", "哪些")):



        return None



    # 已明确给出产出形式(文字类/其他类)则无需确认



    explicit_medium = (



        "文案", "文章", "诗歌", "诗", "小说", "故事", "代码", "程序", "脚本", "总结", "翻译",



        "邮件", "报告", "段落", "正文", "文字", "写作", "ppt", "演示", "思维导图", "表格",



        "图表", "数据",



    )



    if any(m in tl for m in explicit_medium):



        return None



    # 创作类动词



    creative_verbs = (



        "制作", "创作", "做个", "做一个", "搞一个", "产出", "构思", "设想",



        "打造", "塑造", "生成", "帮我做", "帮我生成", "帮我画", "帮我写", "请制作",



        "请创作", "请生成", "请画", "请写", "来一个", "来一张", "来一幅", "做一张",



        "做张", "写一首", "写一段", "写个",



    )



    if not any(v in t for v in creative_verbs):



        return None



    # 模糊：有创作动词但无明确产出形式 → 提供方向选择



    options = [



        {"key": "text", "label": "文案", "desc": "文章 / 故事 / 诗歌等文字内容", "medium": "文案"},



        {"key": "image", "label": "图片", "desc": "插画 / 海报 / 照片等图像", "medium": "图片"},



        {"key": "video", "label": "视频", "desc": "短片 / 动画等视频", "medium": "视频"},



        {"key": "other", "label": "其他", "desc": "语音 / 代码 / 其他形式", "medium": "其他"},



    ]



    return {"options": options}

async def _optimize_media_prompt(raw_prompt: str, media_type: str, db=None) -> str:



    """使用平台文本模型对用户原始输入做「媒体提示词增强优化」。







    AGNES 图像/视频模型对 prompt 质量高度敏感：



    - 用户在多轮对话中说"帮我生成一张刚才讨论的产品图"包含大量会话噪音



    - AGNES 需要的是**详细的、描述性的、英文优先或中英混合**的视觉提示词



    - 缺乏构图/风格/光影/质量等指导词时，生成结果与预期差异巨大







    本函数把用户原文交给文本 LLM 精炼为符合 AGNES 预期的优质 prompt。



    失败时回退到原始输入(不做额外引导，保证必达)。



    """



    if not raw_prompt or len(raw_prompt.strip()) < 2:



        return raw_prompt



    # 如果原始输入已经很长(>200 字)且英文为主，可能本身就是一个不错的 prompt，直接返回



    stripped = raw_prompt.strip()



    ascii_chars = sum(1 for c in stripped if ord(c) < 128)



    if len(stripped) > 200 and ascii_chars / max(len(stripped), 1) > 0.6:



        return stripped



    # 用文本 LLM 做翻译增强



    try:



        from ..capability_registry import resolve_auto_model as _resolve_auto



        from ..models import ModelProvider as MP



        from ..database import SessionLocal as SL



        import httpx as _httpx







        # 独立数据库会话，避免与主流程竞争



        _db = db



        _should_close = False



        if _db is None:



            _db = SL()



            _should_close = True



        try:



            text_model = _resolve_auto("text", _db)



            if not text_model:



                return stripped



            provider = _db.query(MP).filter(MP.id == text_model.provider_id).first()



            api_key = text_model.api_key or (provider.api_key if provider else None)



            api_url = text_model.endpoint



            if not api_url and provider and provider.api_base_url:



                api_url = f"{provider.api_base_url.rstrip('/')}/v1/chat/completions"



            if not (api_key and api_url):



                return stripped







            prompt_zh = (



                "你是一个专业的AI图像/视频提示词工程师。请将以下用户需求提炼成"



                f"一段高质量的{'图像' if media_type == 'image' else '视频'}生成"



                "提示词。要求：\n"



                "1. 用英文书写（AGNES模型对英文响应最佳），核心关键词可保留中文；\n"



                "2. 添加构图/风格/光影/色彩/质量描述（如 highly detailed, 8K, "



                "professional lighting, masterpiece, cinematic等）；\n"



                "3. 去掉与视觉无关的对话噪音；\n"



                "4. 只输出最终prompt文本，不要任何解释、不要引号包裹、不要\"Prompt:\"前缀。\n\n"



                f"用户需求：{stripped}"



            )



            messages = [



                {"role": "system", "content": "You are a professional prompt engineer. Output ONLY the final prompt text, no explanations."},



                {"role": "user", "content": prompt_zh},



            ]



            # max_tokens 调大到 1024：mimo 等推理模型会把 token 消耗在思考链(reasoning_content)



            # 上，512 时 content 常被截断为空；并关闭推理(enable_thinking=False)，让模型直接输出提示词。



            # 对不支持该参数的模型至多优化失败 → 下方 except 兜底退化为原始提示词，不阻断生成。



            body = {"model": text_model.name, "messages": messages, "max_tokens": 1024,



                    "temperature": 0.7, "enable_thinking": False}



            async with _httpx.AsyncClient(



                timeout=30.0, proxy=_proxy_for_url(api_url), trust_env=False



            ) as client:



                resp = await client.post(



                    api_url,



                    headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},



                    json=body,



                )



                if resp.status_code == 200:



                    data = resp.json()



                    opt = data.get("choices", [{}])[0].get("message", {}).get("content", "").strip()



                    if opt and len(opt) > 5:



                        return opt



        finally:



            if _should_close:



                _db.close()



    except Exception:



        pass



    return stripped

def _friendly_media_err(kind: str, raw: str) -> str:



    """把 AGNES 原始错误转译为对用户友好的中文。







    AGNES 常见错误：429 限频(每分钟1次视频) / Server disconnected(连接重置) /



    publish_video_queue_failed + context deadline exceeded(队列超时)。这些都不是



    平台代码问题，统一给出可操作的提示，不把原始 HTTP/JSON 甩给用户。



    """



    r = (raw or "").lower()



    if "429" in raw or "rate limit" in r:



        return (f"⚠️ {kind}生成请求过于频繁（AGNES 限制每分钟 1 次），"



                f"请稍等 1 分钟后再试。")



    if "server disconnected" in r or "连接" in raw or "reset" in r:



        return f"⚠️ {kind}生成服务暂时连接失败，请稍后重试。"



    if "publish_video" in r or "queue" in r or "deadline" in r or "context" in r:



        return f"⚠️ {kind}生成队列繁忙（AGNES 处理超时），请稍后重试。"



    return f"⚠️ {kind}生成失败（请稍后重试）：{(raw or '')[:120]}"

async def _agnes_post_json(api_url: str, headers: dict, payload: dict, timeout) -> tuple:



    """直连 AGNES 发 POST，对 5xx / 连接中断做指数退避重试。







    AGNES 视频/图像队列极不稳定：经常返回 500 publish_video_queue_failed /



    context deadline exceeded，或连接被重置(Server disconnected)。这类错误是 AGNES



    侧瞬时抖动，重试通常立刻成功。4xx（参数错误）不重试，直接返回。



    返回 (resp, err)：成功(resp.status_code<500) 时 err=None；耗尽重试时 resp=None。



    """



    last_err = None



    for attempt in range(3):



        try:



            async with httpx.AsyncClient(



                timeout=timeout, proxy=_proxy_for_url(api_url), trust_env=False



            ) as client:



                resp = await client.post(api_url, headers=headers, json=payload)



            if resp.status_code < 500:



                return resp, None



            last_err = f"HTTP {resp.status_code}: {resp.text[:200]}"



        except Exception as e:



            last_err = str(e)[:200]



        if attempt < 2:



            await asyncio.sleep(2 * (attempt + 1))



    return None, last_err


def _media_api_url(ai_model, provider, media_kind: str) -> str:
    """推导图像/视频生成地址，避免把 chat/completions 端点误用作媒体端点。

    - 若模型 endpoint 已是对应的媒体端点(含 /images/generations 或 /videos)，直接使用；
    - 否则从 chat endpoint 反推 base，再拼 /v1/images|videos/generations；
    - 若 endpoint 为空，则回退供应商 api_base_url + 媒体路径。
    """
    suffix = "/v1/images/generations" if media_kind == "image" else "/v1/videos"
    chat_suffixes = ["/v1/chat/completions", "/chat/completions"]
    ep = (ai_model.endpoint or "").strip()
    if ep:
        if suffix in ep:
            return ep
        base = ep
        for cs in chat_suffixes:
            if base.endswith(cs):
                base = base[: -len(cs)]
                break
        else:
            if "/v1" in base:
                base = base.rsplit("/v1", 1)[0]
        return f"{base.rstrip('/')}{suffix}"
    if provider and provider.api_base_url:
        return f"{provider.api_base_url.rstrip('/')}{suffix}"
    return ""

def _build_media_route(db, capability: str, preferred_id: str = None):
    """构造媒体生成模型路由顺序：
    1) 按能力默认模型(后台"能力默认模型"或全局默认) 作为第一顺位
    2) 调用方显式指定的模型(若不同于默认) 次位
    3) 其余启用且具备该能力的模型，按 is_default 降序、created_at 升序 排列
    去重后返回 AIModel 列表。
    """
    from ..models import AIModel as _AM

    selected = []
    seen = set()

    def _add(m):
        if m and m.id not in seen:
            selected.append(m)
            seen.add(m.id)

    default_model = resolve_auto_model(capability, db)
    if default_model:
        _add(default_model)

    if preferred_id:
        pref = db.query(_AM).filter(_AM.id == preferred_id, _AM.is_enabled == True).first()
        _add(pref)

    others = db.query(_AM).filter(
        _AM.is_enabled == True,
        _AM.capabilities.op("JSON_CONTAINS")(_AM.capabilities, f'"{capability}"'),
    ).order_by(_AM.is_default.desc(), _AM.created_at.asc()).all()
    for m in others:
        _add(m)

    return selected


async def _run_media_route(db, route, prompt, user_id, media_kind):
    """依次尝试路由中的模型生成媒体；首个成功(非 ⚠️ 开头)即返回，否则尝试下一个。"""
    last_err = None
    for idx, m in enumerate(route, 1):
        logger.info(f"[_run_media_route] ({idx}/{len(route)}) 尝试模型 {m.name} ({m.id})")
        try:
            res = await _generate_media(db, m, prompt, user_id)
        except Exception as e:
            last_err = f"模型 {m.name} 异常: {str(e)[:160]}"
            logger.warning(f"[_run_media_route] 模型 {m.name} 异常: {e}")
            continue
        if res and not res.startswith("⚠️"):
            return res
        last_err = res
        logger.warning(f"[_run_media_route] 模型 {m.name} 失败: {str(res)[:160]}")
    return last_err or "⚠️ 所有模型均生成失败"


async def _generate_media(db, ai_model, prompt: str, user_id: str = None) -> str:



    """图像/视频生成模型统一入口：按 capabilities 调用 AGNES 等媒体生成 API。



    返回可直接在 AI 聊天中渲染的富文本(含 <img>/<video> 标签)。"""



    provider = db.query(ModelProvider).filter(ModelProvider.id == ai_model.provider_id).first()



    api_key = ai_model.api_key or (provider.api_key if provider else None)



    if not api_key:



        return "⚠️ 该媒体模型缺少 API Key，请在后台「大模型库」中配置。"







    caps = [c.lower() for c in (ai_model.capabilities or [])]



    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}



    timeout = ai_model.timeout or 120
    # 单次媒体调用超时封顶：避免单个供应商长时间挂起拖垮整条生成链路；
    # 多模型 failover 由 _generate_media_task 外层循环保证
    call_timeout = min(timeout, 300 if "video" in caps else 180)







    # ============ 图像生成 ============



    if "image" in caps:



        api_url = _media_api_url(ai_model, provider, "image")



        if not api_url:



            return "⚠️ 该图像模型未配置 Endpoint。"



        # 先优化 prompt（用户原始输入含多轮对话噪音 → 高质量图像生成提示词）



        # 优化失败绝不中断生成，退化为原始提示词



        try:



            optimized_prompt = await _optimize_media_prompt(prompt, "image", db)



        except Exception:



            optimized_prompt = prompt



        payload = {



            "model": ai_model.name,



            "prompt": optimized_prompt,



            "size": "1024x1024",



        }



        async with span("media.agnes_call"):
            resp, err = await _agnes_post_json(api_url, headers, payload, call_timeout)



        if resp is None:



            # 3 次重试后仍失败 → AGNES 侧持续抖动，提示稍后重试而非甩原始错误



            return _friendly_media_err("图像", err)



        if resp.status_code != 200:



            # AGNES 内容策略拒绝：用原始(未优化)提示词重试一次，避免优化后提示词触发误杀



            if resp.status_code == 400 and "content_policy" in (resp.text or "").lower():



                payload["prompt"] = prompt



                resp2, _ = await _agnes_post_json(api_url, headers, payload, call_timeout)



                if resp2 is not None and resp2.status_code == 200:



                    resp = resp2



                else:



                    return "⚠️ 图像生成被内容策略拒绝，请换一个更中性的描述后重试。"



            else:



                return _friendly_media_err("图像", f"HTTP {resp.status_code}: {resp.text[:200]}")



        try:



            data = resp.json()



        except Exception:



            return f"⚠️ 图像生成失败：响应解析错误 {resp.text[:200]}"



        items = data.get("data") or []



        if not items:



            return f"⚠️ 图像生成失败：响应无图片数据 {str(data)[:200]}"



        item = items[0]



        url = item.get("url") or item.get("b64_json")



        if not url:



            return f"⚠️ 图像生成失败：未返回图片地址 {str(data)[:200]}"



        if not url.startswith("http"):



            url = f"data:image/png;base64,{url}"



        return (f'<img class="md-img" src="{url}" alt="生成图片"/>\n\n'



                f'**提示词：** {_html_escape(prompt)}')







    # ============ 视频生成 (异步任务 + 轮询) ============



    if "video" in caps:



        api_url = _media_api_url(ai_model, provider, "video")



        if not api_url:



            return "⚠️ 该视频模型未配置 Endpoint。"



        # 优化 prompt（用户输入含多轮对话噪音 → 高质量视频提示词）；失败退化为原始提示词



        try:



            optimized_prompt = await _optimize_media_prompt(prompt, "video", db)



        except Exception:



            optimized_prompt = prompt



        base = api_url.split("/v1")[0] if "/v1" in api_url else api_url



        payload = {



            "model": ai_model.name,



            "prompt": optimized_prompt,



            "height": 768,



            "width": 1152,



            "num_frames": 121,



            "frame_rate": 24,



        }



        async with span("media.agnes_call"):
            resp, err = await _agnes_post_json(api_url, headers, payload, call_timeout)



        if resp is None:



            return _friendly_media_err("视频", err)



        if resp.status_code != 200:



            # AGNES 内容策略拒绝：用原始(未优化)提示词重试一次，避免优化后提示词触发误杀



            if resp.status_code == 400 and "content_policy" in (resp.text or "").lower():



                payload["prompt"] = prompt



                resp2, _ = await _agnes_post_json(api_url, headers, payload, call_timeout)



                if resp2 is not None and resp2.status_code == 200:



                    resp = resp2



                else:



                    return "⚠️ 视频生成被内容策略拒绝，请换一个更中性的描述后重试。"



            else:



                return _friendly_media_err("视频", f"HTTP {resp.status_code}: {resp.text[:200]}")



        try:



            data = resp.json()



        except Exception:



            return f"⚠️ 视频生成失败：响应解析错误 {resp.text[:200]}"



        video_id = data.get("video_id") or data.get("task_id") or data.get("id")



        if not video_id:



            return f"⚠️ 视频生成失败：未返回任务ID {str(data)[:200]}"



        # 轮询结果(最多约 5 分钟)。AGNES 视频状态端点为 GET {base}/v1/videos/{video_id}



        # 该端点顶层返回 status(queued/in_progress/completed/failed)，成片地址在 metadata.url



        result_url = None



        err_msg = None



        for _ in range(60):



            await asyncio.sleep(5)



            try:



                async with httpx.AsyncClient(



                    timeout=timeout, proxy=_proxy_for_url(f"{base}/v1/videos/{video_id}"), trust_env=False



                ) as client:



                    r2 = await client.get(f"{base}/v1/videos/{video_id}", headers=headers)



            except Exception:



                continue



            if r2.status_code != 200:



                continue



            rd = r2.json()



            st = rd.get("status")



            if st == "completed":



                result_url = (rd.get("metadata") or {}).get("url") or rd.get("url")



                break



            if st == "failed":



                err_msg = rd.get("error") or (rd.get("metadata") or {}).get("error")



                break



        if result_url:



            return (f'<video class="md-video" controls src="{result_url}"></video>\n\n'



                    f'**提示词：** {_html_escape(prompt)}')



        return f"⚠️ 视频生成失败：{err_msg or '生成超时，请稍后在对话中重试'}"







    return "⚠️ 未知的媒体模型类型。"

async def _generate_media_task(ai_msg_id: str, ai_model_id: str, prompt: str, user_id: str):



    """后台异步执行图像/视频生成，完成后回写消息内容。







    目的是把最长 5 分钟的 AGNES 视频轮询从 HTTP 请求中剥离，避免请求被客户端/



    代理超时断开(表现为前端「发送失败：请求失败」)。send_message 立即返回占位



    内容(generating 标记)，本任务独立在事件循环中轮询并在完成后 merge 回写消息。



    使用独立的 SessionLocal 会话，因为请求级 db 在响应返回后即被 FastAPI 关闭。



    """



    db = SessionLocal()



    try:



        ai_model = db.query(AIModel).filter(AIModel.id == ai_model_id).first()



        if not ai_model:



            logger.warning(f"[_generate_media_task] 模型不存在: {ai_model_id}")



            return







        logger.info(f"[_generate_media_task] 启动 media 生成 msg={ai_msg_id} model={ai_model_id} prompt={prompt[:60]!r}")
        caps = [c.lower() for c in (ai_model.capabilities or [])]
        is_video = "video" in caps
        media_kind = "video" if is_video else "image"

        # 构造模型路由：默认模型第一顺位，之后按路由策略逐一尝试
        route = _build_media_route(db, media_kind, ai_model_id)
        if not route:
            result = f"⚠️ 未找到可用的{'视频' if is_video else '图像'}模型，请在后台「大模型库」中启用具备该能力的模型。"
        else:
            _media_name = "视频" if is_video else "图像"
            per_call = 300 if is_video else 180
            route_timeout = min(per_call * len(route) + 30, 900 if is_video else 600)
            try:
                result = await asyncio.wait_for(
                    _run_media_route(db, route, prompt, user_id, media_kind),
                    timeout=route_timeout,
                )
            except asyncio.TimeoutError:
                logger.warning(f"[_generate_media_task] media 路由全部超时 msg={ai_msg_id} is_video={is_video} timeout={route_timeout}s")
                result = f"⚠️ {_media_name}生成超时：已依次尝试 {len(route)} 个模型仍失败，请稍后重试。"
            except Exception as e:
                logger.warning(f"[_generate_media_task] media 路由异常 msg={ai_msg_id}: {e}", exc_info=True)
                result = f"⚠️ {_media_name}生成失败：{str(e)[:200]}"







        # 回写消息内容(占位 → 最终富文本/错误)



        msg = db.query(Message).filter(Message.id == ai_msg_id).first()



        if msg:



            msg.content = result



            msg.token_used = len(result) // 2



            db.commit()



            logger.info(f"[_generate_media_task] 消息 {ai_msg_id} 已回写内容")



        # 媒体 Token 记账：把图片/视频真实消耗写入 TokenUsageLog，供驾驶舱统计。



        # 主流程对媒体请求只写了占位文本的几 token，此处补写真实媒体消耗（模型名正确、数值合理）。



        _write_media_token_log(db, ai_model, user_id, msg.conversation_id if msg else None, prompt, trace_id=get_trace_id())



    except Exception as e:
        logger.warning(f"[_generate_media_task] 后台生成异常: {e}", exc_info=True)
        # 兜底回写：任何意外异常都不能让消息永久停在「⏳ 生成中…」
        try:
            db.rollback()
            _err = db.query(Message).filter(Message.id == ai_msg_id).first()
            if _err and str(_err.content or "").startswith("⏳"):
                _err.content = "⚠️ 媒体生成失败，请稍后重试"
                db.commit()
        except Exception:
            pass


    finally:



        try:



            db.close()



        except Exception:



            pass

def _write_media_token_log(db, ai_model, user_id: str, conversation_id: str, prompt: str, trace_id: str = None):



    """媒体生成完成后补写 TokenUsageLog（图片/视频），修正此前只记占位文本的偏差。







    媒体 API 不返回 token 用量，采用与 IM Bot 路径一致的估算：



    视频基线 1200、图片基线 400，再加 prompt 长度的 1/4 作为提示词开销。



    """



    if not user_id or not ai_model:



        return



    try:



        is_video = "video" in (ai_model.capabilities or [])



        base = 1200 if is_video else 400



        est = base + (len(prompt or "") // 4)



        db.add(TokenUsageLog(



            id=gen_uuid(),



            user_id=user_id,



            model_id=ai_model.id,



            model_name=ai_model.name,



            conversation_id=conversation_id,



            prompt_tokens=est,



            completion_tokens=0,



            total_tokens=est,



            created_at=cst_now(),
            trace_id=trace_id,



        ))



        db.commit()



    except Exception as e:



        logger.warning(f"[_write_media_token_log] 写入失败(忽略): {e}")



        try:



            db.rollback()



        except Exception:



            pass