影刀RPA进阶教程：企业短信邮件批量通知平台搭建与发送管理

痛点切入

每个企业都有通知需求：系统故障告警、审批超时提醒、合同到期催办、活动通知推送。问题是每个系统各自为战，有的用钉钉机器人，有的发邮件，有的啥也没有全靠人肉通知。运维要记5种通知方式的配置，市场部群发个活动短信还得找IT帮忙。

实际上，这些通知场景可以统一到一个平台上——一个小小的Python服务+影刀RPA，就能做成一个支持短信（阿里云/腾讯云）、邮件（SMTP）、钉钉/飞书消息的统一通知网关。各业务系统只需要调一个接口，剩下的路由、重试、记录全自动。

这篇文章搭建一个完整的"企业级通知中台"，对接阿里云短信+SMTP邮件+钉钉机器人，代码直接可部署。

---

核心架构

业务系统（影刀流程/ERP/CRM）
    ↓ HTTP POST /send
通知网关（Flask服务）
    ↓ 路由判断
├─ 短信通道（阿里云SMS）
├─ 邮件通道（SMTP）
└─ IM通道（钉钉/飞书机器人）
![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/076756fb7ae0426698bace50a4297726.png#pic_center)

    ↓
发送记录 → SQLite/MySQL → 报表
    ↓
失败重试 + 限流保护

---

完整通知平台代码

"""
企业统一通知平台
支持：阿里云短信、腾讯云短信、SMTP邮件、钉钉机器人、飞书消息
作者：林焱

部署方式：
1. 安装依赖：pip install flask requests alibabacloud_sdk_dysmsapi
2. 启动服务：python notification_platform.py
3. 配置为Windows服务或Linux systemd自启动
"""
from flask import Flask, request, jsonify
import requests
import json
import time
import hashlib
import hmac
import base64
import sqlite3
import smtplib
import threading
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders
from datetime import datetime, timedelta
from urllib.parse import quote_plus

app = Flask(__name__)


# ==================== 数据库初始化 ====================

def init_db():
    """初始化通知记录数据库"""
    conn = sqlite3.connect("notification_platform.db", check_same_thread=False)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS notification_log (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            channel TEXT NOT NULL,              -- sms/email/dingtalk/feishu
            recipient TEXT NOT NULL,            -- 接收人
            content TEXT NOT NULL,              -- 消息内容
            status TEXT NOT NULL,               -- sent/failed/retrying
            error_msg TEXT,                     -- 失败原因
            send_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            retry_count INTEGER DEFAULT 0,
            business_id TEXT,                   -- 业务关联ID
            template_id TEXT                    -- 模板ID
        )
    """)
    conn.commit()
    return conn


# ==================== 短信通道：阿里云SMS ====================

class AliyunSMS:
    """
    阿里云短信服务
    前置条件：
    1. 开通阿里云短信服务：https://www.aliyun.com/product/sms
    2. 申请短信签名和模板（需审核）
    3. 获取AccessKey：RAM访问控制 → 创建AccessKey
    """

    def __init__(self, access_key_id, access_key_secret, sign_name):
        """
        :param access_key_id: 阿里云AccessKey ID
        :param access_key_secret: 阿里云AccessKey Secret
        :param sign_name: 短信签名（如"XX科技"）需在阿里云后台审核通过
        """
        from alibabacloud_dysmsapi20170525.client import Client
        from alibabacloud_dysmsapi20170525 import models as dysmsapi_models
        from alibabacloud_tea_openapi import models as open_api_models

        config = open_api_models.Config(
            access_key_id=access_key_id,
            access_key_secret=access_key_secret,
        )
        config.endpoint = "dysmsapi.aliyuncs.com"
        self.client = Client(config)
        self.sign_name = sign_name
        self.models = dysmsapi_models

    def send(self, phone_numbers, template_code, template_params):
        """
        发送短信
        :param phone_numbers: 手机号（字符串，多号码逗号分隔）
        :param template_code: 阿里云短信模板CODE（如"SMS_123456789"）
        :param template_params: 模板变量 {"code": "1234"} 需与模板中变量名一致
        :return: {"success": True/False, "biz_id": "...", "error": "..."}

        踩坑：
        1. 手机号格式：国内号码需加86前缀或不加
           建议统一格式：'13800138000' 或 '8613800138000'
        2. 模板变量值不能含特殊字符（<>等）
        3. 单次最多1000个手机号（实际建议≤200）
        """
        try:
            send_req = self.models.SendSmsRequest(
                phone_numbers=phone_numbers,
                sign_name=self.sign_name,
                template_code=template_code,
                template_param=json.dumps(template_params, ensure_ascii=False)
            )
            response = self.client.send_sms(send_req)

            if response.body.code == "OK":
                return {
                    "success": True,
                    "biz_id": response.body.biz_id or "",
                    "error": ""
                }
            else:
                return {
                    "success": False,
                    "biz_id": "",
                    "error": f"Code:{response.body.code}, Msg:{response.body.message}"
                }
        except Exception as e:
            return {"success": False, "biz_id": "", "error": str(e)}

    def batch_send(self, phone_list, template_code, template_params_func):
        """
        批量发送（逐个发送，支持不同内容）
        :param phone_list: [{"phone": "13800138000", "name": "张三"}, ...]
        :param template_code: 短信模板CODE
        :param template_params_func: 函数，接收phone_info返回模板参数dict
        """
        results = []
        success_count = 0
        fail_count = 0

        for phone_info in phone_list:
            params = template_params_func(phone_info)
            result = self.send(
                phone_numbers=phone_info["phone"],
                template_code=template_code,
                template_params=params
            )
            result["recipient"] = phone_info["phone"]
            result["name"] = phone_info.get("name", "")

            if result["success"]:
                success_count += 1
            else:
                fail_count += 1

            results.append(result)
            # 短信发送限速：阿里云默认限制单号码QPS（300次/分钟）
            time.sleep(0.2)

        return {
            "total": len(phone_list),
            "success": success_count,
            "failed": fail_count,
            "details": results
        }

    def query_send_status(self, phone_number, send_date, biz_id,
                          page_size=10):
        """
        查询短信发送状态（用于确认用户是否收到）
        :param phone_number: 手机号
        :param send_date: 发送日期 "20240115"
        :param biz_id: 发送ID
        """
        try:
            query_req = self.models.QuerySendDetailsRequest(
                phone_number=phone_number,
                send_date=send_date,
                page_size=page_size,
                current_page=1
            )
            if biz_id:
                query_req.biz_id = biz_id

            response = self.client.query_send_details(query_req)

            if response.body.code == "OK":
                details = response.body.sms_send_detail_dtos.sms_send_detail_dto
                return [{
                    "phone": d.phone_num,
                    "status": d.send_status,  # 1=等待回执, 2=发送失败, 3=发送成功
                    "err_code": d.err_code,
                    "content": d.content,
                    "send_date": d.send_date,
                    "receive_date": d.receive_date
                } for d in details]
            return []
        except Exception as e:
            return [{"error": str(e)}]


# ==================== 短信通道：腾讯云SMS ====================

class TencentSMS:
    """
    腾讯云短信服务
    与阿里云的主要区别：
    1. 签名方式：腾讯云使用TC3-HMAC-SHA256签名
    2. 模板ID格式：纯数字（如 1234567）
    3. 手机号格式：必须带+86前缀
    """

    def __init__(self, secret_id, secret_key, app_id, sign_name):
        """
        :param secret_id: 腾讯云SecretId
        :param secret_key: 腾讯云SecretKey
        :param app_id: SMS应用ID（腾讯云控制台 → 短信 → 应用管理）
        :param sign_name: 短信签名内容（需审核通过）
        """
        self.secret_id = secret_id
        self.secret_key = secret_key
        self.app_id = app_id
        self.sign_name = sign_name
        self.endpoint = "sms.tencentcloudapi.com"

    def _sign(self, payload, timestamp):
        """
        腾讯云TC3-HMAC-SHA256签名算法
        这是踩坑最多的地方——签名字符串必须严格按规范拼接
        """
        # 1. 拼接规范请求串
        http_request_method = "POST"
        canonical_uri = "/"
        canonical_querystring = ""
        canonical_headers = (
            f"content-type:application/json\n"
            f"host:{self.endpoint}\n"
        )
        signed_headers = "content-type;host"
        hashed_request_payload = hashlib.sha256(
            payload.encode("utf-8")
        ).hexdigest()
        canonical_request = (
            f"{http_request_method}\n{canonical_uri}\n"
            f"{canonical_querystring}\n{canonical_headers}\n"
            f"{signed_headers}\n{hashed_request_payload}"
        )

        # 2. 拼接待签名字符串
        algorithm = "TC3-HMAC-SHA256"
        date = datetime.utcfromtimestamp(timestamp).strftime("%Y-%m-%d")
        service = "sms"
        credential_scope = f"{date}/{service}/tc3_request"
        hashed_canonical_request = hashlib.sha256(
            canonical_request.encode("utf-8")
        ).hexdigest()
        string_to_sign = (
            f"{algorithm}\n{timestamp}\n{credential_scope}\n"
            f"{hashed_canonical_request}"
        )

        # 3. 计算签名
        def _hmac_sha256(key, msg):
            return hmac.new(key, msg.encode("utf-8"),
                           hashlib.sha256).digest()

        secret_date = _hmac_sha256(f"TC3{self.secret_key}".encode(), date)
        secret_service = _hmac_sha256(secret_date, service)
        secret_signing = _hmac_sha256(secret_service, "tc3_request")
        signature = hmac.new(secret_signing,
                            string_to_sign.encode("utf-8"),
                            hashlib.sha256).hexdigest()

        # 4. 拼接Authorization
        authorization = (
            f"{algorithm} Credential={self.secret_id}/{credential_scope}, "
            f"SignedHeaders={signed_headers}, Signature={signature}"
        )

        return authorization

    def send(self, phone_numbers, template_id, template_params):
        """
        发送短信
        :param phone_numbers: 手机号列表 ["+8613800138000", "+8613800138001"]
              注意：必须带 +86 前缀！
        :param template_id: 模板ID（纯数字，如 "1234567"）
        :param template_params: 模板参数 ["张三", "1234"]（按模板变量顺序）
        """
        timestamp = int(time.time())
        payload = json.dumps({
            "PhoneNumberSet": phone_numbers,
            "SmsSdkAppId": self.app_id,
            "TemplateId": template_id,
            "TemplateParamSet": template_params,
            "SignName": self.sign_name,
        })

        authorization = self._sign(payload, timestamp)

        headers = {
            "Authorization": authorization,
            "Content-Type": "application/json",
            "Host": self.endpoint,
            "X-TC-Action": "SendSms",
            "X-TC-Timestamp": str(timestamp),
            "X-TC-Version": "2021-01-11",
            "X-TC-Region": "ap-guangzhou",  # 国内短信固定为ap-guangzhou
        }

        try:
            resp = requests.post(
                f"https://{self.endpoint}",
                headers=headers,
                data=payload,
                timeout=10
            )
            result = resp.json()
            response = result.get("Response", {})

            if response.get("Error"):
                return {
                    "success": False,
                    "error": f"{response['Error']['Code']}: {response['Error']['Message']}"
                }

            send_status = response.get("SendStatusSet", [{}])[0]
            return {
                "success": send_status.get("Code") == "Ok",
                "serial_no": send_status.get("SerialNo", ""),
                "fee": send_status.get("Fee", 0),
                "error": send_status.get("Message", "")
            }

        except Exception as e:
            return {"success": False, "error": str(e)}


# ==================== 邮件通道：SMTP ====================

class EmailSender:
    """SMTP邮件发送（支持附件）"""

    def __init__(self, smtp_server, smtp_port, username, password,
                 sender_name=""):
        """
        :param smtp_server: SMTP服务器地址
            常见配置：
            - QQ邮箱：smtp.qq.com, 端口465(SSL)或587(TLS)
            - 163邮箱：smtp.163.com, 端口465
            - 企业邮箱：mail.your-company.com, 端口587
            - 腾讯企业邮：smtp.exmail.qq.com, 端口465
        :param smtp_port: 端口号
        :param username: 发件人邮箱
        :param password: SMTP授权码（不是邮箱密码！）
            获取方式：
            - QQ邮箱：设置→账户→POP3/SMTP→生成授权码
            - 163邮箱：设置→POP3/SMTP/IMAP→新增授权码
        :param sender_name: 发件人显示名称
        """
        self.smtp_server = smtp_server
        self.smtp_port = smtp_port
        self.username = username
        self.password = password
        self.sender_name = sender_name or username

    def send(self, to_emails, subject, body, body_type="html",
             cc_emails=None, attachments=None):
        """
        发送邮件
        :param to_emails: 收件人列表 ["a@company.com", "b@company.com"]
        :param subject: 邮件主题
        :param body: 邮件正文（HTML格式）
        :param body_type: "html" 或 "plain"
        :param cc_emails: 抄送列表
        :param attachments: 附件列表 [("文件名.pdf", file_bytes), ...]
        :return: {"success": True/False, "error": ""}
        """
        try:
            msg = MIMEMultipart()
            msg["From"] = f"{self.sender_name} <{self.username}>"
            msg["To"] = ", ".join(to_emails)
            msg["Subject"] = subject
            if cc_emails:
                msg["Cc"] = ", ".join(cc_emails)

            # 添加正文
            msg.attach(MIMEText(body, body_type, "utf-8"))

            # 添加附件
            if attachments:
                for filename, file_content in attachments:
                    # 踩坑：文件名中的中文需要编码
                    part = MIMEBase("application", "octet-stream")
                    part.set_payload(file_content)
                    encoders.encode_base64(part)
                    # RFC 2231 编码处理中文文件名
                    from email.header import Header
                    part.add_header(
                        "Content-Disposition",
                        "attachment",
                        filename=("utf-8", "", filename)
                    )
                    msg.attach(part)

            # 发送
            if self.smtp_port == 465:
                # SSL连接
                server = smtplib.SMTP_SSL(
                    self.smtp_server, self.smtp_port, timeout=30
                )
            else:
                # TLS连接
                server = smtplib.SMTP(
                    self.smtp_server, self.smtp_port, timeout=30
                )
                server.starttls()

            server.login(self.username, self.password)
            all_recipients = to_emails + (cc_emails or [])
            server.sendmail(self.username, all_recipients, msg.as_string())
            server.quit()

            return {"success": True, "error": ""}

        except Exception as e:
            return {"success": False, "error": str(e)}

    def batch_send_from_excel(self, excel_path, template_subject,
                              template_body_func):
        """
        从Excel读取收件人列表批量发送
        Excel格式：姓名 | 邮箱 | 公司 | ...（自定义列）

        :param excel_path: Excel文件路径
        :param template_subject: 邮件主题模板（可用{字段名}占位）
        :param template_body_func: 函数(row_dict) → HTML正文
        """
        import pandas as pd
        df = pd.read_excel(excel_path)

        results = []
        for _, row in df.iterrows():
            row_dict = row.to_dict()
            to_email = row_dict.get("邮箱", "")

            if not to_email or "@" not in str(to_email):
                results.append({
                    "email": to_email,
                    "status": "skipped",
                    "error": "邮箱格式无效"
                })
                continue

            try:
                subject = template_subject
                for key, value in row_dict.items():
                    subject = subject.replace(f"{{{key}}}", str(value))

                body = template_body_func(row_dict)

                result = self.send(
                    to_emails=[str(to_email)],
                    subject=subject,
                    body=body
                )
                result["email"] = str(to_email)
                results.append({
                    "email": str(to_email),
                    "status": "success" if result["success"] else "failed",
                    "error": result.get("error", "")
                })

            except Exception as e:
                results.append({
                    "email": str(to_email),
                    "status": "failed",
                    "error": str(e)
                })

            # 发送间隔（避免触发SMTP限流）
            time.sleep(1)

        return results


# ==================== 钉钉机器人通道 ====================

class DingTalkNotifier:
    """钉钉自定义机器人消息推送"""

    def __init__(self, webhook_url, secret=None):
        """
        :param webhook_url: 机器人Webhook地址
        :param secret: 加签密钥（安全设置中选择"加签"方式）
        """
        self.webhook_url = webhook_url
        self.secret = secret

    def send_text(self, content, at_mobiles=None, at_all=False):
        """发送文本消息"""
        return self._send({
            "msgtype": "text",
            "text": {"content": content},
            "at": {
                "atMobiles": at_mobiles or [],
                "isAtAll": at_all
            }
        })

    def send_markdown(self, title, text):
        """发送Markdown消息"""
        return self._send({
            "msgtype": "markdown",
            "markdown": {
                "title": title,
                "text": text
            }
        })

    def send_link(self, title, text, message_url, pic_url=""):
        """发送链接消息（卡片）"""
        return self._send({
            "msgtype": "link",
            "link": {
                "title": title,
                "text": text,
                "messageUrl": message_url,
                "picUrl": pic_url,
            }
        })

    def _send(self, body):
        """发送消息（含加签逻辑）"""
        url = self.webhook_url

        if self.secret:
            timestamp = str(round(time.time() * 1000))
            string_to_sign = f"{timestamp}\n{self.secret}"
            hmac_code = hmac.new(
                self.secret.encode("utf-8"),
                string_to_sign.encode("utf-8"),
                digestmod=hashlib.sha256
            ).digest()
            sign = quote_plus(base64.b64encode(hmac_code))
            url = f"{self.webhook_url}&timestamp={timestamp}&sign={sign}"

        try:
            resp = requests.post(url, json=body, timeout=10)
            result = resp.json()
            return {
                "success": result.get("errcode") == 0,
                "error": result.get("errmsg", "")
            }
        except Exception as e:
            return {"success": False, "error": str(e)}


# ==================== 通知网关服务 ====================

# 初始化数据库
db_conn = init_db()

# 初始化各通知通道
sms_aliyun = AliyunSMS(
    access_key_id="your_access_key_id",
    access_key_secret="your_access_key_secret",
    sign_name="XX科技"
)

sms_tencent = TencentSMS(
    secret_id="your_secret_id",
    secret_key="your_secret_key",
    app_id="1400xxxxxx",  # 腾讯云SMS应用ID
    sign_name="XX科技"
)

email_sender = EmailSender(
    smtp_server="smtp.exmail.qq.com",
    smtp_port=465,
    username="notify@your-company.com",
    password="your_smtp_auth_code",
    sender_name="XX科技通知中心"
)

dingtalk = DingTalkNotifier(
    webhook_url=(
        "https://oapi.dingtalk.com/robot/send?"
        "access_token=xxxxxx"
    ),
    secret="your_robot_secret"
)


def log_notification(channel, recipient, content, status, error="",
                     business_id="", template_id=""):
    """记录通知日志"""
    cursor = db_conn.cursor()
    cursor.execute(
        "INSERT INTO notification_log "
        "(channel, recipient, content, status, error_msg, business_id, template_id) "
        "VALUES (?, ?, ?, ?, ?, ?, ?)",
        (channel, recipient,
         content[:500] if len(content) > 500 else content,  # 截断过长内容
         status, error, business_id, template_id)
    )
    db_conn.commit()


def retry_failed_notifications(max_retry=3):
    """重试失败的发送"""
    cursor = db_conn.cursor()
    cursor.execute(
        "SELECT id, channel, recipient, content, business_id, template_id, retry_count "
        "FROM notification_log "
        "WHERE status='failed' AND retry_count < ? "
        "AND send_time > datetime('now', '-24 hours')",
        (max_retry,)
    )
    failed_records = cursor.fetchall()

    for record in failed_records:
        rec_id, channel, recipient, content, biz_id, tpl_id, retries = record

        print(f"重试发送: {rec_id} → {recipient} (第{retries+1}次)")

        if channel == "dingtalk":
            result = dingtalk.send_text(content)
        # elif channel == "sms":
        #     result = sms_aliyun.send(...)
        elif channel == "email":
            result = email_sender.send(
                to_emails=[recipient],
                subject="重发通知",
                body=content
            )
        else:
            continue

        if result["success"]:
            cursor.execute(
                "UPDATE notification_log SET status='sent', retry_count=? WHERE id=?",
                (retries + 1, rec_id)
            )
        else:
            cursor.execute(
                "UPDATE notification_log SET retry_count=?, error_msg=? WHERE id=?",
                (retries + 1, result["error"][:200], rec_id)
            )

        db_conn.commit()
        time.sleep(1)  # 重试间隔


# ==================== Flask API端点 ====================

@app.route("/send", methods=["POST"])
def send_notification():
    """
    统一通知接口
    POST /send
    {
        "channel": "sms|email|dingtalk|feishu",
        "recipient": "13800138000 或 user@company.com 或 @all",
        "content": "消息内容",
        "template_id": "模板ID（短信必填）",
        "template_params": {"key": "value"},
        "business_id": "业务关联ID（可选）",
        "immediate": true  // 是否立即发送
    }
    """
    data = request.json
    if not data:
        return jsonify({"code": 400, "msg": "参数不能为空"})

    channel = data.get("channel", "")
    recipient = data.get("recipient", "")
    content = data.get("content", "")
    template_id = data.get("template_id", "")
    template_params = data.get("template_params", {})
    business_id = data.get("business_id", "")

    # 参数校验
    if not channel or not recipient:
        return jsonify({"code": 400, "msg": "channel和recipient为必传参数"})

    # 渠道限流检查（同一接收人10分钟内最多3条）
    cursor = db_conn.cursor()
    cursor.execute(
        "SELECT COUNT(*) FROM notification_log "
        "WHERE recipient=? AND send_time > datetime('now', '-10 minutes')",
        (recipient,)
    )
    recent_count = cursor.fetchone()[0]
    if recent_count >= 3:
        log_notification(channel, recipient, content, "rejected",
                        "10分钟内发送次数超限", business_id, template_id)
        return jsonify({"code": 429, "msg": "发送频率过高，请10分钟后再试"})

    # 路由到对应通道
    try:
        if channel == "sms":
            # 短信发送
            result = sms_aliyun.send(
                phone_numbers=recipient,
                template_code=template_id,
                template_params=template_params
            )

        elif channel == "email":
            # 邮件发送
            result = email_sender.send(
                to_emails=[recipient],
                subject=data.get("subject", "系统通知"),
                body=content,
                body_type="html"
            )

        elif channel == "dingtalk":
            # 钉钉消息
            at_mobiles = data.get("at_mobiles", [])
            if recipient == "@all":
                result = dingtalk.send_text(content, at_all=True)
            else:
                result = dingtalk.send_text(content, at_mobiles=at_mobiles)

        elif channel == "feishu":
            # 飞书消息（简化版）
            # 完整实现参考第5篇文章
            result = {"success": True, "error": ""}

        else:
            return jsonify({"code": 400, "msg": f"不支持的渠道: {channel}"})

        # 记录日志
        status = "sent" if result["success"] else "failed"
        log_notification(channel, recipient, content, status,
                        result.get("error", ""), business_id, template_id)

        if result["success"]:
            return jsonify({
                "code": 200,
                "msg": "发送成功",
                "data": result
            })
        else:
            return jsonify({
                "code": 500,
                "msg": f"发送失败: {result.get('error', '')}"
            })

    except Exception as e:
        log_notification(channel, recipient, content, "failed",
                        str(e), business_id, template_id)
        return jsonify({
            "code": 500,
            "msg": f"发送异常: {str(e)}"
        })


@app.route("/stats", methods=["GET"])
def get_statistics():
    """获取发送统计
    GET /stats?date=2024-01-15&channel=all
    """
    date = request.args.get("date",
        datetime.now().strftime("%Y-%m-%d"))
    channel = request.args.get("channel", "all")

    cursor = db_conn.cursor()

    if channel == "all":
        cursor.execute(
            "SELECT channel, status, COUNT(*) FROM notification_log "
            "WHERE date(send_time)=? GROUP BY channel, status",
            (date,)
        )
    else:
        cursor.execute(
            "SELECT channel, status, COUNT(*) FROM notification_log "
            "WHERE date(send_time)=? AND channel=? GROUP BY status",
            (date, channel)
        )

    rows = cursor.fetchall()
    stats = {}
    for row in rows:
        ch = row[0] if channel == "all" else channel
        if ch not in stats:
            stats[ch] = {"sent": 0, "failed": 0}
        stats[ch][row[1]] = row[2]

    # 发送成功率
    for ch in stats:
        total = stats[ch]["sent"] + stats[ch]["failed"]
        stats[ch]["success_rate"] = (
            stats[ch]["sent"] / total * 100 if total > 0 else 0
        )

    return jsonify({
        "code": 200,
        "data": {
            "date": date,
            "statistics": stats
        }
    })


# ==================== 后台重试任务 ====================

def retry_loop():
    """后台定时重试失败通知"""
    while True:
        try:
            retry_failed_notifications(max_retry=3)
        except Exception as e:
            print(f"重试任务异常: {e}")
        time.sleep(300)  # 每5分钟重试一次


@app.route("/health", methods=["GET"])
def health_check():
    """健康检查"""
    return jsonify({
        "code": 200,
        "status": "running",
        "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    })


if __name__ == "__main__":
    # 启动后台重试线程
    retry_thread = threading.Thread(target=retry_loop, daemon=True)
    retry_thread.start()

    # 启动Flask服务
    print("=" * 50)
    print("企业通知平台已启动")
    print("API地址: http://127.0.0.1:5000/send")
    print("统计接口: http://127.0.0.1:5000/stats")
    print("=" * 50)

    app.run(host="0.0.0.0", port=5000, debug=False)

---

影刀RPA对接方式

影刀流程中调用通知平台：

1. 影刀设计器 → 添加"HTTP请求"活动
2. 配置：
   - URL: http://127.0.0.1:5000/send
   - 方法: POST
   - Content-Type: application/json
   - Body:
     {
       "channel": "dingtalk",
       "recipient": "@all",
       "content": "【RPA通知】流程执行完毕，共同步153条数据"
     }
3. 输出变量: response → 判断code是否为200

---

常见问题速查

#	现象	原因	解决方式
1	阿里云短信返回isv.BUSINESS_LIMIT_CONTROL	触发业务限流（单号码每小时≤5条）	添加频率控制；同一号码1小时内最多发3条
2	腾讯云短信签名请求失败	未使用TC3签名算法或timestamp过期	确认使用TC3-HMAC-SHA256，timestamp为当前时间戳（秒级）
3	SMTP邮件发送535 Authentication Failed	授权码错误，或使用的是邮箱密码而非授权码	QQ/163邮箱需在设置中生成"SMTP授权码"（非密码）
4	钉钉机器人消息发送成功但群内不显示	未将机器人添加到目标群	群设置 → 智能群助手 → 添加机器人 → 复制Webhook地址
5	批量短信发送到第51条时失败	单号码单小时发送量超限（阿里云5条/小时/号码）	批量发送时分批处理，每批间隔≥12分钟
6	Flask启动后外部无法访问	绑定了127.0.0.1	改用host="0.0.0.0"；检查Windows防火墙是否拦截5000端口
7	短信内容变量包含特殊字符报错	模板变量不能含<>等特殊字符	template_params中做过滤re.sub(r'[<>]', '', value)

---

推荐资源

• 阿里云短信服务SDK

• 腾讯云短信API文档

• 钉钉自定义机器人文档

• 

• 影刀社区：搜索"通知平台"可找到HTTP请求配置示例

---

安全提醒：AccessKey和SMTP授权码是敏感凭证，务必使用环境变量管理，不要在代码中硬编码。通知平台应部署在内网，仅允许内网服务调用。短信签名和模板需要人工审核，使用前请提前在云平台完成备案。

内容标签：#影刀RPA #通知平台 #阿里云短信 #SMTP #钉钉机器人

作者：林焱