bwadmin/QueryCarInfo2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

corp0927

Browse Source
This commit is contained in:
bwadmin
2025-09-27 20:20:37 +08:00
parent 6774d1460b
commit 4d7994a2eb
4 changed files with 113 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api.py
+1 -1
View File
@@ -166,7 +166,7 @@ def update_query_result():
if query_info: if query_info:
for car_info in cars_info: for car_info in cars_info:
# 提取必要的字段 # 提取必要的字段
plate_number = car_info.get('plate_number', query_info[0]) plate_number = car_info.get('plate_number', '')
phone = car_info.get('mobilephone','') phone = car_info.get('mobilephone','')
id_card = car_info.get('certno', '') id_card = car_info.get('certno', '')
name = car_info.get('name', '') name = car_info.get('name', '')
car_info.db
BIN
View File
Binary file not shown.
ewcc2.py
+95 -58
View File
@@ -2,6 +2,7 @@ from tkinter import N
from DrissionPage._base.chromium import Chromium from DrissionPage._base.chromium import Chromium
from DrissionPage._configs.chromium_options import ChromiumOptions from DrissionPage._configs.chromium_options import ChromiumOptions
from DrissionPage._pages.mix_tab import MixTab from DrissionPage._pages.mix_tab import MixTab
from applog import logger
import time import time
class Sinopec_ewcc(object): class Sinopec_ewcc(object):
@@ -17,9 +18,11 @@ class Sinopec_ewcc(object):
self.co.ignore_certificate_errors(True) self.co.ignore_certificate_errors(True)
self.browser=Chromium(self.co) self.browser=Chromium(self.co)
self.tab=self.browser.latest_tab self.tab=self.browser.latest_tab
logger.info(f"浏览器初始化完成，当前URL: {self.tab.url}")
def quit(self ): def quit(self ):
self.browser.quit() self.browser.quit()
logger.info(f"浏览器已退出")
def init_cookies(self, cookies): def init_cookies(self, cookies):
"""设置传入的cookies到浏览器 """设置传入的cookies到浏览器
@@ -28,7 +31,7 @@ class Sinopec_ewcc(object):
Returns: Returns:
bool: 设置成功返回True，失败返回False bool: 设置成功返回True，失败返回False
""" """
print('使用传入的cookies登录') logger.info('使用传入的cookies登录')
# 处理字符串类型的cookies # 处理字符串类型的cookies
if isinstance(cookies, str): if isinstance(cookies, str):
@@ -36,21 +39,21 @@ class Sinopec_ewcc(object):
import json import json
# 尝试将字符串解析为JSON对象 # 尝试将字符串解析为JSON对象
parsed_cookies = json.loads(cookies) parsed_cookies = json.loads(cookies)
print('成功将字符串cookies解析为JSON对象') logger.info(f"成功将字符串cookies解析为JSON对象，共 {len(parsed_cookies)} 条")
# 递归调用，使用解析后的数据 # 递归调用，使用解析后的数据
return self.init_cookies(parsed_cookies) return self.init_cookies(parsed_cookies)
except json.JSONDecodeError as e: except json.JSONDecodeError as e:
print(f'字符串cookies解析失败: {str(e)}') logger.error('字符串cookies解析失败: {str(e)}')
return False return False
# 处理字典类型的cookies # 处理字典类型的cookies
elif isinstance(cookies, dict): elif isinstance(cookies, dict):
try: try:
self.browser.set.cookies(cookies) self.browser.set.cookies(cookies)
print('成功设置字典类型的cookies') logger.info('成功设置字典类型的cookies')
return True return True
except Exception as e: except Exception as e:
print(f'设置字典类型的cookies失败: {str(e)}') logger.info(f'设置字典类型的cookies失败: {str(e)}')
return False return False
# 处理列表类型的cookies # 处理列表类型的cookies
@@ -58,15 +61,15 @@ class Sinopec_ewcc(object):
try: try:
for cookie in cookies: for cookie in cookies:
self.browser.set.cookies(cookie) self.browser.set.cookies(cookie)
print('成功设置列表类型的cookies') logger.info('成功设置列表类型的cookies')
return True return True
except Exception as e: except Exception as e:
print(f'设置列表类型的cookies失败: {str(e)}') logger.info(f'设置列表类型的cookies失败: {str(e)}')
return False return False
# 处理未知类型 # 处理未知类型
else: else:
print(f'传入的cookies格式不正确，类型为: {type(cookies)}，应为列表、字典或JSON字符串') logger.info(f'传入的cookies格式不正确，类型为: {type(cookies)}，应为列表、字典或JSON字符串')
return False return False
def save_cookies(self): def save_cookies(self):
@@ -99,16 +102,16 @@ class Sinopec_ewcc(object):
try: try:
find_tab = self.browser.get_tab(url=url) find_tab = self.browser.get_tab(url=url)
print("根据 url 找到标签") logger.info(f"根据 {url} 找到标签")
except: except:
print("根据 url 没有找到标签") logger.error(f"根据 {url} 没有找到标签")
if not find_tab: if not find_tab:
try: try:
find_tab=self.browser.get_tab(title=title) find_tab=self.browser.get_tab(title=title)
print("根据 标题名 找到 标签") logger.info(f"根据 标题名{title} 找到 标签")
except : except :
print("根据 标题名 没有找到标签") logger.exception("根据 标题名: {title} 没有找到标签")
find_tab=None find_tab=None
return find_tab return find_tab
@@ -118,22 +121,20 @@ class Sinopec_ewcc(object):
return False return False
time.sleep(2) time.sleep(2)
cur_login=False cur_login=False
### 判断是否有仍然发生的界面 ### 判断是否有仍然发送的界面
try: try:
ele_process = self.tab.ele('#proceed-button') ele_process = self.tab.ele('#proceed-button')
ele_process.click() ele_process.click()
time.sleep(3) time.sleep(3)
except Exception as e: except Exception as e:
print(e) logger.error(f"处理提示出错：仍然发送 : {e}")
try: try:
e=None e=None
e=self.tab.ele("#userName") e=self.tab.ele("#userName")
print("找到登录成功的标志"+e.text) logger.info(f"找到登录成功的标志: {e.text}")
cur_login=True cur_login=True
except Exception as e: except Exception as e:
print('没有找到登录成功的标志') logger.info(f'没有找到登录成功的标志,错误代码：{e}')
print(f'{e}')
cur_login = False cur_login = False
return cur_login return cur_login
@@ -153,7 +154,7 @@ class Sinopec_ewcc(object):
time.sleep(1) time.sleep(1)
ele_div.ele('@type=submit').click() ele_div.ele('@type=submit').click()
except Exception as e: except Exception as e:
print(e) logger.info(f'登录表单填写异常: {e}')
time.sleep(3) time.sleep(3)
### 判断是否有仍然发生的界面 ### 判断是否有仍然发生的界面
@@ -162,20 +163,20 @@ class Sinopec_ewcc(object):
ele_process.click() ele_process.click()
except Exception as e: except Exception as e:
print(e) logger.info(f'证书继续按钮处理异常: {e}')
time.sleep(3) time.sleep(3)
return True return True
def Autologin(self,username=None,userpass=None,otp=None,cookies=None): def Autologin(self,username=None,userpass=None,otp=None,cookies=None):
# 1 查找现有的标签 # 1 查找现有的标签
print('<1> 查找 现有的标签') logger.info('<1> 查找 现有的标签')
if self.find_tab(): # 新增：查找标签页 if self.find_tab(): # 新增：查找标签页
print("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa") logger.info("找到现有标签页")
# 2 如果没有登录 ，则使用cookie登录 # 2 如果没有登录 ，则使用cookie登录
if not self.is_login(): if not self.is_login():
print('<2> 找不到对应的网站,使用cookies尝试登录') logger.info('<2> 找不到对应的网站,使用cookies尝试登录')
self.init_cookies(cookies) self.init_cookies(cookies)
self.tab.get(self.url) self.tab.get(self.url)
self.tab.refresh() self.tab.refresh()
@@ -183,14 +184,14 @@ class Sinopec_ewcc(object):
### 判断是否有仍然发生的界面 ### 判断是否有仍然发生的界面
try: try:
ele_process = self.tab.ele('#proceed-button') ele_process = self.tab.ele('#proceed-button')
print("**********************************") logger.info("证书继续按钮存在并已点击")
ele_process.click() ele_process.click()
except Exception as e: except Exception as e:
print(e) logger.info(f'证书继续按钮处理异常: {e}')
# 3 判断是否登录，如果登录不成功，则使用帐号和密码登录 # 3 判断是否登录，如果登录不成功，则使用帐号和密码登录
if not self.is_login(): if not self.is_login():
print('<3> 找不对应的网站，使用帐号密码登录') logger.info('<3> 找不对应的网站，使用帐号密码登录')
# 这里要注意12小时才可以登录一次。 # 这里要注意12小时才可以登录一次。
# 这里显示登录界面，并等待输入一分钟 # 这里显示登录界面，并等待输入一分钟
self.login(username=username,userpass=userpass,otp=otp) self.login(username=username,userpass=userpass,otp=otp)
@@ -199,7 +200,7 @@ class Sinopec_ewcc(object):
if self.is_login(): if self.is_login():
self.tab.refresh() self.tab.refresh()
self.save_cookies() self.save_cookies()
print(" 保存cookies") logger.info(" 保存cookies")
return True return True
else: else:
self.send_message() self.send_message()
@@ -208,50 +209,86 @@ class Sinopec_ewcc(object):
def send_message(self): def send_message(self):
pass pass
def query_car_info(self,plate_number=None,phone_number=None,additional_info=None): def query_car_info(self, plate_number=None, phone_number=None, additional_info={}):
print(plate_number) """查询车辆信息
print("ssssssssssssssssssssssssss") Args:
plate_number: 车牌号
phone_number: 手机号
additional_info: 附加查询信息
Returns:
list: 查询到的车辆信息列表
"""
# 参数校验
if not plate_number and not phone_number:
logger.warning("车牌号和手机号至少需要提供一个")
return []
logger.info(f"查询车牌号: {plate_number} 电话：{phone_number}")
try:
# 切换到订单中心
self.tab.ele("@@tag()=li@@title=订单中心").click() self.tab.ele("@@tag()=li@@title=订单中心").click()
time.sleep(2) time.sleep(2) # 可以考虑替换为条件等待
# 点击客户信息查询
self.tab.ele("@@tag()=a@@name=客户信息查询").click() self.tab.ele("@@tag()=a@@name=客户信息查询").click()
time.sleep(2) time.sleep(2) # 可以考虑替换为条件等待
# 获取iframe
myframe = self.tab.get_frame("iframeSOC_CSR_DETAIL") myframe = self.tab.get_frame("iframeSOC_CSR_DETAIL")
### print(myframe.inner_html) time.sleep(2) # 可以考虑替换为条件等待
time.sleep(2)
### print("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa") # 根据输入参数选择查询类型
ele = myframe.ele("#searchType") search_type_ele = myframe.ele("@@tag()=select@@id=searchType")
### print(ele.inner_html) search_value_ele = myframe.ele('#searchValue')
myframe.ele("@@tag()=select@@id=searchType").select.by_text("车牌号") if plate_number:
myframe.ele('#searchValue').input(plate_number, clear=True) 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.listen.start('acc/getAccCsrInfo')
myframe.ele("@text()=查询").click() myframe.ele("@text()=查询").click()
DataPacket = myframe.listen.wait()
### print(DataPacket.response.body)
###print("ccccccccccccccccccccccccccccccccccccccc") try:
rtn={} DataPacket = myframe.listen.wait(timeout=10) # 添加超时参数
if DataPacket.response.body['success']: except Exception as e:
total_num = DataPacket.response.body['data']['total'] logger.error(f"等待查询响应超时: {str(e)}")
# rtn['length']=total_num return []
# rtn['test']='aaa' finally:
rtn=[] myframe.listen.stop()
eles = myframe.eles("@@name=客户信息详情界面@@text()=详情")
for ele in eles: result_list = []
sele = self.tab.ele("@@text()^客户信息查询@@class:J_menuTab") if DataPacket and hasattr(DataPacket, 'response') and hasattr(DataPacket.response, 'body'):
if sele: if DataPacket.response.body.get('success'):
sele.click() 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') self.tab.listen.start('getAccCsrInfoByCsrid')
ele.click() ele.click()
DataPacket = self.tab.listen.wait()
### print(DataPacket.response.body) try:
rtn.append(DataPacket.response.body['data']) 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() self.tab.listen.stop()
### print(ele.inner_html)
return rtn return result_list
except Exception as e:
logger.error(f"查询车辆信息时发生异常: {str(e)}")
return []
query_processor.py
+4
View File
@@ -98,6 +98,10 @@ def update_query_result(query_id, results_count, cars_info=None):
# 如果提供了车辆信息，添加到请求数据中 # 如果提供了车辆信息，添加到请求数据中
if cars_info: if cars_info:
data['cars_info'] = cars_info data['cars_info'] = cars_info
# 显示cars_info信息
logger.info(f"车辆信息: {cars_info}")
else:
logger.info("未提供车辆信息")
response = requests.post(url, json=data) response = requests.post(url, json=data)
response.raise_for_status() response.raise_for_status()