import json import io import uuid from urllib.parse import unquote, urljoin, urlparse import requests from alibabacloud_tea_openapi.client import Client as OpenApiClient from alibabacloud_tea_openapi import models as open_api_models from alibabacloud_tea_openapi import utils_models as open_api_util_models from alibabacloud_tea_util import models as util_models from config import Config from extensions import db, s3_client from models import GenerationRecord from services.generation_service import refund_points, try_deduct_points from services.logger import system_logger from utils import get_proxied_url def _create_client(): config = open_api_models.Config( access_key_id=Config.ALIBABA_CLOUD_ACCESS_KEY_ID, access_key_secret=Config.ALIBABA_CLOUD_ACCESS_KEY_SECRET, region_id=Config.AIGC_DETECTOR_REGION_ID, endpoint=Config.AIGC_DETECTOR_ENDPOINT, ) return OpenApiClient(config) def _normalize_image_url(image_url, request_root): if not image_url: return None if image_url.startswith(("http://", "https://")): return image_url return urljoin(request_root, image_url.lstrip("/")) def _is_own_file_url(image_url, request_root): public_path = Config.MINIO["public_url"].rstrip("/") + "/" parsed = urlparse(image_url) if not parsed.scheme: return image_url.startswith(public_path) root = urlparse(request_root) return parsed.netloc == root.netloc and parsed.path.startswith(public_path) def _extract_minio_object_key(image_url, request_root): public_path = Config.MINIO["public_url"].rstrip("/") + "/" parsed = urlparse(_normalize_image_url(image_url, request_root)) minio_endpoint = urlparse(Config.MINIO["endpoint"]) bucket_path = f"/{Config.MINIO['bucket']}/" if parsed.netloc == urlparse(request_root).netloc and parsed.path.startswith(public_path): return unquote(parsed.path[len(public_path):]) if parsed.netloc == minio_endpoint.netloc and parsed.path.startswith(bucket_path): return unquote(parsed.path[len(bucket_path):]) return None def _get_presigned_minio_url(object_key): return s3_client.generate_presigned_url( "get_object", Params={"Bucket": Config.MINIO["bucket"], "Key": object_key}, ExpiresIn=1800, ) def _mirror_image_for_detection(image_url, request_root): """把外部临时图转存到 MinIO，并返回给阿里云可直接下载的签名 URL。""" normalized_url = _normalize_image_url(image_url, request_root) object_key = _extract_minio_object_key(normalized_url, request_root) if object_key: return _get_presigned_minio_url(object_key) resp = requests.get(get_proxied_url(normalized_url), timeout=Config.PROXY_TIMEOUT_SHORT) resp.raise_for_status() content_type = resp.headers.get("Content-Type") or "image/png" ext = ".png" if "jpeg" in content_type or "jpg" in content_type: ext = ".jpg" elif "webp" in content_type: ext = ".webp" filename = f"aigc-detect-{uuid.uuid4().hex}{ext}" s3_client.upload_fileobj( io.BytesIO(resp.content), Config.MINIO["bucket"], filename, ExtraArgs={"ContentType": content_type}, ) return _get_presigned_minio_url(filename) def _extract_result(resp): body = resp.get("body", resp) data = body.get("Data") or body.get("data") or {} if isinstance(data, str): try: data = json.loads(data) except json.JSONDecodeError: data = {"raw": data} result = data.get("Result") or data.get("result") or [] result_items = result if isinstance(result, list) else [] first = result_items[0] if result_items else {} if not isinstance(first, dict): first = {"raw": first} label = first.get("Label") or first.get("label") or data.get("Label") or data.get("label") confidence = ( first.get("Confidence") or first.get("confidence") or first.get("Rate") or first.get("rate") or data.get("Confidence") or data.get("confidence") ) risk_level = ( first.get("RiskLevel") or first.get("riskLevel") or data.get("RiskLevel") or data.get("riskLevel") ) suggestion = ( first.get("Suggestion") or first.get("suggestion") or data.get("Suggestion") or data.get("suggestion") ) description = ( first.get("Description") or first.get("description") or data.get("Description") or data.get("description") ) details = [] for item in result_items: if not isinstance(item, dict): continue details.append({ "label": item.get("Label") or item.get("label"), "description": item.get("Description") or item.get("description"), "confidence": item.get("Confidence") or item.get("confidence"), "risk_level": item.get("RiskLevel") or item.get("riskLevel") or risk_level, }) return { "label": label, "description": description, "confidence": confidence, "risk_level": risk_level, "suggestion": suggestion, "details": details, "raw": body, } def _ensure_success(resp): body = resp.get("body", resp) code = body.get("Code") if isinstance(body, dict) else None if code is None and isinstance(body, dict): code = body.get("code") if code is None or str(code) == "200": return message = body.get("Msg") or body.get("Message") or body.get("msg") or body.get("message") or "检测接口返回失败" raise RuntimeError(f"{message} (code={code})") def detect_aigc_image(user_id, image_url, request_root): cost = Config.AIGC_DETECTOR_COST if not image_url: return {"error": "缺少图片地址"}, 400 if not try_deduct_points(user_id, cost): return {"error": f"积分不足，本次 AI 生成检测需要 {cost} 积分"}, 400 try: normalized_url = _mirror_image_for_detection(image_url, request_root) client = _create_client() params = open_api_util_models.Params( action="ImageModeration", version="2022-03-02", protocol="HTTPS", pathname="/", method="POST", auth_type="AK", body_type="json", req_body_type="json", style="ROA", ) request = open_api_util_models.OpenApiRequest( body={ "Service": Config.AIGC_DETECTOR_SERVICE, "ServiceParameters": json.dumps( {"imageUrl": normalized_url}, ensure_ascii=False, ), } ) runtime = util_models.RuntimeOptions( read_timeout=Config.PROXY_TIMEOUT_DEFAULT * 1000, connect_timeout=Config.PROXY_TIMEOUT_SHORT * 1000, ) resp = client.do_request(params, request, runtime) _ensure_success(resp) result = _extract_result(resp) record = GenerationRecord( user_id=user_id, prompt=f"AI生成检测: {normalized_url}", model=Config.AIGC_DETECTOR_SERVICE, cost=cost, image_urls=json.dumps([{"type": "aigc_detection", "url": normalized_url, "result": result}], ensure_ascii=False), ) db.session.add(record) db.session.commit() return {"message": "检测完成", "cost": cost, "result": result}, 200 except Exception as e: db.session.rollback() refund_points(user_id, cost) system_logger.error("AI生成检测失败", user_id=user_id, image_url=image_url, error=str(e)) return {"error": f"AI生成检测失败: {str(e)}"}, 500