Ubuntu
68fe1c1406
- gateway/app.py: 接口4 handler 改为直接调用 face_features 模块 - face_features.py: API Key 读取支持 gateway/config.json - gateway/config.example.json: 增加 ark_api_key 配置项
144 lines
6.1 KiB
Python
144 lines
6.1 KiB
Python
"""接口4：用户面部特征分析（调用火山方舟 豆包视觉模型 doubao-seed-1-6-vision）。
|
||
|
||
算法来源：/home/xsl/fuyan（FaceArk.py）。worker 把图片以 base64 data URI 传给方舟
|
||
多模态模型，模型返回一大堆人脸特征 JSON；本模块解析后映射出接口4 的英文优先字段
|
||
（face_shape 等），并保留 doubao 返回的全部中文字段。
|
||
|
||
⚠️ 这是**唯一调外网云模型**的接口（其余接口全本地）。API Key 走配置/环境变量，不入 git。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import json
|
||
import logging
|
||
import os
|
||
|
||
logger = logging.getLogger("hair.worker")
|
||
|
||
ARK_BASE_URL = os.getenv("ARK_BASE_URL", "https://ark.cn-beijing.volces.com/api/v3")
|
||
ARK_MODEL = os.getenv("ARK_MODEL", "doubao-seed-1-6-vision-250815")
|
||
|
||
# doubao 中文键 → 接口4 英文优先字段
|
||
_KEY_MAP = {
|
||
"脸型": "face_shape",
|
||
"眉形": "eyebrow_shape",
|
||
"面部年龄": "facial_age",
|
||
"动静类型": "dynamic_static_type",
|
||
"性别": "gender",
|
||
"基因风格": "gene_style",
|
||
}
|
||
|
||
# 移植自 fuyan FaceArk.GetPicDesc 的特征枚举（去掉身高体重前缀）
|
||
_PROMPT = (
|
||
"分析一下图片告诉我以下特征，只要答案，格式为json字符串，"
|
||
"图片是否有人脸（有人/没人） 三庭五眼特征（答案要有三庭五眼四个字，9个字以内） "
|
||
"面部年龄（给出区间年龄）鼻长（鼻长适中/长鼻/短鼻） "
|
||
"脸型（圆形脸/心形脸/菱形脸/鹅蛋脸/方形脸/长形脸/瓜子脸） 嘴型 "
|
||
"眼袋（答案要有眼袋两个个字） 眼型 鼻型 眼皮（双眼皮/单眼皮） "
|
||
"法令纹（有法令纹/无法令纹） 人中（人中适中/人中长/人中短） 眉形 "
|
||
"瞳色（答案要有瞳色两个字） 脖长（脖长适中/脖子短/脖子长） "
|
||
"肤色（粉一白/粉二白/粉三白/黄一白/黄二白/黄黑皮）"
|
||
"直得分 曲得分 直曲总分（直得分-曲得分） 大量感得分 小量感得分 量感总分（大量感得分-小量感得） "
|
||
"面部立体度（总分十分）瞳距（毫米）对比度（对比度较强/对比度适中/对比度较弱）"
|
||
"鼻子立体度（立体度高/立体度适中/立体度低）"
|
||
"色相（中间表示0,最大值分别是-5和5，负数表示偏冷，正数表示偏暖）"
|
||
"亮度（中间表示0,最大值分别是-5和5；负数表示暗沉，正数表示白皙）"
|
||
"色度（中间表示0,最大值分别是-5和5；负数表示饱和度低，正数表示鲜艳）"
|
||
"面部颜色对比度（10分制）"
|
||
"四季色彩季型（净春型/暖春型/浅春型/浅夏型/冷夏型/柔夏型/柔秋型/暖秋型/深秋型/净冬型/冷冬型/深冬型）"
|
||
"基因风格（戏剧型/睿智型/自然型/古典型/优雅型/浪漫型/前卫型/少女型/少年型）"
|
||
"量感类型（大量感/中量感/小量感）轮廓类型（轮廓偏曲/轮廓适中/轮廓偏直）"
|
||
"动静类型（静态型/动态型）性别（男/女）"
|
||
)
|
||
|
||
_client = None
|
||
|
||
|
||
def _load_api_key() -> str | None:
|
||
"""ARK_API_KEY 环境变量优先，否则读 worker_config.json / gateway/config.json 的 ark_api_key。"""
|
||
key = os.getenv("ARK_API_KEY")
|
||
if key:
|
||
return key
|
||
base = os.path.dirname(__file__)
|
||
for cfg_name in ("worker_config.json", "gateway/config.json"):
|
||
cfg = os.path.join(base, cfg_name)
|
||
if os.path.isfile(cfg):
|
||
try:
|
||
with open(cfg, encoding="utf-8") as f:
|
||
v = json.load(f).get("ark_api_key")
|
||
if v:
|
||
return v
|
||
except Exception as e: # noqa: BLE001
|
||
logger.warning("读取 %s ark_api_key 失败：%s", cfg_name, e)
|
||
return None
|
||
|
||
|
||
def get_client():
|
||
global _client
|
||
if _client is None:
|
||
from volcenginesdkarkruntime import Ark
|
||
key = _load_api_key()
|
||
if not key:
|
||
raise RuntimeError("缺少火山方舟 API Key（设 ARK_API_KEY 或 worker_config.json.ark_api_key）")
|
||
_client = Ark(base_url=ARK_BASE_URL, api_key=key)
|
||
return _client
|
||
|
||
|
||
def _parse_json(text: str) -> dict:
|
||
"""去掉 ```json 包裹后解析。"""
|
||
s = text.strip()
|
||
if s.startswith("```"):
|
||
s = s.strip("`")
|
||
if s[:4].lower() == "json":
|
||
s = s[4:]
|
||
return json.loads(s.strip())
|
||
|
||
|
||
def _image_to_url(image_bytes: bytes = None, image_url: str = None) -> str:
|
||
"""优先用现成 URL；否则把字节转 base64 data URI（doubao 兼容）。"""
|
||
if image_url:
|
||
return image_url
|
||
fmt = "png" if image_bytes[:8] == b"\x89PNG\r\n\x1a\n" else "jpeg"
|
||
return f"data:image/{fmt};base64," + base64.b64encode(image_bytes).decode()
|
||
|
||
|
||
def analyze_features(image_bytes: bytes = None, image_url: str = None) -> dict:
|
||
"""调 doubao 视觉模型分析人脸特征。
|
||
|
||
Returns: dict —— 含接口4 英文优先字段(face_shape 等) + doubao 全部中文字段。
|
||
无人脸时 doubao 的「图片是否有人脸」= 没人，调用方据此可判 1001。
|
||
"""
|
||
url = _image_to_url(image_bytes, image_url)
|
||
resp = get_client().chat.completions.create(
|
||
model=ARK_MODEL,
|
||
messages=[{
|
||
"role": "user",
|
||
"content": [
|
||
{"type": "image_url", "image_url": {"url": url}},
|
||
{"type": "text", "text": _PROMPT},
|
||
],
|
||
}],
|
||
)
|
||
text = resp.choices[0].message.content
|
||
data = _parse_json(text) # doubao 中文字段
|
||
# 英文优先字段映射（doubao 缺某字段则跳过）
|
||
for zh, en in _KEY_MAP.items():
|
||
if zh in data and en not in data:
|
||
data[en] = data[zh]
|
||
return data
|
||
|
||
|
||
def has_face(features: dict) -> bool:
|
||
"""据 doubao 的「图片是否有人脸」判断。"""
|
||
v = features.get("图片是否有人脸") or features.get("是否有人") or ""
|
||
return "没人" not in str(v) and "没有" not in str(v)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import sys
|
||
path = sys.argv[1] if len(sys.argv) > 1 else "tests/fixtures/frontal.jpg"
|
||
with open(path, "rb") as f:
|
||
feats = analyze_features(image_bytes=f.read())
|
||
print("has_face:", has_face(feats))
|
||
print(json.dumps(feats, ensure_ascii=False, indent=2))
|