from DrissionPage._base.chromium import Chromium from DrissionPage._configs.chromium_options import ChromiumOptions from DrissionPage._pages.mix_tab import MixTab from cssselect.parser import parse_simple_selector from config import ConfigManager # 新增导入 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.ignore_certificate_errors(True) self.browser=Chromium(self.co) self.config_manager = ConfigManager() # 新增配置管理器实例 self.ewcc_config = self.config_manager.app_config.get('ewcc', {}) # 修改为通过类访问 def quit(self ): self.browser.quit() def load_cookies(self): cookies = self.ewcc_config.get('cookies', {}) print('使用保存的cookies登录') for cookie in cookies: self.browser.set.cookies(cookie) ### print(cookie) def save_cookies(self): # 保存当前 cookie self.tab.refresh() cookies = self.tab.cookies() ### print(cookies) self.ewcc_config['cookies'] = cookies self.ewcc_config['savetime'] = time.time() print("--------------------") self.config_manager.app_config['ewcc'] = self.ewcc_config # 修改为通过类访问 self.config_manager.save_config() # 修改为调用类方法 def test(self ): # print(self.browser as Chromium.version ) pass 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): if url is None: url=self.url if title is None: title=self.url_info try: self.tab= self.browser.get_tab(url=url) print("根据 url 找到标签") except: print("根据 url 没有找到标签") self.tab=None if not self.tab: try: self.tab=self.browser.get_tab(title=title) print("根据 标题名 找到 标签") except : print("根据 标题名 没有找到标签") self.tab=None return self.tab def is_login(self): if self.tab is None: return False time.sleep(2) cur_login=False 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 # 判断是否登录 ，如果没有登录 ，则提示十次，间断5秒 i = 1 time.sleep(2) while i <= 10: cur_login = False try: e = None e = self.tab.ele('@@tag()=span@@text()^忘记密码') ### print(e.value) print(f'需要登录 <{i}> :') time.sleep(5) except Exception as e: print('已经登录') print(f'{e}') cur_login = True if cur_login: break i += 1 if not cur_login: print('登录 失败，强制退出。') 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(self.ewcc_config.get('username','pengf332.hnsy')) time.sleep(1) ele_div.ele('#fs3_password').set.value(self.ewcc_config.get('password','Pf7842158c')) 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 show_login(self): self.tab.get(self.url) self.tab.wait.eles_loaded('xpath://*[@id="authen1Form"]/button') time.sleep(3) try: ele_div=self.tab.ele('#tab3Con') ele_div.ele('#fs3_username').set.value(self.ewcc_config.get('username','pengf332.hnsy')) ele_div.ele('#fs3_password').set.value(self.ewcc_config.get('password','Pf7842158b')) ele_div.ele('#fs3_otpOrSms').set.value(self.ewcc_config.get('otpOrSms','129727')) # ele_div.ele('@type=submit').click() except Exception as e: print(e) time.sleep(60) return True def send_message(self): pass def query_car_info(self,plate_number=None,phone_number=None,additional_info=None): 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 if __name__ == "__main__": ewcc = Sinopec_ewcc() # 1 查找现有的标签 print('<1> 查找 现有的标签') ewcc.find_tab() # 新增：查找标签页 # 2 如果没有登录 ，则使用cookie登录 if not ewcc.is_login(): print('<2> 找不到对应的网站,使用cookies尝试登录') ewcc.load_cookies() ewcc.new_tab() # 3 判断是否登录，如果登录不成功，则使用帐号和密码登录 if not ewcc.is_login(): print('<3> 找不对应的网站，使用帐号密码登录') # 这里要注意12小时才可以登录一次。 # 这里显示登录界面，并等待输入一分钟 ewcc.show_login() # 4 判断是否登录 ，如果失败，则发送信息到微信上 if ewcc.is_login(): ewcc.tab.refresh() ewcc.save_cookies() print(" 保存cookies") else: ewcc.send_message() # ewcc.query_car_info() ewcc.quit()