xslandClaude Sonnet 4.6
28a6062fea
- face_features.py：_PROMPT 只问6项特征（+有无人脸），analyze_features 只返回6个英文字段；无人脸返回 None（不再依赖 has_face 在外部判定） - gateway/app.py：无脸判定改为 feats is None，去掉 has_face import - app.py / docs / test_interface4.html：Swagger/文档/示例/测试页同步 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
835 lines
33 KiB
Python
835 lines
33 KiB
Python
"""旷视五接口 — worker 侧（高性能 GPU 后端）。
|
||
|
||
接口 1（四庭七眼测量）已为**真实算法实现**（见 face_analysis 包）；接口 2~5 仍为 Mock。
|
||
拆分架构：worker 跑算法、返回 `annotated_image_base64`（不落盘、不拼 URL，由网关完成）。
|
||
worker 对 `/api/*` 校验内网鉴权头 `X-Internal-Token`，`/health` 供网关探测不校验。
|
||
"""
|
||
import base64
|
||
import json
|
||
import logging
|
||
import os
|
||
from contextlib import asynccontextmanager
|
||
from io import BytesIO
|
||
from typing import Any, List, Optional
|
||
|
||
import cv2
|
||
import numpy as np
|
||
from fastapi import FastAPI, File, Form, Request, UploadFile
|
||
from fastapi.responses import JSONResponse
|
||
from fastapi.staticfiles import StaticFiles
|
||
from pydantic import BaseModel, Field
|
||
|
||
logging.basicConfig(level=logging.INFO,
|
||
format="%(asctime)s %(levelname)s %(name)s: %(message)s")
|
||
logger = logging.getLogger("hair.worker")
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# App & 全局常量
|
||
# ---------------------------------------------------------------------------
|
||
|
||
BASE_URL = "https://hair.xiangsilian.com"
|
||
SAMPLE_IMAGE_URL = f"{BASE_URL}/static/sample.jpg"
|
||
|
||
# 分辨率门槛（短边/长边，方向无关），可由环境变量覆盖（技术方案 §8.3）
|
||
MIN_SHORT_SIDE = int(os.getenv("MIN_SHORT_SIDE", "600"))
|
||
MIN_LONG_SIDE = int(os.getenv("MIN_LONG_SIDE", "800"))
|
||
# 文件大小上限（字节），可由环境变量覆盖；设为 0 表示不限制
|
||
MAX_FILE_BYTES = int(os.getenv("MAX_FILE_BYTES", "1000000")) # 默认 1 MB
|
||
if MAX_FILE_BYTES <= 0:
|
||
MAX_FILE_BYTES = float("inf")
|
||
|
||
# 运行期状态：模型就绪标志（/health 据此返回 200/503）
|
||
_STATE = {"ready": False}
|
||
|
||
|
||
def _load_accept_passwords() -> List[str]:
|
||
"""worker 鉴权密码来源：环境变量 WORKER_ACCEPT_PASSWORDS（逗号分隔）优先，
|
||
否则读 worker_config.json 的 accept_passwords 列表。"""
|
||
env = os.getenv("WORKER_ACCEPT_PASSWORDS")
|
||
if env:
|
||
return [p.strip() for p in env.split(",") if p.strip()]
|
||
cfg_path = os.path.join(os.path.dirname(__file__), "worker_config.json")
|
||
if os.path.isfile(cfg_path):
|
||
try:
|
||
with open(cfg_path, encoding="utf-8") as f:
|
||
return list(json.load(f).get("accept_passwords", []))
|
||
except Exception as e: # noqa: BLE001
|
||
logger.warning("读取 worker_config.json 失败：%s", e)
|
||
return []
|
||
|
||
|
||
ACCEPT_PASSWORDS = _load_accept_passwords()
|
||
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(_app: FastAPI):
|
||
"""启动时加载模型单例（detector 必需，segmenter 尽力而为），就绪后才放行 /health。"""
|
||
from face_analysis.detector import detector # 触发 MediaPipe 单例初始化
|
||
_ = detector
|
||
try:
|
||
from face_analysis.hair_segmenter import get_segmenter
|
||
seg = get_segmenter()
|
||
logger.info("BiSeNet 头发分割就绪，device=%s", seg.device)
|
||
except Exception as e: # noqa: BLE001
|
||
# 方案 B 不可用（如 torch 缺失）：降级为方案 A only，不阻塞服务
|
||
logger.warning("头发分割不可用，接口1 将走方案A兜底：%s", e)
|
||
# 接口2（C端生发）单例预热：FaceLandmarker + SegFormer + mesh/贴图映射
|
||
try:
|
||
from hairline.service import get_landmarker, get_parser, get_texture_map
|
||
from hairline.render import load_ext_mesh
|
||
get_landmarker(); get_parser(); load_ext_mesh(); get_texture_map()
|
||
logger.info("接口2 发际线管线就绪")
|
||
except Exception as e: # noqa: BLE001
|
||
logger.warning("接口2 发际线管线初始化失败（该接口将返回错误）：%s", e)
|
||
_STATE["ready"] = True
|
||
yield
|
||
|
||
|
||
app = FastAPI(
|
||
lifespan=lifespan,
|
||
title="旷视五接口",
|
||
version="0.1.0",
|
||
description="""
|
||
## 概述
|
||
|
||
本服务提供五个人像分析接口，当前为 **Mock 第一版**：
|
||
- 接口已全部上线，传任意合法参数均可正常响应
|
||
- 所有字段返回固定示例值，图片字段统一指向示例图片
|
||
- 真实算法逻辑后续接入，字段结构不变
|
||
|
||
## 图片传参说明
|
||
|
||
每个接口的图片参数均支持三种方式，**严格互斥，必须且只能选其一**：
|
||
|
||
| 方式 | 字段名 | 说明 |
|
||
|------|--------|------|
|
||
| 文件上传 | `image_file` | `multipart/form-data`，单文件 ≤ 1 MB |
|
||
| URL | `image_url` | 图片的完整 HTTP/HTTPS 地址 |
|
||
| base64 | `image_base64` | 需携带前缀，如 `data:image/jpeg;base64,xxxx` |
|
||
|
||
> 传 0 个或同时传多个，均返回错误码 `1007`。
|
||
|
||
## 图片要求
|
||
|
||
- 格式：**JPG / PNG**
|
||
- 分辨率：最低 1080×1920，最大 4000×5000
|
||
- 人脸数量：仅支持**单人**，多人返回错误码 `1005`
|
||
- 文件大小：≤ 1 MB（文件上传方式）
|
||
|
||
## 统一响应结构
|
||
|
||
```json
|
||
{
|
||
"code": 0,
|
||
"message": "success",
|
||
"request_id": "唯一请求标识",
|
||
"data": {}
|
||
}
|
||
```
|
||
|
||
`code = 0` 表示成功，非 0 表示失败，`message` 为具体原因。
|
||
|
||
## 错误码一览
|
||
|
||
| code | 说明 |
|
||
|------|------|
|
||
| 1001 | 无法识别人像 |
|
||
| 1002 | 人像分辨率过低 |
|
||
| 1003 | 非正面照 / 角度过大 |
|
||
| 1004 | 性别标签无法判定 |
|
||
| 1005 | 检测到多张人脸（仅支持单人）|
|
||
| 1006 | 文件超出 1 MB 限制 |
|
||
| 1007 | 图片参数错误（未传或同时传多个）|
|
||
| 1008 | 图片格式不支持（仅 JPG / PNG）|
|
||
""",
|
||
)
|
||
|
||
app.mount("/static", StaticFiles(directory="static"), name="static")
|
||
|
||
# 不校验鉴权的路径前缀（供网关探测 / 文档 / 静态）
|
||
_AUTH_EXEMPT = ("/health", "/docs", "/openapi.json", "/redoc", "/static")
|
||
|
||
|
||
@app.middleware("http")
|
||
async def internal_token_auth(request: Request, call_next):
|
||
"""worker 内网鉴权：/api/* 必须带正确的 X-Internal-Token，否则 401。
|
||
|
||
密码列表来自 worker 配置（accept_passwords）。未配置任何密码时放行（仅便于
|
||
本地起步），生产务必通过 worker_config.json / 环境变量配置密码。
|
||
"""
|
||
path = request.url.path
|
||
if path == "/" or path.startswith(_AUTH_EXEMPT):
|
||
return await call_next(request)
|
||
if path.startswith("/api/") and ACCEPT_PASSWORDS:
|
||
token = request.headers.get("X-Internal-Token")
|
||
if token not in ACCEPT_PASSWORDS:
|
||
return JSONResponse(status_code=401, content={
|
||
"code": 1009,
|
||
"message": "未授权：缺少或错误的 X-Internal-Token",
|
||
"request_id": "worker",
|
||
"data": None,
|
||
})
|
||
return await call_next(request)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 通用响应模型
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def ok(data: Any) -> dict:
|
||
return {
|
||
"code": 0,
|
||
"message": "success",
|
||
"request_id": "mock-request-id",
|
||
"data": data,
|
||
}
|
||
|
||
|
||
def err(code: int, message: str) -> dict:
|
||
return {
|
||
"code": code,
|
||
"message": message,
|
||
"request_id": "mock-request-id",
|
||
"data": None,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 图片输入处理（三选一：file / url / base64）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
async def resolve_image_bytes(image_file, image_url, image_base64):
|
||
"""三选一取图，返回 (raw_bytes, error_response)。
|
||
|
||
严格互斥：传 0 个或多个 → 1007；URL 下载失败/base64 解码失败 → 1008。
|
||
大小校验放到上层（统一 1006），此处只负责取到字节。
|
||
"""
|
||
provided = [x for x in (image_file, image_url, image_base64) if x]
|
||
if len(provided) != 1:
|
||
return None, err(1007, "图片参数错误：必须且只能传 image_file / image_url / image_base64 其中一个")
|
||
|
||
if image_file is not None:
|
||
return await image_file.read(), None
|
||
|
||
if image_url:
|
||
try:
|
||
import httpx
|
||
with httpx.Client(timeout=10.0, follow_redirects=True) as client:
|
||
resp = client.get(image_url)
|
||
resp.raise_for_status()
|
||
return resp.content, None
|
||
except Exception: # noqa: BLE001
|
||
return None, err(1008, "图片下载失败或格式不支持")
|
||
|
||
# base64：去掉 data:image/...;base64, 前缀
|
||
try:
|
||
b64 = image_base64.split(",", 1)[1] if image_base64.startswith("data:") else image_base64
|
||
return base64.b64decode(b64), None
|
||
except Exception: # noqa: BLE001
|
||
return None, err(1008, "base64 解码失败")
|
||
|
||
|
||
# 接口2/3/5 返回不透明照片，用 JPG 显著减小体积（接口1 标注图含透明，仍用 PNG）
|
||
_JPG_QUALITY = int(os.getenv("JPG_QUALITY", "90"))
|
||
|
||
|
||
def _jpg_b64(bgr) -> str:
|
||
"""BGR 图 → JPG base64。"""
|
||
_ok, buf = cv2.imencode(".jpg", bgr, [cv2.IMWRITE_JPEG_QUALITY, _JPG_QUALITY])
|
||
return base64.b64encode(buf.tobytes()).decode()
|
||
|
||
|
||
def _png_to_jpg_b64(png_bytes) -> str:
|
||
"""ComfyUI 返回的 PNG 字节 → 重编码为 JPG base64；无法解码则原样透传。"""
|
||
img = cv2.imdecode(np.frombuffer(png_bytes, np.uint8), cv2.IMREAD_COLOR)
|
||
if img is None:
|
||
return base64.b64encode(png_bytes).decode()
|
||
return _jpg_b64(img)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 通用图片请求 Body（JSON 方式，用于 url / base64）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_image_fields_desc = (
|
||
"图片传参方式严格互斥，必须且只能选其一：\n"
|
||
"- **image_file**（multipart/form-data 上传，≤ 1 MB）\n"
|
||
"- **image_url**（完整 HTTP/HTTPS 地址）\n"
|
||
"- **image_base64**（需携带前缀，如 `data:image/jpeg;base64,xxxx`）\n\n"
|
||
"同时传多个或一个都不传，均返回错误码 `1007`。"
|
||
)
|
||
|
||
|
||
class ImageJsonBody(BaseModel):
|
||
image_url: Optional[str] = Field(
|
||
default=None,
|
||
description="图片 URL（与 image_base64 二选一，不可同时传）",
|
||
examples=["https://hair.xiangsilian.com/static/sample.jpg"],
|
||
)
|
||
image_base64: Optional[str] = Field(
|
||
default=None,
|
||
description="图片 base64，需带前缀，如 `data:image/jpeg;base64,xxxx`",
|
||
examples=["data:image/jpeg;base64,/9j/4AAQSkZJRgAB..."],
|
||
)
|
||
|
||
model_config = {
|
||
"json_schema_extra": {
|
||
"examples": [
|
||
{
|
||
"summary": "使用 URL 传图",
|
||
"value": {"image_url": "https://hair.xiangsilian.com/static/sample.jpg"},
|
||
},
|
||
{
|
||
"summary": "使用 base64 传图",
|
||
"value": {"image_base64": "data:image/jpeg;base64,/9j/4AAQSkZJRgAB..."},
|
||
},
|
||
]
|
||
}
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 接口 1：四庭七眼测量标注
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.post(
|
||
"/api/v1/face/measure",
|
||
summary="接口1 四庭七眼测量标注",
|
||
tags=["人脸分析"],
|
||
description=f"""
|
||
输入用户正面照，返回：
|
||
- 标注好四庭七眼数据的 **PNG 图片**（仅标注图层，不含人物）
|
||
- 四庭（顶庭/上庭/中庭/下庭）各段**厘米数值及占比**
|
||
- 七眼（眼宽/脸宽/两眼间距）**厘米数值及占比**
|
||
- 五个关键分界点的**原图像素坐标**（头顶/发际线/眉心/鼻翼下缘/下巴尖）
|
||
|
||
{_image_fields_desc}
|
||
|
||
图片同时支持 `multipart/form-data` 文件上传（字段名 `image_file`），或 JSON Body 传 `image_url` / `image_base64`。
|
||
|
||
---
|
||
|
||
**坐标说明**：所有坐标以原图像素为基准，原点为图片左上角，x 向右，y 向下。
|
||
|
||
**标注图片 UI 规范**（真实版本生效）：
|
||
- 字体/线条颜色：`#FFFFFF 100%`
|
||
- 四庭数值在图片**左侧**呈现，七眼间距**上下穿插**展示
|
||
- 字体：PingFangSC-Regular 10pt，线宽 1pt
|
||
- 横线/竖线渐变消失，虚线两侧带箭头
|
||
""",
|
||
responses={
|
||
200: {
|
||
"description": "成功",
|
||
"content": {
|
||
"application/json": {
|
||
"example": {
|
||
"code": 0,
|
||
"message": "success",
|
||
"request_id": "mock-request-id",
|
||
"data": {
|
||
"annotated_image_url": SAMPLE_IMAGE_URL,
|
||
"face_total_height_cm": 13.76,
|
||
"four_courts": {
|
||
"top_court_cm": 3.44,
|
||
"upper_court_cm": 3.44,
|
||
"middle_court_cm": 3.44,
|
||
"lower_court_cm": 3.44,
|
||
"ratios": {
|
||
"top_court": 0.25,
|
||
"upper_court": 0.25,
|
||
"middle_court": 0.25,
|
||
"lower_court": 0.25,
|
||
},
|
||
},
|
||
"seven_eyes": {
|
||
"eye_width_cm": 3.44,
|
||
"face_width_cm": 24.08,
|
||
"inter_eye_distance_cm": 3.44,
|
||
"ratios": {"eye_width": 0.143, "inter_eye_distance": 0.143},
|
||
},
|
||
"landmarks": {
|
||
"hair_top": {"x": 540, "y": 120},
|
||
"hairline": {"x": 540, "y": 430},
|
||
"brow_center": {"x": 540, "y": 740},
|
||
"nose_bottom": {"x": 540, "y": 1050},
|
||
"chin_tip": {"x": 540, "y": 1360},
|
||
},
|
||
},
|
||
}
|
||
}
|
||
},
|
||
},
|
||
400: {
|
||
"description": "参数错误 / 图片识别失败",
|
||
"content": {
|
||
"application/json": {
|
||
"examples": {
|
||
"图片参数错误": {"value": {"code": 1007, "message": "图片参数错误：必须且只能传 image_file / image_url / image_base64 其中一个", "request_id": "x", "data": None}},
|
||
"无法识别人像": {"value": {"code": 1001, "message": "无法识别人像", "request_id": "x", "data": None}},
|
||
"多张人脸": {"value": {"code": 1005, "message": "检测到多张人脸，仅支持单人照片", "request_id": "x", "data": None}},
|
||
}
|
||
}
|
||
},
|
||
},
|
||
},
|
||
)
|
||
async def face_measure(
|
||
image_file: Optional[UploadFile] = File(default=None, description="上传图片文件（JPG/PNG，≤ 1 MB）"),
|
||
image_url: Optional[str] = Form(default=None, description="图片 URL"),
|
||
image_base64: Optional[str] = Form(default=None, description="图片 base64（需带 data:image/...;base64, 前缀）"),
|
||
):
|
||
# 1. 三选一取图
|
||
raw, e = await resolve_image_bytes(image_file, image_url, image_base64)
|
||
if e is not None:
|
||
return e
|
||
|
||
# 2. 大小校验（≤ 1MB）
|
||
if len(raw) > MAX_FILE_BYTES:
|
||
return err(1006, "文件超出 1 MB 限制")
|
||
|
||
# 3. 解码
|
||
image = cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
|
||
if image is None:
|
||
return err(1008, "图片格式不支持（仅 JPG / PNG）")
|
||
|
||
# 4. 分辨率（短边/长边，方向无关，可配置门槛）
|
||
h, w = image.shape[:2]
|
||
short_side, long_side = min(w, h), max(w, h)
|
||
if short_side < MIN_SHORT_SIDE or long_side < MIN_LONG_SIDE:
|
||
return err(1002, "人像分辨率过低")
|
||
|
||
try:
|
||
from face_analysis.detector import detector
|
||
from face_analysis.pose import estimate_head_pose, check_frontal_face
|
||
from face_analysis.measure import measure_face
|
||
from face_analysis.annotation import create_annotated_image
|
||
|
||
# 5. 人脸检测
|
||
landmarks = detector.detect(image)
|
||
if landmarks is None:
|
||
return err(1001, "无法识别人像")
|
||
|
||
# 6. 姿态校验
|
||
if not check_frontal_face(landmarks, w, h):
|
||
return err(1003, "角度问题，请上传正面照")
|
||
head_pose = estimate_head_pose(landmarks, w, h)
|
||
|
||
# 7. 头发分割（方案 B），失败传 None 由 measure 内部回退方案 A
|
||
hair_mask = None
|
||
try:
|
||
from face_analysis.hair_segmenter import get_segmenter
|
||
hair_mask = get_segmenter().segment_hair(image)
|
||
except Exception as seg_e: # noqa: BLE001
|
||
logger.warning("头发分割失败，回退方案A：%s", seg_e)
|
||
|
||
# 8. 测量 + 标注图
|
||
result = measure_face(landmarks, hair_mask, w, h, head_pose=head_pose)
|
||
annotated = create_annotated_image(image, result)
|
||
buf = BytesIO()
|
||
annotated.save(buf, format="PNG")
|
||
|
||
# 9. 拆分架构：返回 base64，不落盘不拼 URL（落盘改 URL 由网关完成）
|
||
data = result.to_response()
|
||
data["annotated_image_base64"] = base64.b64encode(buf.getvalue()).decode()
|
||
return ok(data)
|
||
except Exception as ex: # noqa: BLE001
|
||
logger.exception("接口1 处理异常")
|
||
return err(1007, f"处理失败：{ex}")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 接口 2：C 端生发
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.post(
|
||
"/api/v1/hair/grow",
|
||
summary="接口2 C端生发",
|
||
tags=["生发"],
|
||
description=f"""
|
||
输入用户正面照 + **性别**，返回该性别对应的多张「建议发际线预览图」（本期为
|
||
**发际线曲线叠加在原照片上的预览图**，非最终文生图生发图）。每个方案包含：
|
||
- 预览图（worker 返回 `image_base64`，网关落盘后改写为 `image_url`）
|
||
- 发际线类型 `hairline_type`（英文 key）
|
||
- 顺序 `order`（本期固定 `1..N`，不排序）
|
||
|
||
{_image_fields_desc}
|
||
|
||
图片同时支持 `multipart/form-data` 文件上传（字段名 `image_file`）。
|
||
|
||
---
|
||
|
||
- **gender**（必填）：`male` / `female`。决定返回的贴图集合（female 5 张 / male 4 张）。
|
||
非法或缺失返回 `1004`。
|
||
- **beauty_enabled**：本期保留但不生效。
|
||
|
||
`hairline_type` 取值：`ellipse` / `flower` / `heart` / `straight` / `wave`（female），
|
||
`ellipse` / `m` / `straight` / `inverse_arc`（male）。
|
||
""",
|
||
responses={
|
||
200: {
|
||
"description": "成功",
|
||
"content": {
|
||
"application/json": {
|
||
"example": {
|
||
"code": 0,
|
||
"message": "success",
|
||
"request_id": "mock-request-id",
|
||
"data": {
|
||
"results": [
|
||
{"image_base64": "iVBORw0KGgo...", "hairline_type": "ellipse", "order": 1},
|
||
{"image_base64": "iVBORw0KGgo...", "hairline_type": "flower", "order": 2},
|
||
]
|
||
},
|
||
}
|
||
}
|
||
},
|
||
},
|
||
400: {
|
||
"description": "参数错误 / 图片识别失败",
|
||
"content": {
|
||
"application/json": {
|
||
"examples": {
|
||
"图片参数错误": {"value": {"code": 1007, "message": "图片参数错误：必须且只能传 image_file / image_url / image_base64 其中一个", "request_id": "x", "data": None}},
|
||
"非正面照": {"value": {"code": 1003, "message": "角度问题，请上传正面照", "request_id": "x", "data": None}},
|
||
}
|
||
}
|
||
},
|
||
},
|
||
},
|
||
)
|
||
async def hair_grow(
|
||
image_file: Optional[UploadFile] = File(default=None, description="上传图片文件（JPG/PNG，≤ 1 MB）"),
|
||
image_url: Optional[str] = Form(default=None, description="图片 URL"),
|
||
image_base64: Optional[str] = Form(default=None, description="图片 base64（需带 data:image/...;base64, 前缀）"),
|
||
gender: Optional[str] = Form(default=None, description="性别 male/female（必填）"),
|
||
beauty_enabled: bool = Form(default=False, description="是否开启美颜（本期不生效）"),
|
||
use_mask: bool = Form(default=True, description="是否启用 inpaint 遮罩（测试对比用）。false 时用干净原图生成（空遮罩，不烧模板线）"),
|
||
prompt: str = Form(default="补充遮罩区域的头发", description="ComfyUI 提示词，会替换工作流节点60的文本"),
|
||
):
|
||
# 1. gender 必填校验（非法/缺失 → 1004）
|
||
if gender not in ("male", "female"):
|
||
return err(1004, "gender 必填且只能为 male / female")
|
||
|
||
# 2. 三选一取图
|
||
raw, e = await resolve_image_bytes(image_file, image_url, image_base64)
|
||
if e is not None:
|
||
return e
|
||
if len(raw) > MAX_FILE_BYTES:
|
||
return err(1006, "文件超出 1 MB 限制")
|
||
|
||
image = cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
|
||
if image is None:
|
||
return err(1008, "图片格式不支持（仅 JPG / PNG）")
|
||
|
||
h, w = image.shape[:2]
|
||
short_side, long_side = min(w, h), max(w, h)
|
||
if short_side < MIN_SHORT_SIDE or long_side < MIN_LONG_SIDE:
|
||
return err(1002, "人像分辨率过低")
|
||
|
||
try:
|
||
from fastapi.concurrency import run_in_threadpool
|
||
from hairline.service import generate_grow_results
|
||
|
||
# 预览 + 生发(ComfyUI) 都是阻塞且较慢，放线程池避免卡住事件循环
|
||
items = await run_in_threadpool(generate_grow_results, image, gender, use_mask, prompt)
|
||
if items is None:
|
||
return err(1001, "无法识别人像")
|
||
|
||
results = []
|
||
for p in items:
|
||
results.append({
|
||
"image_base64": _jpg_b64(p["image_bgr"]), # 预览图 JPG
|
||
"grown_image_base64": (_png_to_jpg_b64(p["grown_png"]) # 生发图 JPG
|
||
if p["grown_png"] else None),
|
||
"hairline_type": p["hairline_type"],
|
||
"order": p["order"],
|
||
})
|
||
return ok({"results": results})
|
||
except Exception as ex: # noqa: BLE001
|
||
logger.exception("接口2 处理异常")
|
||
return err(1007, f"处理失败：{ex}")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 接口 3：B 端生发
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.post(
|
||
"/api/v1/hair/grow-b",
|
||
summary="接口3 B端生发（医生/操作端）",
|
||
tags=["生发"],
|
||
description="""
|
||
医生/操作端在用户照片上**手动用马克笔划线标注**目标发际线后，**只需上传这一张划线图**，返回：
|
||
- 生发后效果图（系统检测划线 → 据此生成「植发 3 个月」效果）
|
||
- 发际线形（手绘为定制，固定 `custom`）
|
||
|
||
**划线图片**：字段名前缀为 `marked_image_`，支持文件/URL/base64 三选一。**不需要原始照片**。
|
||
""",
|
||
responses={
|
||
200: {
|
||
"description": "成功",
|
||
"content": {
|
||
"application/json": {
|
||
"example": {
|
||
"code": 0,
|
||
"message": "success",
|
||
"request_id": "mock-request-id",
|
||
"data": {
|
||
"hair_growth_image_base64": "iVBORw0KGgo...（生发图）",
|
||
"hairline_type": "custom",
|
||
},
|
||
}
|
||
}
|
||
},
|
||
},
|
||
400: {
|
||
"description": "参数错误",
|
||
"content": {
|
||
"application/json": {
|
||
"example": {"code": 1007, "message": "图片参数错误", "request_id": "x", "data": None}
|
||
}
|
||
},
|
||
},
|
||
},
|
||
)
|
||
async def hair_grow_b(
|
||
marked_image_file: Optional[UploadFile] = File(default=None, description="划线图片文件（JPG/PNG，≤ 1 MB）"),
|
||
marked_image_url: Optional[str] = Form(default=None, description="划线图片 URL"),
|
||
marked_image_base64: Optional[str] = Form(default=None, description="划线图片 base64"),
|
||
use_mask: bool = Form(default=True, description="是否画发际线（测试对比用）。false 时跳过划线检测、直接送划线图"),
|
||
prompt: str = Form(default="补充遮罩区域的头发", description="ComfyUI 提示词，会替换工作流节点60的文本"),
|
||
):
|
||
# 划线图三选一取图（只需这一张）
|
||
marked_raw, e = await resolve_image_bytes(marked_image_file, marked_image_url, marked_image_base64)
|
||
if e is not None:
|
||
return e
|
||
if len(marked_raw) > MAX_FILE_BYTES:
|
||
return err(1006, "文件超出 1 MB 限制")
|
||
|
||
marked = cv2.imdecode(np.frombuffer(marked_raw, np.uint8), cv2.IMREAD_COLOR)
|
||
if marked is None:
|
||
return err(1008, "图片格式不支持（仅 JPG / PNG）")
|
||
|
||
h, w = marked.shape[:2]
|
||
short_side, long_side = min(w, h), max(w, h)
|
||
if short_side < MIN_SHORT_SIDE or long_side < MIN_LONG_SIDE:
|
||
return err(1002, "人像分辨率过低")
|
||
|
||
try:
|
||
from fastapi.concurrency import run_in_threadpool
|
||
from hairline.service import generate_grow_b
|
||
|
||
res = await run_in_threadpool(generate_grow_b, marked, use_mask, prompt)
|
||
if res["status"] == "no_face":
|
||
return err(1001, "无法识别人像")
|
||
if res["status"] == "no_line":
|
||
return err(1001, "未检测到发际线划线，请确认划线图额头有清晰的手绘发际线")
|
||
|
||
grown_b64 = _png_to_jpg_b64(res["grown_png"]) if res["grown_png"] else None # 生发图 JPG
|
||
data = {
|
||
"hair_growth_image_base64": grown_b64,
|
||
"hairline_type": "custom",
|
||
}
|
||
return ok(data)
|
||
except Exception as ex: # noqa: BLE001
|
||
logger.exception("接口3 处理异常")
|
||
return err(1007, f"处理失败：{ex}")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 接口 4：用户特征
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.post(
|
||
"/api/v1/face/features",
|
||
summary="接口4 用户特征分析",
|
||
tags=["人脸分析"],
|
||
description=f"""
|
||
输入用户照片，返回 N 个用户面部特征字段。
|
||
|
||
{_image_fields_desc}
|
||
|
||
图片同时支持 `multipart/form-data` 文件上传（字段名 `image_file`）。
|
||
|
||
---
|
||
|
||
由**火山方舟 豆包视觉模型**分析，返回**固定 6 个英文字段**。
|
||
|
||
**返回格式**：`data.features` 为一个 **JSON 字符串**（不是对象），需要在客户端 `JSON.parse()` 后使用。
|
||
|
||
| 字段 | 说明 |
|
||
|------|------|
|
||
| face_shape | 脸形（如"鹅蛋脸"） |
|
||
| eyebrow_shape | 眉形（如"平眉"） |
|
||
| facial_age | 面部年龄（区间，如"18-25岁"） |
|
||
| dynamic_static_type | 动静类型（"静态型"/"动态型"） |
|
||
| gender | 性别（"男"/"女"） |
|
||
| gene_style | 基因风格（如"自然型"） |
|
||
|
||
> 无人脸返回 `1001`。
|
||
""",
|
||
responses={
|
||
200: {
|
||
"description": "成功",
|
||
"content": {
|
||
"application/json": {
|
||
"example": {
|
||
"code": 0,
|
||
"message": "success",
|
||
"request_id": "mock-request-id",
|
||
"data": {
|
||
"features": '{"face_shape":"鹅蛋脸","eyebrow_shape":"平眉","facial_age":"18-25岁","dynamic_static_type":"静态型","gender":"女","gene_style":"少年型"}',
|
||
},
|
||
}
|
||
}
|
||
},
|
||
},
|
||
400: {
|
||
"description": "参数错误 / 图片识别失败",
|
||
"content": {
|
||
"application/json": {
|
||
"example": {"code": 1001, "message": "无法识别人像", "request_id": "x", "data": None}
|
||
}
|
||
},
|
||
},
|
||
},
|
||
)
|
||
async def face_features(
|
||
image_file: Optional[UploadFile] = File(default=None, description="上传图片文件（JPG/PNG，≤ 1 MB）"),
|
||
image_url: Optional[str] = Form(default=None, description="图片 URL"),
|
||
image_base64: Optional[str] = Form(default=None, description="图片 base64（需带 data:image/...;base64, 前缀）"),
|
||
):
|
||
# ⚠️ 接口4 已迁到**网关本机**实现（直接调豆包视觉模型，见 gateway/app.py）。
|
||
# 网关不会把本接口转发到 worker，故此处仅留 Mock 占位、保持 worker 无外网依赖。
|
||
features = json.dumps(
|
||
{
|
||
"face_shape": "鹅蛋脸", "eyebrow_shape": "平眉", "facial_age": "18-25岁",
|
||
"dynamic_static_type": "静态型", "gender": "女", "gene_style": "少年型",
|
||
},
|
||
ensure_ascii=False,
|
||
)
|
||
return ok({"features": features})
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 接口 5：发际线 PNG 生成
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.post(
|
||
"/api/v1/hairline/generate",
|
||
summary="接口5 发际线PNG生成",
|
||
tags=["人脸分析"],
|
||
description=f"""
|
||
输入用户照片，返回 N 张用户发际线的 PNG 图片，并标注最合适发际线的面部中间点坐标。
|
||
|
||
{_image_fields_desc}
|
||
|
||
图片同时支持 `multipart/form-data` 文件上传（字段名 `image_file`）。
|
||
|
||
---
|
||
|
||
**入参**：新增必填 `gender`（`male`/`female`），决定返回的发际线集合（female 5 / male 4）。
|
||
|
||
**返回说明**：
|
||
|
||
- `hairline_images`：发际线叠加图列表（发际线曲线叠加在用户照片上，同接口2预览），
|
||
数量 = 该性别的发际线类型数，本期按贴图顺序 `order=1..N`（暂不计算合适度）。
|
||
worker 返回 `image_base64`，网关落盘后改写为 `image_url`。
|
||
- `best_hairline_center_point`：最佳（`order=1`）发际线曲线的**面部中间点**坐标，
|
||
以**原图像素**为基准（左上角为原点，x 向右，y 向下）。
|
||
""",
|
||
responses={
|
||
200: {
|
||
"description": "成功",
|
||
"content": {
|
||
"application/json": {
|
||
"example": {
|
||
"code": 0,
|
||
"message": "success",
|
||
"request_id": "mock-request-id",
|
||
"data": {
|
||
"hairline_images": [
|
||
{"image_base64": "iVBORw0KGgo...", "order": 1},
|
||
{"image_base64": "iVBORw0KGgo...", "order": 2},
|
||
],
|
||
"best_hairline_center_point": {"x": 540, "y": 430},
|
||
},
|
||
}
|
||
}
|
||
},
|
||
},
|
||
400: {
|
||
"description": "参数错误 / 图片识别失败",
|
||
"content": {
|
||
"application/json": {
|
||
"example": {"code": 1002, "message": "人像分辨率过低", "request_id": "x", "data": None}
|
||
}
|
||
},
|
||
},
|
||
},
|
||
)
|
||
async def hairline_generate(
|
||
image_file: Optional[UploadFile] = File(default=None, description="上传图片文件（JPG/PNG，≤ 1 MB）"),
|
||
image_url: Optional[str] = Form(default=None, description="图片 URL"),
|
||
image_base64: Optional[str] = Form(default=None, description="图片 base64（需带 data:image/...;base64, 前缀）"),
|
||
gender: Optional[str] = Form(default=None, description="性别 male/female（必填）"),
|
||
):
|
||
if gender not in ("male", "female"):
|
||
return err(1004, "gender 必填且只能为 male / female")
|
||
|
||
raw, e = await resolve_image_bytes(image_file, image_url, image_base64)
|
||
if e is not None:
|
||
return e
|
||
if len(raw) > MAX_FILE_BYTES:
|
||
return err(1006, "文件超出 1 MB 限制")
|
||
|
||
image = cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
|
||
if image is None:
|
||
return err(1008, "图片格式不支持（仅 JPG / PNG）")
|
||
|
||
h, w = image.shape[:2]
|
||
short_side, long_side = min(w, h), max(w, h)
|
||
if short_side < MIN_SHORT_SIDE or long_side < MIN_LONG_SIDE:
|
||
return err(1002, "人像分辨率过低")
|
||
|
||
try:
|
||
from fastapi.concurrency import run_in_threadpool
|
||
from hairline.service import generate_hairline_pngs
|
||
|
||
res = await run_in_threadpool(generate_hairline_pngs, image, gender)
|
||
if res is None:
|
||
return err(1001, "无法识别人像")
|
||
|
||
hairline_images = []
|
||
for it in res["images"]:
|
||
hairline_images.append({
|
||
"image_base64": _jpg_b64(it["image_bgr"]), # 发际线叠图 JPG
|
||
"order": it["order"],
|
||
})
|
||
c = res["best_center"]
|
||
data = {
|
||
"hairline_images": hairline_images,
|
||
"best_hairline_center_point": ({"x": c[0], "y": c[1]} if c else None),
|
||
}
|
||
return ok(data)
|
||
except Exception as ex: # noqa: BLE001
|
||
logger.exception("接口5 处理异常")
|
||
return err(1007, f"处理失败：{ex}")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 健康检查
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.get("/health", include_in_schema=False)
|
||
async def health():
|
||
# 模型就绪后才返回 200，供网关探测；未就绪返回 503。
|
||
if not _STATE["ready"]:
|
||
return JSONResponse(status_code=503, content={"status": "starting"})
|
||
return {"status": "ok"}
|
||
|
||
|
||
@app.get("/", include_in_schema=False)
|
||
async def index():
|
||
return {"service": "旷视五接口", "version": "0.1.0", "docs": f"{BASE_URL}/docs"}
|