""" 内容指纹工具 ============ 用于 AI 第三方平台(如 Vidu)的审核结果缓存与防重复提交。 核心逻辑: - 对提交的音频/视频/图片 URL + 任务参数生成 SHA256 指纹。 - 如果相同内容近期因审核失败被缓存,则直接返回错误,不再调用第三方平台。 - 仅规范化 URL(去掉 token 等动态参数),不下载大文件本身。 """ from __future__ import annotations import hashlib from urllib.parse import parse_qs, urlencode, urlparse # Vidu 内容安全/审核类错误码 # 这些错误在内容不变的情况下重试也没用,应该被缓存。 VIDU_AUDIT_ERROR_CODES = frozenset( { "TaskPromptPolicyViolation", # Prompt 触发安审风控 "AuditSubmitIllegal", # 输入未通过安全审核 "CreationPolicyViolation", # 生成物触发风控 "PhotoAuditNotPass", # 图片审核不通过 "AuditFailed", # 审核失败 "ImageCheckBodyJointsFailed", # 输入图人体检测失败 "ImageCheckFaceFailed", # 输入图人脸检测失败 "ImageObjectsUndetected", # 人体或人脸有遮挡 "FaceDetectFailure", # 人脸检测失败 "FaceDetectNotPass", # 人脸检测不通过 "NoFaceDetected", # 未检测到人脸 "MultiFaceDetected", # 检测到多张人脸 } ) # 常见动态 query 参数,生成指纹时应忽略 _DYNAMIC_QUERY_PARAMS = frozenset( { "token", "e", # 七牛过期时间戳 "t", # 时间戳 "sign", "x-oss-signature", "x-oss-expires", "x-oss-access-key-id", "response-content-disposition", "v", # 版本号/缓存戳 } ) def normalize_url(url: str | None) -> str: """规范化 URL,去掉动态参数,确保同一资源不同 token 得到相同指纹。 Args: url: 原始 URL,可能包含七牛私有 token、时间戳等动态参数 Returns: 规范化后的 URL 字符串,None 或空字符串返回空串 """ if not url: return "" parsed = urlparse(url) query_params = parse_qs(parsed.query, keep_blank_values=True) for key in list(query_params.keys()): if key.lower() in _DYNAMIC_QUERY_PARAMS: del query_params[key] query = urlencode(sorted(query_params.items()), doseq=True) path = parsed.path or "/" if query: return f"{parsed.scheme}://{parsed.netloc}{path}?{query}" return f"{parsed.scheme}://{parsed.netloc}{path}" def compute_content_fingerprint( task_type: str, *, video_url: str | None = None, audio_url: str | None = None, ref_photo_url: str | None = None, text: str | None = None, voice_id: str | None = None, ) -> str: """计算内容指纹。 指纹字段选择原则:只包含会影响 Vidu 审核结果的输入内容。 不包含 callback_url、speed、volume、payload 等业务/技术参数。 Args: task_type: 任务类型,如 "lip_sync", "tts", "clone_voice" video_url: 视频 URL audio_url: 音频 URL ref_photo_url: 参考图片 URL text: 文本/Prompt voice_id: 音色 ID Returns: SHA256 十六进制指纹字符串 """ parts = [ task_type.strip().lower(), normalize_url(video_url), normalize_url(audio_url), normalize_url(ref_photo_url), (text or "").strip(), (voice_id or "").strip().lower(), ] raw = "|".join(parts) return hashlib.sha256(raw.encode("utf-8")).hexdigest() def is_vidu_audit_error(err_code: str | None) -> bool: """判断是否为 Vidu 审核类错误码。""" if not err_code: return False return err_code.strip() in VIDU_AUDIT_ERROR_CODES def extract_vidu_error_code(message: str | None) -> str | None: """从 Vidu 错误信息中提取错误码。 Vidu 错误信息格式通常为:"ErrorCode: 中文描述" """ if not message: return None candidate = message.split(":")[0].strip() return candidate or None