spider_work/jiwu_spider/auto_send_wx_msg.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.12.10
# Date   : 2026/08/18
"""企业微信群机器人通用发送模块（文本 / 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/18 从 deca 项目照搬到集物星球项目；WEBHOOK_URL 待替换为本项目专用群机器人。
    2026/08/27 上传/发送增加 tenacity 重试，专治 TLS 瞬断（UNEXPECTED_EOF_WHILE_READING）导致的偶发发送失败。
"""
import os
import re
import json

import requests
from loguru import logger
from tenacity import (retry, stop_after_attempt, wait_exponential,
                      retry_if_exception_type)

# 企业微信群机器人 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=2d41b34f-8e30-479b-a39d-331f1d5c233f"
# 素材上传接口：发文件/图片前先把素材传上去换 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

# ---- 重试配置：仅对网络/TLS 类瞬断重试，业务错误码(errcode!=0)不重试 ----
RETRY_TIMES = 4          # 最多尝试次数（含首次）
RETRY_WAIT_MIN = 2       # 退避最小间隔（秒）
RETRY_WAIT_MAX = 8       # 退避最大间隔（秒），间隔按 2/4/8 递增


def after_log(retry_state):
    """tenacity 重试回调，记录每次尝试的结果。

    约定：被 @retry 装饰的函数首个位置参数为 log（日志对象），本回调据此取用。

    Args:
        retry_state: tenacity 传入的 RetryCallState，含调用参数与结果。
    """
    log = retry_state.args[0] if retry_state.args else logger
    if retry_state.outcome.failed:
        log.warning(f"Function '{retry_state.fn.__name__}', Attempt {retry_state.attempt_number} Times")
    else:
        log.info(f"Function '{retry_state.fn.__name__}', Attempt {retry_state.attempt_number} succeeded")


@retry(stop=stop_after_attempt(RETRY_TIMES),
       wait=wait_exponential(multiplier=1, min=RETRY_WAIT_MIN, max=RETRY_WAIT_MAX),
       retry=retry_if_exception_type(requests.exceptions.RequestException),
       after=after_log, reraise=True)
def _do_upload_media(log, url: str, file_path: str) -> requests.Response:
    """执行素材上传的内部函数，带重试；每次尝试都重新打开文件。

    重点：multipart 上传的文件流被读过一次就到 EOF，重试必须重新 open 文件，
    否则第二次会传空体。故把 open 放进被重试的函数体内，保证每次从头读。

    Args:
        log: 日志对象（首参约定，供 after_log 取用）。
        url (str): upload_media 接口地址（已含 key 与 type）。
        file_path (str): 待上传文件路径。

    Returns:
        requests.Response: HTTP 状态正常的响应对象。

    Raises:
        requests.exceptions.RequestException: 连接/TLS/超时/HTTP 状态异常，触发重试。
    """
    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()  # HTTP 非 2xx 抛 HTTPError(属 RequestException)，触发重试
    return resp


@retry(stop=stop_after_attempt(RETRY_TIMES),
       wait=wait_exponential(multiplier=1, min=RETRY_WAIT_MIN, max=RETRY_WAIT_MAX),
       retry=retry_if_exception_type(requests.exceptions.RequestException),
       after=after_log, reraise=True)
def _post_json_with_retry(log, url: str, **kwargs) -> requests.Response:
    """带重试的 JSON POST（发文本 / 文件消息共用）。

    JSON 请求体是 bytes，可安全重试；仅网络/TLS 类异常触发重试，业务错误码由调用方判断。

    Args:
        log: 日志对象（首参约定，供 after_log 取用）。
        url (str): 请求地址。
        **kwargs: 透传给 requests.post 的参数（headers / data / timeout 等）。

    Returns:
        requests.Response: HTTP 状态正常的响应对象。

    Raises:
        requests.exceptions.RequestException: 连接/TLS/超时/HTTP 状态异常，触发重试。
    """
    resp = requests.post(url, **kwargs)
    resp.raise_for_status()
    return resp


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 = _post_json_with_retry(log, WEBHOOK_URL, headers=headers,
                                     data=json.dumps(data, ensure_ascii=False).encode("utf-8"),
                                     timeout=(5, 30))
        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"企微消息发送失败(重试{RETRY_TIMES}次后仍失败): {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:
        resp = _do_upload_media(log, url, file_path)  # 带重试上传，内部每次重新打开文件
        result = resp.json()
        if result.get("errcode") != 0:  # 业务错误(如 key 失效/文件超限)，重试无意义，直接失败
            log.error(f"企微素材上传失败: {result}")
            return None
        return result.get("media_id")
    except requests.exceptions.RequestException as e:
        log.error(f"企微素材上传异常(重试{RETRY_TIMES}次后仍失败): {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 = _post_json_with_retry(log, WEBHOOK_URL, headers={"Content-Type": "application/json"},
                                     data=json.dumps(data).encode("utf-8"), timeout=(5, 30))
        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"企微文件发送失败(重试{RETRY_TIMES}次后仍失败): {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="🧪 集物星球企微机器人自测",
    )