238 lines
8.0 KiB
Python
238 lines
8.0 KiB
Python
from pathlib import Path
|
|
from io import BytesIO
|
|
import hashlib
|
|
import os
|
|
import re
|
|
from urllib.parse import quote
|
|
|
|
import httpx
|
|
from PIL import Image, ImageDraw, ImageFont
|
|
|
|
from app.config import settings
|
|
|
|
|
|
class WechatConfigError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def _qrcode_file_suffix(content_type: str) -> str:
|
|
normalized = content_type.lower()
|
|
if "image/png" in normalized:
|
|
return ".png"
|
|
if "image/jpeg" in normalized or "image/jpg" in normalized:
|
|
return ".jpg"
|
|
return ".jpg"
|
|
|
|
|
|
def _qrcode_file_name(device_no: str, suffix: str) -> str:
|
|
safe_name = re.sub(r"[^0-9A-Za-z_.-]+", "_", device_no).strip("._")[:80] or "device"
|
|
digest = hashlib.sha1(device_no.encode("utf-8")).hexdigest()[:8]
|
|
return f"{safe_name}-{digest}{suffix}"
|
|
|
|
|
|
def _font_candidates() -> list[Path]:
|
|
configured = os.getenv("JH_QR_FONT_PATH")
|
|
paths = []
|
|
if configured:
|
|
paths.append(Path(configured))
|
|
paths.extend([
|
|
Path("/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc"),
|
|
Path("/usr/share/fonts/opentype/noto/NotoSansCJK-Bold.ttc"),
|
|
Path("/usr/share/fonts/truetype/noto/NotoSansCJK-Regular.ttc"),
|
|
Path("/usr/share/fonts/noto-cjk/NotoSansCJK-Regular.ttc"),
|
|
Path("/usr/local/share/fonts/NotoSansCJK-Regular.ttc"),
|
|
Path("/usr/local/share/fonts/NotoSansSC-Regular.otf"),
|
|
Path("/usr/share/fonts/truetype/wqy/wqy-microhei.ttc"),
|
|
Path("/usr/share/fonts/truetype/arphic/ukai.ttc"),
|
|
Path("/System/Library/Fonts/PingFang.ttc"),
|
|
Path("/System/Library/Fonts/Hiragino Sans GB.ttc"),
|
|
Path("/System/Library/Fonts/STHeiti Medium.ttc"),
|
|
Path("/System/Library/Fonts/Supplemental/Arial Unicode.ttf"),
|
|
Path("/Library/Fonts/Arial Unicode.ttf"),
|
|
Path("C:/Windows/Fonts/msyh.ttc"),
|
|
Path("C:/Windows/Fonts/simhei.ttf"),
|
|
])
|
|
return paths
|
|
|
|
|
|
def _font_can_render_chinese(font) -> bool:
|
|
glyphs: set[bytes] = set()
|
|
for char in "模具宁嘉恒科":
|
|
image = Image.new("L", (140, 140), 0)
|
|
draw = ImageDraw.Draw(image)
|
|
draw.text((8, 8), char, font=font, fill=255)
|
|
bbox = image.getbbox()
|
|
if bbox is None:
|
|
return False
|
|
glyphs.add(image.crop(bbox).tobytes())
|
|
return len(glyphs) >= 5
|
|
|
|
|
|
def _load_label_font(size: int):
|
|
for path in _font_candidates():
|
|
if path.exists():
|
|
try:
|
|
font = ImageFont.truetype(str(path), size=size)
|
|
if _font_can_render_chinese(font):
|
|
return font
|
|
except OSError:
|
|
continue
|
|
raise RuntimeError("服务器缺少可用中文字体，无法生成带中文底部文字的二维码")
|
|
|
|
|
|
def _text_width(draw: ImageDraw.ImageDraw, text: str, font) -> int:
|
|
if not text:
|
|
return 0
|
|
bbox = draw.textbbox((0, 0), text, font=font)
|
|
return bbox[2] - bbox[0]
|
|
|
|
|
|
def _wrap_label_text(draw: ImageDraw.ImageDraw, text: str, font, max_width: int) -> list[str]:
|
|
normalized = " ".join(str(text or "").split())
|
|
if not normalized:
|
|
return []
|
|
lines: list[str] = []
|
|
current = ""
|
|
for char in normalized:
|
|
candidate = f"{current}{char}"
|
|
if current and _text_width(draw, candidate, font) > max_width:
|
|
lines.append(current)
|
|
current = char
|
|
continue
|
|
current = candidate
|
|
if current:
|
|
lines.append(current)
|
|
return lines
|
|
|
|
|
|
def _draw_qrcode_label(content: bytes, suffix: str, label: str) -> bytes:
|
|
label_text = str(label or "").strip()
|
|
if not label_text:
|
|
return content
|
|
|
|
image = Image.open(BytesIO(content)).convert("RGB")
|
|
original_width, original_height = image.size
|
|
width = max(original_width, 860)
|
|
if width != original_width:
|
|
height = round(original_height * width / original_width)
|
|
image = image.resize((width, height), Image.Resampling.LANCZOS)
|
|
else:
|
|
height = original_height
|
|
|
|
font_size = max(54, min(78, width // 12))
|
|
font = _load_label_font(font_size)
|
|
line_gap = max(10, font_size // 4)
|
|
horizontal_padding = max(42, width // 14)
|
|
vertical_padding = max(28, width // 18)
|
|
|
|
measure = ImageDraw.Draw(Image.new("RGB", (width, 10), "white"))
|
|
lines = _wrap_label_text(measure, label_text, font, width - horizontal_padding * 2)
|
|
if len(lines) > 5:
|
|
lines = lines[:5]
|
|
ellipsis = "..."
|
|
while lines[-1] and _text_width(measure, f"{lines[-1]}{ellipsis}", font) > width - horizontal_padding * 2:
|
|
lines[-1] = lines[-1][:-1]
|
|
lines[-1] = f"{lines[-1]}{ellipsis}"
|
|
|
|
line_height = font_size + line_gap
|
|
label_height = vertical_padding * 2 + line_height * max(1, len(lines)) - line_gap
|
|
canvas = Image.new("RGB", (width, height + label_height), "white")
|
|
canvas.paste(image, (0, 0))
|
|
|
|
draw = ImageDraw.Draw(canvas)
|
|
y = height + vertical_padding
|
|
for line in lines:
|
|
draw.text((horizontal_padding, y), line, font=font, fill=(20, 32, 48))
|
|
y += line_height
|
|
|
|
output = BytesIO()
|
|
if suffix.lower() in {".jpg", ".jpeg"}:
|
|
canvas.save(output, format="JPEG", quality=95)
|
|
else:
|
|
canvas.save(output, format="PNG")
|
|
return output.getvalue()
|
|
|
|
|
|
async def get_access_token() -> str:
|
|
if not settings.wechat_app_id or not settings.wechat_app_secret:
|
|
raise WechatConfigError("未配置 WECHAT_APP_ID 或 WECHAT_APP_SECRET")
|
|
|
|
params = {
|
|
"grant_type": "client_credential",
|
|
"appid": settings.wechat_app_id,
|
|
"secret": settings.wechat_app_secret,
|
|
}
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
response = await client.get("https://api.weixin.qq.com/cgi-bin/token", params=params)
|
|
data = response.json()
|
|
token = data.get("access_token")
|
|
if not token:
|
|
raise RuntimeError(data.get("errmsg") or "获取微信 access_token 失败")
|
|
return token
|
|
|
|
|
|
async def get_phone_number(phone_code: str) -> str:
|
|
token = await get_access_token()
|
|
url = f"https://api.weixin.qq.com/wxa/business/getuserphonenumber?access_token={token}"
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
response = await client.post(url, json={"code": phone_code})
|
|
data = response.json()
|
|
phone_info = data.get("phone_info") or {}
|
|
phone = phone_info.get("phoneNumber") or phone_info.get("purePhoneNumber")
|
|
if not phone:
|
|
raise RuntimeError(data.get("errmsg") or "获取微信手机号失败")
|
|
return phone
|
|
|
|
|
|
async def create_device_qrcode(
|
|
device_no: str,
|
|
page: str,
|
|
scene: str,
|
|
public_base_url: str,
|
|
label: str | None = None,
|
|
) -> str | None:
|
|
return await create_miniapp_qrcode(
|
|
key=device_no,
|
|
page=page,
|
|
scene=scene,
|
|
public_base_url=public_base_url,
|
|
directory="qrcodes",
|
|
label=label,
|
|
)
|
|
|
|
|
|
async def create_miniapp_qrcode(
|
|
key: str,
|
|
page: str,
|
|
scene: str,
|
|
public_base_url: str,
|
|
directory: str,
|
|
label: str | None = None,
|
|
) -> str | None:
|
|
token = await get_access_token()
|
|
url = f"https://api.weixin.qq.com/wxa/getwxacodeunlimit?access_token={token}"
|
|
payload = {
|
|
"scene": scene,
|
|
"page": page,
|
|
"check_path": False,
|
|
"env_version": settings.wechat_qr_env_version,
|
|
}
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
response = await client.post(url, json=payload)
|
|
|
|
content_type = response.headers.get("content-type", "")
|
|
if "application/json" in content_type:
|
|
data = response.json()
|
|
raise RuntimeError(data.get("errmsg") or "生成小程序码失败")
|
|
|
|
qr_dir = settings.upload_path / directory
|
|
qr_dir.mkdir(parents=True, exist_ok=True)
|
|
suffix = _qrcode_file_suffix(content_type)
|
|
file_name = _qrcode_file_name(key, suffix)
|
|
file_path = Path(qr_dir) / file_name
|
|
label_text = label if label is not None else key
|
|
file_path.write_bytes(_draw_qrcode_label(response.content, suffix, label_text))
|
|
encoded_directory = quote(directory.strip("/"), safe="")
|
|
encoded_file_name = quote(file_name, safe="")
|
|
return f"{public_base_url.rstrip('/')}/uploads/{encoded_directory}/{encoded_file_name}"
|