#!/usr/bin/env python3 """在服务器上生成通知图片并上传到企微""" import asyncio import sys sys.path.insert(0, '/app') from PIL import Image, ImageDraw, ImageFont from app.services.wecom_service import WecomService async def generate_image(): """生成通知图片""" WIDTH, HEIGHT = 600, 300 img = Image.new('RGB', (WIDTH, HEIGHT)) draw = ImageDraw.Draw(img) # 渐变背景 for y in range(HEIGHT): ratio = y / HEIGHT r = int(0 + (0 - 0) * ratio) g = int(120 + (180 - 120) * ratio) b = int(215 + (160 - 215) * ratio) draw.line([(0, y), (WIDTH, y)], fill=(r, g, b)) # 字体 try: title_font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 32) subtitle_font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", 20) button_font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 24) except: title_font = ImageFont.load_default() subtitle_font = ImageFont.load_default() button_font = ImageFont.load_default() # 标题 draw.text((30, 30), "IT咨询即将关闭", fill=(255, 255, 255), font=title_font) draw.text((30, 85), "请尽快回复坐席,否则将自动结束", fill=(220, 220, 220), font=subtitle_font) # 3D按钮 btn_x, btn_y = 30, 150 btn_w, btn_h = 540, 60 # 阴影 draw.rounded_rectangle([btn_x+3, btn_y+3, btn_x+btn_w+3, btn_y+btn_h+3], radius=12, fill=(100, 100, 100)) # 按钮主体 for i in range(btn_h): ratio = i / btn_h r = int(20 + (100 - 20) * ratio) g = int(150 + (200 - 150) * ratio) b = int(220 + (255 - 220) * ratio) draw.rectangle([btn_x, btn_y+i, btn_x+btn_w, btn_y+i+1], fill=(r, g, b)) # 高光 draw.rounded_rectangle([btn_x, btn_y, btn_x+btn_w, btn_y+btn_h//3], radius=12, fill=(255, 255, 255, 60)) # 按钮文字 draw.text((btn_x + btn_w//2 - 60, btn_y + 18), "立即回复", fill=(255, 255, 255), font=button_font) # 保存到内存 from io import BytesIO buffer = BytesIO() img.save(buffer, format='PNG') buffer.seek(0) return buffer.getvalue() async def upload_and_send(): """上传图片并发送消息""" print("🎨 正在生成通知图片...") image_data = await generate_image() print(f" 图片已生成, 大小: {len(image_data)} bytes") # 创建企微服务 wecom = WecomService() try: print("📤 正在上传图片到企微...") # 上传临时素材 - 传入二进制数据 media_id = await wecom.upload_temp_media("image", image_data, "reminder.png") if not media_id or media_id.startswith("ERROR"): print(f"❌ 上传失败: {media_id}") return print(f" 上传成功! media_id: {media_id}") print("📨 正在发送图片消息...") # 发送图片消息 result = await wecom.send_image_message("sxn", media_id) if result.get("errcode") == 0: print(f"✅ 发送成功! msgid: {result.get('msgid')}") else: print(f"❌ 发送失败: {result}") finally: await wecom.close() if __name__ == "__main__": asyncio.run(upload_and_send())