269 lines
12 KiB
Python
269 lines
12 KiB
Python
"""火山人像素材库审核编排:真人资产静默送审 + 轮询绿/红状态。
|
|
|
|
策略(用户定):一团队一素材组,后台静默上传,前端只显示绿盾(active)/红标(failed,提示改提示词重生)。
|
|
全部 best-effort:审核未启用/出错都不影响主流程(生图/采用照常)。
|
|
"""
|
|
import logging
|
|
from datetime import timedelta
|
|
|
|
from django.db import IntegrityError, transaction
|
|
from django.utils import timezone
|
|
|
|
from apps.assets import assets_client
|
|
from apps.assets.models import Asset, AssetReviewGroup
|
|
from apps.assets.storage import TosStorage
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# processing 超过此时长仍无终态 → 判超时失败,停止再查(防永久 processing + 无限轮询)
|
|
_PROCESSING_TIMEOUT = timedelta(minutes=15)
|
|
|
|
# 火山 Status → 本地 review_status
|
|
_STATUS_MAP = {"Active": "active", "Failed": "failed", "Processing": "processing", "Pending": "processing"}
|
|
|
|
|
|
def _asset_url(asset: Asset) -> str:
|
|
"""资产主图可公开访问 URL(火山要从 URL 抓图;TOS 签名 URL 即可,会立即抓取)。"""
|
|
f = asset.files.filter(is_primary=True).first() or asset.files.first()
|
|
if f is None:
|
|
return ""
|
|
if f.preview_url:
|
|
return f.preview_url
|
|
try:
|
|
return TosStorage().presigned_get_url(object_key=f.object_key)
|
|
except Exception: # noqa: BLE001
|
|
return ""
|
|
|
|
|
|
def get_or_create_team_group(team) -> AssetReviewGroup:
|
|
"""取/建团队的火山素材组(一团队一组)。并发安全:
|
|
① DB 行靠 OneToOne 唯一约束去重(只有一个赢,输的 catch 后复用);
|
|
② 远程建组用 select_for_update 串行化,只有第一个把 remote_group_id 从空写非空的那次才真建组,
|
|
避免并发双建远程组(孤儿)+ 丢送审。"""
|
|
grp = AssetReviewGroup.objects.filter(team=team).first()
|
|
if grp and grp.remote_group_id:
|
|
return grp
|
|
name = f"airshelf-team-{team.id}"
|
|
if grp is None:
|
|
try:
|
|
grp = AssetReviewGroup.objects.create(team=team, name=name, remote_group_id="")
|
|
except IntegrityError:
|
|
grp = AssetReviewGroup.objects.get(team=team)
|
|
if not grp.remote_group_id:
|
|
with transaction.atomic():
|
|
locked = AssetReviewGroup.objects.select_for_update().get(pk=grp.pk)
|
|
if not locked.remote_group_id:
|
|
locked.remote_group_id = assets_client.create_asset_group(name=name, description="AirShelf 真人素材审核")
|
|
locked.save(update_fields=["remote_group_id"])
|
|
grp = locked
|
|
return grp
|
|
|
|
|
|
def _volcano_asset_type(asset: Asset) -> str:
|
|
"""火山 CreateAsset 的 AssetType:视频必须传 Video,不能默认 Image。"""
|
|
if asset.asset_type == Asset.Type.VIDEO:
|
|
return "Video"
|
|
if asset.asset_type == Asset.Type.AUDIO:
|
|
return "Audio"
|
|
return "Image"
|
|
|
|
|
|
def submit_asset_for_review(asset: Asset, *, force: bool = False) -> bool:
|
|
"""真人资产送审:建组(若无)→ 传素材 → 标 processing。出错只记日志,不抛。
|
|
返回是否真正进入审核(True=已标 processing;False=未送审/未配置/失败),
|
|
供手动兜底端点据此如实回报,避免前端把「没送出去」误显示成「审核中」。
|
|
|
|
force=True 跳过 REVIEW_CATEGORIES 白名单:用户上传的资产被拿去当生成参考时,
|
|
我们无从判断里面有没有真人脸,一律登记一次(与人物素材库上传同策略)。"""
|
|
if not assets_client.is_enabled():
|
|
return False
|
|
if not force and asset.category not in Asset.REVIEW_CATEGORIES:
|
|
return False
|
|
if asset.review_remote_id and asset.review_status in ("active", "processing"):
|
|
return True
|
|
url = _asset_url(asset)
|
|
if not url:
|
|
return False
|
|
try:
|
|
grp = get_or_create_team_group(asset.team)
|
|
remote_id = assets_client.create_asset(
|
|
group_id=grp.remote_group_id,
|
|
image_url=url,
|
|
name=(asset.name or "person")[:64],
|
|
asset_type=_volcano_asset_type(asset),
|
|
)
|
|
if not remote_id:
|
|
# 火山没回 Id:不要标 processing(否则 remote_id 为空、poll 永远早退、卡死黄),留空可重试
|
|
logger.warning("create_asset 返回空 id,asset %s 暂不送审(可重试)", asset.id)
|
|
_record_submit_error(asset, "审核服务未返回素材编号")
|
|
return False
|
|
asset.review_remote_id = remote_id
|
|
asset.review_status = "processing"
|
|
asset.review_error = ""
|
|
asset.save(update_fields=["review_remote_id", "review_status", "review_error", "updated_at"])
|
|
return True
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.warning("submit_asset_for_review failed for asset %s: %s", asset.id, exc)
|
|
# 只写日志等于把原因埋了:调用方(视频复刻等)只能报「素材提交审核失败」这种没信息量的话,
|
|
# 用户和排查都无从下手。原因落到 review_error,让上层能带出来。
|
|
_record_submit_error(asset, str(exc))
|
|
return False
|
|
|
|
|
|
# 火山素材审核的常见拒因:原文是英文错误码,直接怼给用户看不懂也不知道怎么办。
|
|
_SUBMIT_ERROR_HINTS = (
|
|
("WidthTooSmall", "图片太小（宽高需在 300–6000 像素之间），请换一张更大的图"),
|
|
("HeightTooSmall", "图片太小（宽高需在 300–6000 像素之间），请换一张更大的图"),
|
|
("WidthTooLarge", "图片太大（宽高需在 300–6000 像素之间），请压缩后重试"),
|
|
("HeightTooLarge", "图片太大（宽高需在 300–6000 像素之间），请压缩后重试"),
|
|
("SizeTooLarge", "图片文件太大，请压缩后重试"),
|
|
("InvalidImageFormat", "图片格式不支持，请改用 JPG / PNG / WebP"),
|
|
("DownloadFailed", "审核服务拉取不到这张图，请重新上传"),
|
|
)
|
|
|
|
|
|
def humanize_submit_error(reason: str) -> str:
|
|
"""把火山的英文错误码换成能照着做的中文。认不出的原样返回。"""
|
|
text = reason or ""
|
|
for code, hint in _SUBMIT_ERROR_HINTS:
|
|
if code in text:
|
|
return hint
|
|
return text
|
|
|
|
|
|
def _record_submit_error(asset: Asset, reason: str) -> None:
|
|
"""把送审失败原因写回资产。写失败不抛 —— 审计不能反过来搞挂主流程。"""
|
|
try:
|
|
asset.review_error = humanize_submit_error(reason)[:2000]
|
|
asset.save(update_fields=["review_error", "updated_at"])
|
|
except Exception: # noqa: BLE001
|
|
logger.warning("记录 asset %s 送审失败原因时出错", asset.id, exc_info=True)
|
|
|
|
|
|
# 平台自己生成的资产,提示词与生成链路都在我们手里,视为免审;用户上传的必须真过一遍审核。
|
|
# 判据是 Asset.source 而不是「在不在某个库里」—— 按库免审等于把审核架空:
|
|
# 用户上传一张图进资产库,再从自由创作引用出去,就绕过了整套人像审核。
|
|
SELF_TRUSTED_SOURCES = (Asset.Source.AI_GENERATED, Asset.Source.SYSTEM)
|
|
|
|
|
|
def reference_review_state(asset: Asset) -> str:
|
|
"""引用一个平台资产(自由创作 @引用三库)前的审核判定。
|
|
|
|
返回 allowed / processing / failed / unsubmitted 四态之一,由调用方决定放行还是给提示。
|
|
未送审(unsubmitted)不代表拒绝到底 —— 调用方应顺手送一次审,让用户等一会儿再来。
|
|
"""
|
|
if not assets_client.is_enabled():
|
|
# 审核整套机制没配置时不能拿它拦人,否则一关审核全平台引用都用不了
|
|
return "allowed"
|
|
if asset.source in SELF_TRUSTED_SOURCES:
|
|
return "allowed"
|
|
if asset.review_status == "active":
|
|
return "allowed"
|
|
if asset.review_status == "processing":
|
|
return "processing"
|
|
if asset.review_status == "failed":
|
|
return "failed"
|
|
return "unsubmitted"
|
|
|
|
|
|
def _unsubmitted_review_qs(*, team=None):
|
|
qs = Asset.objects.filter(
|
|
category__in=Asset.REVIEW_CATEGORIES,
|
|
review_status="",
|
|
is_deleted=False,
|
|
)
|
|
if team is not None:
|
|
qs = qs.filter(team=team)
|
|
return qs
|
|
|
|
|
|
def _processing_review_qs(*, team=None):
|
|
# 不限 REVIEW_CATEGORIES:force 送审的上传视频/临时图也是 processing,worker 得盯到绿/红
|
|
qs = Asset.objects.filter(
|
|
review_status="processing",
|
|
is_deleted=False,
|
|
).exclude(review_remote_id="")
|
|
if team is not None:
|
|
qs = qs.filter(team=team)
|
|
return qs
|
|
|
|
|
|
def submit_unsubmitted_reviews(*, team=None, limit: int = 25) -> int:
|
|
"""把未送审的人脸资产自动丢给火山,不用人去后台点「批量送审」。"""
|
|
submitted = 0
|
|
cap = max(1, min(int(limit), 50))
|
|
for asset in _unsubmitted_review_qs(team=team).order_by("created_at")[:cap]:
|
|
if submit_asset_for_review(asset):
|
|
submitted += 1
|
|
return submitted
|
|
|
|
|
|
def poll_processing_reviews(*, team=None, limit: int = 80) -> dict[str, str]:
|
|
cap = max(1, min(int(limit), 200))
|
|
out: dict[str, str] = {}
|
|
for asset in _processing_review_qs(team=team).order_by("updated_at")[:cap]:
|
|
out[str(asset.id)] = poll_asset_review(asset)
|
|
return out
|
|
|
|
|
|
def drain_platform_reviews(*, submit_limit: int = 25, poll_limit: int = 80) -> dict:
|
|
"""全平台:未送审自动送 + 审核中自动拉状态。worker 兜底循环,人不在后台页也会跑。"""
|
|
submitted = submit_unsubmitted_reviews(limit=submit_limit)
|
|
statuses = poll_processing_reviews(limit=poll_limit)
|
|
return {"submitted": submitted, "polled": len(statuses), "statuses": statuses}
|
|
|
|
|
|
def ensure_review_drain_loop() -> None:
|
|
"""第一次有人轮询审核时,顺手把 worker 兜底循环拉起来(没有 beat 也能自转)。"""
|
|
import sys
|
|
|
|
from django.conf import settings as dj_settings
|
|
from django.core.cache import cache
|
|
|
|
if getattr(dj_settings, "CELERY_TASK_ALWAYS_EAGER", False) or "test" in sys.argv:
|
|
return
|
|
if not cache.add("asset-review-drain-kick", "1", timeout=300):
|
|
return
|
|
try:
|
|
from apps.ai.tasks import drain_asset_reviews_task
|
|
|
|
drain_asset_reviews_task.delay()
|
|
except Exception: # noqa: BLE001
|
|
cache.delete("asset-review-drain-kick")
|
|
logger.warning("kick asset review drain loop failed", exc_info=True)
|
|
|
|
|
|
def poll_asset_review(asset: Asset) -> str:
|
|
"""查单个真人资产审核状态并更新 review_status。返回最新状态。
|
|
只在状态变化时落库(保留 updated_at 作为「进入 processing 的时刻」);processing 超时兜底为 failed。"""
|
|
if not assets_client.is_enabled() or not asset.review_remote_id:
|
|
return asset.review_status
|
|
# 超时兜底:processing 太久(火山卡住 / remote_id 失效每次抛错)→ 判失败,退出永久轮询
|
|
if asset.review_status == "processing" and asset.updated_at and (timezone.now() - asset.updated_at) > _PROCESSING_TIMEOUT:
|
|
asset.review_status = "failed"
|
|
asset.review_error = "审核超时,请重新生成"
|
|
asset.save(update_fields=["review_status", "review_error", "updated_at"])
|
|
return "failed"
|
|
try:
|
|
data = assets_client.get_asset(asset.review_remote_id)
|
|
raw = data.get("Status")
|
|
status = _STATUS_MAP.get(raw)
|
|
if status is None:
|
|
logger.warning("未知审核 Status %r(asset %s),暂按 processing 处理", raw, asset.id)
|
|
status = "processing"
|
|
if status != asset.review_status: # 只在变化时写,避免每次 poll 刷新 updated_at 让超时永不触发
|
|
asset.review_status = status
|
|
asset.review_error = (data.get("ErrorMessage") or "") if status == "failed" else ""
|
|
asset.save(update_fields=["review_status", "review_error", "updated_at"])
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.warning("poll_asset_review failed for asset %s: %s", asset.id, exc)
|
|
return asset.review_status
|
|
|
|
|
|
def poll_team_reviews(team) -> dict:
|
|
"""先把本团队未送审的人脸资产送出去,再轮询审核中状态。
|
|
前端基础资产趴定时打这里:用户不用去后台点送审,也不用等管理员。"""
|
|
ensure_review_drain_loop()
|
|
submit_unsubmitted_reviews(team=team, limit=20)
|
|
return poll_processing_reviews(team=team, limit=80)
|