这是图片自动审核服务代码提供上传接口自动检测违规内容并分流存储。python图片上传自动审核服务pip install fastapi uvicorn python-multipart pillow transformers torchimport ioimport uuidimport loggingfrom enum import Enumfrom pathlib import Pathfrom typing import Optionalimport torchfrom PIL import Image, ImageOpsfrom fastapi import FastAPI, File, UploadFile, HTTPExceptionfrom fastapi.concurrency import run_in_threadpoolfrom fastapi.staticfiles import StaticFilesfrom pydantic import BaseModel# ---------------------- 配置 ----------------------UPLOAD_DIR Path(storage/uploads) # 审核通过QUARANTINE_DIR Path(storage/quarantine) # 疑似 / 违规不对外暴露MAX_FILE_SIZE 10 * 1024 * 1024 # 10MBALLOWED_FORMATS {JPEG, PNG, WEBP, GIF, BMP}MAX_SIDE 1024 # 送审前缩放减少推理耗时NSFW_BLOCK_THRESHOLD 0.85 # 直接拒绝NSFW_REVIEW_THRESHOLD 0.50 # 转人工复审for _d in (UPLOAD_DIR, QUARANTINE_DIR):_d.mkdir(parentsTrue, exist_okTrue)logging.basicConfig(levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s)logger logging.getLogger(audit)# ---------------------- 数据结构 ----------------------class AuditLabel(str, Enum):PASS pass # 通过REVIEW review # 转人工BLOCK block # 拒绝class AuditResult(BaseModel):label: AuditLabelscore: float # 违规置信度 0~1detail: dict {}class UploadResponse(BaseModel):image_id: strfilename: straudit: AuditResulturl: Optional[str] None # 仅 pass 时返回可访问地址message: str# ---------------------- 审核器策略模式可插拔 ----------------------class BaseAuditor:name basedef audit(self, image: Image.Image) - AuditResult:raise NotImplementedErrorclass LocalNSFWAuditor(BaseAuditor):本地模型审核无需联网、无调用费用适合中小流量name local-nsfwdef __init__(self, model_name: str Falconsai/nsfw_image_detection):from transformers import AutoImageProcessor, AutoModelForImageClassificationself.device cuda if torch.cuda.is_available() else cpuself.processor AutoImageProcessor.from_pretrained(model_name)self.model (AutoModelForImageClassification.from_pretrained(model_name).to(self.device).eval())self.labels self.model.config.id2labellogger.info(审核模型加载完成 model%s labels%s device%s,model_name, self.labels, self.device)torch.inference_mode()def audit(self, image: Image.Image) - AuditResult:inputs self.processor(imagesimage, return_tensorspt).to(self.device)probs torch.softmax(self.model(**inputs).logits, dim-1)[0]scores {self.labels[i]: round(float(p), 4) for i, p in enumerate(probs)}nsfw_score max((v for k, v in scores.items() if nsfw in k.lower()), default0.0)if nsfw_score NSFW_BLOCK_THRESHOLD:label AuditLabel.BLOCKelif nsfw_score NSFW_REVIEW_THRESHOLD:label AuditLabel.REVIEWelse:label AuditLabel.PASSreturn AuditResult(labellabel,scorensfw_score,detail{auditor: self.name, scores: scores},)class CloudAuditor(BaseAuditor):云厂商审核适配器骨架阿里云内容安全 / 腾讯云 IMS / 百度 AI 均可按所选厂商文档补全 audit() 即可其余业务代码无需改动。name clouddef __init__(self, endpoint: str, access_key: str, secret_key: str):self.endpoint endpointself.access_key access_keyself.secret_key secret_keydef audit(self, image: Image.Image) - AuditResult:import requestsbuf io.BytesIO()image.save(buf, formatJPEG, quality90)# resp requests.post(self.endpoint, databuf.getvalue(),# headers{...}, timeout5)# 将云厂商返回的 label / suggestion 映射成 AuditResultraise NotImplementedError(按云厂商文档实现)# ---------------------- FastAPI 应用 ----------------------app FastAPI(title图片自动审核服务)auditor: BaseAuditor LocalNSFWAuditor() # 想换成云审核改这一行即可# 只暴露审核通过的目录quarantine 不可直接访问app.mount(/static, StaticFiles(directorystr(UPLOAD_DIR)), namestatic)async def _read_limited(file: UploadFile, limit: int) - bytes:分块读取防止超大文件打爆内存chunks, size [], 0while chunk : await file.read(1024 * 1024):size len(chunk)if size limit:raise HTTPException(status_code413, detailf图片超过 {limit // 1024 // 1024}MB 限制)chunks.append(chunk)if size 0:raise HTTPException(status_code400, detail上传内容为空)return b.join(chunks)def _prepare_audit_image(raw: bytes):校验图片合法性并返回 (原图格式, 送审缩略图)try:img Image.open(io.BytesIO(raw))img.load()except Exception:raise HTTPException(status_code400, detail不是有效的图片文件)fmt (img.format or ).upper()if fmt not in ALLOWED_FORMATS:raise HTTPException(status_code400, detailf不支持的图片格式{fmt or 未知})audit_img ImageOps.exif_transpose(img) # 修正手机拍照方向if audit_img.mode ! RGB:audit_img audit_img.convert(RGB)audit_img.thumbnail((MAX_SIDE, MAX_SIDE), Image.LANCZOS)return fmt, audit_imgapp.post(/api/v1/images/upload, response_modelUploadResponse)async def upload_image(file: UploadFile File(...)):# 1. 读取 基础校验raw await _read_limited(file, MAX_FILE_SIZE)fmt, audit_img _prepare_audit_image(raw)# 2. 审核同步推理丢到线程池避免阻塞事件循环try:result await run_in_threadpool(auditor.audit, audit_img)except Exception as exc:logger.exception(审核失败)raise HTTPException(status_code502, detailf审核服务异常{exc})# 3. 按结果分流落盘文件名用 UUID杜绝路径穿越image_id uuid.uuid4().hexext jpg if fmt JPEG else fmt.lower()filename f{image_id}.{ext}target_dir UPLOAD_DIR if result.label AuditLabel.PASS else QUARANTINE_DIR(target_dir / filename).write_bytes(raw)logger.info(上传完成 id%s label%s score%.4f, image_id, result.label.value, result.score)# 4. 返回结果if result.label AuditLabel.PASS:return UploadResponse(image_idimage_id, filenamefilename, auditresult,urlf/static/{filename}, message审核通过,)if result.label AuditLabel.BLOCK:return UploadResponse(image_idimage_id, filenamefilename, auditresult,urlNone, message图片包含违规内容已拒绝,)return UploadResponse(image_idimage_id, filenamefilename, auditresult,urlNone, message图片待人工复审,)app.get(/health)async def health():return {status: ok, auditor: auditor.name}图片审核与分流流程上传的图片会经过读取校验、模型审核和结果分流最终给出清晰反馈。· 上传与校验分块读取限制10MB校验图片格式自动修正方向并转为RGB。· 自动审核本地模型推理判定违规置信度分通过、转人工、拒绝三档。· 分流存储通过图片存入公开目录并返回URL疑似或违规存入隔离目录不对外暴露。· 结果响应返回审核标签、置信度及提示通过和拒绝均明确反馈。---优化建议 可以调整 NSFW_BLOCK_THRESHOLD 和 NSFW_REVIEW_THRESHOLD 来改变审核严格程度或替换 CloudAuditor 接入第三方服务。此文章仅供学习参考。
