Report_Generate_Server/tools/Get_Json.py

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")