450 lines
16 KiB
Python
450 lines
16 KiB
Python
import re
|
||
import os
|
||
from urllib import response
|
||
import yaml
|
||
import requests
|
||
from app.util.status import TemperatureStatus
|
||
|
||
class Kangda:
|
||
def __init__(self, config_path="app/config/config.yaml"):
|
||
|
||
self.config_path = config_path
|
||
self.config = self._parse_config()
|
||
|
||
|
||
# 获取所有事件列表
|
||
def get_event_list(self, filter = False):
|
||
'''
|
||
filter : true 只返回 日常巡检
|
||
false 都返回
|
||
'''
|
||
token = self._login_gettoken()
|
||
event_list = self._event_get_list(token)
|
||
|
||
if filter:
|
||
event_list_filted = list()
|
||
for event in event_list:
|
||
if event.etypeName == "日常巡检":
|
||
event_list_filted.append(event)
|
||
return event_list_filted
|
||
|
||
return event_list
|
||
|
||
|
||
|
||
# 获取事件列表详情
|
||
def get_event_list_detail(self, results):
|
||
|
||
token = self._login_gettoken()
|
||
detail_list = self._event_get_detail(results, token)
|
||
return detail_list
|
||
|
||
def _parse_config(self):
|
||
with open(self.config_path, "r", encoding="utf-8") as f:
|
||
config = yaml.safe_load(f)
|
||
return config
|
||
|
||
def _event_get_list(self, token):
|
||
headers = {
|
||
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36",
|
||
"accept-language": "zh-CN,zh;q=0.9",
|
||
"connections": "keep-alive",
|
||
"Host": "erpapi.concoai.com",
|
||
"accept-encoding": "gzip, deflate",
|
||
"upgrade-insecure-requests": "1",
|
||
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
|
||
"token": token,
|
||
"refreshToken": token
|
||
}
|
||
|
||
parse_body = {"pageNo": 1, "pageSize": 10000}
|
||
|
||
try:
|
||
response = requests.post(url=self.config["url_event_list"],
|
||
headers=headers,
|
||
json=parse_body)
|
||
# 检查请求是否成功
|
||
response.raise_for_status()
|
||
|
||
# 解析JSON响应
|
||
result = response.json()
|
||
|
||
return result["rows"]
|
||
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print("获取事件列表请求出错:", e)
|
||
except ValueError as e:
|
||
print("解析JSON失败:", e)
|
||
|
||
def _event_get_detail(self, results, token):
|
||
|
||
# 请求头
|
||
headers = {
|
||
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36",
|
||
"accept-language": "zh-CN,zh;q=0.9",
|
||
"connections": "keep-alive",
|
||
"Host": "erpapi.concoai.com",
|
||
"accept-encoding": "gzip, deflate",
|
||
"upgrade-insecure-requests": "1",
|
||
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
|
||
"token": token,
|
||
"refreshToken": token
|
||
}
|
||
|
||
# 事件详情列表
|
||
back = list()
|
||
|
||
for result in results:
|
||
# print(result["eventId"])
|
||
try:
|
||
response = requests.get(
|
||
url=self.config["url_event_detail"]+result["eventId"],
|
||
headers=headers
|
||
)
|
||
response.raise_for_status()
|
||
|
||
# 解析JSON响应
|
||
result = response.json()
|
||
|
||
back.append(result["data"])
|
||
|
||
except Exception as e:
|
||
print("获取事件详情失败", e)
|
||
# break
|
||
|
||
|
||
return back
|
||
def _get_task(self, token, robotIds):
|
||
# 请求头
|
||
headers = {
|
||
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36",
|
||
"accept-language": "zh-CN,zh;q=0.9",
|
||
"connections": "keep-alive",
|
||
"Host": "erpapi.concoai.com",
|
||
"accept-encoding": "gzip, deflate",
|
||
"upgrade-insecure-requests": "1",
|
||
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
|
||
"token": token,
|
||
"refreshToken": token
|
||
}
|
||
|
||
try:
|
||
task_list = list()
|
||
for robotId in robotIds:
|
||
response = requests.get(
|
||
url=self.config["url_robot_task"] + robotId,
|
||
headers=headers
|
||
)
|
||
|
||
response.raise_for_status()
|
||
response = response.json()
|
||
task_list.append(response.get("data"))
|
||
return task_list
|
||
except requests.exceptions.RequestException as e:
|
||
print("获取机器人任务请求出错:", e)
|
||
except ValueError as e:
|
||
print("机器人任务请求解析JSON失败:", e)
|
||
def _login_gettoken(self):
|
||
|
||
headers = {
|
||
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36",
|
||
"accept-language": "zh-CN,zh;q=0.9",
|
||
"connections": "keep-alive",
|
||
"Host": "erpapi.concoai.com",
|
||
"accept-encoding": "gzip, deflate",
|
||
"upgrade-insecure-requests": "1",
|
||
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7"
|
||
}
|
||
|
||
try:
|
||
# 发送POST请求，params设置查询参数，json设置JSON请求体
|
||
response = requests.get(
|
||
self.config["url_login"],
|
||
params=self.config["kangda"],
|
||
headers=headers
|
||
# json=payload
|
||
)
|
||
|
||
# 检查请求是否成功
|
||
response.raise_for_status()
|
||
|
||
# 解析JSON响应
|
||
result = response.json()
|
||
|
||
return result.get("data").get("token")
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print("登录请求出错:", e)
|
||
except ValueError as e:
|
||
print("解析JSON失败:", e)
|
||
|
||
def _login_front(self):
|
||
headers = {
|
||
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36",
|
||
"accept-language": "zh-CN,zh;q=0.9",
|
||
"connections": "keep-alive",
|
||
"Host": "erpapi.concoai.com",
|
||
"accept-encoding": "gzip, deflate",
|
||
"upgrade-insecure-requests": "1",
|
||
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7"
|
||
}
|
||
try:
|
||
response = requests.post(
|
||
url=self.config["url_login_front"],
|
||
json=self.config["kangda_front"],
|
||
headers=headers
|
||
)
|
||
|
||
# 检查请求是否成功
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
|
||
return {
|
||
"tenantInfoId": result.get("data").get("tenantInfoId"),
|
||
"token": result.get("data").get("token")
|
||
}
|
||
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print("前端登录请求出错:", e)
|
||
except ValueError as e:
|
||
print("前端解析JSON失败:", e)
|
||
|
||
# 获取机器人班组信息
|
||
def _get_robot_current_duty(self, token, robotId):
|
||
|
||
headers = {
|
||
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36",
|
||
"accept-language": "zh-CN,zh;q=0.9",
|
||
"connections": "keep-alive",
|
||
"Host": "erpapi.concoai.com",
|
||
"accept-encoding": "gzip, deflate",
|
||
"upgrade-insecure-requests": "1",
|
||
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
|
||
"Authorization": "Bearer "+token,
|
||
|
||
}
|
||
|
||
try:
|
||
response = requests.get(
|
||
url=self.config["url_robot_current_duty"] + robotId,
|
||
headers=headers
|
||
)
|
||
|
||
response.raise_for_status()
|
||
response = response.json()
|
||
|
||
return response.get("data")
|
||
except requests.exceptions.RequestException as e:
|
||
print("获取机器人班组信息出错:", e)
|
||
except ValueError as e:
|
||
print("获取机器人班组信息解析JSON失败:", e)
|
||
|
||
|
||
# 获取机器人详情
|
||
def _get_robot_detail(self, token, robotId):
|
||
|
||
headers = {
|
||
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36",
|
||
"accept-language": "zh-CN,zh;q=0.9",
|
||
"connections": "keep-alive",
|
||
"Host": "erpapi.concoai.com",
|
||
"accept-encoding": "gzip, deflate",
|
||
"upgrade-insecure-requests": "1",
|
||
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
|
||
"Authorization": "Bearer "+token,
|
||
|
||
}
|
||
|
||
try:
|
||
response = requests.get(
|
||
url=self.config["url_robot_detail"] + robotId,
|
||
headers=headers
|
||
)
|
||
|
||
response.raise_for_status()
|
||
response = response.json()
|
||
|
||
return response.get("data")
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print("获取机器人详情出错:", e)
|
||
except ValueError as e:
|
||
print("获取机器人详情解析JSON失败:", e)
|
||
|
||
# 根据groupingId 查看组内机器人videoList
|
||
def _get_robot_video_list(self, token, groupingIds):
|
||
|
||
headers = {
|
||
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36",
|
||
"accept-language": "zh-CN,zh;q=0.9",
|
||
"connections": "keep-alive",
|
||
"Host": "erpapi.concoai.com",
|
||
"accept-encoding": "gzip, deflate",
|
||
"upgrade-insecure-requests": "1",
|
||
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
|
||
"Authorization": "Bearer "+token,
|
||
|
||
}
|
||
|
||
try:
|
||
response = requests.post(
|
||
url=self.config["url_robot_video_list"],
|
||
json = {
|
||
"groupingIds":groupingIds
|
||
},
|
||
headers=headers
|
||
)
|
||
|
||
response.raise_for_status()
|
||
response = response.json()
|
||
|
||
# 返回video_list
|
||
return response.get("data")
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print("获取video_list出错:", e)
|
||
except ValueError as e:
|
||
print("获取video_listJSON失败:", e)
|
||
|
||
# 查看租户名下所有分组
|
||
def _get_robot_group(self, tenantInfoId, token):
|
||
# 请求头
|
||
headers = {
|
||
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36",
|
||
"accept-language": "zh-CN,zh;q=0.9",
|
||
"connections": "keep-alive",
|
||
"Host": "erpapi.concoai.com",
|
||
"accept-encoding": "gzip, deflate",
|
||
"upgrade-insecure-requests": "1",
|
||
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
|
||
"Authorization": "Bearer "+token,
|
||
|
||
}
|
||
|
||
try:
|
||
response = requests.post(
|
||
url=self.config["url_groupy"],
|
||
json={
|
||
"tenantInfoId": tenantInfoId
|
||
},
|
||
headers=headers
|
||
)
|
||
|
||
response.raise_for_status()
|
||
|
||
response = response.json()
|
||
|
||
# 返回租户名下的任务列表, 204==>未安排值班==>data为空字符串
|
||
if response.get("code") == 204:
|
||
return None
|
||
return response.get("data").get("dutyList")
|
||
|
||
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print("查询分组出错:", e)
|
||
except ValueError as e:
|
||
print("查询分组解析JSON失败:", e)
|
||
|
||
|
||
|
||
|
||
|
||
def _download_image(self, url, save_image=True):
|
||
'''
|
||
url: 图片下载地址
|
||
save_path: 保存文件目录
|
||
|
||
return: 返回保存的本地文件路径
|
||
'''
|
||
save_path = self.config["image_save_path"]
|
||
|
||
# 确保目录存在
|
||
os.makedirs(save_path, exist_ok=True)
|
||
|
||
file_name = url.split("/")[-1]
|
||
|
||
|
||
try:
|
||
|
||
img_save_path = os.path.join(save_path, file_name)
|
||
|
||
if save_image:
|
||
response = requests.get(url, stream=True)
|
||
response.raise_for_status() # 检查请求是否成功, 不成功会raise错误.
|
||
|
||
with open(img_save_path, "wb") as f:
|
||
for chunk in response.iter_content(1024):
|
||
f.write(chunk)
|
||
return img_save_path
|
||
|
||
except Exception as e:
|
||
print("fail to download image ", url," ", e)
|
||
return None
|
||
|
||
def parse_value(self, values):
|
||
'''
|
||
解析温度是否正常,
|
||
'''
|
||
|
||
# 定义正则表达式模式
|
||
temperature_pattern = re.compile(r'^[+-]?\d+\.\d+C$') # 匹配温度数据
|
||
al_pattern = re.compile(r'^AL\d+$') # 匹配AL后跟数字
|
||
cn_pattern = re.compile(r'[\u4e00-\u9fff]') # 判断是否有中文
|
||
|
||
start = self.config["temperature_range"]["start"]
|
||
end = self.config["temperature_range"]["end"]
|
||
|
||
status = list()
|
||
|
||
for value in values:
|
||
if cn_pattern.match(value):
|
||
status.append(TemperatureStatus.CN)
|
||
|
||
# ocr字符长度小于3
|
||
elif len(value) < 3:
|
||
status.append(TemperatureStatus.CN)
|
||
# 图像中的日期编号
|
||
elif value == "2025" or value == "2026" or value == "2027":
|
||
status.append(TemperatureStatus.CN)
|
||
|
||
# 检查是否为正常数据
|
||
elif temperature_pattern.match(value):
|
||
# 处理温度数据
|
||
temp_str = value[:-1] # 去掉末尾的'C'
|
||
try:
|
||
temp_value = float(temp_str)
|
||
# 判断温度是否在指定范围内
|
||
if start <= temp_value <= end:
|
||
status.append(TemperatureStatus.NORMAL)
|
||
else:
|
||
status.append(TemperatureStatus.UNNORMAL)
|
||
except ValueError:
|
||
status.append(TemperatureStatus.BLURRY)
|
||
elif al_pattern.match(value):
|
||
status.append(TemperatureStatus.ALDATE)
|
||
else:
|
||
status.append(TemperatureStatus.BLURRY)
|
||
|
||
if TemperatureStatus.CN in status:
|
||
return TemperatureStatus.CN
|
||
elif TemperatureStatus.UNNORMAL in status:
|
||
return TemperatureStatus.UNNORMAL
|
||
elif TemperatureStatus.BLURRY in status:
|
||
return TemperatureStatus.BLURRY
|
||
elif TemperatureStatus.ALDATE in status:
|
||
return TemperatureStatus.ALDATE
|
||
|
||
return TemperatureStatus.NORMAL
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|