Files
hair/app.py
T
2026-06-23 22:29:41 +08:00

1097 lines
45 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""旷视五接口 — 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)
def _parse_hair_styles(raw: Optional[str], max_styles: int) -> Optional[list[int]]:
"""解析逗号分隔的发型序号字符串 → 去重排序列表。非法返回 None。
"1,2,3" → [1, 2, 3] "3,1" → [1, 3] "1" → [1]
"""
if raw is None or not isinstance(raw, str) or not raw.strip():
return None
try:
styles = []
for part in raw.split(","):
part = part.strip()
if not part:
continue
v = int(part)
if v < 1 or v > max_styles:
return None
styles.append(v)
if not styles:
return None
# 去重保持顺序
seen = set()
unique = [s for s in styles if not (s in seen or seen.add(s))] # type: ignore[func-returns-value]
return unique
except (ValueError, TypeError):
return None
# ---------------------------------------------------------------------------
# 通用图片请求 BodyJSON 方式,用于 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/6 共用实现
# ---------------------------------------------------------------------------
async def _face_measure_impl(image_file, image_url, image_base64):
"""接口1/6 共用实现:四庭七眼测量 + 标注图生成。返回 (ok_dict, err_dict)。"""
# 1. 三选一取图
raw, e = await resolve_image_bytes(image_file, image_url, image_base64)
if e is not None:
return None, e
# 2. 大小校验(≤ 1MB
if len(raw) > MAX_FILE_BYTES:
return None, err(1006, "文件超出 1 MB 限制")
# 3. 解码
image = cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
if image is None:
return None, 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 None, 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 None, err(1001, "无法识别人像")
# 6. 姿态校验
if not check_frontal_face(landmarks, w, h):
return None, 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, hair_mask=hair_mask)
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), None
except Exception as ex: # noqa: BLE001
logger.exception("接口1/6 处理异常")
return None, err(1007, f"处理失败:{ex}")
@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:四庭七眼测量标注"""
ok_data, err_data = await _face_measure_impl(image_file, image_url, image_base64)
return ok_data if ok_data is not None else err_data
# ---------------------------------------------------------------------------
# 接口 6:四庭七眼测量标注 v2(复刻接口1)
# ---------------------------------------------------------------------------
@app.post(
"/api/v1/face/measure-v2",
summary="接口6 四庭七眼测量标注 v2",
tags=["人脸分析"],
description=f"""
输入用户正面照,返回:
- 标注好四庭七眼数据的 **PNG 图片**(仅标注图层,不含人物)
- 四庭(顶庭/上庭/中庭/下庭)各段**厘米数值及占比**
- 七眼(眼宽/脸宽/两眼间距)**厘米数值及占比**
- 五个关键分界点的**原图像素坐标**(头顶/发际线/眉心/鼻翼下缘/下巴尖)
功能与接口1 完全一致,复刻实现。
{_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_v2(
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, 前缀)"),
):
"""接口6:四庭七眼测量标注 v2(复刻接口1"""
ok_data, err_data = await _face_measure_impl(image_file, image_url, image_base64)
return ok_data if ok_data is not None else err_data
# ---------------------------------------------------------------------------
# 接口 2C 端生发
# ---------------------------------------------------------------------------
@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`。
- **hair_style**(必填):发型序号,**逗号分隔多选**(如 `1,2,3`),最多不超过该性别的预设数量。
`female`1=ellipse, 2=flower, 3=heart, 4=straight, 5=wave
`male`1=ellipse, 2=inverse_arc, 3=m, 4=straight。越界/非法返回 `1007`。
- **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(必填)"),
hair_style: Optional[str] = Form(default=None, description="发型序号逗号分隔(必填),如 1,2,3。female:1-5 male:1-4"),
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. hair_style 必填校验(解析逗号分隔,越界 → 1007)
max_styles = {"female": 5, "male": 4}[gender]
hair_styles = _parse_hair_styles(hair_style, max_styles)
if hair_styles is None:
return err(1007, f"hair_style 必填且为 1..{max_styles} 的整数(逗号分隔),收到 {hair_style!r}")
# 3. 三选一取图
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, hair_styles)
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}")
# ---------------------------------------------------------------------------
# 接口 7C 端生发 v2add_hair2.json 工作流)
# ---------------------------------------------------------------------------
_WORKFLOW2_PATH = os.path.join(os.path.dirname(__file__), "add_hair2.json")
@app.post(
"/api/v1/hair/grow-v2",
summary="接口7 C端生发 v2add_hair2 工作流)",
tags=["生发"],
description=f"""
输入用户正面照 + **性别** + **发型序号**,使用 add_hair2.json 工作流生成指定发际线类型的预览图与生发图。
功能与接口2 完全一致,仅 ComfyUI 工作流不同。
{_image_fields_desc}
图片同时支持 `multipart/form-data` 文件上传(字段名 `image_file`)。
---
- **gender**(必填):`male` / `female`。决定返回的贴图集合(female 5 张 / male 4 张)。
非法或缺失返回 `1004`。
- **hair_style**(必填):`int`,发型序号。`female`1=ellipse, 2=flower, 3=heart, 4=straight, 5=wave
`male`1=ellipse, 2=inverse_arc, 3=m, 4=straight。越界返回 `1007`。
- **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},
]
},
}
}
},
},
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_v2(
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(必填)"),
hair_style: Optional[str] = Form(default=None, description="发型序号逗号分隔(必填),如 1,2,3。female:1-5 male:1-4"),
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. hair_style 必填校验(解析逗号分隔,越界 → 1007)
max_styles = {"female": 5, "male": 4}[gender]
hair_styles = _parse_hair_styles(hair_style, max_styles)
if hair_styles is None:
return err(1007, f"hair_style 必填且为 1..{max_styles} 的整数(逗号分隔),收到 {hair_style!r}")
# 3. 三选一取图
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, hair_styles, _WORKFLOW2_PATH)
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("接口7 处理异常")
return err(1007, f"处理失败:{ex}")
# ---------------------------------------------------------------------------
# 接口 3B 端生发
# ---------------------------------------------------------------------------
@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"}