本地搭建合规换脸实验服务：Python API 接口完整教程

import os

API_TOKEN = os.getenv("API_TOKEN", "change-me")
OUTPUT_BASE_URL = os.getenv("OUTPUT_BASE_URL", "http://127.0.0.1:8000")

UPLOAD_DIR = "data/uploads"
OUTPUT_DIR = "data/outputs"

from fastapi import Header, HTTPException
from app.config import API_TOKEN


def verify_token(authorization: str | None = Header(default=None)):
    if not authorization:
        raise HTTPException(status_code=401, detail="Missing Authorization header")

    prefix = "Bearer "
    if not authorization.startswith(prefix):
        raise HTTPException(status_code=401, detail="Invalid Authorization format")

    token = authorization[len(prefix):]
    if token != API_TOKEN:
        raise HTTPException(status_code=403, detail="Invalid API token")

from fastapi import HTTPException, UploadFile

MAX_FILE_SIZE = 8 * 1024 * 1024
ALLOWED_TYPES = {"image/jpeg", "image/png", "image/webp"}


async def validate_upload(file: UploadFile):
    if file.content_type not in ALLOWED_TYPES:
        raise HTTPException(status_code=400, detail="Only jpg/png/webp images are allowed")

    content = await file.read()
    if len(content) > MAX_FILE_SIZE:
        raise HTTPException(status_code=400, detail="Image is too large")

    await file.seek(0)


def reject_nsfw_or_unconsented():
    """
    这里预留安全检查。

    实际项目中，你可以接入：
    1. 自己的内容审核模型
    2. 第三方审核服务
    3. 人工审核流程

    规则建议：
    - 明确拒绝 NSFW
    - 明确拒绝名人、公众人物、未授权人物素材
    - 记录审计日志
    """
    return True

import os
import uuid
from fastapi import UploadFile
from app.config import UPLOAD_DIR, OUTPUT_DIR, OUTPUT_BASE_URL


os.makedirs(UPLOAD_DIR, exist_ok=True)
os.makedirs(OUTPUT_DIR, exist_ok=True)


def new_task_id() -> str:
    return uuid.uuid4().hex


async def save_upload(file: UploadFile, task_id: str, name: str) -> str:
    ext = os.path.splitext(file.filename or "")[1].lower() or ".png"
    path = os.path.join(UPLOAD_DIR, f"{task_id}_{name}{ext}")
    with open(path, "wb") as f:
        f.write(await file.read())
    return path


def output_path(task_id: str) -> str:
    return os.path.join(OUTPUT_DIR, f"{task_id}.png")


def output_url(task_id: str) -> str:
    return f"{OUTPUT_BASE_URL}/outputs/{task_id}.png"

from PIL import Image, ImageDraw


def process_face_image(source_path: str, target_path: str, output_path: str):
    """
    合规示例：
    这里不提供任何 NSFW 能力，也不绑定某个具体换脸模型。

    你可以把这里替换成：
    - 自己训练/授权使用的模型
    - 只处理本人素材的内部模型
    - 企业内部合规图像处理流程
    """
    target = Image.open(target_path).convert("RGB")

    # 示例：只做一个占位水印，证明接口流程跑通。
    draw = ImageDraw.Draw(target)
    draw.rectangle((20, 20, 320, 72), fill=(15, 118, 110))
    draw.text((32, 36), "Processed by local API", fill=(255, 255, 255))

    target.save(output_path)

from fastapi import Depends, FastAPI, File, UploadFile
from fastapi.staticfiles import StaticFiles

from app.security import verify_token
from app.safety import validate_upload, reject_nsfw_or_unconsented
from app.storage import new_task_id, save_upload, output_path, output_url
from app.processor import process_face_image

app = FastAPI(title="Local Consent Face Processing API")
app.mount("/outputs", StaticFiles(directory="data/outputs"), name="outputs")


@app.get("/health")
def health():
    return {"status": "ok"}


@app.post("/api/v1/face/process")
async def process_face(
    source_image: UploadFile = File(...),
    target_image: UploadFile = File(...),
    _=Depends(verify_token),
):
    await validate_upload(source_image)
    await validate_upload(target_image)
    reject_nsfw_or_unconsented()

    task_id = new_task_id()
    source_path = await save_upload(source_image, task_id, "source")
    target_path = await save_upload(target_image, task_id, "target")
    result_path = output_path(task_id)

    process_face_image(source_path, target_path, result_path)

    return {
        "task_id": task_id,
        "status": "completed",
        "result_url": output_url(task_id),
    }

import requests

url = "http://127.0.0.1:8000/api/v1/face/process"
headers = {"Authorization": "Bearer dev-token"}
files = {
    "source_image": open("source.png", "rb"),
    "target_image": open("target.png", "rb"),
}

res = requests.post(url, headers=headers, files=files, timeout=60)
print(res.json())