# -*- coding: utf-8 -*- # Author : Charley # Python : 3.12.10 # Date : 2026/08/04 """企业微信群机器人通用发送模块（文本 / markdown_v2 / 文件）。 对外三个入口： - send_wechat_group_msg ：发文本或 markdown_v2 文案，items 元素可为字符串或 (名称, 链接) 元组。 - send_wechat_group_file：发文件（先 upload_media 拿 media_id，再发 file 消息），供报表脚本发 Excel。 作为工具库被其它脚本 import，不在模块级配置 loguru sink，日志默认落到调用方的 logger。 变更记录： 2026/08/11 从 PC 版微信(wxauto)迁到企业微信群机器人；新增 send_wechat_group_file 发文件能力。 """ import os import re import json import requests from loguru import logger # 企业微信群机器人 Webhook 地址（key 为群机器人凭证，按需替换；当前为测试群） # WEBHOOK_URL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=b8d398e2-f27e-42ce-af78-336867460122" # 测试 WEBHOOK_URL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=fc5619c2-5699-485f-bed1-a5fdc79ca513" # 素材上传接口：发文件/图片前先把素材传上去换 media_id（type=file/voice；文件 5B~20MB，media_id 有效期 3 天） UPLOAD_URL_TMPL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/upload_media?key={key}&type={media_type}" FILE_MIN_BYTES = 5 # 企微限制：文件不得小于 5 字节 FILE_MAX_BYTES = 20 * 1024 * 1024 # 企微限制：文件不得大于 20MB def _extract_key(webhook_url: str) -> str | None: """从群机器人 Webhook 地址里抽出 key（上传素材接口要单独拼 key）。 Args: webhook_url (str): 形如 ...webhook/send?key=xxxx 的 Webhook 地址。 Returns: str | None: 抽到的 key；地址不含 key 时返回 None。 """ m = re.search(r"key=([0-9a-fA-F\-]+)", webhook_url) return m.group(1) if m else None def build_markdown_content(items: list, title: str) -> str: """把列表拼成 markdown 文案（每条编号，条间加分割线）。 Args: items (list): 元素为字符串或 (名称, 链接) 元组。 title (str): 文案标题（渲染为四级标题）。 Returns: str: 拼好的 markdown 文本。 """ md_content = f"#### {title}\n" for i, item in enumerate(items, 1): if isinstance(item, tuple) and len(item) == 2: name, link = item md_content += f"{i}. [{name}]({link})\n" elif isinstance(item, str): md_content += f"{i}. {item}\n" else: md_content += f"{i}. {str(item)}\n" if i < len(items): md_content += "\n---\n\n" return md_content def build_text_content(items: list) -> str: """把列表拼成纯文本文案（每条编号，条间加分割线）。 Args: items (list): 元素为字符串或 (名称, 链接) 元组。 Returns: str: 拼好的纯文本。 """ content = "" for i, item in enumerate(items, 1): if isinstance(item, tuple) and len(item) == 2: name, link = item content += f"{i}. {name}: {link}\n" elif isinstance(item, str): content += f"{i}. {item}\n" else: content += f"{i}. {str(item)}\n" if i < len(items): content += "----------------------------------\n" return content def send_wechat_group_msg(log=None, items=None, mentioned_list=None, msg_type="markdown", title="🚀 提醒通知") -> dict | None: """发送文本 / markdown_v2 消息到企业微信群机器人。 Args: log (loguru.Logger, optional): 日志对象；不传用全局 logger。Defaults to None。 items (list, optional): 消息条目，元素为字符串或 (名称, 链接) 元组。Defaults to None。 mentioned_list (list, optional): text 类型下 @ 的成员手机号/@all 列表。Defaults to None。 msg_type (str, optional): 消息类型 text / markdown。Defaults to "markdown"。 title (str, optional): markdown 文案标题。Defaults to "🚀 提醒通知"。 Returns: dict | None: 企微返回的 JSON；发送失败返回 None。 """ if items is None: items = [] if log is None: log = logger headers = {"Content-Type": "application/json"} if msg_type == "text": data = { "msgtype": "text", "text": { "content": build_text_content(items), "mentioned_list": mentioned_list if mentioned_list else [], }, } else: # 默认 markdown_v2 data = { "msgtype": "markdown_v2", "markdown_v2": {"content": build_markdown_content(items, title)}, } try: log.info(f"正在发送企微消息: {title}") resp = requests.post(WEBHOOK_URL, headers=headers, data=json.dumps(data, ensure_ascii=False).encode("utf-8"), timeout=(5, 30)) resp.raise_for_status() result = resp.json() if result.get("errcode") not in (0, None): # 企微业务错误码非 0 也算失败 log.error(f"企微消息发送失败: {result}") return None log.success("企微消息发送成功") return result except requests.exceptions.RequestException as e: log.error(f"企微消息发送失败: {e}") return None def _upload_media(log, file_path: str, media_type: str = "file") -> str | None: """把本地文件上传到企微群机器人素材接口，换取 media_id。 Args: log (loguru.Logger): 日志对象。 file_path (str): 本地文件绝对/相对路径。 media_type (str, optional): 素材类型 file / voice。Defaults to "file"。 Returns: str | None: 上传成功返回 media_id（有效期 3 天）；文件不存在/超限/上传失败返回 None。 """ key = _extract_key(WEBHOOK_URL) if not key: log.error("Webhook 地址里没解析到 key，无法上传素材") return None size = os.path.getsize(file_path) if not (FILE_MIN_BYTES <= size <= FILE_MAX_BYTES): # 企微限制 5B~20MB log.error(f"文件大小 {size} 字节超出企微限制(5B~20MB)：{file_path}") return None url = UPLOAD_URL_TMPL.format(key=key, media_type=media_type) try: with open(file_path, "rb") as f: # 素材字段名必须为 media，且要带文件名（群里展示的就是这个名字） files = {"media": (os.path.basename(file_path), f, "application/octet-stream")} resp = requests.post(url, files=files, timeout=(5, 60)) resp.raise_for_status() result = resp.json() if result.get("errcode") != 0: log.error(f"企微素材上传失败: {result}") return None return result.get("media_id") except requests.exceptions.RequestException as e: log.error(f"企微素材上传异常: {e}") return None def send_wechat_group_file(log=None, file_path: str = None) -> dict | None: """发送一个本地文件到企业微信群机器人（自动先上传素材换 media_id 再发 file 消息）。 Args: log (loguru.Logger, optional): 日志对象；不传用全局 logger。Defaults to None。 file_path (str, optional): 待发送文件路径（如 Excel 报表）。Defaults to None。 Returns: dict | None: 企微返回的 JSON；文件不存在/上传失败/发送失败返回 None。 """ if log is None: log = logger if not file_path or not os.path.isfile(file_path): log.error(f"待发送文件不存在: {file_path}") return None media_id = _upload_media(log, file_path, "file") if not media_id: return None data = {"msgtype": "file", "file": {"media_id": media_id}} try: log.info(f"正在发送企微文件: {os.path.basename(file_path)}") resp = requests.post(WEBHOOK_URL, headers={"Content-Type": "application/json"}, data=json.dumps(data).encode("utf-8"), timeout=(5, 30)) resp.raise_for_status() result = resp.json() if result.get("errcode") != 0: log.error(f"企微文件发送失败: {result}") return None log.success(f"企微文件发送成功: {os.path.basename(file_path)}") return result except requests.exceptions.RequestException as e: log.error(f"企微文件发送失败: {e}") return None if __name__ == "__main__": # 自测：发送一条示例 markdown 消息 import sys logger.remove() logger.add(sys.stderr, level="INFO") send_wechat_group_msg( items=["示例商品A（单价¥199，100份）", "示例商品B（单价¥299，50份）"], title="🧪 企微机器人自测", )