158 lines
5.1 KiB
Python
158 lines
5.1 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 + "/defect/list"
|
||
return get_data(url)
|
||
|
||
def get_part_list(turbineId : str) -> dict:
|
||
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], list[dict]]:
|
||
"""获取对应机组所有图片
|
||
Args:
|
||
turbineId (str): 机组ID
|
||
Return:
|
||
list: 包含所有对应机组的部件信息的列表
|
||
list: 包含所有图片信息的列表(按部件分list)
|
||
[
|
||
{
|
||
"imageId": "图片id",
|
||
"imageName": "图片名",
|
||
"imagePath": "图片路径",
|
||
"partId": "部件id",
|
||
"partName": "部件名",
|
||
"imageResolution": "字符串",
|
||
"focalDistance": "字符串/None",
|
||
"shootingTime": "字符串/None",
|
||
"cameraManufacturer": "字符串/None",
|
||
"cameraModel": "字符串/None",
|
||
"weather": "字符串/None",
|
||
"weatherLabel": "字符串/None",
|
||
"humidness": "数值/None",
|
||
"temperature": "字符串/None",
|
||
"windLevel": "数值/None",
|
||
"shootingMethod": "字符串/None",
|
||
"shootingMethodLabel": "字符串/None",
|
||
"shootingDistance": "数值/None",
|
||
"collectorName": "字符串/None",
|
||
"imageType": "None",
|
||
"imageTypeLabel": "None",
|
||
"audioList": "None",
|
||
"gps": "字符串/None"
|
||
},
|
||
...
|
||
]
|
||
"""
|
||
picturerul = DTURL + GETPICTURELIST
|
||
part_data, part_list = get_part_list(turbineId)
|
||
result = []
|
||
for part_id in part_list:
|
||
params = {
|
||
"partId" : part_id,
|
||
}
|
||
for images in get_data(picturerul, params = params):
|
||
result.append(images)
|
||
print(f"图片数据获取成功：{result}")
|
||
return part_data, result
|
||
|
||
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}")
|
||
pic_url = get_full_picture_url(pic_url)
|
||
if pic_url is None:
|
||
return False
|
||
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 |