Files
font2pic/apiserver/server.py
2026-02-08 18:28:39 +08:00

333 lines
11 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Font2SVG API 服务。"""
import argparse
import json
import logging
import os
import threading
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
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 _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 _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 do_OPTIONS(self): # noqa: N802
self.send_response(204)
self._set_cors_headers()
self.end_headers()
def do_GET(self): # noqa: N802
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"})
def do_POST(self): # noqa: N802
parsed = urlparse(self.path)
if parsed.path != "/api/render-svg":
self._send_json(404, {"ok": False, "error": "Not Found"})
return
try:
content_length = int(self.headers.get("Content-Length", "0") or "0")
except ValueError:
self._send_json(400, {"ok": False, "error": "请求长度无效"})
return
if content_length <= 0 or content_length > 256 * 1024:
self._send_json(400, {"ok": False, "error": "请求体大小无效"})
return
try:
raw_body = self.rfile.read(content_length)
payload = json.loads(raw_body.decode("utf-8"))
except json.JSONDecodeError:
self._send_json(400, {"ok": False, "error": "请求体不是有效 JSON"})
return
if not isinstance(payload, dict):
self._send_json(400, {"ok": False, "error": "请求体格式错误"})
return
font_id = str(payload.get("fontId") or "").strip()
text = str(payload.get("text") or "")
if not font_id:
self._send_json(400, {"ok": False, "error": "缺少 fontId"})
return
if not text.strip():
self._send_json(400, {"ok": False, "error": "文本内容不能为空"})
return
font_size = _safe_number(payload.get("fontSize"), 120.0, 8.0, 1024.0, float)
letter_spacing = _safe_number(payload.get("letterSpacing"), 0.0, -2.0, 5.0, float)
max_chars_per_line = _safe_number(
payload.get("maxCharsPerLine"),
MAX_CHARS_PER_LINE,
1,
300,
int,
)
fill_color = _normalize_hex_color(payload.get("fillColor"), "#000000")
try:
font_info = self.catalog.get(font_id)
result = render_svg_from_font_file(
font_info["path"],
text,
font_size=font_size,
fill_color=fill_color,
letter_spacing=letter_spacing,
max_chars_per_line=max_chars_per_line,
)
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
response_data = {
"svg": result["svg"],
"width": result["width"],
"height": result["height"],
"fontName": result.get("fontName") or font_info.get("name") or "Unknown",
"fontId": font_id,
}
self._send_json(200, {"ok": True, "data": response_data})
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/ 与 fonts.json",
)
parser.add_argument(
"--manifest",
default=os.getenv("FONT2SVG_MANIFEST_PATH", ""),
help="字体清单路径,默认使用 <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)
manifest_path = args.manifest.strip() or os.path.join(static_root, "fonts.json")
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())