spider_work/cxx_spider/stats/auto_send_wx_msg.py

# -*- 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="🧪 企微机器人自测",
    )