from tkinter import N 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 logger.info(f"浏览器初始化完成，当前URL: {self.tab.url}") 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("根据 标题名: {title} 没有找到标签") find_tab=None return find_tab def is_login(self): if self.tab is None: return False time.sleep(2) cur_login=False ### 判断是否有仍然发送的界面 try: ele_process = self.tab.ele('#proceed-button') ele_process.click() time.sleep(3) except Exception as e: logger.error(f"处理提示出错：仍然发送 : {e}") try: e=None e=self.tab.ele("#userName") logger.info(f"找到登录成功的标志: {e.text}") cur_login=True except Exception as e: logger.info(f'没有找到登录成功的标志,错误代码：{e}') cur_login = False return cur_login def login(self,username=None,userpass=None,otp=None): 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 = self.tab.ele('#proceed-button') ele_process.click() except Exception as e: logger.info(f'证书继续按钮处理异常: {e}') time.sleep(3) return True def Autologin(self,username=None,userpass=None,otp=None,cookies=None): # 1 查找现有的标签 logger.info('<1> 查找 现有的标签') if self.find_tab(): # 新增：查找标签页 logger.info("找到现有标签页") # 2 如果没有登录 ，则使用cookie登录 if not self.is_login(): logger.info('<2> 找不到对应的网站,使用cookies尝试登录') self.init_cookies(cookies) self.tab.get(self.url) self.tab.refresh() time.sleep(3) ### 判断是否有仍然发生的界面 try: ele_process = self.tab.ele('#proceed-button') logger.info("证书继续按钮存在并已点击") ele_process.click() except Exception as e: logger.info(f'证书继续按钮处理异常: {e}') # 3 判断是否登录，如果登录不成功，则使用帐号和密码登录 if not self.is_login(): logger.info('<3> 找不对应的网站，使用帐号密码登录') # 这里要注意12小时才可以登录一次。 # 这里显示登录界面，并等待输入一分钟 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() 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