"""旷视五接口 — 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" # 运行期状态：模型就绪标志（/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") @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 _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 共用实现 # --------------------------------------------------------------------------- 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）") h, w = image.shape[:2] 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 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) # 8. 测量 + 标注图（hair_mask 定发际线/头顶；ear_mask 定人头最左/最右竖线） result = measure_face(landmarks, hair_mask, w, h, head_pose=head_pose) annotated = create_annotated_image( image, result, ear_mask=ear_mask, hair_mask=hair_mask, variant=variant) buf = BytesIO() annotated.save(buf, format="PNG") # 9. 拆分架构：返回 base64，不落盘不拼 URL（落盘改 URL 由网关完成） data = result.to_response() if variant == "v6": # 接口6：数据去顶庭（重算三庭比例与总高，移除顶庭/头顶字段） vd = result.vertical base_px = vd["upper_court_px"] + vd["middle_court_px"] + vd["lower_court_px"] 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) data["landmarks"].pop("hair_top", None) 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}, }, "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 # --------------------------------------------------------------------------- # 接口 6：四庭七眼测量标注 v2（复刻接口1） # --------------------------------------------------------------------------- @app.post( "/api/v1/face/measure-v2", summary="接口6 四庭七眼测量标注 v2", tags=["人脸分析"], description=f""" 输入用户正面照，返回： - 标注好四庭七眼数据的 **PNG 图片**（仅标注图层，不含人物） - 三庭（上庭/中庭/下庭）各段**厘米数值及占比**（**不含顶庭**） - 七眼（眼宽/脸宽/两眼间距）**厘米数值及占比** - 四个关键分界点的**原图像素坐标**（发际线/眉心/鼻翼下缘/下巴尖） 基于接口1 的变体，与接口1 的差异： - **去顶庭**：不画头顶横线、不返回顶庭数据；`face_total_height_cm` 为三庭之和 - **竖线范围**：纵向竖线从发际线画到下巴尖（接口1 为头顶→下巴尖） - **不画人头最左/最右端线**：仅七眼 6 点共 5 段标尺，不取头发轮廓端线（接口1 为 8 线 7 段） 其余（箭头/虚线/字体/单位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}, }, "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`，不排序） {_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的文本"), ): # 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 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}") # --------------------------------------------------------------------------- # 接口 7：C 端生发 v2（add_hair2.json 工作流） # --------------------------------------------------------------------------- _WORKFLOW2_PATH = os.path.join(os.path.dirname(__file__), "add_hair2.json") @app.post( "/api/v1/hair/grow-v2", summary="接口7 C端生发 v2（add_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）"), 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 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_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}") # --------------------------------------------------------------------------- # 接口 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""" 输入用户照片，返回 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": 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（必填）"), ): 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 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) 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}") # --------------------------------------------------------------------------- # 接口 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}") # --------------------------------------------------------------------------- # 健康检查 # --------------------------------------------------------------------------- @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"}