"""旷视五接口 — 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", "/api/v1/debug") @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""" 输入用户照片 + 性别 + 多选发型，对每个选中发型返回 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。 **返回说明**： - `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 向下）。 """, 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}, }, } } }, }, 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的文本"), ): 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) 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": _jpg_b64(ov["middle"]), # 发际线叠图 middle 档 JPG "image_high_base64": _jpg_b64(ov["high"]), # 发际线叠图 high 档 JPG "image_low_base64": _jpg_b64(ov["low"]), # 发际线叠图 low 档 JPG "grown_image_base64": (_png_to_jpg_b64(it["grown_png"]) # 生发图 JPG（失败为 null） if it.get("grown_png") else None), "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}") # --------------------------------------------------------------------------- # 接口 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"), ): """接口11：发际线生发 + 分步可视化。遮罩固定 pushed、融合固定 multiband。""" 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", rid, hairline_push_cm, hairline_edge, mb_levels) 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, 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 固定参数精简版） # --------------------------------------------------------------------------- @app.post( "/api/v1/hairline/grow_v2", summary="接口12 发际线生发（固定 pushed 遮罩 + 多频段融合，仅返回最终图）", tags=["生发"], description=f""" 接口11 的固定参数精简版，适合生产直调。与接口11 共用同一管线，遮罩固定 pushed（发际线外推）， 融合固定 multiband（多频段金字塔）。本接口固定 `erode_cm=0.6`、`mb_levels=5` 不暴露， **只返回 `final_base64`**（最终合成图），不附带分步可视化。 其余参数（hairline_id、seg_model、gen_backend、is_hr、denoising_strength、edge_erode_px、 hairline_push_cm、hairline_edge 等）保留为可选 Form，调用方可按需覆盖。 {_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）"), 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"), hairline_push_cm: float = Form(default=1.0, description="发际线外推距离（厘米），默认 1.0"), hairline_edge: str = Form(default="column", description="发际线提取方式：column | contour，默认 column"), ): """接口12：发际线生发（固定 pushed 遮罩 + multiband 融合，仅返回最终图）。""" 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 try: # 固定：mask_type=pushed、blend_method=multiband、erode_cm=0.6、mb_levels=5 data = await run_in_threadpool( generate_hairline_grow, image, hairline_id, is_hr=is_hr, seg_model=seg_model, erode_cm=0.6, swap_mode=swap_mode, edge_erode_px=edge_erode_px, denoising_strength=denoising_strength, gen_backend=gen_backend, hairgrow_strength=hairgrow_strength, mb_levels=5, hairline_push_cm=hairline_push_cm, hairline_edge=hairline_edge) except NoFaceError: return err(1001, "无法识别人像") except SwapError as se: return err(1007, f"换发型失败：{se}") # 精简返回：只取最终合成图，丢掉接口11 的全部分步可视化 return ok({ "hairline_id": data["hairline_id"], "image_size": data["image_size"], "final_base64": data["steps"]["final_base64"], }) except Exception as ex: # noqa: BLE001 logger.exception("接口12 处理异常") return err(1007, f"处理失败：{ex}") # --------------------------------------------------------------------------- # 调试：下载后端日志（接口11 遮罩计算全过程） # --------------------------------------------------------------------------- @app.get("/api/v1/debug/hairline_log", include_in_schema=False) async def download_hairline_log(rid: Optional[str] = None, tail: int = 500): """返回 /home/xsl/hair/log/hairline_grow.log 的内容。 rid 非空时只返回该 request id 相关的行；tail 限制返回最后 N 行（默认 500）。 供调试页"下载日志"按钮调用。 """ from fastapi.responses import PlainTextResponse log_path = "/home/xsl/hair/log/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"}