企业官网建设流程全解析

这是图片自动审核服务代码，提供上传接口，自动检测违规内容并分流存储。

```python

"""

图片上传自动审核服务

pip install fastapi uvicorn python-multipart pillow transformers torch

"""

import io

import uuid

import logging

from enum import Enum

from pathlib import Path

from typing import Optional

import torch

from PIL import Image, ImageOps

from fastapi import FastAPI, File, UploadFile, HTTPException

from fastapi.concurrency import run_in_threadpool

from fastapi.staticfiles import StaticFiles

from pydantic import BaseModel

# ---------------------- 配置 ----------------------

UPLOAD_DIR = Path("storage/uploads") # 审核通过

QUARANTINE_DIR = Path("storage/quarantine") # 疑似 / 违规（不对外暴露）

MAX_FILE_SIZE = 10 * 1024 * 1024 # 10MB

ALLOWED_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(parents=True, exist_ok=True)

logging.basicConfig(level=logging.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: AuditLabel

score: float # 违规置信度 0~1

detail: dict = {}

class UploadResponse(BaseModel):

image_id: str

filename: str

audit: AuditResult

url: Optional[str] = None # 仅 pass 时返回可访问地址

message: str

# ---------------------- 审核器（策略模式，可插拔） ----------------------

class BaseAuditor:

name = "base"

def audit(self, image: Image.Image) -> AuditResult:

raise NotImplementedError

class LocalNSFWAuditor(BaseAuditor):

"""本地模型审核：无需联网、无调用费用，适合中小流量"""

name = "local-nsfw"

def __init__(self, model_name: str = "Falconsai/nsfw_image_detection"):

from transformers import AutoImageProcessor, AutoModelForImageClassification

self.device = "cuda" if torch.cuda.is_available() else "cpu"

self.processor = AutoImageProcessor.from_pretrained(model_name)

self.model = (

AutoModelForImageClassification.from_pretrained(model_name)

.to(self.device)

.eval()

)

self.labels = self.model.config.id2label

logger.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(images=image, return_tensors="pt").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()), default=0.0

)

if nsfw_score >= NSFW_BLOCK_THRESHOLD:

label = AuditLabel.BLOCK

elif nsfw_score >= NSFW_REVIEW_THRESHOLD:

label = AuditLabel.REVIEW

else:

label = AuditLabel.PASS

return AuditResult(

label=label,

score=nsfw_score,

detail={"auditor": self.name, "scores": scores},

)

class CloudAuditor(BaseAuditor):

"""

云厂商审核适配器骨架（阿里云内容安全 / 腾讯云 IMS / 百度 AI 均可）

按所选厂商文档补全 audit() 即可，其余业务代码无需改动。

"""

name = "cloud"

def __init__(self, endpoint: str, access_key: str, secret_key: str):

self.endpoint = endpoint

self.access_key = access_key

self.secret_key = secret_key

def audit(self, image: Image.Image) -> AuditResult:

import requests

buf = io.BytesIO()

image.save(buf, format="JPEG", quality=90)

# resp = requests.post(self.endpoint, data=buf.getvalue(),

# headers={...}, timeout=5)

# 将云厂商返回的 label / suggestion 映射成 AuditResult

raise NotImplementedError("按云厂商文档实现")

# ---------------------- FastAPI 应用 ----------------------

app = FastAPI(title="图片自动审核服务")

auditor: BaseAuditor = LocalNSFWAuditor() # 想换成云审核，改这一行即可

# 只暴露审核通过的目录，quarantine 不可直接访问

app.mount("/static", StaticFiles(directory=str(UPLOAD_DIR)), name="static")

async def _read_limited(file: UploadFile, limit: int) -> bytes:

"""分块读取，防止超大文件打爆内存"""

chunks, size = [], 0

while chunk := await file.read(1024 * 1024):

size += len(chunk)

if size > limit:

raise HTTPException(status_code=413, detail=f"图片超过 {limit // 1024 // 1024}MB 限制")

chunks.append(chunk)

if size == 0:

raise HTTPException(status_code=400, 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_code=400, detail="不是有效的图片文件")

fmt = (img.format or "").upper()

if fmt not in ALLOWED_FORMATS:

raise HTTPException(status_code=400, detail=f"不支持的图片格式：{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_img

@app.post("/api/v1/images/upload", response_model=UploadResponse)

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_code=502, detail=f"审核服务异常：{exc}")

# 3. 按结果分流落盘（文件名用 UUID，杜绝路径穿越）

image_id = uuid.uuid4().hex

ext = "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_id=image_id, filename=filename, audit=result,

url=f"/static/{filename}", message="审核通过",

)

if result.label == AuditLabel.BLOCK:

return UploadResponse(

image_id=image_id, filename=filename, audit=result,

url=None, message="图片包含违规内容，已拒绝",

)

return UploadResponse(

image_id=image_id, filename=filename, audit=result,

url=None, message="图片待人工复审",

)

@app.get("/health")

async def health():

return {"status": "ok", "auditor": auditor.name}

```

图片审核与分流流程

上传的图片会经过读取校验、模型审核和结果分流，最终给出清晰反馈。

· 上传与校验：分块读取限制10MB，校验图片格式，自动修正方向并转为RGB。

· 自动审核：本地模型推理，判定违规置信度，分通过、转人工、拒绝三档。

· 分流存储：通过图片存入公开目录并返回URL，疑似或违规存入隔离目录，不对外暴露。

· 结果响应：返回审核标签、置信度及提示，通过和拒绝均明确反馈。

---

优化建议： 可以调整 NSFW_BLOCK_THRESHOLD 和 NSFW_REVIEW_THRESHOLD 来改变审核严格程度，或替换 CloudAuditor 接入第三方服务。

此文章仅供学习参考。