1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
|
class OptimizedRPAFlow:
"""优化后的RPA流程框架"""
def __init__(self):
self.max_retries = 3
self.base_timeout = 10
self.performance_monitor = RPAPerformanceMonitor()
def smart_wait_for_element(self, selectors, timeout=30, check_interval=1):
"""智能元素等待机制"""
start_time = time.time()
if isinstance(selectors, str):
selectors = [selectors]
while time.time() - start_time < timeout:
for selector in selectors:
try:
element = find_element(selector, timeout=1)
if element and self.is_element_interactable(element):
log_info(f"元素找到并可交互: {selector}")
return element
except:
continue
time.sleep(check_interval)
log_error(f"所有选择器都失败: {selectors}")
return None
def is_element_interactable(self, element):
"""检查元素是否可交互"""
try:
if not element.is_displayed():
return False
if not element.is_enabled():
return False
if element.get_attribute("style") and "display: none" in element.get_attribute("style"):
return False
return True
except:
return False
def retry_on_failure(self, max_retries=3, delay=2):
"""失败重试装饰器"""
def decorator(func):
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_retries + 1):
try:
if attempt > 0:
log_info(f"第{attempt}次重试: {func.__name__}")
time.sleep(delay * attempt)
result = func(*args, **kwargs)
if attempt > 0:
log_info(f"重试成功: {func.__name__}")
return result
except Exception as e:
last_exception = e
log_warning(f"第{attempt + 1}次尝试失败: {func.__name__}, 错误: {str(e)}")
if attempt < max_retries:
self.attempt_recovery()
log_error(f"所有重试失败: {func.__name__}")
raise last_exception
return wrapper
return decorator
def attempt_recovery(self):
"""尝试恢复操作"""
try:
log_info("尝试系统恢复...")
refresh_page()
time.sleep(3)
if not self.is_logged_in():
log_info("检测到登录失效，重新登录")
self.login_to_system()
time.sleep(2)
except Exception as e:
log_error(f"恢复操作失败: {str(e)}")
@retry_on_failure(max_retries=3, delay=2)
def robust_order_processing(self):
"""健壮的订单处理流程"""
if not self.is_logged_in():
self.login_to_system()
order_selectors = [
"//a[contains(text(),'订单管理')]",
"//li[@data-menu='orders']/a",
"//nav//a[@href='/orders']"
]
order_menu = self.smart_wait_for_element(order_selectors, timeout=15)
if not order_menu:
raise Exception("无法找到订单菜单")
click_element(order_menu)
list_selectors = [
"//table[contains(@class,'order-list')]",
"//div[@id='orderTable']//tbody",
"//div[contains(@class,'ant-table-tbody')]"
]
order_table = self.smart_wait_for_element(list_selectors, timeout=30)
if not order_table:
raise Exception("订单列表加载失败")
self.process_order_data(order_table)
log_info("订单处理流程完成")
def process_order_data(self, table_element):
"""处理订单数据"""
try:
order_rows = find_elements("//tr[contains(@class,'order-row')]")
if not order_rows:
log_warning("未找到待处理订单")
return
log_info(f"找到 {len(order_rows)} 个待处理订单")
for i, row in enumerate(order_rows):
try:
self.process_single_order(row, i + 1)
except Exception as e:
log_error(f"处理第{i+1}个订单失败: {str(e)}")
continue
except Exception as e:
log_error(f"订单数据处理异常: {str(e)}")
raise
@performance_monitor.measure_operation_time("单订单处理")
def process_single_order(self, order_row, order_index):
"""处理单个订单"""
order_id = get_text_from_element(order_row, ".//td[1]")
current_status = get_text_from_element(order_row, ".//td[5]")
log_info(f"处理订单 {order_id}, 当前状态: {current_status}")
edit_button = find_element(order_row, ".//button[contains(text(),'编辑')]")
if not edit_button:
raise Exception(f"订单 {order_id} 无编辑按钮")
click_element(edit_button)
modal_selectors = [
"//div[contains(@class,'modal')]//form",
"//div[@role='dialog']//form",
"//div[contains(@class,'ant-modal')]//form"
]
edit_modal = self.smart_wait_for_element(modal_selectors, timeout=10)
if not edit_modal:
raise Exception("编辑弹窗未出现")
self.update_order_status(edit_modal, order_id)
log_info(f"订单 {order_id} 处理完成")
def main():
rpa_flow = OptimizedRPAFlow()
try:
rpa_flow.robust_order_processing()
report = rpa_flow.performance_monitor.get_performance_report()
log_info(f"性能报告: {report}")
except Exception as e:
log_error(f"流程执行失败: {str(e)}")
capture_screenshot("final_error")
|
