如何用腾讯云 SCF 部署企微机器人定时推送小程序云测结果（亲测无坑分享）

# 从 utils 模块导入四个工具函数，分别用于读取配置、获取任务信息、构建消息和发送消息
from utils import load_config, get_latest_task_info, build_markdown_message, send_wecom_message_with_retry


# SCF 云函数主入口函数，event 和 context 是平台传入的参数
def main_handler(event=None, context=None):
    # 读取本地配置文件 config.yaml
    config = load_config("config.yaml")

    # 调用接口获取最新的测试任务信息
    task_info = get_latest_task_info(config)

    if not task_info:
        print("未获取到最新任务信息")
        return

    # 构建消息内容
    message = build_markdown_message(config, task_info)

    # 发送消息到企业微信，失败时自动重试
    send_wecom_message_with_retry(config, message)

import yaml
import time
import requests


def load_config(path):
    with open(path, 'r', encoding='utf-8') as f:
        return yaml.safe_load(f)


def get_latest_task_info(config):
    # 任务列表接口
    url = "https://minitest.weixin.qq.com/thirdapi/plan_list"
    params = {
        "token": config["minitest"]["token"],
        "group_en_id": config["minitest"]["group_en_id"],
        "page_index": 1,
        "page_size": 1  # 只取最新一条
    }
    try:
        # 发 GET 请求拿到任务列表
        resp = requests.get(url, params=params)
        data = resp.json()

        # 如果接口返回成功且有任务
        if data["rtn"] == 0 and data["data"]["plan_list"]:
            # 取第一条任务
            task = data["data"]["plan_list"][0]
            # 再查这个任务的测试报告链接
            report = requests.get("https://minitest.weixin.qq.com/thirdapi/share_url", params={
                "token": config["minitest"]["token"],
                "group_en_id": config["minitest"]["group_en_id"],
                "plan_id": task["id"]
            }).json()
            # 如果拿到了报告链接，存到任务里
            task["report_url"] = report["data"]["share_url"] if report["rtn"] == 0 else ""
            return task
    except Exception as e:
        print("获取任务失败：", e)
    return None


def build_markdown_message(config, task):
    # 从配置中读取消息模板
    msg_tpl = config["message"]["template"]
    # 获取要 @ 的人列表
    at_list = config["message"].get("at_list", [])
    # 转成企业微信支持的 @ 格式
    at_str = " ".join([f"<@{uid}>" for uid in at_list])

    # 填充消息模板
    message = msg_tpl.format(
        id=task["id"],
        platform=task.get("platforms", ""),
        rate=f"{task.get('success_rate', 0)*100:.1f}%" if task.get("success_rate") else "",
        url=task.get("report_url", ""),
        at=at_str
    )
    return message


def send_wecom_message_with_retry(config, content, retries=3):
    webhook = config["wecom"]["webhook"]
    data = {
        "msgtype": "markdown",
        "markdown": {"content": content}
    }

    # 最多重试3次
    for attempt in range(retries):
        try:
            # 发 POST 请求发送消息
            resp = requests.post(webhook, json=data)
            if resp.status_code == 200 and resp.json().get("errcode") == 0:
                print("消息发送成功")
                return
        except Exception as e:
            print(f"发送失败，重试中（{attempt+1}/{retries}）：", e)
        # 每次失败等待 3 秒后重试
        time.sleep(3)
    print("消息发送失败")