from tkinter import N, NO from DrissionPage._base.chromium import Chromium from DrissionPage._configs.chromium_options import ChromiumOptions from DrissionPage._pages.mix_tab import MixTab from applog import logger import time class Sinopec_ewcc(object): tab:MixTab # browser:Chromium def __init__(self): self.url = 'https://ewcc.sinopec.com' self.url_info= 'ewcc.sinopec' self.browser_path= r'C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe' self.co = ChromiumOptions().set_browser_path(self.browser_path) #self.co.headless(True) self.co.ignore_certificate_errors(True) self.browser=Chromium(self.co) self.tab=self.browser.latest_tab #判断对应的任务是否完成 self.is_query_member=False self.is_query_oil_fw_tj=False self.is_query_oil_current_month_lj=False self.is_query_client_emp_tj=False def quit(self): self.browser.quit() logger.info(f"浏览器已退出") def init_cookies(self, cookies): """设置传入的cookies到浏览器 Args: cookies: 要设置的cookies（可以是列表、字典或JSON字符串） Returns: bool: 设置成功返回True，失败返回False """ logger.info('使用传入的cookies登录') # 处理字符串类型的cookies if isinstance(cookies, str): try: import json # 尝试将字符串解析为JSON对象 parsed_cookies = json.loads(cookies) logger.info(f"成功将字符串cookies解析为JSON对象，共 {len(parsed_cookies)} 条") # 递归调用，使用解析后的数据 return self.init_cookies(parsed_cookies) except json.JSONDecodeError as e: logger.error('字符串cookies解析失败: {str(e)}') return False # 处理字典类型的cookies elif isinstance(cookies, dict): try: self.browser.set.cookies(cookies) logger.info('成功设置字典类型的cookies') return True except Exception as e: logger.info(f'设置字典类型的cookies失败: {str(e)}') return False # 处理列表类型的cookies elif isinstance(cookies, list): try: for cookie in cookies: self.browser.set.cookies(cookie) logger.info('成功设置列表类型的cookies') return True except Exception as e: logger.info(f'设置列表类型的cookies失败: {str(e)}') return False # 处理未知类型 else: logger.info(f'传入的cookies格式不正确，类型为: {type(cookies)}，应为列表、字典或JSON字符串') return False def save_cookies(self): """获取并返回当前浏览器tab的cookies Returns: list: 当前浏览器tab的cookies列表 """ # 保存当前 cookie self.tab.refresh() cookies = self.tab.cookies() return cookies def open_web(self,url=None): if url is None: url=self.url self.browser.get_tab(url) def new_tab(self,url=None): if url is None: url=self.url self.tab=self.browser.new_tab(url) def find_tab(self,url=None,title=None): find_tab=None if url is None: url=self.url if title is None: title=self.url_info try: find_tab = self.browser.get_tab(url=url) logger.info(f"根据 {url} 找到标签") except: logger.error(f"根据 {url} 没有找到标签") if not find_tab: try: find_tab=self.browser.get_tab(title=title) logger.info(f"根据 标题名{title} 找到 标签") except : logger.exception(f"根据 标题名: {title} 没有找到标签") find_tab=None return find_tab def is_login(self): logger.info(f"判断是否登录") try: e=self.tab.ele("#userName") if e: logger.info(f"找到登录成功的标志: {e.text}") cur_login=True else: logger.info('没有找到登录成功的标志') cur_login=False except Exception as e: cur_login = False return cur_login def login(self,username=None,userpass=None,otp=None): self.browser.set.cookies.clear self.tab.get(self.url) self.tab.wait.eles_loaded('xpath://*[@id="authen1Form"]/button') time.sleep(5) try: ele_div=self.tab.ele('#tab3Con') ele_div.ele('#fs3_username').set.value(username) time.sleep(1) ele_div.ele('#fs3_password').set.value(userpass) time.sleep(1) ele_div.ele('#fs3_otpOrSms').set.value(otp) time.sleep(1) ele_div.ele('@type=submit').click() except Exception as e: logger.info(f'登录表单填写异常: {e}') time.sleep(3) ### 判断是否有仍然发生的界面 try: ele_process=None ele_process = self.tab.ele('#proceed-button') time.sleep(2) if ele_process: ele_process.click(by_js=True) logger.info("继续发送按钮存在并已点击") except Exception as e: logger.error(f'继续发送按钮处理异常: {e}') time.sleep(3) return True def Autologin(self,username=None,userpass=None,otp=None,cookies=None): # 2 如果没有登录 ，则使用cookie登录 if not self.is_login(): logger.info('使用cookies尝试登录') self.browser.set.cookies.clear self.init_cookies(cookies) self.tab.get(self.url) self.tab.refresh() time.sleep(3) ### 判断是否有仍然发生的界面 try: ele_process=None ele_process = self.tab.ele('#proceed-button') if ele_process: time.sleep(5) ele_process.click(by_js=True) logger.info({"ele_process":ele_process}) logger.info("继续发送按钮存在并已点击") except Exception as e: logger.info(f'继续发送按钮处理异常: {e}') # 3 判断是否登录，如果登录不成功，则使用帐号和密码登录 if not self.is_login(): logger.info('<2> 使用帐号密码登录') # 这里显示登录界面，并等待输入一分钟 self.login(username=username,userpass=userpass,otp=otp) # 4 判断是否登录 ，如果失败，则发送信息到微信上 if self.is_login(): self.tab.refresh() self.save_cookies() logger.info(" 保存cookies") return True else: self.send_message() return False def send_message(self): pass def query_car_info(self, plate_number=None, phone_number=None, result_list=None): """查询车辆信息 Args: plate_number: 车牌号 phone_number: 手机号 result_list: 用于存储查询结果的列表（通过参数返回） Returns: bool: 查询是否成功，成功返回True，失败返回False """ # 参数校验 if not plate_number and not phone_number: logger.warning("车牌号和手机号至少需要提供一个") return False # 初始化结果列表 if result_list is None: result_list = [] logger.info(f"查询车牌号: {plate_number} 电话：{phone_number}") try: # 切换到订单中心 self.tab.ele("@@tag()=li@@title=订单中心").click() time.sleep(2) # 可以考虑替换为条件等待 # 点击客户信息查询 self.tab.ele("@@tag()=a@@name=客户信息查询").click() time.sleep(2) # 可以考虑替换为条件等待 # 获取iframe myframe = self.tab.get_frame("iframeSOC_CSR_DETAIL") time.sleep(2) # 可以考虑替换为条件等待 # 根据输入参数选择查询类型 search_type_ele = myframe.ele("@@tag()=select@@id=searchType") search_value_ele = myframe.ele('#searchValue') if plate_number: search_type_ele.select.by_text("车牌号") search_value_ele.input(plate_number, clear=True) elif phone_number: search_type_ele.select.by_text("手机号") search_value_ele.input(phone_number, clear=True) # 执行查询 myframe.listen.start('acc/getAccCsrInfo') myframe.ele("@text()=查询").click() try: DataPacket = myframe.listen.wait(timeout=10) # 添加超时参数 except Exception as e: logger.error(f"等待查询响应超时: {str(e)}") return False finally: myframe.listen.stop() if DataPacket and hasattr(DataPacket, 'response') and hasattr(DataPacket.response, 'body'): if DataPacket.response.body.get('success'): total_num = DataPacket.response.body['data'].get('total', 0) logger.info(f"查询到 {total_num} 条记录") # 获取详情 detail_eles = myframe.eles("@@name=客户信息详情界面@@text()=详情") for ele in detail_eles: # 执行详情点击操作并获取数据 self.tab.listen.start('getAccCsrInfoByCsrid') ele.click(by_js=True) try: detail_packet = self.tab.listen.wait(timeout=10) if detail_packet and hasattr(detail_packet, 'response') and hasattr(detail_packet.response, 'body'): if detail_packet.response.body.get('success'): result_list.append(detail_packet.response.body['data']) except Exception as e: logger.error(f"获取详情失败: {str(e)}") finally: self.tab.listen.stop() return True except Exception as e: logger.error(f"查询车辆信息时发生异常: {str(e)}") return False def query_memeber(self): try: # 切换到积分中心 self.tab.ele("@@tag()=li@@title=积分中心").click() logger.info("点击积分中心") topele=self.tab.ele("#childMenu") aele=topele.ele("@@tag()=li@@onclick:综合管理") aele.click(by_js=True) logger.info("点击综合管理") #time.sleep(2) # 可以考虑替换为条件等待 # 点击客户信息查询 self.tab.ele("@@tag()=a@@name=员工拉新权益会员明细").click() logger.info("点击客户拉新权益会员明细") time.sleep(2) # 可以考虑替换为条件等待 # 获取iframe myframe = self.tab.get_frame("iframeemp_rec_sw") time.sleep(2) # 可以考虑替换为条件等待 # 根据输入参数选择查询类型 search_type_ele = myframe.eles("@@tag()=input@@type=text") #取得开始日期，并获得焦点 ele=search_type_ele[1] logger.info(f"元素信息：{ele}") ele.focus() #查找 弹出的日期窗体 data_ele=myframe.ele("@@tag()=div@@class=el-picker-panel__body-wrapper") ele_date_head=data_ele.ele("@@tag()=div@@class=el-picker-panel__header") ele_date_content=data_ele.ele("@@tag()=div@@class=el-picker-panel__content") logger.info(f"日期元素信息：{ele_date_head}") logger.info(f"日期元素信息：{ele_date_content}") ele_date_day=ele_date_content.eles("@@tag()=td") for a in ele_date_day: day=a.text if day.isdigit() and int(day)==1: a.click(by_js=True) ele=myframe.eles("@@tag()=div@@class=el-form-item__content") ele[1].click(by_js=True) logger.info(f"myframe元素信息 :{ele[1]}") #logger.info(f"日期选择器元素信息：{ele[0]}") time.sleep(5) # 可以考虑替换为条件等待 search_type_ele = myframe.ele("@@tag()=span@@text():导出") if search_type_ele: self.tab.listen.start('exportUcRcodeSw') search_type_ele.click(by_js=True) export_packet = self.tab.listen.wait(timeout=10) if export_packet and hasattr(export_packet, 'response') and hasattr(export_packet.response, 'body'): if export_packet.response.body.get('success'): logger.info("成功导出 员工拉新权益会员明细 ") #logger.info(export_packet.request.postData) self.tab.listen.stop() return True except Exception as e: logger.exception(f"员工拉新权益会员明细查询异常: {str(e)}") return False def query_oil_fw_tj(self): try: # 切换到积分中心 self.tab.ele("@@tag()=li@@title=评价中心").click() logger.info("点击评价中心") topele=self.tab.ele("#childMenu") aele=topele.ele("@@tag()=li@@onclick:综合管理") aele.click(by_js=True) logger.info("点击综合管理") #time.sleep(2) # 可以考虑替换为条件等待 # 点击客户信息查询 self.tab.ele("@@tag()=a@@name=油站服务号应用统计").click() logger.info("点击 油站服务号应用统计") time.sleep(2) # 可以考虑替换为条件等待 # 获取iframe myframe = self.tab.get_frame("iframeservice_no_apply_static") time.sleep(2) # 可以考虑替换为条件等待 # 根据输入参数选择查询类型 search_type_ele = myframe.eles("@@tag()=input@@type=text") for a in search_type_ele: logger.info(f"元素信息：{a}") #取得开始日期，并获得焦点 ele=search_type_ele[3] logger.info(f"元素信息：{ele}") ele.focus() #查找 弹出的日期窗体 data_ele=myframe.ele("@@tag()=div@@class=el-picker-panel__body-wrapper") ele_date_head=data_ele.ele("@@tag()=div@@class=el-picker-panel__header") ele_date_content=data_ele.ele("@@tag()=div@@class=el-picker-panel__content") logger.info(f"日期元素信息：{ele_date_head}") logger.info(f"日期元素信息：{ele_date_content}") ele_date_day=ele_date_content.eles("@@tag()=td") for a in ele_date_day: day=a.text if day.isdigit() and int(day)==1: a.click(by_js=True) ele=myframe.eles("@@tag()=div@@class=el-form-item__content") ele[1].click(by_js=True) logger.info(f"myframe元素信息 :{ele[1]}") #logger.info(f"日期选择器元素信息：{ele[0]}") time.sleep(5) # 可以考虑替换为条件等待 search_type_ele = myframe.ele("@@tag()=span@@text():导出") if search_type_ele: self.tab.listen.start('exportUcRcodeSw') search_type_ele.click(by_js=True) export_packet = self.tab.listen.wait(timeout=10) if export_packet and hasattr(export_packet, 'response') and hasattr(export_packet.response, 'body'): if export_packet.response.body.get('success'): logger.info("成功导出 员工拉新权益会员明细 ") #logger.info(export_packet.request.postData) self.tab.listen.stop() return True except Exception as e: logger.exception(f"评价中心 员工拉新权益会员明细查询异常: {str(e)}") return False def query_client_emp_tj(self): try: # 切换到积分中心 self.tab.ele("@@tag()=li@@title=评价中心").click() logger.info("点击评价中心") topele=self.tab.ele("#childMenu") aele=topele.ele("@@tag()=li@@onclick:统一客户服务评价") aele.click(by_js=True) logger.info("统一客户服务评价") #time.sleep(2) # 可以考虑替换为条件等待 # 点击客户信息查询 self.tab.ele("@@tag()=a@@name=按油站汇总本月累计").click() logger.info("点击 油站汇总本月累计") time.sleep(2) # 可以考虑替换为条件等待 # 获取iframe myframe = self.tab.get_frame("iframestation_refueling_total") time.sleep(2) # 可以考虑替换为条件等待 # 根据输入参数选择查询类型 search_type_ele = myframe.eles("@@tag()=input@@type=text") for a in search_type_ele: logger.info(f"元素信息：{a}") #取得开始日期，并获得焦点 ele=search_type_ele[1] logger.info(f"元素信息：{ele}") ele.focus() #查找 弹出的日期窗体 data_ele=myframe.ele("@@tag()=div@@class=el-picker-panel__body-wrapper") ele_date_head=data_ele.ele("@@tag()=div@@class=el-picker-panel__header") ele_date_content=data_ele.ele("@@tag()=div@@class=el-picker-panel__content") logger.info(f"日期元素信息：{ele_date_head}") logger.info(f"日期元素信息：{ele_date_content}") ele_date_day=ele_date_content.eles("@@tag()=td") for a in ele_date_day: day=a.text if day.isdigit() and int(day)==1: a.click(by_js=True) ele=myframe.eles("@@tag()=div@@class=el-form-item__content") ele[1].click(by_js=True) logger.info(f"myframe元素信息 :{ele[1]}") #logger.info(f"日期选择器元素信息：{ele[0]}") time.sleep(5) # 可以考虑替换为条件等待 search_type_ele = myframe.ele("@@tag()=span@@text():导出") if search_type_ele: self.tab.listen.start('exportUcRcodeSw') search_type_ele.click(by_js=True) export_packet = self.tab.listen.wait(timeout=10) if export_packet and hasattr(export_packet, 'response') and hasattr(export_packet.response, 'body'): if export_packet.response.body.get('success'): logger.info("成功导出 油站汇总本月累计 ") #logger.info(export_packet.request.postData) self.tab.listen.stop() return True except Exception as e: logger.exception(f"评价中心 油站汇总本月累计查询异常: {str(e)}") return False def query_oil_current_month_lj(self): try: # 切换到积分中心 self.tab.ele("@@tag()=li@@title=评价中心").click() logger.info("点击评价中心") topele=self.tab.ele("#childMenu") aele=topele.ele("@@tag()=li@@onclick:统一客户服务评价") aele.click(by_js=True) logger.info("统一客户服务评价") #time.sleep(2) # 可以考虑替换为条件等待 # 点击客户信息查询 self.tab.ele("@@tag()=a@@name=客户评价与员工关联统计").click() logger.info("点击 客户评价与员工关联统计") time.sleep(2) # 可以考虑替换为条件等待 # 获取iframe myframe = self.tab.get_frame("iframestation_refueling_total") time.sleep(2) # 可以考虑替换为条件等待 # 根据输入参数选择查询类型 search_type_ele = myframe.eles("@@tag()=input@@type=text") for a in search_type_ele: logger.info(f"元素信息：{a}") #取得开始日期，并获得焦点 ele=search_type_ele[1] logger.info(f"元素信息：{ele}") ele.focus() #查找 弹出的日期窗体 data_ele=myframe.ele("@@tag()=div@@class=el-picker-panel__body-wrapper") ele_date_head=data_ele.ele("@@tag()=div@@class=el-picker-panel__header") ele_date_content=data_ele.ele("@@tag()=div@@class=el-picker-panel__content") logger.info(f"日期元素信息：{ele_date_head}") logger.info(f"日期元素信息：{ele_date_content}") ele_date_day=ele_date_content.eles("@@tag()=td") for a in ele_date_day: day=a.text if day.isdigit() and int(day)==1: a.click(by_js=True) ele=myframe.eles("@@tag()=div@@class=el-form-item__content") ele[1].click(by_js=True) logger.info(f"myframe元素信息 :{ele[1]}") #logger.info(f"日期选择器元素信息：{ele[0]}") time.sleep(5) # 可以考虑替换为条件等待 search_type_ele = myframe.ele("@@tag()=span@@text():导出") if search_type_ele: self.tab.listen.start('exportUcRcodeSw') search_type_ele.click(by_js=True) export_packet = self.tab.listen.wait(timeout=10) if export_packet and hasattr(export_packet, 'response') and hasattr(export_packet.response, 'body'): if export_packet.response.body.get('success'): logger.info("成功导出 客户评价与员工关联统计 ") #logger.info(export_packet.request.postData) self.tab.listen.stop() return True except Exception as e: logger.exception(f"评价中心 客户评价与员工关联统计查询异常: {str(e)}") return False