294 lines
10 KiB
Python
294 lines
10 KiB
Python
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
|
|
|
|
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 = self.tab.ele('#proceed-button')
|
|
time.sleep(2)
|
|
if ele_process:
|
|
ele_process.click()
|
|
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 = self.tab.ele('#proceed-button')
|
|
if ele_process:
|
|
ele_process.click()
|
|
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()
|
|
|
|
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
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|