24024
3d479c52e7
feat(api): 新增AI生成内容检测功能 - 添加aigc-detect路由接口，用于检测图像是否为AI生成内容 - 集成阿里云AIGC检测服务，配置相关API参数 - 实现积分扣除机制，每次检测消耗1积分 feat(frontend): 前端集成AI生成检测UI组件 - 添加formatAigcDetectionResult函数用于格式化检测结果显示 - 实现detectAigcImage函数处理检测请求和响应 - 在图片展示区域添加AI生成检测按钮和结果展示框 - 更新图片容器布局，优化下载按钮显示效果 refactor(service): 优化积分管理逻辑 - 新增try_deduct_points函数实现积分预检查和扣除 - 改进错误处理机制，确保数据库事务安全 refactor(history): 过滤AI检测记录 - 在历史记录查询中排除AI生成检测类型的数据 - 使用image_urls字段过滤检测相关的记录 ```
229 lines
7.6 KiB
Python
229 lines
7.6 KiB
Python
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
|