167 lines
5.6 KiB
Python
167 lines
5.6 KiB
Python
import requests
|
||
import json
|
||
from tools.API import *
|
||
|
||
def get_data(url : str, data_type : str = "data", params : dict = None ) -> dict:
|
||
headers = {
|
||
"content_type": "application/x-www-form-urlencoded"
|
||
}
|
||
try:
|
||
response = requests.get(url, headers=headers, params=params)
|
||
if response.status_code == 200:
|
||
data = json.loads(response.text)
|
||
return data[data_type]
|
||
else:
|
||
print(f"请求数据失败，状态码：{response.status_code}")
|
||
return None
|
||
except Exception as e:
|
||
print(f"请求数据失败，异常：{e}")
|
||
return None
|
||
|
||
def get_project_info(projectId : str) -> dict:
|
||
projecturl = DTURL + GETPROJECTINFO.format(projectId=projectId)
|
||
return get_data(projecturl, "data")
|
||
|
||
def get_jizu_info(turbineId : str) -> dict:
|
||
jizuurl = DTURL + GETJIZUINFO.format(turbineId=turbineId)
|
||
return get_data(jizuurl, "data")
|
||
|
||
def get_jizu_shigong_info(turbineId : str) -> dict:
|
||
jizuurl = DTURL + GETSHIGONGINFO
|
||
params = {
|
||
"rows": {
|
||
"turbineId": turbineId
|
||
}
|
||
}
|
||
result = get_data(jizuurl, "rows", params)
|
||
return result[0] if result else None
|
||
|
||
def get_weather(weatherid : str) -> dict:
|
||
weatherurl = DTURL + GETWEATHERINFO.format(weatherCode=weatherid)
|
||
return get_data(weatherurl, "data")
|
||
|
||
def get_defect_record_list() -> list[dict]:
|
||
url = DTURL + DEFECTRECORDLIST
|
||
return get_data(url)
|
||
|
||
def get_part_list(turbineId : str) -> dict:
|
||
"""获取对应机组所有部件信息
|
||
Args:
|
||
turbineId (str): 机组ID
|
||
Return:
|
||
list: 包含所有部件完整信息的列表
|
||
|
||
list: 包含所有部件id的列表
|
||
"""
|
||
parturl = DTURL + GETPARTLIST
|
||
params = {
|
||
"turbineId" : turbineId
|
||
}
|
||
result = get_data(parturl, params = params)
|
||
print(f"获取到部件{result}")
|
||
return result, [item["partId"] for item in result]
|
||
|
||
def get_part_picture(turbineId : str) -> tuple[list[dict], dict[list], list[str]]:
|
||
"""获取对应机组所有图片
|
||
Args:
|
||
turbineId (str): 机组ID
|
||
Return:
|
||
list: 包含所有对应机组的部件信息的列表
|
||
dict[list]: 包含所有图片信息的列表(按部件分list)
|
||
Yepians: 包含所有叶片信息的列表
|
||
"""
|
||
picturerul = DTURL + GETPICTURELIST
|
||
part_data, part_list = get_part_list(turbineId)
|
||
#先按顺序找叶片123
|
||
Yepians = []
|
||
yepian_to_find = ["叶片1","叶片2","叶片3"]
|
||
#依次获取叶片1，2，3的信息（未考虑多个同叶片的信息情况，正常情况下不会发生）
|
||
try:
|
||
part_result = {}
|
||
yepian_part_result = {}
|
||
for name in yepian_to_find:
|
||
# 找到第一个匹配的部件并添加到列表中
|
||
for part in part_data:
|
||
if part['partName'] == name:
|
||
Yepians.append(part)
|
||
for image in get_data(picturerul, params = {"partId" : part["partId"]}):
|
||
if part["partName"] not in yepian_part_result:
|
||
yepian_part_result[part["partName"]] = []
|
||
yepian_part_result[part["partName"]].append(image)
|
||
elif part['partName'] not in part_result and part['partName'] not in yepian_to_find:
|
||
for image in get_data(picturerul, params = {"partId" : part["partId"]}):
|
||
if part["partName"] not in part_result:
|
||
part_result[part["partName"]] = []
|
||
part_result[part["partName"]].append(image)
|
||
except Exception as e:
|
||
print(f"获取叶片信息失败，异常：{e}")
|
||
return None, None, None
|
||
return part_data, yepian_part_result, part_result, Yepians
|
||
|
||
def get_yepian_xiangqing(yepian_id : str) -> dict:
|
||
url = GETYEPIANINFO.format(partId=yepian_id)
|
||
return get_data(DTURL + url, "data")
|
||
|
||
def find_defect_record(defect_pictures : list[dict]):
|
||
"""查找缺陷图片的缺陷记录
|
||
|
||
Args:
|
||
defect_pictures (list[dict]): 缺陷图列表
|
||
"""
|
||
for pic in defect_pictures:
|
||
pic_id = pic["imageId"]
|
||
|
||
def get_full_picture_url(pic_url : str) -> str:
|
||
"""获取完整的图片URL
|
||
|
||
Args:
|
||
pic_url (str): 图片URL
|
||
|
||
Returns:
|
||
str: 完整的图片URL
|
||
"""
|
||
return DTURL + pic_url
|
||
|
||
def check_pic_url(pic_url: str, timeout: int = 5) -> bool:
|
||
"""检查图片URL是否有效
|
||
仅验证了头部和Content-Type
|
||
|
||
Args:
|
||
pic_url: 图片URL
|
||
timeout: 请求超时时间（秒）
|
||
|
||
Returns:
|
||
bool: True 如果是有效图片，否则 False
|
||
"""
|
||
print(f"检查图片URL：{pic_url}")
|
||
if pic_url is None:
|
||
return False
|
||
pic_url = get_full_picture_url(pic_url)
|
||
try:
|
||
# 发起HEAD请求（更快，仅获取头部信息）
|
||
response = requests.head(pic_url, timeout=timeout, allow_redirects=True)
|
||
if response.status_code != 200:
|
||
print(f"请求图片失败，状态码：{response.status_code}")
|
||
return False
|
||
|
||
# 检查Content-Type是否是图片类型
|
||
content_type = response.headers.get("Content-Type", "").lower()
|
||
if not content_type.startswith("image/"):
|
||
print(f"图片类型错误：{content_type}")
|
||
return False
|
||
return True
|
||
except Exception as e:
|
||
print(f"请求图片失败, 未知错误{e}")
|
||
return False
|
||
|
||
def get_defect_detail(defectId : str) -> dict:
|
||
"""获取缺陷详情
|
||
|
||
Args:
|
||
defectId (str): 缺陷ID
|
||
|
||
Returns:
|
||
dict: 缺陷详情
|
||
"""
|
||
url = DTURL + DEFECTDETAIL.format(defectId=defectId)
|
||
return get_data(url, "data") |