RPA流程稳定性优化：异常捕获、超时重试、日志监控、定时触发器部署与EXE离线自动化最佳实践

# RPA异常捕获代码示例：订单处理流程
import time
from datetime import datetime

def process_order(order_id):
    """订单处理主流程，包含完整的异常捕获机制"""
    log_event(f"开始处理订单: {order_id}")
    
    try:
        # 第一层：登录系统
        login_result = safe_login()
        if not login_result:
            raise SystemError("系统登录失败")
        
        # 第二层：获取订单数据
        try:
            order_data = capture_order_data(order_id)
        except ElementNotFound:
            log_event(f"订单元素未找到: {order_id}", level="warning")
            # 尝试备用定位方案
            order_data = capture_order_data_backup(order_id)
        except TimeoutException:
            log_event(f"订单数据获取超时: {order_id}", level="error")
            return {"status": "timeout", "order_id": order_id}
        
        # 第三层：数据写入
        try:
            write_to_database(order_data)
        except DatabaseError as e:
            log_event(f"数据库写入失败: {str(e)}", level="error")
            # 保存到本地队列，稍后重试
            save_to_local_queue(order_data)
        
        log_event(f"订单处理完成: {order_id}")
        return {"status": "success", "order_id": order_id}
        
    except Exception as e:
        # 兜底异常捕获
        log_event(f"订单处理异常: {order_id}, 错误: {str(e)}", level="critical")
        send_alert(f"订单处理流程异常: {order_id}")
        return {"status": "failed", "order_id": order_id, "error": str(e)}

def safe_login():
    """带异常捕获的登录操作"""
    try:
        # 执行登录
        click_element("login_button")
        input_text("username_field", "admin")
        input_text("password_field", "***")
        click_element("submit_button")
        
        # 验证登录结果
        if wait_for_element("dashboard_indicator", timeout=10):
            return True
        return False
    except Exception as e:
        log_event(f"登录异常: {str(e)}", level="error")
        return False

def capture_order_data(order_id):
    """获取订单数据"""
    # 使用智能元素定位，提升稳定性
    search_box = find_element("order_search_input", 
                             smart_generate=True,  # 本地智能生成元素路径
                             timeout=15)
    input_text(search_box, order_id)
    click_element("search_button")
    
    # 等待结果加载
    result_table = wait_for_element("order_result_table", timeout=10)
    return extract_data(result_table)

def log_event(message, level="info"):
    """统一的日志记录函数"""
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    log_entry = f"[{timestamp}] [{level.upper()}] {message}"
    
    # 写入本地日志文件，数据不出本地
    with open("rpa_execution.log", "a", encoding="utf-8") as f:
        f.write(log_entry + "\n")
    
    # 同时输出到控制台
    print(log_entry)

# RPA超时重试代码示例：带指数退避的重试机制
import time
import random

def retry_with_backoff(func, max_retries=3, base_delay=2, max_delay=30):
    """
    带指数退避的超时重试机制
    
    Args:
        func: 要执行的函数
        max_retries: 最大重试次数
        base_delay: 基础延迟秒数
        max_delay: 最大延迟秒数
    """
    for attempt in range(max_retries + 1):
        try:
            result = func()
            if attempt > 0:
                log_event(f"第{attempt}次重试成功", level="info")
            return result
        except Exception as e:
            if attempt == max_retries:
                log_event(f"重试耗尽，最终失败: {str(e)}", level="error")
                raise
            
            # 计算退避延迟：2^attempt * base_delay + 随机抖动
            delay = min(base_delay * (2 ** attempt) + random.uniform(0, 1), max_delay)
            log_event(f"第{attempt+1}次尝试失败，{delay:.1f}秒后重试...", level="warning")
            time.sleep(delay)

def fetch_api_data(api_url, timeout=10):
    """带超时的API数据获取"""
    try:
        # 设置请求超时时间
        response = http_request(api_url, timeout=timeout)
        if response.status == 200:
            return response.json()
        else:
            raise TimeoutException(f"API返回非200状态码: {response.status}")
    except TimeoutException:
        raise
    except Exception as e:
        raise TimeoutException(f"请求异常: {str(e)}")

# 使用示例：获取订单数据
def get_order_details(order_id):
    api_url = f"https://api.example.com/orders/{order_id}"
    
    # 使用超时重试机制获取数据
    return retry_with_backoff(
        lambda: fetch_api_data(api_url, timeout=15),
        max_retries=3,
        base_delay=2
    )

# RPA日志监控代码示例：结构化日志与监控告警
import json
import os
from datetime import datetime, timedelta
from collections import defaultdict

class RPALogMonitor:
    """RPA日志监控系统"""
    
    def __init__(self, log_dir="./logs"):
        self.log_dir = log_dir
        self.error_threshold = 5  # 5分钟内错误超过5次触发告警
        self.time_window = 300    # 5分钟窗口
        self.error_count = defaultdict(list)
        
        # 确保日志目录存在
        os.makedirs(log_dir, exist_ok=True)
    
    def log(self, flow_name, step_name, status, message="", details=None):
        """记录结构化日志"""
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "flow_name": flow_name,
            "step_name": step_name,
            "status": status,  # success, warning, error, critical
            "message": message,
            "details": details or {}
        }
        
        # 按流程分文件存储，便于管理
        log_file = os.path.join(self.log_dir, f"{flow_name}_{datetime.now().strftime('%Y%m%d')}.log")
        with open(log_file, "a", encoding="utf-8") as f:
            f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")
        
        # 实时检查是否需要告警
        if status in ["error", "critical"]:
            self._check_alert(flow_name, step_name, message)
    
    def _check_alert(self, flow_name, step_name, message):
        """检查是否需要触发告警"""
        now = datetime.now()
        key = f"{flow_name}:{step_name}"
        
        # 清理过期记录
        self.error_count[key] = [t for t in self.error_count[key] 
                                 if now - t < timedelta(seconds=self.time_window)]
        
        # 添加当前错误
        self.error_count[key].append(now)
        
        # 检查阈值
        if len(self.error_count[key]) >= self.error_threshold:
            self._send_alert(flow_name, step_name, len(self.error_count[key]))
            # 重置计数，避免重复告警
            self.error_count[key] = []
    
    def _send_alert(self, flow_name, step_name, error_count):
        """发送告警通知"""
        alert_message = f"""
【RPA流程告警】
流程名称: {flow_name}
异常步骤: {step_name}
错误次数: {error_count}次/{self.time_window//60}分钟
时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
建议: 请立即检查流程运行状态
        """
        
        # 多渠道告警：钉钉、飞书、企微、个人微信
        send_dingtalk_alert(alert_message)
        send_feishu_alert(alert_message)
        send_wecom_alert(alert_message)
        send_wechat_alert(alert_message)
        
        log_event(f"告警已发送: {flow_name}/{step_name}", level="critical")
    
    def get_flow_stats(self, flow_name, days=7):
        """获取流程执行统计"""
        stats = {
            "total_runs": 0,
            "success_count": 0,
            "error_count": 0,
            "avg_duration": 0,
            "error_types": defaultdict(int)
        }
        
        for i in range(days):
            date_str = (datetime.now() - timedelta(days=i)).strftime('%Y%m%d')
            log_file = os.path.join(self.log_dir, f"{flow_name}_{date_str}.log")
            
            if not os.path.exists(log_file):
                continue
            
            with open(log_file, "r", encoding="utf-8") as f:
                for line in f:
                    try:
                        entry = json.loads(line.strip())
                        stats["total_runs"] += 1
                        
                        if entry["status"] == "success":
                            stats["success_count"] += 1
                        elif entry["status"] in ["error", "critical"]:
                            stats["error_count"] += 1
                            error_type = entry.get("details", {}).get("error_type", "unknown")
                            stats["error_types"][error_type] += 1
                    except:
                        continue
        
        # 计算成功率
        if stats["total_runs"] > 0:
            stats["success_rate"] = stats["success_count"] / stats["total_runs"] * 100
        else:
            stats["success_rate"] = 0
        
        return stats

# 使用示例
monitor = RPALogMonitor()

# 记录流程步骤
monitor.log("订单处理流程", "登录系统", "success", "登录成功")
monitor.log("订单处理流程", "获取数据", "warning", "元素加载较慢，已等待")
monitor.log("订单处理流程", "提交订单", "error", "网络超时", 
            {"error_type": "timeout", "retry_count": 2})

# 查看统计
stats = monitor.get_flow_stats("订单处理流程", days=7)
print(f"成功率: {stats['success_rate']:.1f}%")
print(f"错误分布: {dict(stats['error_types'])}")

# RPA定时触发器代码示例：灵活的定时任务调度
import schedule
import time
import threading
from datetime import datetime

class RPAScheduler:
    """RPA定时触发器管理器"""
    
    def __init__(self):
        self.jobs = {}
        self.running = False
        self.scheduler_thread = None
    
    def add_daily_job(self, job_id, flow_func, hour, minute, 
                      args=None, kwargs=None):
        """添加每日定时任务"""
        job = schedule.every().day.at(f"{hour:02d}:{minute:02d}").do(
            self._wrap_job, job_id, flow_func, args, kwargs
        )
        self.jobs[job_id] = {
            "type": "daily",
            "schedule": f"每天 {hour:02d}:{minute:02d}",
            "job": job,
            "status": "active"
        }
        log_event(f"定时任务已添加: {job_id}, 执行时间: 每天 {hour:02d}:{minute:02d}")
        return job_id
    
    def add_interval_job(self, job_id, flow_func, interval_minutes,
                         args=None, kwargs=None):
        """添加间隔定时任务"""
        job = schedule.every(interval_minutes).minutes.do(
            self._wrap_job, job_id, flow_func, args, kwargs
        )
        self.jobs[job_id] = {
            "type": "interval",
            "schedule": f"每 {interval_minutes} 分钟",
            "job": job,
            "status": "active"
        }
        log_event(f"定时任务已添加: {job_id}, 执行间隔: 每 {interval_minutes} 分钟")
        return job_id
    
    def add_cron_job(self, job_id, flow_func, cron_expr,
                     args=None, kwargs=None):
        """添加自定义Cron定时任务（支持复杂调度规则）"""
        # 解析cron表达式并创建调度
        # 支持：每周几、每月几号、特定时间段等
        job = self._parse_cron(cron_expr, flow_func, args, kwargs)
        self.jobs[job_id] = {
            "type": "cron",
            "schedule": cron_expr,
            "job": job,
            "status": "active"
        }
        log_event(f"定时任务已添加: {job_id}, Cron规则: {cron_expr}")
        return job_id
    
    def _wrap_job(self, job_id, flow_func, args, kwargs):
        """包装任务执行，添加异常捕获和日志"""
        log_event(f"定时任务触发: {job_id}, 时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        
        try:
            # 执行任务
            result = flow_func(*(args or []), **(kwargs or {}))
            log_event(f"定时任务完成: {job_id}, 结果: {result}")
            
            # 回调通知响应执行结果
            self._notify_result(job_id, "success", result)
            
        except Exception as e:
            log_event(f"定时任务失败: {job_id}, 错误: {str(e)}", level="error")
            self._notify_result(job_id, "failed", str(e))
    
    def _notify_result(self, job_id, status, result):
        """任务执行结果回调通知"""
        notification = {
            "job_id": job_id,
            "status": status,
            "result": result,
            "timestamp": datetime.now().isoformat()
        }
        
        # 支持在钉钉、飞书、企微、个人微信内控制应用执行
        # 并接收回调通知响应执行结果
        send_notification(notification)
    
    def start(self):
        """启动定时调度器"""
        self.running = True
        self.scheduler_thread = threading.Thread(target=self._run_scheduler)
        self.scheduler_thread.daemon = True
        self.scheduler_thread.start()
        log_event("定时触发器调度器已启动")
    
    def _run_scheduler(self):
        """调度器主循环"""
        while self.running:
            schedule.run_pending()
            time.sleep(1)
    
    def stop(self):
        """停止调度器"""
        self.running = False
        if self.scheduler_thread:
            self.scheduler_thread.join()
        log_event("定时触发器调度器已停止")
    
    def list_jobs(self):
        """列出所有定时任务"""
        return {
            job_id: {
                "type": info["type"],
                "schedule": info["schedule"],
                "status": info["status"]
            }
            for job_id, info in self.jobs.items()
        }
    
    def remove_job(self, job_id):
        """移除定时任务"""
        if job_id in self.jobs:
            schedule.cancel_job(self.jobs[job_id]["job"])
            del self.jobs[job_id]
            log_event(f"定时任务已移除: {job_id}")

# 使用示例
scheduler = RPAScheduler()

# 场景1：每天早上8点执行报表获取
scheduler.add_daily_job("daily_report", fetch_daily_report, hour=8, minute=0)

# 场景2：每30分钟检查一次库存
scheduler.add_interval_job("inventory_check", check_inventory, interval_minutes=30)

# 场景3：每周一凌晨2点执行数据备份
scheduler.add_cron_job("weekly_backup", backup_data, cron_expr="0 2 * * 1")

# 启动调度器
scheduler.start()

# 查看所有定时任务
print(scheduler.list_jobs())

# RPA流程稳定性优化完整实战代码
import time
import json
from datetime import datetime, timedelta
import schedule

class StableOrderProcessor:
    """稳定的订单处理流程"""
    
    def __init__(self):
        self.monitor = RPALogMonitor()
        self.retry_config = {
            "max_retries": 3,
            "base_delay": 2,
            "max_delay": 30
        }
    
    def execute_full_flow(self):
        """执行完整流程，包含所有稳定性机制"""
        flow_name = "电商订单处理"
        start_time = datetime.now()
        
        try:
            # 步骤1：登录系统（带异常捕获）
            self.monitor.log(flow_name, "登录系统", "info", "开始登录")
            login_result = self._safe_login()
            if not login_result:
                self.monitor.log(flow_name, "登录系统", "error", "登录失败")
                return {"status": "failed", "step": "login"}
            self.monitor.log(flow_name, "登录系统", "success", "登录成功")
            
            # 步骤2：获取订单数据（带超时重试）
            self.monitor.log(flow_name, "获取订单", "info", "开始获取订单数据")
            try:
                orders = self._fetch_orders_with_retry()
                self.monitor.log(flow_name, "获取订单", "success", f"获取到 {len(orders)} 条订单")
            except Exception as e:
                self.monitor.log(flow_name, "获取订单", "error", f"获取失败: {str(e)}")
                return {"status": "failed", "step": "fetch_orders"}
            
            # 步骤3：处理订单（带异常捕获）
            processed_count = 0
            for order in orders:
                try:
                    self._process_single_order(order)
                    processed_count += 1
                except Exception as e:
                    self.monitor.log(flow_name, "处理订单", "warning", 
                                   f"订单 {order['id']} 处理异常: {str(e)}")
                    # 继续处理下一个订单，不中断整个流程
            
            self.monitor.log(flow_name, "处理订单", "success", 
                           f"成功处理 {processed_count}/{len(orders)} 条订单")
            
            # 步骤4：同步到ERP（带超时重试）
            self.monitor.log(flow_name, "同步ERP", "info", "开始同步数据")
            try:
                self._sync_to_erp_with_retry(orders)
                self.monitor.log(flow_name, "同步ERP", "success", "同步完成")
            except Exception as e:
                self.monitor.log(flow_name, "同步ERP", "error", f"同步失败: {str(e)}")
                # 保存到本地队列，稍后手动处理
                self._save_to_local_queue(orders)
            
            # 步骤5：发送通知
            duration = (datetime.now() - start_time).total_seconds()
            self._send_result_notification(processed_count, len(orders), duration)
            
            self.monitor.log(flow_name, "流程完成", "success", 
                           f"总耗时: {duration:.1f}秒")
            
            return {
                "status": "success",
                "processed": processed_count,
                "total": len(orders),
                "duration": duration
            }
            
        except Exception as e:
            self.monitor.log(flow_name, "流程异常", "critical", f"未预期异常: {str(e)}")
            return {"status": "failed", "error": str(e)}
    
    def _safe_login(self):
        """安全的登录操作"""
        try:
            # 执行登录流程
            # 使用稳定的元素路径，支持本地智能生成
            return True
        except Exception as e:
            return False
    
    def _fetch_orders_with_retry(self):
        """带重试的订单获取"""
        return retry_with_backoff(
            self._fetch_orders,
            max_retries=self.retry_config["max_retries"],
            base_delay=self.retry_config["base_delay"]
        )
    
    def _fetch_orders(self):
        """实际获取订单"""
        # 模拟订单获取
        time.sleep(1)
        return [
            {"id": "ORD001", "amount": 100},
            {"id": "ORD002", "amount": 200}
        ]
    
    def _process_single_order(self, order):
        """处理单个订单"""
        # 订单处理逻辑
        time.sleep(0.5)
    
    def _sync_to_erp_with_retry(self, orders):
        """带重试的ERP同步"""
        return retry_with_backoff(
            lambda: self._sync_to_erp(orders),
            max_retries=self.retry_config["max_retries"],
            base_delay=self.retry_config["base_delay"]
        )
    
    def _sync_to_erp(self, orders):
        """实际同步到ERP"""
        time.sleep(1)
    
    def _save_to_local_queue(self, orders):
        """保存到本地队列"""
        queue_file = "pending_orders.json"
        with open(queue_file, "w", encoding="utf-8") as f:
            json.dump(orders, f, ensure_ascii=False)
    
    def _send_result_notification(self, processed, total, duration):
        """发送结果通知"""
        message = f"""
订单处理完成通知：
- 处理成功: {processed}/{total} 条
- 总耗时: {duration:.1f} 秒
- 时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
        """
        # 支持多渠道通知
        send_notification(message)

# 部署定时触发器
def deploy_scheduler():
    """部署定时触发器"""
    processor = StableOrderProcessor()
    scheduler = RPAScheduler()
    
    # 每天凌晨2点执行
    scheduler.add_daily_job(
        "order_processing",
        processor.execute_full_flow,
        hour=2,
        minute=0
    )
    
    # 同时支持API触发，方便手动补跑
    # 也支持通过钉钉、飞书、企微、个人微信内控制应用执行
    
    scheduler.start()
    
    # 保持运行
    try:
        while True:
            time.sleep(60)
    except KeyboardInterrupt:
        scheduler.stop()

if __name__ == "__main__":
    deploy_scheduler()