404 lines
14 KiB
Python
404 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""Font2SVG API 服务。"""
|
||
|
||
import argparse
|
||
import json
|
||
import logging
|
||
import os
|
||
import threading
|
||
import time
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
from urllib.parse import urlparse
|
||
|
||
try:
|
||
from .renderer import MAX_CHARS_PER_LINE, render_svg_from_font_file
|
||
except ImportError:
|
||
from renderer import MAX_CHARS_PER_LINE, render_svg_from_font_file
|
||
try:
|
||
from .png_renderer import render_png_from_svg
|
||
except ImportError:
|
||
from png_renderer import render_png_from_svg
|
||
|
||
LOGGER = logging.getLogger("font2svg.api")
|
||
|
||
|
||
def _normalize_hex_color(color, fallback="#000000"):
|
||
value = str(color or "").strip()
|
||
if len(value) == 7 and value.startswith("#"):
|
||
hex_part = value[1:]
|
||
if all(ch in "0123456789abcdefABCDEF" for ch in hex_part):
|
||
return value
|
||
return fallback
|
||
|
||
|
||
def _safe_number(value, fallback, min_value=None, max_value=None, cast=float):
|
||
try:
|
||
parsed = cast(value)
|
||
except (TypeError, ValueError):
|
||
parsed = fallback
|
||
if min_value is not None:
|
||
parsed = max(min_value, parsed)
|
||
if max_value is not None:
|
||
parsed = min(max_value, parsed)
|
||
return parsed
|
||
|
||
|
||
def _is_within_root(path, root):
|
||
try:
|
||
root_abs = os.path.abspath(root)
|
||
path_abs = os.path.abspath(path)
|
||
return os.path.commonpath([root_abs, path_abs]) == root_abs
|
||
except ValueError:
|
||
return False
|
||
|
||
|
||
def _normalize_manifest_path(path):
|
||
value = str(path or "").strip()
|
||
if not value:
|
||
return ""
|
||
|
||
if value.startswith("//"):
|
||
parsed = urlparse(f"https:{value}")
|
||
return parsed.path
|
||
|
||
if value.startswith("http://") or value.startswith("https://"):
|
||
parsed = urlparse(value)
|
||
return parsed.path
|
||
|
||
return value
|
||
|
||
|
||
def _resolve_font_path(item, static_root):
|
||
path = _normalize_manifest_path(item.get("path"))
|
||
filename = str(item.get("filename") or "").strip()
|
||
category = str(item.get("category") or "").strip()
|
||
candidates = []
|
||
|
||
if path:
|
||
cleaned = path.split("?", 1)[0].strip()
|
||
candidates.append(os.path.join(static_root, cleaned.lstrip("/")))
|
||
|
||
if category and filename:
|
||
candidates.append(os.path.join(static_root, "fonts", category, filename))
|
||
|
||
if filename:
|
||
candidates.append(os.path.join(static_root, "fonts", filename))
|
||
|
||
for candidate in candidates:
|
||
absolute = os.path.abspath(candidate)
|
||
if _is_within_root(absolute, static_root) and os.path.isfile(absolute):
|
||
return absolute
|
||
|
||
return None
|
||
|
||
|
||
class FontCatalog:
|
||
def __init__(self, static_root, manifest_path):
|
||
self.static_root = os.path.abspath(static_root)
|
||
self.manifest_path = os.path.abspath(manifest_path)
|
||
self._manifest_mtime = None
|
||
self._font_map = {}
|
||
self._lock = threading.Lock()
|
||
|
||
def _reload(self):
|
||
if not os.path.isfile(self.manifest_path):
|
||
raise FileNotFoundError(f"未找到字体清单: {self.manifest_path}")
|
||
|
||
mtime = os.path.getmtime(self.manifest_path)
|
||
if self._manifest_mtime == mtime and self._font_map:
|
||
return
|
||
|
||
with open(self.manifest_path, "r", encoding="utf-8") as file:
|
||
payload = json.load(file)
|
||
|
||
if not isinstance(payload, list):
|
||
raise ValueError("fonts.json 格式无效，必须是数组")
|
||
|
||
font_map = {}
|
||
for item in payload:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
|
||
font_id = str(item.get("id") or "").strip()
|
||
if not font_id:
|
||
continue
|
||
|
||
font_path = _resolve_font_path(item, self.static_root)
|
||
if not font_path:
|
||
continue
|
||
|
||
font_map[font_id] = {
|
||
"path": font_path,
|
||
"name": str(item.get("name") or "").strip(),
|
||
"category": str(item.get("category") or "").strip(),
|
||
}
|
||
|
||
if not font_map:
|
||
raise ValueError("未解析到可用字体路径，请检查 fonts.json 与 fonts/ 目录")
|
||
|
||
self._font_map = font_map
|
||
self._manifest_mtime = mtime
|
||
LOGGER.info("字体清单已加载: %s 个字体", len(font_map))
|
||
|
||
def get(self, font_id):
|
||
with self._lock:
|
||
self._reload()
|
||
item = self._font_map.get(font_id)
|
||
|
||
if not item:
|
||
raise KeyError(f"字体不存在: {font_id}")
|
||
if not os.path.isfile(item["path"]):
|
||
raise FileNotFoundError(f"字体文件不存在: {item['path']}")
|
||
|
||
return item
|
||
|
||
def health(self):
|
||
with self._lock:
|
||
self._reload()
|
||
return {
|
||
"manifestPath": self.manifest_path,
|
||
"fontCount": len(self._font_map),
|
||
}
|
||
|
||
|
||
class RenderHandler(BaseHTTPRequestHandler):
|
||
catalog = None
|
||
|
||
def send_response(self, code, message=None):
|
||
self._response_status = code
|
||
super().send_response(code, message)
|
||
|
||
def _set_cors_headers(self):
|
||
self.send_header("Access-Control-Allow-Origin", "*")
|
||
self.send_header("Access-Control-Allow-Methods", "GET,POST,OPTIONS")
|
||
self.send_header("Access-Control-Allow-Headers", "Content-Type,Authorization")
|
||
self.send_header("Access-Control-Max-Age", "86400")
|
||
|
||
def _log_request_timing(self, start_time):
|
||
elapsed_ms = (time.perf_counter() - start_time) * 1000
|
||
status = getattr(self, "_response_status", "-")
|
||
client_ip = self.client_address[0] if self.client_address else "-"
|
||
LOGGER.info(
|
||
"请求完成 method=%s path=%s status=%s duration_ms=%.2f client=%s",
|
||
self.command,
|
||
self.path,
|
||
status,
|
||
elapsed_ms,
|
||
client_ip,
|
||
)
|
||
|
||
def _send_json(self, status_code, payload):
|
||
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
||
self.send_response(status_code)
|
||
self._set_cors_headers()
|
||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
|
||
def _send_binary(self, status_code, body, content_type):
|
||
self.send_response(status_code)
|
||
self._set_cors_headers()
|
||
self.send_header("Content-Type", content_type)
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
|
||
def _parse_render_payload(self):
|
||
try:
|
||
content_length = int(self.headers.get("Content-Length", "0") or "0")
|
||
except ValueError:
|
||
raise ValueError("请求长度无效") from None
|
||
|
||
if content_length <= 0 or content_length > 256 * 1024:
|
||
raise ValueError("请求体大小无效")
|
||
|
||
try:
|
||
raw_body = self.rfile.read(content_length)
|
||
payload = json.loads(raw_body.decode("utf-8"))
|
||
except json.JSONDecodeError:
|
||
raise ValueError("请求体不是有效 JSON") from None
|
||
|
||
if not isinstance(payload, dict):
|
||
raise ValueError("请求体格式错误")
|
||
|
||
font_id = str(payload.get("fontId") or "").strip()
|
||
text = str(payload.get("text") or "")
|
||
if not font_id:
|
||
raise ValueError("缺少 fontId")
|
||
if not text.strip():
|
||
raise ValueError("文本内容不能为空")
|
||
|
||
return {
|
||
"fontId": font_id,
|
||
"text": text,
|
||
"fontSize": _safe_number(payload.get("fontSize"), 120.0, 8.0, 1024.0, float),
|
||
"letterSpacing": _safe_number(payload.get("letterSpacing"), 0.0, -2.0, 5.0, float),
|
||
"maxCharsPerLine": _safe_number(
|
||
payload.get("maxCharsPerLine"),
|
||
MAX_CHARS_PER_LINE,
|
||
1,
|
||
300,
|
||
int,
|
||
),
|
||
"fillColor": _normalize_hex_color(payload.get("fillColor"), "#000000"),
|
||
}
|
||
|
||
def _render_svg_core(self, render_params):
|
||
font_info = self.catalog.get(render_params["fontId"])
|
||
result = render_svg_from_font_file(
|
||
font_info["path"],
|
||
render_params["text"],
|
||
font_size=render_params["fontSize"],
|
||
fill_color=render_params["fillColor"],
|
||
letter_spacing=render_params["letterSpacing"],
|
||
max_chars_per_line=render_params["maxCharsPerLine"],
|
||
)
|
||
return font_info, result
|
||
|
||
def do_OPTIONS(self): # noqa: N802
|
||
start_time = time.perf_counter()
|
||
self._response_status = None
|
||
try:
|
||
self.send_response(204)
|
||
self._set_cors_headers()
|
||
self.end_headers()
|
||
finally:
|
||
self._log_request_timing(start_time)
|
||
|
||
def do_GET(self): # noqa: N802
|
||
start_time = time.perf_counter()
|
||
self._response_status = None
|
||
try:
|
||
parsed = urlparse(self.path)
|
||
if parsed.path == "/healthz":
|
||
try:
|
||
data = self.catalog.health()
|
||
self._send_json(200, {"ok": True, "data": data})
|
||
except Exception as error:
|
||
LOGGER.exception("健康检查失败")
|
||
self._send_json(500, {"ok": False, "error": str(error)})
|
||
return
|
||
|
||
self._send_json(404, {"ok": False, "error": "Not Found"})
|
||
finally:
|
||
self._log_request_timing(start_time)
|
||
|
||
def do_POST(self): # noqa: N802
|
||
start_time = time.perf_counter()
|
||
self._response_status = None
|
||
try:
|
||
parsed = urlparse(self.path)
|
||
if parsed.path not in ("/api/render-svg", "/api/render-png"):
|
||
self._send_json(404, {"ok": False, "error": "Not Found"})
|
||
return
|
||
|
||
try:
|
||
render_params = self._parse_render_payload()
|
||
font_info, result = self._render_svg_core(render_params)
|
||
except KeyError as error:
|
||
self._send_json(404, {"ok": False, "error": str(error)})
|
||
return
|
||
except FileNotFoundError as error:
|
||
self._send_json(404, {"ok": False, "error": str(error)})
|
||
return
|
||
except ValueError as error:
|
||
self._send_json(400, {"ok": False, "error": str(error)})
|
||
return
|
||
except Exception as error:
|
||
LOGGER.exception("渲染失败")
|
||
self._send_json(500, {"ok": False, "error": str(error)})
|
||
return
|
||
|
||
if parsed.path == "/api/render-png":
|
||
try:
|
||
png_bytes = render_png_from_svg(
|
||
result["svg"],
|
||
result["width"],
|
||
result["height"],
|
||
)
|
||
except Exception as error:
|
||
LOGGER.exception("PNG 渲染失败")
|
||
self._send_json(500, {"ok": False, "error": str(error)})
|
||
return
|
||
|
||
self._send_binary(200, png_bytes, "image/png")
|
||
return
|
||
|
||
response_data = {
|
||
"svg": result["svg"],
|
||
"width": result["width"],
|
||
"height": result["height"],
|
||
"fontName": result.get("fontName") or font_info.get("name") or "Unknown",
|
||
"fontId": render_params["fontId"],
|
||
}
|
||
self._send_json(200, {"ok": True, "data": response_data})
|
||
finally:
|
||
self._log_request_timing(start_time)
|
||
|
||
def log_message(self, format_str, *args):
|
||
LOGGER.info("%s - %s", self.address_string(), format_str % args)
|
||
|
||
|
||
def build_server(host, port, static_root, manifest_path):
|
||
catalog = FontCatalog(static_root=static_root, manifest_path=manifest_path)
|
||
RenderHandler.catalog = catalog
|
||
return ThreadingHTTPServer((host, port), RenderHandler)
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="Font2SVG 渲染 API 服务")
|
||
parser.add_argument("--host", default=os.getenv("FONT2SVG_API_HOST", "0.0.0.0"))
|
||
parser.add_argument("--port", type=int, default=int(os.getenv("FONT2SVG_API_PORT", "9300")))
|
||
parser.add_argument(
|
||
"--static-root",
|
||
default=os.getenv("FONT2SVG_STATIC_ROOT", os.getcwd()),
|
||
help="静态资源根目录（包含 fonts/ 与配置清单）",
|
||
)
|
||
parser.add_argument(
|
||
"--manifest",
|
||
default=os.getenv("FONT2SVG_MANIFEST_PATH", ""),
|
||
help="字体清单路径，默认优先使用 <static-root>/miniprogram/assets/fonts.json，其次 <static-root>/fonts.json",
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
|
||
)
|
||
|
||
static_root = os.path.abspath(args.static_root)
|
||
if args.manifest.strip():
|
||
manifest_path = args.manifest.strip()
|
||
else:
|
||
manifest_candidates = [
|
||
os.path.join(static_root, "miniprogram", "assets", "fonts.json"),
|
||
os.path.join(static_root, "fonts.json"),
|
||
]
|
||
manifest_path = next((item for item in manifest_candidates if os.path.isfile(item)), manifest_candidates[0])
|
||
|
||
server = build_server(
|
||
host=args.host,
|
||
port=args.port,
|
||
static_root=static_root,
|
||
manifest_path=manifest_path,
|
||
)
|
||
|
||
LOGGER.info("Font2SVG API 服务启动: http://%s:%s", args.host, args.port)
|
||
LOGGER.info("静态目录: %s", static_root)
|
||
LOGGER.info("字体清单: %s", manifest_path)
|
||
|
||
try:
|
||
server.serve_forever()
|
||
except KeyboardInterrupt:
|
||
LOGGER.info("收到退出信号，正在停止服务")
|
||
finally:
|
||
server.server_close()
|
||
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|