xsl
48c9875381
- measure.py: MeasureResult 增 hairline_discarded 判定(顶庭<0.7cm)，弃用时 顶/上庭字段置null、face_total只算中下庭、hairline_source=discarded - annotation.py: 弃用时保留头顶线、去掉发际线、只标中/下庭 - app.py: 接口1/6 弃用分支 + eye1/7竖向范围改用眉心 + 接口1调试页(measure-debug) - 新增 static/test_interface1_debug.html 分步可视化页(9步原理)
2291 lines
112 KiB
Python
2291 lines
112 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 PIL import Image, ImageDraw, ImageFont
|
||
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"
|
||
|
||
# 运行期状态：模型就绪标志（/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`，单文件上传 |
|
||
| URL | `image_url` | 图片的完整 HTTP/HTTPS 地址 |
|
||
| base64 | `image_base64` | 需携带前缀，如 `data:image/jpeg;base64,xxxx` |
|
||
|
||
> 传 0 个或同时传多个，均返回错误码 `1007`。
|
||
|
||
## 图片要求
|
||
|
||
- 格式：**JPG / PNG**
|
||
- 分辨率：不限制
|
||
- 人脸数量：仅支持**单人**，多人返回错误码 `1005`
|
||
- 文件大小：不限制
|
||
|
||
## 统一响应结构
|
||
|
||
```json
|
||
{
|
||
"code": 0,
|
||
"message": "success",
|
||
"request_id": "唯一请求标识",
|
||
"data": {}
|
||
}
|
||
```
|
||
|
||
`code = 0` 表示成功，非 0 表示失败，`message` 为具体原因。
|
||
|
||
## 错误码一览
|
||
|
||
| code | 说明 |
|
||
|------|------|
|
||
| 1001 | 无法识别人像 |
|
||
| 1003 | 非正面照 / 角度过大 |
|
||
| 1004 | 性别标签无法判定 |
|
||
| 1005 | 检测到多张人脸（仅支持单人）|
|
||
| 1007 | 图片参数错误（未传或同时传多个）|
|
||
| 1008 | 图片格式不支持（仅 JPG / PNG）|
|
||
""",
|
||
)
|
||
|
||
app.mount("/static", StaticFiles(directory="static"), name="static")
|
||
|
||
# 不校验鉴权的路径前缀（供网关探测 / 文档 / 静态）
|
||
_AUTH_EXEMPT = ("/health", "/docs", "/openapi.json", "/redoc", "/static",
|
||
"/api/v1/debug", "/api/v1/redraw",
|
||
"/api/swapHair", "/hairColor")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# change_hair 代理路由（解决 CORS 问题）
|
||
# ---------------------------------------------------------------------------
|
||
_CHANGE_HAIR_BASE = os.getenv("CHANGE_HAIR_BASE", "http://127.0.0.1:8801")
|
||
|
||
|
||
@app.post("/api/swapHair/v1", tags=["change_hair"])
|
||
async def proxy_swap_hair(request: Request):
|
||
"""代理转发到 change_hair /api/swapHair/v1（换发型）"""
|
||
try:
|
||
import httpx
|
||
body = await request.body()
|
||
async with httpx.AsyncClient(timeout=300.0) as client:
|
||
resp = await client.post(f"{_CHANGE_HAIR_BASE}/api/swapHair/v1",
|
||
content=body,
|
||
headers={"Content-Type": "application/json"})
|
||
return JSONResponse(content=resp.json(), status_code=resp.status_code)
|
||
except Exception as e:
|
||
logger.exception("代理 swapHair 失败")
|
||
return err(1007, f"换发型服务异常：{e}")
|
||
|
||
|
||
@app.post("/hairColor/v2", tags=["change_hair"])
|
||
async def proxy_hair_color(request: Request):
|
||
"""代理转发到 change_hair /hairColor/v2（换发色）"""
|
||
try:
|
||
import httpx
|
||
body = await request.body()
|
||
async with httpx.AsyncClient(timeout=300.0) as client:
|
||
resp = await client.post(f"{_CHANGE_HAIR_BASE}/hairColor/v2",
|
||
content=body,
|
||
headers={"Content-Type": "application/json"})
|
||
return JSONResponse(content=resp.json(), status_code=resp.status_code)
|
||
except Exception as e:
|
||
logger.exception("代理 hairColor 失败")
|
||
return err(1007, f"换发色服务异常：{e}")
|
||
|
||
|
||
@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。
|
||
不限制文件大小，此处只负责取到字节。
|
||
"""
|
||
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 _rgba_png_b64(rgba) -> str:
|
||
"""(H,W,4) float32/uint8 RGBA 透明层 → PNG base64（保留 alpha 通道）。
|
||
|
||
用于发际线叠图/预览图：只含发际线曲线像素、背景透明，前端叠加到原图上显示。
|
||
"""
|
||
arr = np.asarray(rgba)
|
||
if arr.dtype != np.uint8:
|
||
arr = np.clip(arr, 0, 255).astype(np.uint8)
|
||
img = Image.fromarray(arr, mode="RGBA")
|
||
buf = BytesIO()
|
||
img.save(buf, format="PNG")
|
||
return base64.b64encode(buf.getvalue()).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
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 通用图片请求 Body（JSON 方式，用于 url / base64）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_image_fields_desc = (
|
||
"图片传参方式严格互斥，必须且只能选其一：\n"
|
||
"- **image_file**（multipart/form-data 上传）\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 共用实现
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _run_face_measure_data(image, variant="v1"):
|
||
"""接口1测量数值核心：detect → 姿态校验 → 头发/耳朵分割 → measure_face → eye1~eye7。
|
||
|
||
返回 (data_dict, result, hair_mask, ear_mask)，或检测/姿态失败时返回 None。
|
||
不含标注图（annotated_image_*），供接口5 容错复用；接口1 调用后再自行生成标注图。
|
||
任何分割/七眼计算异常均内部吞掉（七眼相关字段不出现），不影响主测量结果。
|
||
|
||
variant="v1"（接口1/接口5）：完整四庭七眼 + eye1~eye7（含头部端线段）。
|
||
variant="v6"（接口6）：去顶庭、不画端线、无 eye1~eye7。
|
||
"""
|
||
h, w = image.shape[:2]
|
||
from face_analysis.detector import detector
|
||
from face_analysis.pose import estimate_head_pose, check_frontal_face
|
||
from face_analysis.measure import measure_face
|
||
|
||
landmarks = detector.detect(image)
|
||
if landmarks is None:
|
||
return None
|
||
if not check_frontal_face(landmarks, w, h):
|
||
return None
|
||
head_pose = estimate_head_pose(landmarks, w, h)
|
||
|
||
# 头发/耳朵分割（方案 B，单次推理），失败传 None 由 measure 内部回退方案 A。
|
||
hair_mask = None
|
||
ear_mask = None
|
||
try:
|
||
from face_analysis.hair_segmenter import get_segmenter
|
||
from face_analysis.calibration import normalized_to_pixel
|
||
pxs = [normalized_to_pixel(p, w, h) for p in landmarks.landmark]
|
||
face_box = (min(p[0] for p in pxs), min(p[1] for p in pxs),
|
||
max(p[0] for p in pxs), max(p[1] for p in pxs))
|
||
hair_mask, ear_mask = get_segmenter().segment_hair_and_ears(image, face_box=face_box)
|
||
except Exception as seg_e: # noqa: BLE001
|
||
logger.warning("头发/耳朵分割失败，回退方案A：%s", seg_e)
|
||
|
||
result = measure_face(landmarks, hair_mask, w, h, head_pose=head_pose)
|
||
discarded = result.hairline_discarded
|
||
data = result.to_response()
|
||
vd = result.vertical
|
||
if variant == "v6":
|
||
if discarded:
|
||
# 发际线弃用：接口6 的上庭也依赖发际线，一并置 null；只保留中/下庭。
|
||
base_px = vd["middle_court_px"] + vd["lower_court_px"]
|
||
data["four_courts"]["upper_court_cm"] = None
|
||
data["four_courts"]["ratios"] = {
|
||
"upper_court": None,
|
||
"middle_court": round(vd["middle_court_px"] / base_px, 3),
|
||
"lower_court": round(vd["lower_court_px"] / base_px, 3),
|
||
}
|
||
data["four_courts"].pop("top_court_cm", None)
|
||
data["face_total_height_cm"] = round(
|
||
result.middle_cm + result.lower_cm, 2)
|
||
data["landmarks"]["hairline"] = None
|
||
else:
|
||
base_px = vd["upper_court_px"] + vd["middle_court_px"] + vd["lower_court_px"]
|
||
# 接口6 是三庭：去掉顶庭相关字段（top_court_cm / ratios.top_court / landmarks.hair_top）
|
||
data["four_courts"]["ratios"] = {
|
||
"upper_court": round(vd["upper_court_px"] / base_px, 3),
|
||
"middle_court": round(vd["middle_court_px"] / base_px, 3),
|
||
"lower_court": round(vd["lower_court_px"] / base_px, 3),
|
||
}
|
||
data["four_courts"].pop("top_court_cm", None)
|
||
data["face_total_height_cm"] = round(
|
||
result.upper_cm + result.middle_cm + result.lower_cm, 2)
|
||
# 注：landmarks.hair_top 保留返回（供前端/下游定位头顶），但顶庭数值、
|
||
# 占比、标注图仍按三庭处理，显示效果不变。
|
||
|
||
# 七眼段宽度（cm）。eye1=左耳外段 eye2=左脸颊 eye3=左眼 eye4=两眼间距 eye5=右眼 eye6=右脸颊 eye7=右耳外段。
|
||
# eye2~eye6（5段）只用内部分点，接口1/6 共用；eye1/eye7 需耳朵分割端线，仅接口1 有。
|
||
try:
|
||
epts = result.eyes["points"]
|
||
lcx, rcx = epts["left_cheek"][0], epts["right_cheek"][0]
|
||
pc = result.px_per_cm
|
||
inner_xs = [lcx, epts["left_outer"][0], epts["left_inner"][0],
|
||
epts["right_inner"][0], epts["right_outer"][0], rcx]
|
||
for i in range(5):
|
||
a, b = inner_xs[i], inner_xs[i + 1]
|
||
data["seven_eyes"][f"eye{i + 2}"] = (
|
||
None if (a is None or b is None) else round((b - a) / pc, 2))
|
||
if variant != "v6":
|
||
# 接口1 额外算 eye1/eye7（左/右耳外段），需耳朵分割端线。
|
||
# 竖向范围：发际线弃用时用眉心做上界（hair_top 不可靠），否则用头顶。
|
||
from face_analysis.annotation import _ear_edges_from_mask
|
||
top_y = (vd["brow_center"][1] if result.hairline_discarded
|
||
else vd["hair_top"][1])
|
||
head_l, head_r = _ear_edges_from_mask(
|
||
ear_mask, hair_mask,
|
||
top_y, vd["chin_tip"][1],
|
||
lcx, rcx, (lcx + rcx) / 2)
|
||
data["seven_eyes"]["eye1"] = (
|
||
None if (head_l is None) else round((lcx - head_l) / pc, 2))
|
||
data["seven_eyes"]["eye7"] = (
|
||
None if (head_r is None) else round((head_r - rcx) / pc, 2))
|
||
except Exception as seg_e: # noqa: BLE001
|
||
logger.warning("七眼段宽度计算失败：%s", seg_e)
|
||
|
||
return data, result, hair_mask, ear_mask
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 接口1 调试：分步可视化（每一步的中间产物图）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# 调试接口 9 张分步图的 key（与前端 STEPS 一一对应）
|
||
_DEBUG_STEP_KEYS = [
|
||
"input", "landmarks", "pose", "segmentation",
|
||
"hairline", "vertical", "seven_eyes", "scale", "final",
|
||
]
|
||
|
||
|
||
def _overlay_mask(image_bgr, mask, color, alpha=0.45):
|
||
"""在 BGR 图上把 mask 区域以 color(BGR) 半透明叠加。mask 为 bool/uint8。"""
|
||
out = image_bgr.copy()
|
||
m = np.asarray(mask).astype(bool)
|
||
if m.shape[:2] != out.shape[:2]:
|
||
return out
|
||
overlay = out[m]
|
||
# alpha 混合
|
||
overlay = (overlay * (1 - alpha) + np.array(color, dtype=np.float32) * alpha)
|
||
out[m] = np.clip(overlay, 0, 255).astype(np.uint8)
|
||
return out
|
||
|
||
|
||
_DEBUG_FONT_PATH = os.path.join(
|
||
os.path.dirname(__file__), "face_analysis", "fonts", "NotoSansCJKsc-Regular.otf")
|
||
_debug_font_cache = {}
|
||
|
||
|
||
def _debug_font(size):
|
||
f = _debug_font_cache.get(size)
|
||
if f is None:
|
||
f = ImageFont.truetype(_DEBUG_FONT_PATH, size)
|
||
_debug_font_cache[size] = f
|
||
return f
|
||
|
||
|
||
def _draw_text_cv2(img, text, org, color=(255, 255, 255), scale=None, thickness=None,
|
||
bg=True, anchor="lt"):
|
||
"""在 BGR 图上绘制文字（支持中文，用 PIL + 思源黑体）。org=(x,y)。
|
||
|
||
cv2.putText 不支持中文（会显示成问号），故统一改用 PIL 渲染。color 为 BGR 三元组。
|
||
anchor: lt=左上角对齐 org / lb=左下角 / ct=水平垂直居中。bg=True 时画黑色背景框。
|
||
"""
|
||
h, w = img.shape[:2]
|
||
s = min(w, h)
|
||
scale = scale if scale else max(0.4, s * 0.0016)
|
||
thickness = thickness if thickness else max(1, round(s * 0.0022))
|
||
# PIL 字号与 cv2 scale 大致对应（cv2 scale≈字号/30）
|
||
font_size = max(10, round(scale * 30))
|
||
font = _debug_font(font_size)
|
||
# BGR → RGB
|
||
rgb = (int(color[2]), int(color[1]), int(color[0]))
|
||
pil_img = Image.fromarray(cv2.cvtColor(img, cv2.COLOR_BGR2RGB))
|
||
draw = ImageDraw.Draw(pil_img)
|
||
bbox = draw.textbbox((0, 0), text, font=font)
|
||
tw, th = bbox[2] - bbox[0], bbox[3] - bbox[1]
|
||
x, y = org
|
||
if anchor == "lb":
|
||
text_y = y - th
|
||
elif anchor == "ct":
|
||
x = x - tw // 2
|
||
text_y = y - th // 2
|
||
else:
|
||
text_y = y
|
||
if bg:
|
||
pad = max(2, round(thickness * 1.2))
|
||
draw.rectangle(
|
||
[max(0, x - pad), max(0, text_y - pad),
|
||
min(w, x + tw + pad), min(h, text_y + th + pad)],
|
||
fill=(0, 0, 0))
|
||
# PIL text 的 y 是文字顶部基线，bbox 偏移需校正
|
||
draw.text((x, text_y - bbox[1]), text, fill=rgb, font=font)
|
||
img[:] = cv2.cvtColor(np.asarray(pil_img), cv2.COLOR_RGB2BGR)
|
||
return img
|
||
|
||
|
||
def _run_face_measure_data_debug(image):
|
||
"""接口1 调试：产出 9 步中间图 + 数值，逐步塞进返回 dict。
|
||
|
||
与 _run_face_measure_data 同链路，但每步把中间产物渲染成叠加图（JPG base64）
|
||
放进 data["steps"][key + "_base64"]，关键数值放进 data["debug"]。
|
||
检测/姿态失败时，仍返回已完成的步骤图 + 对应错误码，供前端展示「卡在哪一步」。
|
||
|
||
返回 (data, error_code_or_None, error_msg_or_None)。
|
||
"""
|
||
h, w = image.shape[:2]
|
||
from face_analysis.detector import detector
|
||
from face_analysis.pose import estimate_head_pose, check_frontal_face
|
||
from face_analysis.measure import measure_face, _brow_center
|
||
from face_analysis.calibration import (
|
||
normalized_to_pixel, estimate_scale_factor,
|
||
_iris_diameter_px, _eye_width_px, _lm_list,
|
||
AVG_IRIS_DIAMETER_CM, AVG_EYE_WIDTH_CM,
|
||
)
|
||
from face_analysis.face_mesh_landmarks import (
|
||
GLABELLA_9, GLABELLA_151, NOSE_BOTTOM, CHIN_TIP,
|
||
LEFT_EYE_OUTER, LEFT_EYE_INNER, RIGHT_EYE_INNER, RIGHT_EYE_OUTER,
|
||
LEFT_CHEEK, RIGHT_CHEEK, LEFT_POSITION, RIGHT_POSITION,
|
||
IRIS_LEFT_LEFT, IRIS_LEFT_RIGHT, IRIS_RIGHT_LEFT, IRIS_RIGHT_RIGHT,
|
||
PNP_INDICES,
|
||
)
|
||
from face_analysis.hair_segmenter import locate_hairline_by_segmentation
|
||
|
||
data = {"steps": {}, "debug": {"image_width": w, "image_height": h}}
|
||
steps = data["steps"]
|
||
dbg = data["debug"]
|
||
|
||
def put(key, bgr_img):
|
||
steps[key + "_base64"] = "data:image/jpeg;base64," + _jpg_b64(bgr_img)
|
||
|
||
# ① 输入原图
|
||
put("input", image)
|
||
|
||
# ② 人脸关键点检测
|
||
landmarks = detector.detect(image)
|
||
if landmarks is None:
|
||
dbg["num_landmarks"] = 0
|
||
return data, 1001, "无法识别人像"
|
||
lm = _lm_list(landmarks)
|
||
dbg["num_landmarks"] = len(lm)
|
||
|
||
vis_lm = image.copy()
|
||
# 先画全部 478 点（小白点）
|
||
s = min(w, h)
|
||
r_all = max(1, round(s * 0.0018))
|
||
for p in lm:
|
||
px = normalized_to_pixel(p, w, h)
|
||
cv2.circle(vis_lm, (int(px[0]), int(px[1])), r_all, (220, 220, 220), -1)
|
||
# 虹膜点 468~477（青色稍大）
|
||
r_iris = max(2, round(s * 0.0035))
|
||
for idx in [468, 469, 470, 471, 472, 473, 474, 475, 476, 477]:
|
||
if idx < len(lm):
|
||
px = normalized_to_pixel(lm[idx], w, h)
|
||
cv2.circle(vis_lm, (int(px[0]), int(px[1])), r_iris, (255, 200, 0), -1)
|
||
# 七眼 6 点 + 5 纵向点（红色 + 标号）
|
||
key_pts = {
|
||
"头顶(推算)": None, # 纵向点除眉心外由后续 measure 给出，这里只画能拿到的
|
||
"眉心": GLABELLA_9,
|
||
}
|
||
important = [
|
||
(GLABELLA_9, "眉间9"), (GLABELLA_151, "眉间151"), (NOSE_BOTTOM, "鼻翼下94"),
|
||
(CHIN_TIP, "下巴152"), (LEFT_EYE_OUTER, "左眼外33"), (LEFT_EYE_INNER, "左眼内133"),
|
||
(RIGHT_EYE_INNER, "右眼内362"), (RIGHT_EYE_OUTER, "右眼外263"),
|
||
(LEFT_CHEEK, "左脸234"), (RIGHT_CHEEK, "右脸454"),
|
||
]
|
||
r_imp = max(3, round(s * 0.005))
|
||
for idx, name in important:
|
||
px = normalized_to_pixel(lm[idx], w, h)
|
||
cv2.circle(vis_lm, (int(px[0]), int(px[1])), r_imp, (0, 0, 255), -1)
|
||
_draw_text_cv2(vis_lm, name, (int(px[0]) + r_imp + 2, int(px[1])),
|
||
color=(0, 255, 255), scale=max(0.35, s * 0.0013))
|
||
put("landmarks", vis_lm)
|
||
|
||
# ③ 头部姿态校验
|
||
head_pose = estimate_head_pose(lm, w, h) if hasattr(landmarks, "landmark") else estimate_head_pose(lm, w, h)
|
||
frontal = check_frontal_face(landmarks, w, h)
|
||
vis_pose = image.copy()
|
||
# 画 6 个 PnP 点（黄）
|
||
nose_tip_px = None
|
||
for idx in PNP_INDICES:
|
||
px = normalized_to_pixel(lm[idx], w, h)
|
||
cv2.circle(vis_pose, (int(px[0]), int(px[1])), max(3, round(s * 0.004)), (0, 255, 255), -1)
|
||
if idx == 1:
|
||
nose_tip_px = (int(px[0]), int(px[1]))
|
||
# 三轴箭头（鼻尖为原点）
|
||
if nose_tip_px is not None and head_pose is not None:
|
||
L = max(30, round(s * 0.08))
|
||
# yaw 绕 Y(竖轴) → 在屏幕上表现为左右；pitch 绕 X → 上下；roll 绕 Z → 面内旋转
|
||
yaw, pitch, roll = head_pose
|
||
import math
|
||
# 简化：用 roll 直接旋转 X/Y 轴示意，yaw 投影到横向、pitch 到纵向
|
||
cosr, sinr = math.cos(math.radians(roll)), math.sin(math.radians(roll))
|
||
# X 轴（红，向右）
|
||
cv2.arrowedLine(vis_pose, nose_tip_px,
|
||
(int(nose_tip_px[0] + L * cosr), int(nose_tip_px[1] + L * sinr)),
|
||
(0, 0, 255), max(2, round(s * 0.003)), tipLength=0.2)
|
||
# Y 轴（绿，向下）
|
||
cv2.arrowedLine(vis_pose, nose_tip_px,
|
||
(int(nose_tip_px[0] - L * sinr), int(nose_tip_px[1] + L * cosr)),
|
||
(0, 255, 0), max(2, round(s * 0.003)), tipLength=0.2)
|
||
# Z 轴（青，向内用圆圈示意）
|
||
cv2.circle(vis_pose, nose_tip_px, max(6, round(s * 0.012)), (255, 255, 0), max(1, round(s * 0.002)))
|
||
# 角度文字
|
||
txt = f"yaw={yaw:.1f} pitch={pitch:.1f} roll={roll:.1f}"
|
||
_draw_text_cv2(vis_pose, txt, (10, 10), color=(50, 255, 50),
|
||
scale=max(0.5, s * 0.0022))
|
||
_draw_text_cv2(vis_pose, f"frontal={'YES' if frontal else 'NO'}", (10, 40),
|
||
color=(50, 255, 50) if frontal else (50, 50, 255),
|
||
scale=max(0.5, s * 0.0022))
|
||
dbg["head_pose"] = {"yaw": round(yaw, 2), "pitch": round(pitch, 2),
|
||
"roll": round(roll, 2), "frontal": bool(frontal)}
|
||
put("pose", vis_pose)
|
||
|
||
if not frontal:
|
||
return data, 1003, "角度问题，请上传正面照"
|
||
|
||
# ④ 头发/耳朵分割
|
||
hair_mask = None
|
||
ear_mask = None
|
||
try:
|
||
from face_analysis.hair_segmenter import get_segmenter
|
||
pxs = [normalized_to_pixel(p, w, h) for p in lm]
|
||
face_box = (min(p[0] for p in pxs), min(p[1] for p in pxs),
|
||
max(p[0] for p in pxs), max(p[1] for p in pxs))
|
||
hair_mask, ear_mask = get_segmenter().segment_hair_and_ears(image, face_box=face_box)
|
||
except Exception as seg_e: # noqa: BLE001
|
||
logger.warning("[debug] 头发/耳朵分割失败：%s", seg_e)
|
||
|
||
vis_seg = image.copy()
|
||
if hair_mask is not None:
|
||
vis_seg = _overlay_mask(vis_seg, hair_mask, (0, 200, 0), alpha=0.45)
|
||
dbg["hair_pixels"] = int(np.asarray(hair_mask).astype(bool).sum())
|
||
if ear_mask is not None:
|
||
vis_seg = _overlay_mask(vis_seg, ear_mask, (200, 80, 0), alpha=0.5)
|
||
dbg["ear_pixels"] = int(np.asarray(ear_mask).astype(bool).sum())
|
||
_draw_text_cv2(vis_seg, "绿=头发(hair=17) 蓝=耳朵(ear=7/8)", (10, 10),
|
||
color=(50, 255, 50), scale=max(0.45, s * 0.0018))
|
||
put("segmentation", vis_seg)
|
||
|
||
# 主测量（复用 measure_face，内部含 ⑤ 纵向决策 + 七眼 + 尺度）
|
||
result = measure_face(landmarks, hair_mask, w, h, head_pose=head_pose)
|
||
v = result.vertical
|
||
dbg["hairline_source"] = result.hairline_source
|
||
|
||
# ⑤ 纵向定位（发际线/头顶）—— 复刻方案 B 的中轴线扫描
|
||
vis_hl = image.copy()
|
||
if hair_mask is not None:
|
||
vis_hl = _overlay_mask(vis_hl, hair_mask, (0, 180, 0), alpha=0.3)
|
||
brow_x, brow_y = _brow_center(lm, w, h)
|
||
# 中轴线 ±3px 列带高亮（黄）
|
||
cx = int(round(brow_x))
|
||
cv2.line(vis_hl, (max(0, cx - 3), 0), (max(0, cx - 3), h), (0, 230, 255), 1)
|
||
cv2.line(vis_hl, (min(w - 1, cx + 3), 0), (min(w - 1, cx + 3), h), (0, 230, 255), 1)
|
||
# 画 hairline_y / hair_top_y 两条横线
|
||
hairline_y = int(v["hairline"][1])
|
||
hair_top_y = int(v["hair_top"][1])
|
||
cv2.line(vis_hl, (0, hair_top_y), (w, hair_top_y), (255, 255, 0), max(2, round(s * 0.0025)))
|
||
cv2.line(vis_hl, (0, hairline_y), (w, hairline_y), (0, 100, 255), max(2, round(s * 0.0025)))
|
||
_draw_text_cv2(vis_hl, f"hair_top_y={hair_top_y}", (hair_top_y if hair_top_y < h - 40 else h - 40, 0),
|
||
color=(255, 255, 0), scale=max(0.4, s * 0.0015))
|
||
# 文字标注位置：hairline_y 行右侧
|
||
_draw_text_cv2(vis_hl, f"hairline_y={hairline_y} (source={result.hairline_source})",
|
||
(hairline_y, w - int(s * 0.5)), color=(0, 200, 255),
|
||
scale=max(0.4, s * 0.0015))
|
||
# 发际线弃用提示：顶庭 < 0.7cm 视为贴近头顶、不可靠
|
||
if result.hairline_discarded:
|
||
gap_cm = result.top_cm
|
||
_draw_text_cv2(vis_hl,
|
||
f"⚠️ 发际线离头顶仅 {gap_cm:.2f}cm (<0.7cm)，已弃用",
|
||
(10, 10), color=(40, 40, 255), scale=max(0.5, s * 0.0022))
|
||
put("hairline", vis_hl)
|
||
|
||
# ⑥ 四庭纵向点
|
||
vis_v = image.copy()
|
||
v_names = ["hair_top", "hairline", "brow_center", "nose_bottom", "chin_tip"]
|
||
v_labels = ["头顶", "发际线", "眉心", "鼻翼下缘", "下巴尖"]
|
||
v_court_px = [v["top_court_px"], v["upper_court_px"], v["middle_court_px"], v["lower_court_px"]]
|
||
court_names = ["顶庭", "上庭", "中庭", "下庭"]
|
||
court_cm = [result.top_cm, result.upper_cm, result.middle_cm, result.lower_cm]
|
||
vx0 = min(int(v[n][0]) for n in v_names)
|
||
for i, name in enumerate(v_names):
|
||
x, y = int(v[name][0]), int(v[name][1])
|
||
cv2.circle(vis_v, (x, y), max(3, round(s * 0.004)), (0, 0, 255), -1)
|
||
# 画一条横线
|
||
cv2.line(vis_v, (vx0 - max(20, round(s * 0.04)), y),
|
||
(min(w - 1, vx0 + int(s * 0.02)), y), (0, 200, 255), 1)
|
||
_draw_text_cv2(vis_v, v_labels[i], (min(w - 60, x + 8), y),
|
||
color=(50, 255, 255), scale=max(0.4, s * 0.0015))
|
||
# 各庭段高（竖向虚线 + cm 文字）
|
||
for i in range(4):
|
||
y_a = int(v[v_names[i]][1])
|
||
y_b = int(v[v_names[i + 1]][1])
|
||
lx = max(10, vx0 - max(40, round(s * 0.08)))
|
||
cv2.line(vis_v, (lx, y_a), (lx, y_b), (0, 255, 100), max(2, round(s * 0.0025)))
|
||
cv2.circle(vis_v, (lx, y_a), 3, (0, 255, 100), -1)
|
||
cv2.circle(vis_v, (lx, y_b), 3, (0, 255, 100), -1)
|
||
_draw_text_cv2(vis_v, f"{court_names[i]} {court_cm[i]:.2f}cm",
|
||
(lx - int(s * 0.18), (y_a + y_b) // 2),
|
||
color=(100, 255, 100), scale=max(0.4, s * 0.0015))
|
||
dbg["vertical_points"] = {n: {"x": int(v[n][0]), "y": int(v[n][1])} for n in v_names}
|
||
put("vertical", vis_v)
|
||
|
||
# ⑦ 七眼横向点
|
||
vis_e = image.copy()
|
||
epts = result.eyes["points"]
|
||
seven_keys = ["left_cheek", "left_outer", "left_inner", "right_inner", "right_outer", "right_cheek"]
|
||
seven_labels = ["左脸颊", "左眼外", "左眼内", "右眼内", "右眼外", "右脸颊"]
|
||
ey0 = min(int(epts[k][1]) for k in seven_keys)
|
||
for i, k in enumerate(seven_keys):
|
||
x, y = int(epts[k][0]), int(epts[k][1])
|
||
cv2.circle(vis_e, (x, y), max(3, round(s * 0.004)), (0, 0, 255), -1)
|
||
cv2.line(vis_e, (x, max(0, ey0 - 20)), (x, min(h - 1, ey0 + 20)),
|
||
(0, 200, 255), 1)
|
||
_draw_text_cv2(vis_e, seven_labels[i], (x, ey0 - max(25, round(s * 0.04))),
|
||
color=(50, 255, 255), scale=max(0.4, s * 0.0015), anchor="ct")
|
||
# 头部最左/最右端线（耳朵外缘）
|
||
try:
|
||
from face_analysis.annotation import _ear_edges_from_mask
|
||
lcx, rcx = epts["left_cheek"][0], epts["right_cheek"][0]
|
||
head_l, head_r = _ear_edges_from_mask(
|
||
ear_mask, hair_mask, v["hair_top"][1], v["chin_tip"][1],
|
||
lcx, rcx, (lcx + rcx) / 2)
|
||
if head_l is not None:
|
||
cv2.line(vis_e, (int(head_l), 0), (int(head_l), h), (255, 100, 255), max(1, round(s * 0.002)))
|
||
_draw_text_cv2(vis_e, "人头最左", (int(head_l), 10),
|
||
color=(255, 150, 255), scale=max(0.35, s * 0.0013))
|
||
if head_r is not None:
|
||
cv2.line(vis_e, (int(head_r), 0), (int(head_r), h), (255, 100, 255), max(1, round(s * 0.002)))
|
||
_draw_text_cv2(vis_e, "人头最右", (int(head_r), 10),
|
||
color=(255, 150, 255), scale=max(0.35, s * 0.0013))
|
||
dbg["head_left_x"] = head_l
|
||
dbg["head_right_x"] = head_r
|
||
except Exception as e: # noqa: BLE001
|
||
logger.warning("[debug] 七眼端线绘制失败：%s", e)
|
||
dbg["seven_eye_points"] = {k: {"x": int(epts[k][0]), "y": int(epts[k][1])} for k in seven_keys}
|
||
put("seven_eyes", vis_e)
|
||
|
||
# ⑧ 尺度校准
|
||
vis_sc = image.copy()
|
||
px_per_cm = result.px_per_cm
|
||
iris_px = _iris_diameter_px(lm, w, h)
|
||
if iris_px is not None and iris_px > 0:
|
||
# 画左右虹膜直径线（青）
|
||
for (li, ri) in [(IRIS_LEFT_LEFT, IRIS_LEFT_RIGHT), (IRIS_RIGHT_LEFT, IRIS_RIGHT_RIGHT)]:
|
||
p1 = normalized_to_pixel(lm[li], w, h)
|
||
p2 = normalized_to_pixel(lm[ri], w, h)
|
||
cv2.line(vis_sc, (int(p1[0]), int(p1[1])), (int(p2[0]), int(p2[1])),
|
||
(255, 200, 0), max(2, round(s * 0.004)))
|
||
cv2.circle(vis_sc, (int(p1[0]), int(p1[1])), max(2, round(s * 0.003)), (255, 200, 0), -1)
|
||
cv2.circle(vis_sc, (int(p2[0]), int(p2[1])), max(2, round(s * 0.003)), (255, 200, 0), -1)
|
||
method = "iris"
|
||
_draw_text_cv2(vis_sc, f"虹膜直径法: {iris_px:.1f}px / {AVG_IRIS_DIAMETER_CM}cm", (10, 10),
|
||
color=(255, 200, 0), scale=max(0.45, s * 0.0018))
|
||
else:
|
||
# 降级眼宽法（黄）
|
||
eye_px = _eye_width_px(lm, w, h)
|
||
method = "eye_width"
|
||
for (oi, ii) in [(LEFT_EYE_OUTER, LEFT_EYE_INNER), (RIGHT_EYE_INNER, RIGHT_EYE_OUTER)]:
|
||
p1 = normalized_to_pixel(lm[oi], w, h)
|
||
p2 = normalized_to_pixel(lm[ii], w, h)
|
||
cv2.line(vis_sc, (int(p1[0]), int(p1[1])), (int(p2[0]), int(p2[1])),
|
||
(0, 255, 255), max(2, round(s * 0.004)))
|
||
_draw_text_cv2(vis_sc, f"眼宽法(降级): {eye_px:.1f}px / {AVG_EYE_WIDTH_CM}cm", (10, 10),
|
||
color=(0, 255, 255), scale=max(0.45, s * 0.0018))
|
||
_draw_text_cv2(vis_sc, f"px_per_cm = {px_per_cm:.3f}", (10, 40),
|
||
color=(50, 255, 50), scale=max(0.5, s * 0.002))
|
||
dbg["px_per_cm"] = round(px_per_cm, 4)
|
||
dbg["scale_method"] = method
|
||
put("scale", vis_sc)
|
||
|
||
# 把 to_response 的数值并入 data（前端指标速览复用）
|
||
data.update(result.to_response())
|
||
# 七眼段宽
|
||
try:
|
||
pc = result.px_per_cm
|
||
inner_xs = [epts["left_cheek"][0], epts["left_outer"][0], epts["left_inner"][0],
|
||
epts["right_inner"][0], epts["right_outer"][0], epts["right_cheek"][0]]
|
||
data.setdefault("seven_eyes", {})
|
||
for i in range(5):
|
||
a, b = inner_xs[i], inner_xs[i + 1]
|
||
data["seven_eyes"][f"eye{i + 2}"] = (
|
||
None if (a is None or b is None) else round((b - a) / pc, 2))
|
||
from face_analysis.annotation import _ear_edges_from_mask as _eef
|
||
lcx, rcx = epts["left_cheek"][0], epts["right_cheek"][0]
|
||
# 弃用时用眉心做上界（与 _run_face_measure_data 一致）
|
||
top_y = v["brow_center"][1] if result.hairline_discarded else v["hair_top"][1]
|
||
head_l, head_r = _eef(ear_mask, hair_mask, top_y, v["chin_tip"][1],
|
||
lcx, rcx, (lcx + rcx) / 2)
|
||
data["seven_eyes"]["eye1"] = None if head_l is None else round((lcx - head_l) / pc, 2)
|
||
data["seven_eyes"]["eye7"] = None if head_r is None else round((head_r - rcx) / pc, 2)
|
||
except Exception as seg_e: # noqa: BLE001
|
||
logger.warning("[debug] 七眼段宽计算失败：%s", seg_e)
|
||
|
||
# ⑨ 最终标注图（原图 + 标注层叠加）
|
||
try:
|
||
from face_analysis.annotation import create_annotated_image
|
||
annotated = create_annotated_image(image, result, ear_mask=ear_mask, hair_mask=hair_mask)
|
||
anno_rgba = np.asarray(annotated)
|
||
# 叠加到原图
|
||
vis_final = image.copy()
|
||
alpha = anno_rgba[:, :, 3:4].astype(np.float32) / 255.0
|
||
vis_final = (vis_final.astype(np.float32) * (1 - alpha)
|
||
+ anno_rgba[:, :, :3].astype(np.float32) * alpha)
|
||
vis_final = np.clip(vis_final, 0, 255).astype(np.uint8)
|
||
put("final", vis_final)
|
||
except Exception as e: # noqa: BLE001
|
||
logger.warning("[debug] 标注图叠加失败：%s", e)
|
||
|
||
return data, None, None
|
||
|
||
|
||
async def _face_measure_impl(image_file, image_url, image_base64, variant="v1"):
|
||
"""接口1/6 共用实现：四庭七眼测量 + 标注图生成。返回 (ok_dict, err_dict)。
|
||
|
||
variant="v1"（接口1）：完整四庭七眼 + 头部端线。
|
||
variant="v6"（接口6）：去掉头顶/顶庭、不画人头最左/最右端线、数据去顶庭。
|
||
"""
|
||
# 1. 三选一取图
|
||
raw, e = await resolve_image_bytes(image_file, image_url, image_base64)
|
||
if e is not None:
|
||
return None, e
|
||
|
||
# 2. 解码（不限制文件大小 / 分辨率）
|
||
image = cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
|
||
if image is None:
|
||
return None, err(1008, "图片格式不支持（仅 JPG / PNG）")
|
||
|
||
try:
|
||
ret = _run_face_measure_data(image, variant=variant)
|
||
if ret is None:
|
||
# 区分错误码：未检出人脸 vs 非正面。重新检测一次以判断。
|
||
from face_analysis.detector import detector
|
||
from face_analysis.pose import check_frontal_face
|
||
if detector.detect(image) is None:
|
||
return None, err(1001, "无法识别人像")
|
||
return None, err(1003, "角度问题，请上传正面照")
|
||
data, result, hair_mask, ear_mask = ret
|
||
|
||
# 标注图（接口1/6 才需要；接口5 复用 _run_face_measure_data 时不生成）
|
||
from face_analysis.annotation import create_annotated_image
|
||
annotated = create_annotated_image(
|
||
image, result, ear_mask=ear_mask, hair_mask=hair_mask, variant=variant)
|
||
buf = BytesIO()
|
||
annotated.save(buf, format="PNG")
|
||
|
||
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%`，透明底
|
||
- 字号/线宽/虚线/箭头按图片短边自适应缩放
|
||
- 四庭（名 + 数值带cm + 百分比 三行）在图片**左侧**呈现，七眼段宽**上下穿插**展示（数值带cm，下方另起一行标占头宽百分比）
|
||
- 横线/竖线渐变消失并略超出端点；段宽/庭高用虚线 + 实心三角双箭头标示
|
||
- 竖线含人头最左/最右端线（取自头发分割轮廓），共 8 线 7 段
|
||
""",
|
||
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},
|
||
"eye1": 3.44, "eye2": 3.44, "eye3": 3.44, "eye4": 3.44,
|
||
"eye5": 3.44, "eye6": 3.44, "eye7": 3.44,
|
||
},
|
||
"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）"),
|
||
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
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 接口1 调试：分步可视化（每一步中间产物图 + 原理）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.post("/api/v1/face/measure-debug", include_in_schema=False)
|
||
async def face_measure_debug(
|
||
image_file: Optional[UploadFile] = File(default=None),
|
||
image_url: Optional[str] = Form(default=None),
|
||
image_base64: Optional[str] = Form(default=None),
|
||
):
|
||
"""接口1 调试：返回算法每一步的中间产物图（data.steps.*_base64）+ 数值（data.debug）。
|
||
|
||
与正式接口同链路，但额外产出 9 张分步叠加图（输入/关键点/姿态/分割/发际线/
|
||
四庭/七眼/尺度/最终标注），供调试页分步可视化。错误时仍返回已完成的步骤图。
|
||
"""
|
||
raw, e = await resolve_image_bytes(image_file, image_url, image_base64)
|
||
if e is not None:
|
||
return e
|
||
image = cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
|
||
if image is None:
|
||
return err(1008, "图片格式不支持（仅 JPG / PNG）")
|
||
try:
|
||
data, code, msg = _run_face_measure_data_debug(image)
|
||
if code is not None:
|
||
# 仍带分步图返回，前端可展示卡在哪一步
|
||
return {"code": code, "message": msg,
|
||
"request_id": "mock-request-id", "data": data}
|
||
return ok(data)
|
||
except Exception as ex: # noqa: BLE001
|
||
logger.exception("接口1 调试处理异常")
|
||
return err(1007, f"处理失败：{ex}")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 接口 6：四庭七眼测量标注 v2（复刻接口1）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.post(
|
||
"/api/v1/face/measure-v2",
|
||
summary="接口6 四庭七眼测量标注 v2",
|
||
tags=["人脸分析"],
|
||
description=f"""
|
||
输入用户正面照，返回：
|
||
- 标注好四庭七眼数据的 **PNG 图片**（仅标注图层，不含人物）
|
||
- 三庭（上庭/中庭/下庭）各段**厘米数值及占比**（**不含顶庭**）
|
||
- 五眼段宽（eye2~eye6：左脸颊/左眼/两眼间距/右眼/右脸颊）**厘米数值**，另含眼宽/脸宽/两眼间距
|
||
- 四个关键分界点的**原图像素坐标**（发际线/眉心/鼻翼下缘/下巴尖）
|
||
|
||
基于接口1 的变体，与接口1 的差异：
|
||
- **去顶庭**：不画头顶横线、不返回顶庭数据；`face_total_height_cm` 为三庭之和
|
||
- **竖线范围**：纵向竖线从发际线画到下巴尖（接口1 为头顶→下巴尖）
|
||
- **不画人头最左/最右端线**：仅七眼 6 点共 5 段标尺（eye2~eye6），不取头发轮廓端线（接口1 为 8 线 7 段 eye1~eye7）
|
||
|
||
其余（箭头/虚线/字体/单位cm/七眼数据）与接口1 一致。
|
||
|
||
{_image_fields_desc}
|
||
|
||
图片同时支持 `multipart/form-data` 文件上传（字段名 `image_file`），或 JSON Body 传 `image_url` / `image_base64`。
|
||
|
||
---
|
||
|
||
**坐标说明**：所有坐标以原图像素为基准，原点为图片左上角，x 向右，y 向下。
|
||
|
||
**标注图片 UI 规范**（真实版本生效）：
|
||
- 字体/线条/箭头颜色：`#FFFFFF 100%`，透明底
|
||
- 字号/线宽/虚线/箭头按图片短边自适应缩放
|
||
- 三庭（名 + 数值带cm + 百分比 三行）在图片**左侧**呈现，七眼段宽**上下穿插**展示（数值带cm，下方另起一行标占头宽百分比）
|
||
- 段宽/庭高用虚线 + 实心三角双箭头标示
|
||
""",
|
||
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": 10.32,
|
||
"four_courts": {
|
||
"upper_court_cm": 3.44,
|
||
"middle_court_cm": 3.44,
|
||
"lower_court_cm": 3.44,
|
||
"ratios": {
|
||
"upper_court": 0.333,
|
||
"middle_court": 0.333,
|
||
"lower_court": 0.333,
|
||
},
|
||
},
|
||
"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},
|
||
"eye2": 3.44, "eye3": 3.44, "eye4": 3.44, "eye5": 3.44, "eye6": 3.44,
|
||
},
|
||
"landmarks": {
|
||
"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）"),
|
||
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, variant="v6")
|
||
return ok_data if ok_data is not None else err_data
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 接口 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`，不排序）
|
||
|
||
> **female 走「换发型」模式**：生发图 `grown_image_base64` 由换发型（change_hair）
|
||
> + Flux-2 整帧重绘（= 接口12 final 管线，整帧美颜+整帧重绘）生成，其余参数用固化默认值。
|
||
> **male 仍走原生发（ComfyUI add_hair）管线**。入参与返回结构不变。
|
||
> female 依赖 change_hair 与 ComfyUI(:8188) 均在跑。
|
||
|
||
{_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）"),
|
||
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的文本"),
|
||
flux_model: Optional[str] = Form(default=None, description="Flux 模型文件名（切换模型用）。None=工作流默认；如 flux-2-klein-9b-Q5_K_M.gguf / flux-2-klein-9b-Q4_K_M.gguf / flux2.0/flux-2-klein-9b-fp8.safetensors"),
|
||
redraw_max_side: Optional[int] = Form(default=None, description="重绘压图长边像素。None=默认896；0=不缩图（原图直送）；其他如 768/640/1024"),
|
||
):
|
||
# 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
|
||
|
||
image = cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
|
||
if image is None:
|
||
return err(1008, "图片格式不支持（仅 JPG / PNG）")
|
||
|
||
try:
|
||
from fastapi.concurrency import run_in_threadpool
|
||
|
||
# 预览 + 生发/换发型 都是阻塞且较慢，放线程池避免卡住事件循环。
|
||
# female：换发型 + Flux-2 整帧重绘（= 接口12 final 管线）；male：仍走原生发管线。
|
||
if gender == "female":
|
||
from hairline.service import generate_grow_results_swap
|
||
items = await run_in_threadpool(
|
||
generate_grow_results_swap, image, hair_styles, _V2_FINAL_DEFAULTS,
|
||
redraw_max_side=redraw_max_side, unet_name=flux_model)
|
||
else:
|
||
from hairline.service import generate_grow_results
|
||
items = await run_in_threadpool(
|
||
generate_grow_results, image, gender, use_mask, prompt, hair_styles,
|
||
unet_name=flux_model)
|
||
if items is None:
|
||
return err(1001, "无法识别人像")
|
||
|
||
results = []
|
||
for p in items:
|
||
results.append({
|
||
"image_base64": _rgba_png_b64(p["overlay"]), # 发际线曲线透明 PNG
|
||
"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}")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 调试接口：接口2 女性生发 分步计时
|
||
# ---------------------------------------------------------------------------
|
||
@app.post(
|
||
"/api/v1/debug/grow-timing",
|
||
summary="调试-接口2女性生发分步计时",
|
||
tags=["调试"],
|
||
include_in_schema=False,
|
||
)
|
||
async def debug_grow_timing(
|
||
image_file: Optional[UploadFile] = File(default=None),
|
||
image_url: Optional[str] = Form(default=None),
|
||
image_base64: Optional[str] = Form(default=None),
|
||
hair_style: str = Form(default="2", description="发型序号(花瓣=2)，逗号分隔多选"),
|
||
webui_steps: Optional[int] = Form(default=None, description="swapHair webui img2img 采样步数，None=服务端默认(15)，可填10/15/20/25对比"),
|
||
redraw_max_side: Optional[int] = Form(default=None, description="ComfyUI重绘分辨率(长边像素)。None=默认896；0=原图不缩；其他如640/768/1024"),
|
||
redraw_prompt: Optional[str] = Form(default=None, description="ComfyUI重绘提示词，None=默认'填充遮罩区域的头发'"),
|
||
):
|
||
"""单图跑接口2女性生发，返回每个步骤的耗时 + 结果图，用于定位性能瓶颈。
|
||
|
||
步骤拆分：
|
||
1. extract_context：人脸关键点检测 + 头发分割 + 发际线几何
|
||
2. [每个发型] generate_hairline_redraw：
|
||
2a. compute_mask：发际线遮罩计算
|
||
2b. _call_swap：调 change_hair 换发型（内含 webui SD1.5 推理，远程或本机）
|
||
2c. _composite：接缝融合（多频段/羽化）
|
||
3. [每个发型] _call_local_redraw：调本机 ComfyUI 用 Flux.2 重绘
|
||
"""
|
||
import time as _time
|
||
from fastapi.concurrency import run_in_threadpool
|
||
|
||
raw, e = await resolve_image_bytes(image_file, image_url, image_base64)
|
||
if e is not None:
|
||
return e
|
||
image = cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
|
||
if image is None:
|
||
return err(1008, "图片格式不支持（仅 JPG / PNG）")
|
||
|
||
try:
|
||
max_styles = 5
|
||
hair_styles = _parse_hair_styles(hair_style, max_styles)
|
||
if hair_styles is None:
|
||
return err(1007, f"hair_style 必须为 1..{max_styles}")
|
||
|
||
t_total0 = _time.perf_counter()
|
||
timings = {"total_ms": 0, "extract_context_ms": 0, "per_hairstyle": []}
|
||
|
||
# 步骤1: extract_context
|
||
t0 = _time.perf_counter()
|
||
from hairline.service import extract_context as _ec, _call_local_redraw, _REDRAW_MAX_SIDE # noqa
|
||
from face_analysis.hairline_grow import generate_hairline_redraw, NoFaceError # noqa
|
||
from face_analysis.head_mask import SEGFORMER_HAIR # noqa
|
||
ctx = await run_in_threadpool(_ec, image)
|
||
timings["extract_context_ms"] = int((_time.perf_counter() - t0) * 1000)
|
||
if ctx is None:
|
||
return err(1001, "无法识别人像")
|
||
|
||
hair_mask_reuse = (ctx["parse_map"] == SEGFORMER_HAIR)
|
||
h, w = image.shape[:2]
|
||
eff_side = _REDRAW_MAX_SIDE if redraw_max_side is None else redraw_max_side
|
||
redraw_img, hair_mask_redraw = image, hair_mask_reuse
|
||
downscale_info = None
|
||
if eff_side > 0 and max(h, w) > eff_side:
|
||
from hairline.service import _downscale_max_side
|
||
redraw_img, _rs = _downscale_max_side(image, eff_side)
|
||
_nh, _nw = redraw_img.shape[:2]
|
||
if hair_mask_redraw is not None:
|
||
hair_mask_redraw = cv2.resize(hair_mask_reuse.astype(np.uint8), (_nw, _nh),
|
||
interpolation=cv2.INTER_NEAREST).astype(bool)
|
||
downscale_info = {"from": f"{w}x{h}", "to": f"{_nw}x{_nh}", "max_side": eff_side}
|
||
|
||
textures_map = None
|
||
from hairline.service import get_texture_map, _FEMALE_KEY_TO_CHANG, load_texture_rgba, build_overlay_layer, load_ext_mesh
|
||
textures = get_texture_map()["female"]
|
||
items = [(s, textures[s - 1]) for s in hair_styles]
|
||
|
||
for order, (key, white_path) in items:
|
||
hs_t0 = _time.perf_counter()
|
||
entry = {"hairline_type": key, "order": order}
|
||
chang_id = _FEMALE_KEY_TO_CHANG.get(key)
|
||
entry["chang_id"] = chang_id
|
||
entry["ok"] = False
|
||
entry["error"] = None
|
||
entry["grown_b64"] = None
|
||
if chang_id is None:
|
||
entry["error"] = f"无对应 chang_id"
|
||
timings["per_hairstyle"].append(entry)
|
||
continue
|
||
try:
|
||
# 2a/2b/2c: generate_hairline_redraw (内部含 mask+swap+blend)
|
||
t0 = _time.perf_counter()
|
||
data = await run_in_threadpool(
|
||
generate_hairline_redraw, redraw_img, chang_id,
|
||
hair_mask=hair_mask_redraw, webui_steps=webui_steps, **_V2_FINAL_DEFAULTS)
|
||
t_redraw_pipeline = _time.perf_counter() - t0
|
||
_tm = data.get("timings_ms") or {}
|
||
entry["mask_ms"] = _tm.get("mask", 0)
|
||
entry["swap_ms"] = _tm.get("swap", 0)
|
||
entry["blend_ms"] = _tm.get("blend", 0)
|
||
entry["redraw_pipeline_ms"] = int(t_redraw_pipeline * 1000)
|
||
|
||
steps = data.get("steps") or {}
|
||
final_b64 = steps.get("final_base64") or ""
|
||
mask_b64 = steps.get("redraw_band_mask_base64") or ""
|
||
if not final_b64 or not mask_b64:
|
||
entry["error"] = f"final/遮罩缺失(final={len(final_b64)} mask={len(mask_b64)})"
|
||
timings["per_hairstyle"].append(entry)
|
||
continue
|
||
if final_b64.startswith("data:"):
|
||
final_b64 = final_b64.split(",", 1)[1]
|
||
if mask_b64.startswith("data:"):
|
||
mask_b64 = mask_b64.split(",", 1)[1]
|
||
|
||
# 3: ComfyUI 重绘
|
||
t0 = _time.perf_counter()
|
||
# max_side: 0 或 None 都让 _call_local_redraw 用默认逻辑(外层已控制分辨率)
|
||
_ms = redraw_max_side if redraw_max_side is not None and redraw_max_side > 0 else None
|
||
grown_png = await run_in_threadpool(
|
||
_call_local_redraw,
|
||
base64.b64decode(final_b64), base64.b64decode(mask_b64),
|
||
max_side=_ms, prompt=redraw_prompt)
|
||
entry["comfyui_redraw_ms"] = int((_time.perf_counter() - t0) * 1000)
|
||
if grown_png:
|
||
entry["grown_b64"] = "data:image/jpeg;base64," + _png_to_jpg_b64(grown_png)
|
||
entry["ok"] = True
|
||
else:
|
||
entry["error"] = "ComfyUI 重绘返回空"
|
||
except NoFaceError:
|
||
entry["error"] = "未检出人脸"
|
||
except Exception as ex: # noqa: BLE001
|
||
entry["error"] = str(ex)[:150]
|
||
entry["hairstyle_total_ms"] = int((_time.perf_counter() - hs_t0) * 1000)
|
||
timings["per_hairstyle"].append(entry)
|
||
|
||
timings["total_ms"] = int((_time.perf_counter() - t_total0) * 1000)
|
||
timings["downscale"] = downscale_info
|
||
timings["image_size"] = f"{w}x{h}"
|
||
return ok(timings)
|
||
except Exception as ex: # noqa: BLE001
|
||
logger.exception("debug/grow-timing 异常")
|
||
return err(1007, f"处理失败：{ex}")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 接口 7：C 端生发 v2 —— 已弃用（add_hair2.json 用 Klein-9b 大模型，会把常驻的
|
||
# Klein-4b/Flux 挤出显存，导致接口2/3/5 耗时抖动；且业务已不再调用）。
|
||
# 保留路由返回明确错误，避免老客户端拿到裸 404。
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.post("/api/v1/hair/grow-v2", include_in_schema=False, deprecated=True)
|
||
async def hair_grow_v2():
|
||
"""接口7 已弃用：请改用 /api/v1/hair/grow（接口2）。"""
|
||
return err(1007, "接口7（/api/v1/hair/grow-v2）已弃用，请使用 /api/v1/hair/grow")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 接口 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）"),
|
||
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
|
||
|
||
marked = cv2.imdecode(np.frombuffer(marked_raw, np.uint8), cv2.IMREAD_COLOR)
|
||
if marked is None:
|
||
return err(1008, "图片格式不支持（仅 JPG / PNG）")
|
||
|
||
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）"),
|
||
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"""
|
||
输入用户照片 + 性别 + 多选发型，对每个选中发型返回 middle/high/low 三档发际线叠图与生发图，
|
||
并标注最合适发际线的面部中间点坐标。
|
||
|
||
{_image_fields_desc}
|
||
|
||
图片同时支持 `multipart/form-data` 文件上传（字段名 `image_file`）。
|
||
|
||
---
|
||
|
||
**入参**（同接口2：先选性别，再多选发型）：
|
||
- 必填 `gender`（`male`/`female`），决定发型集合（female 5 / male 4）。
|
||
- 必填 `hair_style`（发型序号，逗号分隔如 `1,2,3`），决定返回哪些发际线类型。缺失/越界/非法返回 `1007`。
|
||
`female`：1=ellipse,2=flower,3=heart,4=straight,5=wave；`male`：1=ellipse,2=inverse_arc,3=m,4=straight。
|
||
- 可选 `use_mask` / `prompt`：同接口2 的生发控制参数。
|
||
注：生发黑模板固定取 `hairline_texture_black/`（middle 档），即三档叠图分别用各自贴图、但生发目标固定 middle。
|
||
- 可选 `generate_grow_image`（默认 `true`）：是否生成生发效果图（ComfyUI 生发，全流程最耗时）。
|
||
`false` 时跳过生发，各发型 `grown_image_*` 恒为 `null`，仅返回三档发际线叠图与中心点，大幅降低耗时。
|
||
|
||
**返回说明**：
|
||
|
||
- `hairline_images`：**选中发型**的列表，数量 = 所选发型数，`order` = 发型序号。每项含：
|
||
- `image_middle_url` / `image_high_url` / `image_low_url`：该发型 middle/high/low 三档发际线叠图（同接口2预览）。
|
||
- `grown_image_url`：该发型的**生发图**（生发失败时为 `null`）。
|
||
- `hairline_type`：发际线类型 key。
|
||
worker 返回 `*_base64`，网关落盘后改写为 `*_url`。
|
||
- `best_hairline_center_point`：**首个选中发型**的 **middle 档**发际线曲线**面部中间点**坐标，
|
||
以**原图像素**为基准（左上角为原点，x 向右，y 向下）。
|
||
- `high_hairline_center_point`：同上，**high 档**发际线中点。
|
||
- `low_hairline_center_point`：同上，**low 档**发际线中点。
|
||
""",
|
||
responses={
|
||
200: {
|
||
"description": "成功",
|
||
"content": {
|
||
"application/json": {
|
||
"example": {
|
||
"code": 0,
|
||
"message": "success",
|
||
"request_id": "mock-request-id",
|
||
"data": {
|
||
"hairline_images": [
|
||
{"hairline_type": "ellipse", "image_middle_base64": "iVBORw0KGgo...", "image_high_base64": "iVBORw0KGgo...", "image_low_base64": "iVBORw0KGgo...", "grown_image_base64": "iVBORw0KGgo...", "order": 1},
|
||
{"hairline_type": "flower", "image_middle_base64": "iVBORw0KGgo...", "image_high_base64": "iVBORw0KGgo...", "image_low_base64": "iVBORw0KGgo...", "grown_image_base64": None, "order": 2},
|
||
],
|
||
"best_hairline_center_point": {"x": 540, "y": 430},
|
||
"high_hairline_center_point": {"x": 540, "y": 380},
|
||
"low_hairline_center_point": {"x": 540, "y": 480},
|
||
"face_measure": {
|
||
"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},
|
||
"eye1": 3.44, "eye2": 3.44, "eye3": 3.44, "eye4": 3.44,
|
||
"eye5": 3.44, "eye6": 3.44, "eye7": 3.44,
|
||
},
|
||
"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},
|
||
},
|
||
"hairline_source": "segmentation",
|
||
"head_pose": {"yaw": -1.39, "pitch": 2.49, "roll": -0.06},
|
||
},
|
||
},
|
||
}
|
||
}
|
||
},
|
||
},
|
||
400: {
|
||
"description": "参数错误 / 图片识别失败",
|
||
"content": {
|
||
"application/json": {
|
||
"example": {"code": 1001, "message": "无法识别人像", "request_id": "x", "data": None}
|
||
}
|
||
},
|
||
},
|
||
},
|
||
)
|
||
async def hairline_generate(
|
||
image_file: Optional[UploadFile] = File(default=None, description="上传图片文件（JPG/PNG）"),
|
||
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"),
|
||
use_mask: bool = Form(default=True, description="生发是否启用 inpaint 遮罩（同接口2，测试对比用）"),
|
||
prompt: str = Form(default="填充遮罩区域的头发", description="ComfyUI 提示词（同接口2），会替换工作流节点60的文本"),
|
||
generate_grow_image: bool = Form(default=True, description="是否生成生发效果图（ComfyUI 生发，最耗时）。默认 true 出图；false 时跳过生发，各发型 grown_image 恒为 null，仅返回三档发际线叠图与中心点"),
|
||
):
|
||
if gender not in ("male", "female"):
|
||
return err(1004, "gender 必填且只能为 male / female")
|
||
|
||
# hair_style 必填（同接口2）：解析逗号分隔，缺失/越界/非法 → 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}")
|
||
|
||
raw, e = await resolve_image_bytes(image_file, image_url, image_base64)
|
||
if e is not None:
|
||
return e
|
||
|
||
image = cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
|
||
if image is None:
|
||
return err(1008, "图片格式不支持（仅 JPG / PNG）")
|
||
|
||
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, hair_styles, use_mask, prompt,
|
||
generate_grow_image=generate_grow_image)
|
||
if res is None:
|
||
return err(1001, "无法识别人像")
|
||
|
||
hairline_images = []
|
||
for it in res["images"]:
|
||
ov = it["overlays"]
|
||
hairline_images.append({
|
||
"hairline_type": it["hairline_type"],
|
||
"image_middle_base64": _rgba_png_b64(ov["middle"]), # 发际线曲线透明 PNG middle 档
|
||
"image_high_base64": _rgba_png_b64(ov["high"]), # 发际线曲线透明 PNG high 档
|
||
"image_low_base64": _rgba_png_b64(ov["low"]), # 发际线曲线透明 PNG low 档
|
||
"grown_image_base64": (_png_to_jpg_b64(it["grown_png"]) # 生发图 JPG（失败为 null）
|
||
if it.get("grown_png") else None),
|
||
"order": it["order"],
|
||
})
|
||
c = res.get("best_centers") or {}
|
||
def _pt(p):
|
||
return ({"x": p[0], "y": p[1]} if p else None)
|
||
data = {
|
||
"hairline_images": hairline_images,
|
||
"best_hairline_center_point": _pt(c.get("middle")),
|
||
"high_hairline_center_point": _pt(c.get("high")),
|
||
"low_hairline_center_point": _pt(c.get("low")),
|
||
}
|
||
|
||
# face_measure：复用接口1测量数值（四庭/七眼 eye1~eye7/landmarks/姿态），不含标注图。
|
||
# 独立流程，容错：测量失败（无人脸/非正面/分割失败）→ null，不影响发际线主结果。
|
||
try:
|
||
fm = _run_face_measure_data(image, variant="v1")
|
||
data["face_measure"] = fm[0] if fm is not None else None
|
||
except Exception as fm_e: # noqa: BLE001
|
||
logger.warning("接口5 face_measure 失败：%s", fm_e)
|
||
data["face_measure"] = None
|
||
|
||
return ok(data)
|
||
except Exception as ex: # noqa: BLE001
|
||
logger.exception("接口5 处理异常")
|
||
return err(1007, f"处理失败：{ex}")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 接口 9：头发遮罩生成
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.post(
|
||
"/api/v1/head/mask",
|
||
summary="接口9 头发遮罩生成",
|
||
tags=["人脸分析"],
|
||
description=f"""
|
||
输入一张含人头的照片，生成头发遮罩，并返回**每一步的可视化图**（供对比调试）：
|
||
|
||
1. MediaPipe 关键点 → 底部分割线（关键点 21,68,104,69,108,151,337,299,333,298,251 的连线，
|
||
21 水平延伸到最左边、251 延伸到最右边）。
|
||
2. 分割线以上为「上半区」。
|
||
3. 头发分割（**BiSeNet** 与 **SegFormer** 两套）；从每列最顶端头发向下填充到分割线，
|
||
得到**含额头**的闭合区域（头发+额头，不割断）。
|
||
4. 外缘朝中心点 151 内缩 **erode_cm（默认 1.2cm，可调）**（虹膜标定换算像素）、底线不动 → 最终遮罩。
|
||
|
||
{_image_fields_desc}
|
||
|
||
返回 `data` 内含 `steps_common`（关键点/分割线、上半区）与 `bisenet` / `segformer`
|
||
两组结果（各含 `hair_mask` / `closed_region` / `final_overlay` / `mask`）。经网关时所有
|
||
`*_base64` 图片字段会被落盘改写为 `*_url`。
|
||
""",
|
||
responses={
|
||
200: {
|
||
"description": "成功",
|
||
"content": {
|
||
"application/json": {
|
||
"example": {
|
||
"code": 0,
|
||
"message": "success",
|
||
"request_id": "mock-request-id",
|
||
"data": {
|
||
"px_per_cm": 48.12,
|
||
"erode_cm": 1.2,
|
||
"erode_px": 58,
|
||
"image_size": {"width": 1080, "height": 1440},
|
||
"steps_common": {"landmarks_baseline_url": SAMPLE_IMAGE_URL,
|
||
"upper_region_url": SAMPLE_IMAGE_URL},
|
||
"bisenet": {"hair_pixels": 123456, "closed_pixels": 110000, "mask_pixels": 98765,
|
||
"hair_mask_url": SAMPLE_IMAGE_URL,
|
||
"closed_region_url": SAMPLE_IMAGE_URL,
|
||
"final_overlay_url": SAMPLE_IMAGE_URL,
|
||
"mask_url": SAMPLE_IMAGE_URL},
|
||
"segformer": {"hair_pixels": 130000, "closed_pixels": 112000, "mask_pixels": 99000,
|
||
"hair_mask_url": SAMPLE_IMAGE_URL,
|
||
"closed_region_url": SAMPLE_IMAGE_URL,
|
||
"final_overlay_url": SAMPLE_IMAGE_URL,
|
||
"mask_url": SAMPLE_IMAGE_URL},
|
||
},
|
||
}
|
||
}
|
||
},
|
||
},
|
||
},
|
||
)
|
||
async def head_mask(
|
||
image_file: Optional[UploadFile] = File(default=None, description="上传图片文件（JPG/PNG）"),
|
||
image_url: Optional[str] = Form(default=None, description="图片 URL"),
|
||
image_base64: Optional[str] = Form(default=None, description="图片 base64（需带 data:image/...;base64, 前缀）"),
|
||
erode_cm: float = Form(default=1.2, description="外缘朝中心151内缩的距离（厘米），默认 1.2"),
|
||
):
|
||
"""接口9：头发遮罩生成 + 分步可视化"""
|
||
raw, e = await resolve_image_bytes(image_file, image_url, image_base64)
|
||
if e is not None:
|
||
return e
|
||
|
||
image = cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
|
||
if image is None:
|
||
return err(1008, "图片格式不支持（仅 JPG / PNG）")
|
||
|
||
try:
|
||
from fastapi.concurrency import run_in_threadpool
|
||
from face_analysis.head_mask import generate_head_mask, NoFaceError
|
||
try:
|
||
data = await run_in_threadpool(generate_head_mask, image, erode_cm)
|
||
except NoFaceError:
|
||
return err(1001, "无法识别人像")
|
||
return ok(data)
|
||
except Exception as ex: # noqa: BLE001
|
||
logger.exception("接口9 处理异常")
|
||
return err(1007, f"处理失败：{ex}")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 接口 10：头部外缘膨胀带遮罩
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.post(
|
||
"/api/v1/head/band",
|
||
summary="接口10 头部外缘膨胀带遮罩",
|
||
tags=["人脸分析"],
|
||
description=f"""
|
||
先和接口9 一样得到**内缩后的基准遮罩**（含额头的闭合区域外缘朝151内缩 erode_cm、底线不动，默认1.2cm），
|
||
在此基础上生成一条沿头部外缘的膨胀带遮罩：
|
||
|
||
1. 取基准遮罩的**外轮廓线**，去掉贴着底部分割线的那一段（只留头发/头部外缘弧线）。
|
||
2. 把外轮廓线膨胀成带子（**总宽 dilate_cm，默认 2cm，可调**；半径=总宽/2，虹膜标定换算像素）。
|
||
3. 裁到分割线以上（不越过底线）。
|
||
|
||
{_image_fields_desc}
|
||
|
||
返回 `data` 内含 `steps_common`（关键点/分割线、上半区）与 `bisenet` / `segformer`
|
||
两组结果（各含 `base_mask` / `contour` / `band_overlay` / `mask`）。经网关时所有
|
||
`*_base64` 图片字段会被落盘改写为 `*_url`。
|
||
""",
|
||
responses={
|
||
200: {
|
||
"description": "成功",
|
||
"content": {
|
||
"application/json": {
|
||
"example": {
|
||
"code": 0,
|
||
"message": "success",
|
||
"request_id": "mock-request-id",
|
||
"data": {
|
||
"px_per_cm": 48.12,
|
||
"erode_cm": 1.2,
|
||
"erode_px": 58,
|
||
"dilate_cm": 2.0,
|
||
"dilate_radius_px": 48,
|
||
"image_size": {"width": 1080, "height": 1440},
|
||
"steps_common": {"landmarks_baseline_url": SAMPLE_IMAGE_URL,
|
||
"upper_region_url": SAMPLE_IMAGE_URL},
|
||
"bisenet": {"base_pixels": 96000, "band_pixels": 42000,
|
||
"base_mask_url": SAMPLE_IMAGE_URL,
|
||
"contour_url": SAMPLE_IMAGE_URL,
|
||
"band_overlay_url": SAMPLE_IMAGE_URL,
|
||
"mask_url": SAMPLE_IMAGE_URL},
|
||
"segformer": {"base_pixels": 97000, "band_pixels": 43000,
|
||
"base_mask_url": SAMPLE_IMAGE_URL,
|
||
"contour_url": SAMPLE_IMAGE_URL,
|
||
"band_overlay_url": SAMPLE_IMAGE_URL,
|
||
"mask_url": SAMPLE_IMAGE_URL},
|
||
},
|
||
}
|
||
}
|
||
},
|
||
},
|
||
},
|
||
)
|
||
async def head_band(
|
||
image_file: Optional[UploadFile] = File(default=None, description="上传图片文件（JPG/PNG）"),
|
||
image_url: Optional[str] = Form(default=None, description="图片 URL"),
|
||
image_base64: Optional[str] = Form(default=None, description="图片 base64（需带 data:image/...;base64, 前缀）"),
|
||
erode_cm: float = Form(default=1.2, description="基准遮罩外缘朝151内缩距离（厘米，同接口9），默认 1.2"),
|
||
dilate_cm: float = Form(default=2.0, description="外轮廓线膨胀后带子总宽（厘米），默认 2.0"),
|
||
):
|
||
"""接口10：头部外缘膨胀带遮罩 + 分步可视化"""
|
||
raw, e = await resolve_image_bytes(image_file, image_url, image_base64)
|
||
if e is not None:
|
||
return e
|
||
|
||
image = cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
|
||
if image is None:
|
||
return err(1008, "图片格式不支持（仅 JPG / PNG）")
|
||
|
||
try:
|
||
from fastapi.concurrency import run_in_threadpool
|
||
from face_analysis.head_band import generate_head_band, NoFaceError
|
||
try:
|
||
data = await run_in_threadpool(generate_head_band, image, erode_cm, dilate_cm)
|
||
except NoFaceError:
|
||
return err(1001, "无法识别人像")
|
||
return ok(data)
|
||
except Exception as ex: # noqa: BLE001
|
||
logger.exception("接口10 处理异常")
|
||
return err(1007, f"处理失败：{ex}")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 接口 11：发际线生发（接口9 遮罩 + change_hair 换发型 + 按遮罩羽化贴回）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.post(
|
||
"/api/v1/hairline/grow",
|
||
summary="接口11 发际线生发",
|
||
tags=["生发"],
|
||
description=f"""
|
||
输入一张**发际线较高 / 头发稀少**的正脸图 + **发际线类型 ID**（= change_hair 的 `hair_id`，
|
||
如 `chang_tuoyuan`/`chang_bolang`/`chang_zhixian`/`chang_xinxing`/`chang_huaban`），
|
||
输出同一个人、同一发型、按该发际线类型压低发际线后的图片，并返回**每一步可视化**。
|
||
|
||
管线（见 `docs/发际线增强算法.md`）：
|
||
1. 用接口9 算法算头发遮罩（含额头闭合区域，外缘内缩 `erode_cm`）。`seg_model` 选
|
||
BiSeNet/SegFormer，`mask_type` 选 eroded(内缩)/closed(闭合区域)。
|
||
2. 调 change_hair 换发型服务生成该发际线类型图（返回图已与原图同分辨率同对齐）。
|
||
`swap_mode=ext_mask`：把接口9 遮罩作为 `ext_mask` 传给换发型，webui 精确重绘该区域
|
||
（忠于算法文档）；`swap_mode=as_is`：不改换发型，贴回时再裁到接口9 遮罩。
|
||
3. 严格按接口9 遮罩把生成图贴回原图（遮罩外=原图不动）。
|
||
4. 接缝融合：`blend_method` 选 feather(高斯羽化)/alpha_gradient(距离变换内渐变)/
|
||
seamless(泊松无缝克隆)/multiband(多频段金字塔融合)；`feather_px`、`edge_erode_px` 控制过渡细节
|
||
（feather_px 仅 feather/alpha_gradient 用；multiband 用 `mb_levels` 控制金字塔层数 2~6）。
|
||
`color_match=true` 时先在遮罩区做 Reinhard 颜色统计迁移，消除生成图与原图的整体色差
|
||
（对 feather/alpha_gradient/multiband 有效；seamless 自带色彩调和，自动跳过）。
|
||
|
||
{_image_fields_desc}
|
||
|
||
返回 `data.steps` 含 `input`/`baseline`/`hair_mask`/`mask_overlay`/`mask`/`swap_raw`/
|
||
`alpha`/`final` 各步图（`*_base64`，经网关改写为 `*_url`）。
|
||
""",
|
||
)
|
||
async def hairline_grow(
|
||
image_file: Optional[UploadFile] = File(default=None, description="上传图片文件（JPG/PNG）"),
|
||
image_url: Optional[str] = Form(default=None, description="图片 URL"),
|
||
image_base64: Optional[str] = Form(default=None, description="图片 base64（需带 data:image/...;base64, 前缀）"),
|
||
hairline_id: str = Form(..., description="发际线类型 ID（= change_hair hair_id，如 chang_tuoyuan）"),
|
||
gen_backend: str = Form(default="swaphair", description="生成后端：swaphair(换发型LoRA) | hairgrow(区域生发inpaint，压低发际线)（默认 swaphair）"),
|
||
hairgrow_strength: float = Form(default=0.75, description="区域生发强度（仅 hairgrow 后端），默认 0.75"),
|
||
is_hr: bool = Form(default=False, description="高清模式（换发型输出 1152×1536，否则 576×768）"),
|
||
seg_model: str = Form(default="segformer", description="头发分割模型：bisenet | segformer（默认 segformer）"),
|
||
erode_cm: float = Form(default=0.6, description="baseline 参考内缩距离（厘米），默认 0.6"),
|
||
hairline_push_cm: float = Form(default=1.0, description="发际线外推距离（厘米，往头发方向推进），默认 1.0"),
|
||
hairline_edge: str = Form(default="column", description="发际线提取方式：column | contour，默认 column"),
|
||
swap_mode: str = Form(default="ext_mask", description="换发型取图模式：ext_mask | as_is（默认 ext_mask）"),
|
||
edge_erode_px: int = Form(default=3, description="贴图前遮罩内缩像素（防边缘露皮/光晕），默认 3"),
|
||
denoising_strength: float = Form(default=0.6, description="换发型 webui 重绘强度（越大生发越激进），默认 0.6"),
|
||
mb_levels: int = Form(default=5, description="多频段金字塔层数（2~6，越大低频色差抹得越宽），默认 5"),
|
||
blend_method: str = Form(default="multiband", description="接缝融合方法：multiband(多频段金字塔,默认) | seamless(泊松无缝克隆) | two_stage(泊松→多频段两段式,大色差场景) | feather(高斯羽化) | alpha_gradient(距离变换内渐变)"),
|
||
color_match: bool = Form(default=True, description="融合前 Reinhard 颜色迁移消除整体色差（对 multiband/feather/alpha_gradient 有效；seamless/two_stage 自带调色故跳过），默认 True"),
|
||
color_match_strength: float = Form(default=1.0, description="颜色迁移强度（0~1，1=全迁移，<1 只迁移部分防过度改色），默认 1.0"),
|
||
mb_feather_px: int = Form(default=1, description="多频段最细层掩码轻羽化像素（0=不羽化，消除发丝边缘锯齿），默认 1"),
|
||
transition_band_px: int = Form(default=-1, description="keep-region 过渡带边距（-1=自动按层数 2**n，>=0 用绝对像素与层数解耦），默认 -1"),
|
||
inpainting_fill: int = Form(default=1, description="change_hair服务端重绘填充：0=保留原图(治染绿) | 1=填充噪声(默认/原始) | 2=纯色 | 3=潜变量噪声。默认 1"),
|
||
mask_blur: int = Form(default=11, description="change_hair服务端遮罩边缘模糊像素（原始11，越大颜色越易从边缘渗透），默认 11"),
|
||
mask_dilate_scale: float = Form(default=1.0, description="change_hair服务端遮罩膨胀缩放（1.0=原始核尺寸，<1收缩防越界），默认 1.0"),
|
||
):
|
||
"""接口11：发际线生发 + 分步可视化（**不含重绘**，重绘见接口12）。遮罩固定 pushed，融合默认 multiband（可选 seamless/two_stage/feather）。"""
|
||
raw, e = await resolve_image_bytes(image_file, image_url, image_base64)
|
||
if e is not None:
|
||
return e
|
||
|
||
image = cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
|
||
if image is None:
|
||
return err(1008, "图片格式不支持（仅 JPG / PNG）")
|
||
|
||
try:
|
||
from fastapi.concurrency import run_in_threadpool
|
||
from face_analysis.hairline_grow import generate_hairline_grow, NoFaceError, SwapError
|
||
from uuid import uuid4 as _uuid4
|
||
rid = _uuid4().hex[:8]
|
||
logger.info("[%s] 接口11 收到请求: hairline_push_cm=%s hairline_edge=%s mb_levels=%s "
|
||
"blend=%s color_match=%s cm_strength=%s mb_feather_px=%s transition_band_px=%s "
|
||
"inpainting_fill=%s mask_blur=%s mask_dilate_scale=%s",
|
||
rid, hairline_push_cm, hairline_edge, mb_levels, blend_method,
|
||
color_match, color_match_strength, mb_feather_px, transition_band_px,
|
||
inpainting_fill, mask_blur, mask_dilate_scale)
|
||
try:
|
||
data = await run_in_threadpool(
|
||
generate_hairline_grow, image, hairline_id,
|
||
is_hr=is_hr, seg_model=seg_model, erode_cm=erode_cm, swap_mode=swap_mode,
|
||
edge_erode_px=edge_erode_px, denoising_strength=denoising_strength,
|
||
gen_backend=gen_backend, hairgrow_strength=hairgrow_strength,
|
||
mb_levels=mb_levels, hairline_push_cm=hairline_push_cm,
|
||
hairline_edge=hairline_edge, blend_method=blend_method,
|
||
color_match=color_match, color_match_strength=color_match_strength,
|
||
mb_feather_px=mb_feather_px, transition_band_px=transition_band_px,
|
||
inpainting_fill=inpainting_fill, mask_blur=mask_blur,
|
||
mask_dilate_scale=mask_dilate_scale, rid=rid)
|
||
except NoFaceError:
|
||
return err(1001, "无法识别人像")
|
||
except SwapError as se:
|
||
return err(1007, f"换发型失败：{se}")
|
||
logger.info("[%s] 接口11 成功返回", rid)
|
||
return ok(data)
|
||
except Exception as ex: # noqa: BLE001
|
||
logger.exception("接口11 处理异常")
|
||
return err(1007, f"处理失败：{ex}")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 接口 12：发际线带重绘（调用接口11 的 ④final 作输入 + ⑤-①重绘带作遮罩，Flux-2 重绘）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.post(
|
||
"/api/v1/hairline/grow_v2",
|
||
summary="接口12 发际线带重绘（接口11 final + 发际线重绘带 → Flux-2 保色重绘）",
|
||
tags=["生发"],
|
||
description=f"""
|
||
接口12 是接口11 的**下游重绘阶段**。内部先跑接口11 核心管线拿到 **④ 接缝融合最终图（final）**，
|
||
再取 **⑤-① 发际线重绘带**（发际线沿外推方向 `band_lo_mult×push` ~ `band_hi_mult×push`
|
||
之间、经 ①-a baseline 截断只留上部的带状区域，默认 0.5×~1.5×push）
|
||
作为遮罩，用 **Flux-2（ComfyUI）** 做 reference-latent 保色重绘（不易染绿），重绘结果再与 final 融合。
|
||
|
||
与接口11 的关系：接口11 只负责生成 final（不再含重绘）；接口12 负责在 final 上做发际线带重绘。
|
||
接口11 的可调参数（seg_model/gen_backend/hairline_push_cm/blend_method/color_match 等）
|
||
在本接口同样暴露，用于内部生成 final 与重绘带；另有 `comfyui_prompt` 控制 Flux-2 提示词。
|
||
|
||
⚠️ 依赖 ComfyUI（默认 :8188）在跑，否则重绘失败会在 `data.redraw.c_error` 报告。
|
||
|
||
**同时产出两版结果供对比**：
|
||
- `redraw_full`：ComfyUI 整帧输出（全脸美颜 + 全脸重绘），与手动跑 ComfyUI 一致。
|
||
- `redraw_band`：加发只在发际线带、美颜保留全脸（band 内用 ComfyUI 重绘，band 外 = final 结构
|
||
+ 按 `beauty_alpha` 融入全脸美颜）。
|
||
|
||
返回 `data.steps`：`input`（原图）/ `final`（接口11 的 ④，重绘输入基底）/
|
||
`redraw_band_overlay`（⑤-① 重绘带可视化）/ `redraw_full`（A 整帧）/ `redraw_band`（B 局部加发+全脸美颜）/
|
||
`redraw_c`（兼容旧字段，=redraw_full）。
|
||
|
||
{_image_fields_desc}
|
||
""",
|
||
)
|
||
async def hairline_grow_v2(
|
||
image_file: Optional[UploadFile] = File(default=None, description="上传图片文件（JPG/PNG）"),
|
||
image_url: Optional[str] = Form(default=None, description="图片 URL"),
|
||
image_base64: Optional[str] = Form(default=None, description="图片 base64（需带 data:image/...;base64, 前缀）"),
|
||
hairline_id: str = Form(..., description="发际线类型 ID（= change_hair hair_id，如 chang_tuoyuan）"),
|
||
gen_backend: str = Form(default="swaphair", description="生成后端：swaphair(换发型LoRA) | hairgrow(区域生发inpaint，压低发际线)（默认 swaphair）"),
|
||
hairgrow_strength: float = Form(default=0.75, description="区域生发强度（仅 hairgrow 后端），默认 0.75"),
|
||
is_hr: bool = Form(default=False, description="高清模式（换发型输出 1152×1536，否则 576×768）"),
|
||
seg_model: str = Form(default="segformer", description="头发分割模型：bisenet | segformer（默认 segformer）"),
|
||
erode_cm: float = Form(default=0.6, description="baseline 参考内缩距离（厘米），默认 0.6"),
|
||
swap_mode: str = Form(default="ext_mask", description="换发型取图模式：ext_mask | as_is（默认 ext_mask）"),
|
||
edge_erode_px: int = Form(default=3, description="贴图前遮罩内缩像素（防边缘露皮/光晕），默认 3"),
|
||
denoising_strength: float = Form(default=0.6, description="换发型 webui 重绘强度（越大生发越激进），默认 0.6"),
|
||
mb_levels: int = Form(default=5, description="多频段金字塔层数（2~6），默认 5"),
|
||
hairline_push_cm: float = Form(default=1.0, description="发际线外推距离（厘米），默认 1.0"),
|
||
hairline_edge: str = Form(default="column", description="发际线提取方式：column | contour，默认 column"),
|
||
blend_method: str = Form(default="multiband", description="接缝融合方法：multiband | seamless | two_stage | feather | alpha_gradient"),
|
||
color_match: bool = Form(default=True, description="融合前 Reinhard 颜色迁移消除整体色差，默认 True"),
|
||
color_match_strength: float = Form(default=1.0, description="颜色迁移强度（0~1），默认 1.0"),
|
||
mb_feather_px: int = Form(default=1, description="多频段最细层掩码轻羽化像素，默认 1"),
|
||
transition_band_px: int = Form(default=-1, description="keep-region 过渡带边距（-1=自动），默认 -1"),
|
||
inpainting_fill: int = Form(default=1, description="change_hair服务端重绘填充：0=保留原图 | 1=噪声 | 2=纯色 | 3=潜变量。默认 1"),
|
||
mask_blur: int = Form(default=11, description="change_hair服务端遮罩边缘模糊像素，默认 11"),
|
||
mask_dilate_scale: float = Form(default=1.0, description="change_hair服务端遮罩膨胀缩放，默认 1.0"),
|
||
comfyui_prompt: Optional[str] = Form(default=None, description="Flux-2 重绘提示词，None 用默认「填充遮罩区域的头发」"),
|
||
beauty_alpha: float = Form(default=0.6, description="redraw_band 版 band 外的全脸美颜融入强度（0=band外无美颜纯用final，1≈整帧版），默认 0.6"),
|
||
band_lo_mult: float = Form(default=0.5, description="重绘带外推倍率下限（相对 hairline_push_cm，内轮廓=0×、原外推线=1.0×），默认 0.5"),
|
||
band_hi_mult: float = Form(default=1.5, description="重绘带外推倍率上限（相对 hairline_push_cm），默认 1.5"),
|
||
):
|
||
"""接口12：发际线带重绘。同时产出 redraw_full(整帧美颜) 与 redraw_band(局部加发+全脸美颜) 两版对比。"""
|
||
raw, e = await resolve_image_bytes(image_file, image_url, image_base64)
|
||
if e is not None:
|
||
return e
|
||
|
||
image = cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
|
||
if image is None:
|
||
return err(1008, "图片格式不支持（仅 JPG / PNG）")
|
||
|
||
try:
|
||
from fastapi.concurrency import run_in_threadpool
|
||
from face_analysis.hairline_grow import generate_hairline_redraw, NoFaceError, SwapError
|
||
from uuid import uuid4 as _uuid4
|
||
rid = _uuid4().hex[:8]
|
||
logger.info("[%s] 接口12 收到请求: hairline_id=%s hairline_push_cm=%s blend=%s comfyui_prompt=%r",
|
||
rid, hairline_id, hairline_push_cm, blend_method, comfyui_prompt)
|
||
try:
|
||
data = await run_in_threadpool(
|
||
generate_hairline_redraw, image, hairline_id,
|
||
is_hr=is_hr, seg_model=seg_model, erode_cm=erode_cm, swap_mode=swap_mode,
|
||
edge_erode_px=edge_erode_px, denoising_strength=denoising_strength,
|
||
gen_backend=gen_backend, hairgrow_strength=hairgrow_strength,
|
||
mb_levels=mb_levels, hairline_push_cm=hairline_push_cm,
|
||
hairline_edge=hairline_edge, blend_method=blend_method,
|
||
color_match=color_match, color_match_strength=color_match_strength,
|
||
mb_feather_px=mb_feather_px, transition_band_px=transition_band_px,
|
||
inpainting_fill=inpainting_fill, mask_blur=mask_blur,
|
||
mask_dilate_scale=mask_dilate_scale, comfyui_prompt=comfyui_prompt,
|
||
beauty_alpha=beauty_alpha, band_lo_mult=band_lo_mult,
|
||
band_hi_mult=band_hi_mult, rid=rid)
|
||
except NoFaceError:
|
||
return err(1001, "无法识别人像")
|
||
except SwapError as se:
|
||
return err(1007, f"换发型失败：{se}")
|
||
logger.info("[%s] 接口12 成功返回", rid)
|
||
return ok(data)
|
||
except Exception as ex: # noqa: BLE001
|
||
logger.exception("接口12 处理异常")
|
||
return err(1007, f"处理失败：{ex}")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 接口 12 final：精简版发际线带重绘（仅需图片 + 发型 ID，其余参数全用默认值）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# 接口12 final 固化的默认参数（= test_interface12.html 当前默认值，color_match 关闭）
|
||
_V2_FINAL_DEFAULTS = dict(
|
||
gen_backend="swaphair", hairgrow_strength=0.75, is_hr=False, seg_model="segformer",
|
||
erode_cm=0.6, swap_mode="ext_mask", edge_erode_px=3, denoising_strength=0.6,
|
||
mb_levels=5, hairline_push_cm=0.8, hairline_edge="column", blend_method="two_stage",
|
||
color_match=False, color_match_strength=0.4, mb_feather_px=1, transition_band_px=-1,
|
||
inpainting_fill=1, mask_blur=11, mask_dilate_scale=1.0, comfyui_prompt=None,
|
||
beauty_alpha=0.6, band_lo_mult=0.5, band_hi_mult=1.5,
|
||
)
|
||
|
||
|
||
async def _run_v2_final(image_file, image_url, image_base64, hairline_id, tag):
|
||
"""接口12 final / final v2 共用：仅需图片 + hairline_id，其余用固化默认值。
|
||
两者后端计算完全一致（同一次 ComfyUI 输出同时含 A 整帧与 B 局部+美颜），
|
||
差异仅在配套测试页展示哪一版。"""
|
||
raw, e = await resolve_image_bytes(image_file, image_url, image_base64)
|
||
if e is not None:
|
||
return e
|
||
|
||
image = cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR)
|
||
if image is None:
|
||
return err(1008, "图片格式不支持（仅 JPG / PNG）")
|
||
|
||
try:
|
||
from fastapi.concurrency import run_in_threadpool
|
||
from face_analysis.hairline_grow import generate_hairline_redraw, NoFaceError, SwapError
|
||
from uuid import uuid4 as _uuid4
|
||
rid = _uuid4().hex[:8]
|
||
logger.info("[%s] %s 收到请求: hairline_id=%s（其余用默认值）", rid, tag, hairline_id)
|
||
try:
|
||
data = await run_in_threadpool(
|
||
generate_hairline_redraw, image, hairline_id,
|
||
rid=rid, **_V2_FINAL_DEFAULTS)
|
||
except NoFaceError:
|
||
return err(1001, "无法识别人像")
|
||
except SwapError as se:
|
||
return err(1007, f"换发型失败：{se}")
|
||
logger.info("[%s] %s 成功返回", rid, tag)
|
||
return ok(data)
|
||
except Exception as ex: # noqa: BLE001
|
||
logger.exception("%s 处理异常", tag)
|
||
return err(1007, f"处理失败：{ex}")
|
||
|
||
|
||
@app.post(
|
||
"/api/v1/hairline/grow_v2_final",
|
||
summary="接口12 final 精简重绘（整帧重绘；仅图片 + 发型 ID）",
|
||
tags=["生发"],
|
||
description=f"""
|
||
接口12 的**精简/生产版**：只需上传图片 + 选择 `hairline_id`，其余所有参数固化为当前调优默认值
|
||
（hairline_push_cm=0.8 / blend_method=two_stage / **color_match=关闭** / color_match_strength=0.4 /
|
||
beauty_alpha=0.6 / band_lo_mult=0.5 / band_hi_mult=1.5 等）。
|
||
|
||
**最终重绘取整帧重绘**（`data.steps.redraw_full`，全脸美颜 + 全脸重绘）。
|
||
返回结构与接口12 一致（`data.steps` 同时含 `redraw_full`（A 整帧）/ `redraw_band`（B 局部+美颜））。
|
||
|
||
⚠️ 依赖 ComfyUI（默认 :8188）在跑。
|
||
|
||
{_image_fields_desc}
|
||
""",
|
||
)
|
||
async def hairline_grow_v2_final(
|
||
image_file: Optional[UploadFile] = File(default=None, description="上传图片文件（JPG/PNG）"),
|
||
image_url: Optional[str] = Form(default=None, description="图片 URL"),
|
||
image_base64: Optional[str] = Form(default=None, description="图片 base64（需带 data:image/...;base64, 前缀）"),
|
||
hairline_id: str = Form(..., description="发际线类型 ID（= change_hair hair_id，如 chang_bolang）"),
|
||
):
|
||
"""接口12 final：整帧重绘版，仅需图片 + hairline_id。"""
|
||
return await _run_v2_final(image_file, image_url, image_base64, hairline_id, "接口12final")
|
||
|
||
|
||
@app.post(
|
||
"/api/v1/hairline/grow_v2_final_v2",
|
||
summary="接口12 final v2 精简重绘（B 局部加发+全脸美颜；仅图片 + 发型 ID）",
|
||
tags=["生发"],
|
||
description=f"""
|
||
接口12 final 的**局部加发版**：参数与 `grow_v2_final` 完全相同（color_match 关闭等），
|
||
唯一区别是**最终重绘取 B 局部加发+全脸美颜**（`data.steps.redraw_band`：
|
||
加发只在发际线带内、band 外保留 final 结构并按 beauty_alpha 融入全脸美颜）。
|
||
|
||
返回结构与接口12 一致（`data.steps` 同时含 `redraw_full`（A 整帧）/ `redraw_band`（B 局部+美颜））。
|
||
|
||
⚠️ 依赖 ComfyUI（默认 :8188）在跑。
|
||
|
||
{_image_fields_desc}
|
||
""",
|
||
)
|
||
async def hairline_grow_v2_final_v2(
|
||
image_file: Optional[UploadFile] = File(default=None, description="上传图片文件（JPG/PNG）"),
|
||
image_url: Optional[str] = Form(default=None, description="图片 URL"),
|
||
image_base64: Optional[str] = Form(default=None, description="图片 base64（需带 data:image/...;base64, 前缀）"),
|
||
hairline_id: str = Form(..., description="发际线类型 ID（= change_hair hair_id，如 chang_bolang）"),
|
||
):
|
||
"""接口12 final v2：B 局部加发+全脸美颜版，仅需图片 + hairline_id。"""
|
||
return await _run_v2_final(image_file, image_url, image_base64, hairline_id, "接口12finalv2")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 重绘端点（替代 local_test /api/generate）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.post(
|
||
"/api/v1/redraw",
|
||
summary="ComfyUI 重绘",
|
||
tags=["重绘"],
|
||
description="""
|
||
传入人物图片 + 遮罩图片，直接调 ComfyUI（0716add-hair 工作流）执行局部重绘。
|
||
替代原 local_test :8899 的 /api/generate 接口。
|
||
|
||
**遮罩图片格式**：支持红色遮罩(R=255)、白色遮罩(R=G=B=255)、Alpha遮罩(A=255)，服务取所有通道最大值。
|
||
**遮罩区域**表示需要重绘的部分，非遮罩区域保持原图不变。
|
||
""",
|
||
)
|
||
async def api_redraw(
|
||
image_file: UploadFile = File(..., description="人物图片（JPG/PNG）"),
|
||
mask_file: UploadFile = File(..., description="遮罩图片（PNG，支持红/白/alpha 格式）"),
|
||
prompt: str = Form(default="填充遮罩区域的头发",
|
||
description="ComfyUI 提示词"),
|
||
):
|
||
image_bytes = await image_file.read()
|
||
mask_bytes = await mask_file.read()
|
||
from fastapi.concurrency import run_in_threadpool
|
||
from hairline.redraw import run_redraw
|
||
try:
|
||
png_bytes = await run_in_threadpool(
|
||
run_redraw, image_bytes, mask_bytes, prompt)
|
||
except Exception as e: # noqa: BLE001
|
||
logger.warning("重绘失败: %s", e)
|
||
return err(500, f"重绘失败: {e}")
|
||
b64 = base64.b64encode(png_bytes).decode()
|
||
return ok({"image_base64": f"data:image/png;base64,{b64}"})
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 调试：下载后端日志（接口11 遮罩计算全过程）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.get("/api/v1/debug/hairline_log", include_in_schema=False)
|
||
async def download_hairline_log(rid: Optional[str] = None, tail: int = 500):
|
||
"""返回 <仓库根>/log/hairline_grow.log 的内容。
|
||
|
||
rid 非空时只返回该 request id 相关的行；tail 限制返回最后 N 行（默认 500）。
|
||
供调试页"下载日志"按钮调用。
|
||
"""
|
||
from fastapi.responses import PlainTextResponse
|
||
log_path = os.getenv(
|
||
"HAIR_LOG_DIR",
|
||
os.path.join(os.path.dirname(os.path.abspath(__file__)), "log"),
|
||
)
|
||
log_path = os.path.join(log_path, "hairline_grow.log")
|
||
try:
|
||
with open(log_path, encoding="utf-8") as fh:
|
||
lines = fh.readlines()
|
||
except FileNotFoundError:
|
||
return PlainTextResponse("(日志文件不存在，可能服务还没处理过请求)", media_type="text/plain")
|
||
if rid:
|
||
lines = [l for l in lines if f"[{rid}]" in l]
|
||
lines = lines[-tail:] if tail > 0 else lines
|
||
return PlainTextResponse("".join(lines), media_type="text/plain; charset=utf-8")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 健康检查
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@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"}
|