import os import json import base64 import requests from datetime import datetime from flask import Flask, request, jsonify, render_template_string from threading import Lock app = Flask(__name__) moderation_data = {} data_lock = Lock() NVIDIA_API_KEY = os.environ.get("NVIDIA_API_KEY", "") NVIDIA_ENDPOINT = "https://integrate.api.nvidia.com/v1/chat/completions" MODEL_NAME = "qwen/qwen3.5-397b-a17b" SYSTEM_PROMPT = """你是一个图片内容审核AI助手。请分析用户提供的图片内容，判断是否包含以下违规内容： 1. 色情/低俗内容 2. 暴力/血腥内容 3. 政治敏感内容 4. 恐怖/惊悚内容 5. 赌博/诈骗内容 6. 其他违反法律法规的内容请基于图片分析结果，严格判断： - 如果图片内容安全，不包含任何违规元素，返回 "TRUE" - 如果图片包含任何违规内容或无法确定安全性，返回 "FALSE" 注意：只返回TRUE或FALSE，不要返回其他内容。""" HTML_TEMPLATE = """ 图片审核系统

🖼️ 图片审核系统

📁 选择图片审核

🤖 AI正在审核中，请稍候...

📋 审核记录

🔌 API 接口文档

审核图片: POST /api/moderate

人工通过: POST /api/approve/{id}

人工拒绝: POST /api/reject/{id}

查看详情: GET /api/status/{id}

""" def encode_image_to_base64(image_data): try: return base64.b64decode(image_data) except: return None def call_nvidia_api(image_base64): if not NVIDIA_API_KEY: return None, "API密钥未配置，请在Space设置中添加NVIDIA_API_KEY环境变量" headers = { "Authorization": f"Bearer {NVIDIA_API_KEY}", "Content-Type": "application/json", } payload = { "model": MODEL_NAME, "messages": [ { "role": "user", "content": [ {"type": "text", "text": SYSTEM_PROMPT}, { "type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_base64}"}, }, ], } ], "max_tokens": 10, "thinking": {"type": "off"}, } try: response = requests.post( NVIDIA_ENDPOINT, headers=headers, json=payload, timeout=120 ) if response.status_code == 200: result = response.json() content = result["choices"][0]["message"]["content"].strip().upper() return content, None else: return None, f"API错误: {response.status_code} - {response.text}" except Exception as e: return None, str(e) @app.route("/") def index(): return render_template_string(HTML_TEMPLATE) @app.route("/api/moderate", methods=["POST"]) def moderate(): data = request.json image_data = data.get("image", "") if not image_data: return jsonify({"error": "没有图片数据"}), 400 image_bytes = encode_image_to_base64(image_data) if not image_bytes: return jsonify({"error": "图片格式错误"}), 400 ai_result, error = call_nvidia_api(image_data) if error: return jsonify({"error": error}), 500 with data_lock: import uuid item_id = str(uuid.uuid4())[:8] status = "ai-pass" if ai_result == "TRUE" else "ai-reject" moderation_data[item_id] = { "id": item_id, "image": f"data:image/jpeg;base64,{image_data}", "ai_result": ai_result, "status": status, "final_status": None, "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), } return jsonify({"id": item_id, "ai_result": ai_result, "status": status}) @app.route("/api/approve/", methods=["POST"]) def approve(item_id): with data_lock: if item_id in moderation_data: moderation_data[item_id]["status"] = "approved" moderation_data[item_id]["final_status"] = "approved" return jsonify({"success": True, "status": "approved"}) return jsonify({"error": "未找到记录"}), 404 @app.route("/api/reject/", methods=["POST"]) def reject(item_id): with data_lock: if item_id in moderation_data: moderation_data[item_id]["status"] = "rejected" moderation_data[item_id]["final_status"] = "rejected" return jsonify({"success": True, "status": "rejected"}) return jsonify({"error": "未找到记录"}), 404 @app.route("/api/status/", methods=["GET"]) def get_status(item_id): with data_lock: if item_id in moderation_data: item = moderation_data[item_id] return jsonify( { "id": item["id"], "ai_result": item["ai_result"], "status": item["final_status"] if item["final_status"] else item["status"], "time": item["time"], } ) return jsonify({"error": "未找到记录"}), 404 @app.route("/api/history", methods=["GET"]) def history(): with data_lock: items = list(moderation_data.values())[-20:][::-1] status_text = { "ai-pass": "AI通过", "ai-reject": "待复审", "approved": "人工通过", "rejected": "人工拒绝", } return jsonify( { "items": [ { "id": item["id"], "image": item["image"], "ai_result": item["ai_result"], "status": item["final_status"] if item["final_status"] else item["status"], "status_text": status_text.get( item["final_status"] if item["final_status"] else item["status"], "未知", ), "time": item["time"], } for item in items ] } ) if __name__ == "__main__": port = int(os.environ.get("PORT", 7860)) app.run(host="0.0.0.0", port=port)