"""旷视五接口 — 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 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 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 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 # --------------------------------------------------------------------------- # 接口 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}") # --------------------------------------------------------------------------- # 接口 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 的生发控制参数（仅 male 路径生效）。 - 可选 `generate_grow_image`（默认 `true`）：是否生成生发效果图（最耗时）。 `false` 时跳过生发，各发型 `grown_image_*` 恒为 `null`，仅返回三档发际线叠图与中心点，大幅降低耗时。 - **生发机制（同接口2，按性别分流）**： `female` 走「换发型 + Flux-2 整帧重绘」（依赖 change_hair:8801 与 ComfyUI:8188）； `male` 走 ComfyUI `add_hair` 原生 inpaint。 - 可选 `flux_model` / `redraw_max_side`：同接口2（仅 female 路径生效）。 **返回说明**： - `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，仅 male 路径生效）"), prompt: str = Form(default="填充遮罩区域的头发", description="ComfyUI 提示词（同接口2，仅 male 路径生效），会替换工作流节点60的文本"), generate_grow_image: bool = Form(default=True, description="是否生成生发效果图（最耗时）。默认 true 出图；false 时跳过生发，各发型 grown_image 恒为 null，仅返回三档发际线叠图与中心点"), flux_model: Optional[str] = Form(default=None, description="Flux 模型文件名（同接口2，切换模型用）。None=工作流默认"), redraw_max_side: Optional[int] = Form(default=None, description="重绘压图长边像素（同接口2，仅 female 路径生效）。None=默认896；0=不缩图（原图直送）；其他如 768/640/1024"), ): 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, redraw_max_side=redraw_max_side, unet_name=flux_model, v2_defaults=_V2_FINAL_DEFAULTS) 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"}