chat-ai-api/app/routers/moderation.py

import os
import tempfile
from urllib.parse import urlparse

import requests
from fastapi import APIRouter, HTTPException

from ..core.ark_client import get_ark_client
from ..config.config import Config
from ..schemas.chat import ModerationRequest, ModerationResponse
from ..db.mongo import save_chat_log

router = APIRouter()

# 待审核文本最大长度限制（超出则拒绝，避免超长文本拖慢模型响应）
MAX_CONTENT_LENGTH = 20000
# 远程媒体下载大小限制（火山方舟视频上限 50MB）
MAX_VIDEO_SIZE = 50 * 1024 * 1024
MAX_IMAGE_SIZE = 20 * 1024 * 1024
# 下载超时（连接超时, 读取超时）
_DOWNLOAD_TIMEOUT = (10, 60)

# 内容审核系统提示词：约束模型仅输出 true / false，降低解析歧义并提升响应速度
_MODERATION_PROMPT = """你是严格的内容安全审核员。判断用户提供的内容（文本/图片/视频画面）是否违规。
违规类别（命中任一即违规）：1.涉黄、低俗、性暗示、裸露；2.暴力、恐怖、血腥、极端主义；3.涉政敏感、反动、危害国家安全；4.违法犯罪、赌博、诈骗、毒品、辱骂、封建迷信；5.广告营销、引流、联系方式、二维码。
图片/视频结合画面主体、文字、场景综合判断；仅依据所给内容判断，不做联想或补全。
只输出一个单词：true（违规）或 false（合规），禁止输出任何其他字符。"""


def _parse_moderation_result(text: str) -> bool:
    """将模型输出解析为布尔值：True=违规，False=合规"""
    cleaned = text.strip().lower()
    if cleaned.startswith("true"):
        return True
    if cleaned.startswith("false"):
        return False
    # 容错：模型可能附带少量多余字符，取首次出现的关键词
    has_true = "true" in cleaned
    has_false = "false" in cleaned
    if has_true and not has_false:
        return True
    if has_false and not has_true:
        return False
    # 无法明确解析时抛错，交由调用方重试，避免误判放行违规内容
    raise ValueError(f"无法解析审核结果: {text[:50]}")


def _try_early_decide(accumulated: str) -> bool | None:
    """流式场景下的提前判定：true/false 首字母不同（t vs f），
    读到第一个非空字符即可判定，无需等模型吐完整个单词。
    返回 None 表示暂无法判定（如首字符是空白/不明字符）。"""
    stripped = accumulated.strip().lower()
    if not stripped:
        return None
    first = stripped[0]
    if first == "t":
        return True
    if first == "f":
        return False
    return None


def _resolve_ext(url: str, content: bytes, default: str) -> str:
    """推断文件扩展名：优先取 URL 路径后缀，其次按文件魔数嗅探，最后用默认值。
    Files API 依据扩展名识别媒体类型，因此需给出正确后缀。"""
    ext = os.path.splitext(urlparse(url).path)[1].lower()
    known = (".mp4", ".mov", ".webm", ".mkv", ".avi", ".flv",
             ".jpg", ".jpeg", ".png", ".webp", ".gif", ".bmp")
    if ext in known:
        return ext
    # 文件魔数嗅探（URL 无有效后缀时，如 COS 的 ?imageMogr2 处理链接）
    if content[:3] == b"\xff\xd8\xff":
        return ".jpg"
    if content[:8] == b"\x89PNG\r\n\x1a\n":
        return ".png"
    if content[:4] == b"RIFF" and content[8:12] == b"WEBP":
        return ".webp"
    if content[:6] in (b"GIF87a", b"GIF89a"):
        return ".gif"
    if content[4:8] == b"ftyp":
        return ".mp4"
    return default


def _upload_remote_to_ark(client, url: str, kind: str) -> str:
    """下载远程 URL 并上传到火山方舟 Files API，返回 file_id。

    直接把 COS 等第三方 URL 传给模型时，若源站返回的 Content-Type 为
    application/octet-stream，方舟会因无法识别媒体类型而报 400。通过
    下载 + Files API 上传（按扩展名识别类型）可彻底绕开该限制。

    kind: 'video' | 'image'，用于大小限制与默认扩展名。
    """
    max_size = MAX_VIDEO_SIZE if kind == "video" else MAX_IMAGE_SIZE
    try:
        resp = requests.get(url, timeout=_DOWNLOAD_TIMEOUT, stream=True)
        resp.raise_for_status()
        content = b""
        for chunk in resp.iter_content(chunk_size=1 << 20):
            if not chunk:
                continue
            content += chunk
            if len(content) > max_size:
                raise HTTPException(
                    status_code=400,
                    detail=f"{kind}文件超过大小限制({max_size // 1024 // 1024}MB)",
                )
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"无法下载{kind}资源: {str(e)}")

    if not content:
        raise HTTPException(status_code=400, detail=f"{kind}资源内容为空")

    ext = _resolve_ext(url, content, ".mp4" if kind == "video" else ".jpg")
    filename = f"moderation_{kind}{ext}"
    tmp_path = None
    try:
        with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as tmp:
            tmp.write(content)
            tmp_path = tmp.name
        with open(tmp_path, "rb") as f:
            uploaded = client.files.create(file=(filename, f), purpose="user_data")
        return uploaded.id
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"上传{kind}到审核服务失败: {str(e)}")
    finally:
        if tmp_path and os.path.exists(tmp_path):
            os.unlink(tmp_path)


def _resolve_type(request: ModerationRequest) -> str:
    """确定审核类型：优先使用显式 type，否则根据已传参数自动推断"""
    if request.type:
        t = request.type.strip().lower()
        if t not in ("text", "image", "video"):
            raise HTTPException(status_code=400, detail=f"不支持的审核类型: {request.type}")
        return t
    if request.video_url:
        return "video"
    if request.image_url:
        return "image"
    if request.content:
        return "text"
    raise HTTPException(status_code=400, detail="请提供待审核的内容（content / image_url / video_url）")


def _build_input_content(mod_type: str, request: ModerationRequest, client) -> str | list:
    """根据审核类型构造 Chat Completions API 的 content 参数，并做参数校验。
    文本审核返回字符串，图片/视频审核返回多模态 content 数组。"""
    if mod_type == "text":
        content = (request.content or "").strip()
        if not content:
            raise HTTPException(status_code=400, detail="待审核文本不能为空")
        if len(content) > MAX_CONTENT_LENGTH:
            raise HTTPException(
                status_code=400,
                detail=f"待审核文本超过长度限制（{MAX_CONTENT_LENGTH} 字符），请分段提交",
            )
        return content

    if mod_type == "image":
        image_url = (request.image_url or "").strip()
        if not image_url:
            raise HTTPException(status_code=400, detail="图片审核需提供 image_url")
        return [
            {"type": "image_url", "image_url": {"url": image_url}},
            {"type": "text", "text": "请审核这张图片的画面内容是否违规。"},
        ]

    # video：对视频画面 + 封面图一起审核
    video_url = (request.video_url or "").strip()
    thumb_url = (request.thumb_url or "").strip()
    if not video_url:
        raise HTTPException(status_code=400, detail="视频审核需提供 video_url")
    if not thumb_url:
        raise HTTPException(status_code=400, detail="视频审核需提供封面图 thumb_url")
    video_file_id = _upload_remote_to_ark(client, video_url, "video")
    thumb_file_id = _upload_remote_to_ark(client, thumb_url, "image")
    return [
        {"type": "file", "file_id": video_file_id},
        {"type": "file", "file_id": thumb_file_id},
        {"type": "text", "text": "请综合审核这段视频的画面及封面图内容是否违规。"},
    ]


# 火山方舟输入/输出安全护栏拦截时返回的错误码统一含该关键字
_SENSITIVE_ERROR_KEYWORD = "SensitiveContentDetected"


def _is_sensitive_block(error: Exception) -> bool:
    """判断异常是否为方舟安全护栏的敏感内容拦截。

    当输入（文本/图片/视频）命中敏感信息时，方舟会在模型处理前直接返回
    形如 InputTextSensitiveContentDetected 的 400 错误。对审核接口而言，
    这本身即表明内容违规，应判定为 sensitive=true，而非当作服务异常报错。
    """
    code = getattr(error, "code", None)
    if code and _SENSITIVE_ERROR_KEYWORD in str(code):
        return True
    return _SENSITIVE_ERROR_KEYWORD in str(error)


def _log_target(mod_type: str, request: ModerationRequest) -> str:
    """生成审核日志的目标标识（文本截断 / 媒体 URL）"""
    if mod_type == "text":
        return (request.content or "")[:500]
    if mod_type == "image":
        return f"[image] {request.image_url}"
    return f"[video] video={request.video_url} thumb={request.thumb_url}"


@router.post("/check", response_model=ModerationResponse)
async def moderate_content(request: ModerationRequest):
    """
    公共内容审核接口（无需登录认证），供外部系统调用。
    支持三种审核类型，利用 AI 模型检测是否包含涉黄、暴恐、涉政、其他不良、广告推广等违规内容：
    - text：审核文本内容（content）
    - image：审核图片（image_url）
    - video：审核视频画面 + 封面图（video_url + thumb_url）

    图片/视频通过 Files API 上传换取 file_id 后送审，规避第三方 URL 的 mimetype 校验问题。
    未显式传 type 时，根据已传参数自动推断（video_url > image_url > content）。
    返回：sensitive=true 表示检测到违规内容，false 表示合规内容。
    """
    mod_type = _resolve_type(request)
    log_target = _log_target(mod_type, request)

    try:
        client = get_ark_client()
    except Exception as e:
        raise HTTPException(status_code=503, detail=f"审核服务初始化失败: {str(e)}")

    # 图片/视频会在此处下载远程资源并上传 Files API（可能抛 400/502）
    input_content = _build_input_content(mod_type, request, client)

    system_msg = {"role": "system", "content": _MODERATION_PROMPT}
    user_msg = {"role": "user", "content": input_content}

    accumulated = ""
    is_sensitive = None
    try:
        stream = client.chat.completions.create(
            model="doubao-seed-2-0-mini-260428",
            messages=[system_msg, user_msg],
            # service_tier="fast",
            stream=True,
            thinking={"type": "disabled"},  # 通过 extra_body 透传给火山方舟
            response_format={"type": "text"}
        )
        try:
            for event in stream:
                # 豆包API使用标准的OpenAI兼容格式：event.choices[0].delta.content
                # 检查是否命中内容过滤器（护栏拦截）
                if hasattr(event, 'choices') and event.choices:
                    choice = event.choices[0]
                    # 如果 finish_reason 是 content_filter，说明内容被护栏拦截
                    if choice.finish_reason == 'content_filter':
                        is_sensitive = True
                        print("检测到内容过滤器拦截 (finish_reason=content_filter)，判定为违规")
                        break

                    delta = choice.delta.content
                    if delta:
                        accumulated += delta
                        # true/false 首字母不同，读到第一个非空字符即可判定
                        is_sensitive = _try_early_decide(accumulated)
                        if is_sensitive is not None:
                            break
        finally:
            stream.close()  # 提前判定后主动关闭连接，释放资源
    except Exception as e:
        # 护栏拦截即视为命中违规内容，直接返回 sensitive=true
        if _is_sensitive_block(e):
            save_chat_log(
                user_id="moderation",
                question=log_target,
                stream_mode=True,
                status="blocked",
                error=str(e),
            )
            return ModerationResponse(sensitive=True)
        save_chat_log(
            user_id="moderation",
            question=log_target,
            stream_mode=True,
            status="error",
            error=str(e),
        )
        raise HTTPException(status_code=500, detail=f"内容审核服务异常: {str(e)}")

    if is_sensitive is None:
        # 未能提前判定（如首字符异常），兜底解析已累积到的文本
        try:
            is_sensitive = _parse_moderation_result(accumulated)
        except ValueError as e:
            save_chat_log(
                user_id="moderation",
                question=log_target,
                stream_mode=True,
                raw_response=accumulated,
                status="error",
                error=str(e),
            )
            raise HTTPException(status_code=500, detail="内容审核结果解析失败，请重试")

    return ModerationResponse(sensitive=is_sensitive)


router.tags = ["内容审核"]