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. 使用赌博行业术语：包括“28”、“壹号”、“问鼎”、“时时彩”、“体彩”、“发布博彩内幕”、“真人娱乐城”、“澳门娱乐城”、“加拿大”、“接龙”、“扫雷”、“红蓝”等词汇及其变体或拆分形式。“加拿大28”是一种网络赌博玩法，“接龙”和“扫雷”是微信红包赌博游戏，“壹号”和“问鼎”是博彩网站，“红蓝”通常与百家乐等赌博游戏相关。
2. 诱导点击的博彩网址：博彩网站通常使用特定的域名后缀，如“.top”和“.vip”等。
3. 与赌博相关的内容：包括赌博游戏和平台（如私盘赌博、红包赌博、棋牌赌博等）、赌博行为（如开盘、投注、公布结果等）、赌博组织和人员（如赌客、财务等），以及与赌博相关的字词（如赢钱、德州扑克、老虎机等）。

【低俗色情】
1. 低俗性暗示和不伦关系：使用直接或间接指向性行为、性饥渴、局部敏感部位或暗示有色情资源的关键词和语句，如“馋”、“难忍”、“需求大”、“放开”、“放纵”、“有水”、“带小雨伞”、“想吃香蕉”、“你懂得”等。包括宣扬乱伦关系如兄妹、爷孙、父女等，或描述相关性行为、性暗示、擦边行为色情裸露等内容。
2. 敏感部位和性行为的描述：使用指代词、谐音、低俗别称等对人体敏感部位进行形容和描述，对性行为、性描写、性方式、性场景进行直观展示或描述，包括对人体性部位的直接暴露和描写，涉及特殊性癖如“SM”、“绿帽癖”等。
3. 对女性外貌和气质的低俗描述：使用带有性暗示或低俗内容的语言，可能涉及贬低女性的词汇，如“母狗”、“婊子”等。
4. 色情服务和物品交易：使用隐晦或黑话传播招嫖信息，或直接发布提供色情服务的内容，包括描述身体特征的词汇如“大胸”、“爆乳”、“36D”等，或提供“上门服务”、“陪洗澡”、“按摩”等服务。涉及色情物品的买卖行为，如原味或二手内衣、袜子、真人情趣玩具等，以及传播和寻求色情资源的行为，如分享色情资源的获取手段、app、网站等，例如“91”、“吃瓜群”、“黄色网站”、“app”、“网盘链接”等。
5. 以性行为为目的的交友行为：以发生性行为、性关系为目的的交友行为，常见情形如约炮、一夜情、床伴等。

【谩骂引战】
1. 人身攻击与辱骂：通过伦理身份、人格等进行攻击侮辱，编造网络黑话、恶意造梗，通过拼音、谐音、指代词等方式，恶意编造低俗烂梗、使用污言秽语侮辱谩骂他人。
2. 发布对立和歧视内容：包括性别对立、阶层对立、地域歧视等，污名化特定群体，煽动职业、性别、阶级、地域、宗族等歧视与对立，激化社会矛盾。

【欺诈】
1. 高佣金兼职诈骗：以高佣金、高薪，夸大行为等诱导话术发布网赚、兼职任务。
2. 金融诈骗：包括股票投资诈骗，通过推荐个股或投资产品诱导他人投资；贷款诈骗，仿冒正规机构进行贷款服务欺诈；回款清退诈骗，以投资平台“清退兑付”的名义进行引流，后续进行投资诈骗。
3. 身份仿冒诈骗：仿冒公检法等国家权力机关，如刑侦、检察院、法院、银行年检专员身份诈骗、电商、快递客服等，编造理由进行欺诈。
4. 免费赠品诈骗：以活动中奖、粉丝回馈等理由免费赠送高价值礼品，包括免费领游戏皮肤，索要微信号密码、下单物品免费但需交邮费等。

【黑产广告】
1. 低佣金招募闲散人员提供简单任务：包括使用“日结”、“一单一结”、“立返佣”等字眼，提供超简单任务如点赞、关注、评论、打字聊天、抄小说等，并承诺低额佣金。提供淘宝等平台的刷单、销量、知名度、增加人气等服务，或引导用户去电商平台刷单。
2. 提供账号相关服务：包括账号交易、账号代实名、代批量注册等服务，发布可通过技术手段解封社交账号、游戏账号的有偿服务内容，以及代注册、代认证服务（如帮助做人脸实名认证类）。
3. 非法金融和技术服务：提供涉及技术手段的非法操作和规避服务，包括恶意举报、一键群发、钻漏洞低价、套现、洗钱等。宣传不看信用、不看征信的借贷服务，提供非正常提额、刷信用、代开、强开服务或教程。还包括黑户漂白、信用卡逾期处理、征信修复、养卡服务，以及交易需实名的卡证信息如银行卡、信用卡、SIM卡、营业执照、对公账户等。
4. 提供接码和引流服务：包括提供接码任务（利用普通人的手机号接收验证码）或相关技术服务，利用工具人的设备、账号达到引流目的。出现收售特定群体粉丝如色情粉、赌博粉、兼职粉、交友粉等，或为黑灰产业提供引流账号、设备或技术支持等。

判断规则：图片/视频结合画面主体、文字、场景综合判断；注意谐音、拼音、指代词、拆字、变体等规避手段；仅依据所给内容判断，不做联想或补全。
只输出一个单词：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="内容审核结果解析失败，请重试")

    # 成功路径：记录审核结果
    save_chat_log(
        user_id="moderation",
        question=log_target,
        stream_mode=True,
        raw_response=accumulated,
        status="success",
    )

    return ModerationResponse(sensitive=is_sensitive)


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