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: 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 / "qrcodes" qr_dir.mkdir(parents=True, exist_ok=True) suffix = _qrcode_file_suffix(content_type) file_name = _qrcode_file_name(device_no, suffix) file_path = Path(qr_dir) / file_name file_path.write_bytes(_draw_qrcode_label(response.content, suffix, label or device_no)) return f"{public_base_url.rstrip('/')}/uploads/qrcodes/{quote(file_name, safe='')}"