影刀RPA 系统通知与消息提醒流程执行结果推送作者林焱流程跑完了你不知道跑出错了你也不知道——这是RPA新手最常遇到的问题。你在工位上忙别的事流程在后台默默跑着失败了也没人管等发现时已经耽误了半天。这篇文章讲清楚影刀RPA怎么推送执行结果通知——从简单的声音提醒到企业微信/钉钉/邮件通知让你不用盯着屏幕也能第一时间知道流程状态。一、什么情况用什么场景推荐方式说明个人电脑上跑流程声音提示弹窗最简单不需要额外配置手机也要收到通知Server酱/ Bark推送到微信/手机团队都要收到通知企业微信/钉钉机器人群里统一通知正式报告要发邮件邮件通知带附件发送流程出错要紧急处理钉钉电话/短信最紧急场景多个流程统一监控Webhook推送到监控面板集中管理二、怎么做2.1 最简单的通知声音弹窗影刀内置设置在客户端设置里开启声音提示上一篇文章讲过流程完成/出错时自动播放声音。用Python发送Windows通知拼多多店群自动化上架方案importctypesimportwinsounddefnotify(title,message,soundTrue):Windows系统通知ifsound:# 播放系统提示音winsound.Beep(1000,500)# 频率1000Hz持续500ms# 显示Windows通知弹窗Toast通知ctypes.windll.user32.MessageBoxW(0,message,title,0x40|0x1000# 信息图标 置顶)# 流程成功notify(流程完成,商品采集流程执行完成共采集50条数据)# 流程失败# notify(流程出错, 翻页采集第3页时超时请检查网络)用Windows Toast通知更现代importsubprocessimportosdefsend_toast_notification(title,message):发送Windows Toast通知右下角弹窗# 用PowerShell发送Toast通知ps_scriptf [Windows.UI.Notifications.ToastNotificationManager, Windows.UI.Notifications, ContentType WindowsRuntime] | Out-Null $template [Windows.UI.Notifications.ToastNotificationManager]::GetTemplateContent([Windows.UI.Notifications.ToastTemplateType]::ToastText02) $textNodes $template.GetElementsByTagName(text) $textNodes.Item(0).AppendChild($template.CreateTextNode({title})) | Out-Null $textNodes.Item(1).AppendChild($template.CreateTextNode({message})) | Out-Null $toast [Windows.UI.Notifications.ToastNotification]::new($template) [Windows.UI.Notifications.ToastNotificationManager]::CreateToastNotifier(RPA).Show($toast) subprocess.run([powershell,-Command,ps_script],capture_outputTrue,timeout10)print(f通知已发送:{title})# send_toast_notification(RPA通知, 流程执行完成)2.2 Server酱推送到微信Server酱是一个免费的服务可以把消息推送到微信。importrequestsdefsend_serverchan(title,content,sckey):通过Server酱推送到微信urlfhttps://sctapi.ftqq.com/{sckey}.senddata{title:title,desp:content}try:responserequests.post(url,datadata,timeout10)resultresponse.json()ifresult.get(code)0:print(微信推送成功)returnTrueelse:print(f推送失败:{result.get(message,未知错误)})returnFalseexceptExceptionase:print(f推送异常:{e})returnFalse# 使用前需要先在 sct.ftqq.com 注册获取SCKEY# send_serverchan(# RPA流程完成,# 商品采集完成共50条数据耗时3分20秒,# your_sckey_here# )2.3 企业微信机器人团队用企业微信的话用群机器人推送通知最方便importrequestsimportjsonfromdatetimeimportdatetimedefsend_wechat_work_bot(webhook_url,title,content,msg_typemarkdown):企业微信群机器人通知ifmsg_typemarkdown:data{msgtype:markdown,markdown:{content:f###{title}\n\n{content}\n\n 时间:{datetime.now().strftime(%Y-%m-%d %H:%M:%S)}}}elifmsg_typetext:data{msgtype:text,text:{content:f{title}\n{content}\n时间:{datetime.now().strftime(%H:%M:%S)}}}try:responserequests.post(webhook_url,jsondata,timeout10,headers{Content-Type:application/json})resultresponse.json()ifresult.get(errcode)0:print(企业微信通知发送成功)returnTrueelse:print(f发送失败:{result})returnFalseexceptExceptionase:print(f发送异常:{e})returnFalse# 使用示例# webhook https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyyour-key# send_wechat_work_bot(# webhook,# RPA采集完成,# **采集结果**\n- 总数: font color\info\50/font条\n- 成功: font color\info\48/font条\n- 失败: font color\warning\2/font条,# markdown# )2.4 钉钉机器人importrequestsimportjsonimporthashlibimporthmacimportbase64importurllib.parseimporttimefromdatetimeimportdatetimedefsend_dingtalk_bot(webhook_url,title,content,secretNone):钉钉群机器人通知timestampstr(round(time.time()*1000))# 构建URLurlwebhook_urlifsecret:string_to_signf{timestamp}\n{secret}hmac_codehmac.new(secret.encode(utf-8),string_to_sign.encode(utf-8),digestmodhashlib.sha256).digest()signurllib.parse.quote_plus(base64.b64encode(hmac_code))urlf{webhook_url}timestamp{timestamp}sign{sign}data{msgtype:markdown,markdown:{title:title,text:f###{title}\n\n{content}\n\n{datetime.now().strftime(%Y-%m-%d %H:%M:%S)}}}try:responserequests.post(url,jsondata,timeout10,headers{Content-Type:application/json})resultresponse.json()ifresult.get(errcode)0:print(钉钉通知发送成功)returnTrueelse:print(f发送失败:{result})returnFalseexceptExceptionase:print(f发送异常:{e})returnFalse# 使用示例# webhook https://oapi.dingtalk.com/robot/send?access_tokenyour-token# send_dingtalk_bot(# webhook,# RPA流程完成,# **执行结果**\n- 状态: ✅ 成功\n- 数据量: 50条\n- 耗时: 3分20秒,# secretyour-secret# )2.5 邮件通知importsmtplibfromemail.mime.textimportMIMETextfromemail.mime.multipartimportMIMEMultipartfromemail.mime.applicationimportMIMEApplicationfromdatetimeimportdatetimeimportosdefsend_email_notification(smtp_host,smtp_port,sender_email,sender_password,receiver_emails,subject,body,attachmentsNone):发送邮件通知msgMIMEMultipart()msg[From]sender_email msg[To], .join(receiver_emails)msg[Subject]f[RPA通知]{subject}-{datetime.now().strftime(%Y-%m-%d %H:%M)}# HTML邮件内容html_bodyf html body h2{subject}/h2 p{body}/p hr p stylecolor: gray; font-size: 12px; 此邮件由RPA流程自动发送 | 时间:{datetime.now().strftime(%Y-%m-%d %H:%M:%S)}/p /body /html msg.attach(MIMEText(html_body,html,utf-8))# 添加附件ifattachments:forfile_pathinattachments:ifos.path.exists(file_path):withopen(file_path,rb)asf:attachmentMIMEApplication(f.read())attachment.add_header(Content-Disposition,attachment,filenameos.path.basename(file_path))msg.attach(attachment)print(f附件已添加:{os.path.basename(file_path)})# 发送try:serversmtplib.SMTP(smtp_host,smtp_port)server.starttls()server.login(sender_email,sender_password)server.sendmail(sender_email,receiver_emails,msg.as_string())server.quit()print(邮件发送成功)returnTrueexceptExceptionase:print(f邮件发送失败:{e})returnFalse# 使用示例QQ邮箱# send_email_notification(# smtp_hostsmtp.qq.com,# smtp_port587,# sender_emailyour_emailqq.com,# sender_passwordyour_auth_code, # QQ邮箱用授权码不是密码# receiver_emails[receiverexample.com],# subject商品采集完成,# body采集流程已执行完成共采集50条数据。详见附件报告。,# attachments[rD:\report\daily_report.xlsx]# )2.6 统一通知管理器实际项目中可能需要同时用多种通知方式。写一个统一的通知管理器importjsonimportosfromdatetimeimportdatetimeclassNotificationManager:统一通知管理器def__init__(self,config_pathnotification_config.json):self.configself._load_config(config_path)def_load_config(self,path):加载通知配置default_config{channels:{sound:{enabled:True},email:{enabled:False,smtp_host:,smtp_port:587,sender:,password:,receivers:[]},wechat_work:{enabled:False,webhook_url:},dingtalk:{enabled:False,webhook_url:,secret:}},rules:{success:[sound],warning:[sound,wechat_work],error:[sound,wechat_work,email]}}ifos.path.exists(path):withopen(path,r,encodingutf-8)asf:user_configjson.load(f)# 合并配置forkeyinuser_config:ifisinstance(user_config[key],dict)andkeyindefault_config:default_config[key].update(user_config[key])else:default_config[key]user_config[key]returndefault_configdefnotify(self,level,title,message,attachmentsNone): 发送通知 level: success / warning / error channelsself.config[rules].get(level,[sound])print(f\n{*50})print(f通知级别:{level.upper()})print(f标题:{title})print(f内容:{message})print(f通知渠道:{channels})print(f{*50})forchannelinchannels:configself.config[channels].get(channel,{})ifnotconfig.get(enabled,False):continuetry:ifchannelsound:self._notify_sound(title,message)elifchannelemail:self._notify_email(config,title,message,attachments)elifchannelwechat_work:self._notify_wechat_work(config,title,message)elifchanneldingtalk:self._notify_dingtalk(config,title,message)exceptExceptionase:print(f{channel}通知失败:{e})def_notify_sound(self,title,message):importwinsound winsound.Beep(1000,300)print(✅ 声音通知已发送)def_notify_email(self,config,title,message,attachments):# 调用前面定义的邮件发送函数print(f✅ 邮件通知已发送给:{config.get(receivers,[])})def_notify_wechat_work(self,config,title,message):# 调用前面定义的企业微信函数print(✅ 企业微信通知已发送)def_notify_dingtalk(self,config,title,message):# 调用前面定义的钉钉函数print(✅ 钉钉通知已发送)# 使用示例# nm NotificationManager()# nm.notify(success, 流程完成, 采集50条数据耗时3分钟)# nm.notify(error, 流程出错, 第3页翻页超时, attachments[rD:\logs\error.log])三、有什么坑坑1通知消息太多被忽略每次流程成功都发通知一天几十条最后大家都无视了。等真正出问题时反而没人看。解决分级通知——成功不发通知或只发声音警告发群消息错误才发邮件群消息。坑2SMTP密码用错QQ邮箱、163邮箱需要用授权码不是登录密码。直接用登录密码会报认证失败。解决到邮箱设置里开启SMTP服务并获取授权码用授权码代替密码。坑3钉钉机器人被限流TEMU店群如何管理运营钉钉机器人每分钟最多发20条消息超了会被限流1小时。解决合并通知内容减少发送频率。或者用关键词过滤——只有包含特定关键词的消息才发。坑4网络问题导致通知发不出去通知依赖网络如果流程所在的机器网络不稳定通知可能发不出去。解决通知发送加重试机制失败后本地记录日志defnotify_with_retry(notify_func,max_retries3,**kwargs):带重试的通知forattemptinrange(max_retries):ifnotify_func(**kwargs):returnTruetime.sleep(2**attempt)# 指数退避# 全部失败记录到本地日志log_pathrD:\logs\notification_failed.logwithopen(log_path,a,encodingutf-8)asf:f.write(f{datetime.now()}- 通知发送失败:{kwargs}\n)returnFalse坑5通知里包含敏感信息流程出错时把完整的错误日志发到群里可能包含文件路径、用户名等敏感信息。解决发送前过滤敏感信息importredefsanitize_message(message):过滤敏感信息# 过滤文件路径中的用户名messagere.sub(rC:\\Users\\[^\\],C:\\Users\\***,message)# 过滤密码messagere.sub(rpassword[\]?\s*[:]\s*[\]?[^\\s],password***,message,flagsre.IGNORECASE)# 过滤Tokenmessagere.sub(rtoken[\]?\s*[:]\s*[\]?[\w-]{20,},token***,message,flagsre.IGNORECASE)returnmessage总结通知系统的核心设计分级通知——成功只声音警告发群消息错误才发邮件多渠道冗余——重要通知同时走多个渠道一个挂了还有备用的带重试机制——网络不稳定时自动重试过滤敏感信息——通知内容不暴露密码、路径等配置化管理——通知渠道和规则放配置文件不改代码就能调整
