from tkinter import N from DrissionPage._base.chromium import Chromium from DrissionPage._configs.chromium_options import ChromiumOptions from DrissionPage._pages.mix_tab import MixTab 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 def quit(self ): self.browser.quit() def init_cookies(self, cookies): """设置传入的cookies到浏览器 Args: cookies: 要设置的cookies（可以是列表、字典或JSON字符串） Returns: bool: 设置成功返回True，失败返回False """ print('使用传入的cookies登录') # 处理字符串类型的cookies if isinstance(cookies, str): try: import json # 尝试将字符串解析为JSON对象 parsed_cookies = json.loads(cookies) print('成功将字符串cookies解析为JSON对象') # 递归调用，使用解析后的数据 return self.init_cookies(parsed_cookies) except json.JSONDecodeError as e: print(f'字符串cookies解析失败: {str(e)}') return False # 处理字典类型的cookies elif isinstance(cookies, dict): try: self.browser.set.cookies(cookies) print('成功设置字典类型的cookies') return True except Exception as e: print(f'设置字典类型的cookies失败: {str(e)}') return False # 处理列表类型的cookies elif isinstance(cookies, list): try: for cookie in cookies: self.browser.set.cookies(cookie) print('成功设置列表类型的cookies') return True except Exception as e: print(f'设置列表类型的cookies失败: {str(e)}') return False # 处理未知类型 else: print(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) print("根据 url 找到标签") except: print("根据 url 没有找到标签") if not find_tab: try: find_tab=self.browser.get_tab(title=title) print("根据 标题名 找到 标签") except : print("根据 标题名 没有找到标签") 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: print(e) try: e=None e=self.tab.ele("#userName") print("找到登录成功的标志"+e.text) cur_login=True except Exception as e: print('没有找到登录成功的标志') print(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: print(e) time.sleep(3) ### 判断是否有仍然发生的界面 try: ele_process = self.tab.ele('#proceed-button') ele_process.click() except Exception as e: print(e) time.sleep(3) return True def Autologin(self,username=None,userpass=None,otp=None,cookies=None): # 1 查找现有的标签 print('<1> 查找 现有的标签') if self.find_tab(): # 新增：查找标签页 print("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa") # 2 如果没有登录 ，则使用cookie登录 if not self.is_login(): print('<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') print("**********************************") ele_process.click() except Exception as e: print(e) # 3 判断是否登录，如果登录不成功，则使用帐号和密码登录 if not self.is_login(): print('<3> 找不对应的网站，使用帐号密码登录') # 这里要注意12小时才可以登录一次。 # 这里显示登录界面，并等待输入一分钟 self.login(username=username,userpass=userpass,otp=otp) # 4 判断是否登录 ，如果失败，则发送信息到微信上 if self.is_login(): self.tab.refresh() self.save_cookies() print(" 保存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,additional_info=None): print(plate_number) print("ssssssssssssssssssssssssss") self.tab.ele("@@tag()=li@@title=订单中心").click() time.sleep(2) self.tab.ele("@@tag()=a@@name=客户信息查询").click() time.sleep(2) myframe = self.tab.get_frame("iframeSOC_CSR_DETAIL") ### print(myframe.inner_html) time.sleep(2) ### print("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa") ele = myframe.ele("#searchType") ### print(ele.inner_html) myframe.ele("@@tag()=select@@id=searchType").select.by_text("车牌号") myframe.ele('#searchValue').input(plate_number, clear=True) myframe.listen.start('acc/getAccCsrInfo') myframe.ele("@text()=查询").click() DataPacket = myframe.listen.wait() ### print(DataPacket.response.body) ###print("ccccccccccccccccccccccccccccccccccccccc") rtn={} if DataPacket.response.body['success']: total_num = DataPacket.response.body['data']['total'] # rtn['length']=total_num # rtn['test']='aaa' rtn=[] eles = myframe.eles("@@name=客户信息详情界面@@text()=详情") for ele in eles: sele = self.tab.ele("@@text()^客户信息查询@@class:J_menuTab") if sele: sele.click() self.tab.listen.start('getAccCsrInfoByCsrid') ele.click() DataPacket = self.tab.listen.wait() ### print(DataPacket.response.body) rtn.append(DataPacket.response.body['data']) self.tab.listen.stop() ### print(ele.inner_html) return rtn