from datetime import datetime, timedelta from common.sqlite_handler import SQLiteHandler from typing import Optional, List from .api_http_client import api_http_client from busisness.models import ArtifactInfo, TaskInfo, LoginRequest, LEDInfo from config.ini_manager import ini_manager class MouldService: """模具服务类，提供模具相关的API调用""" def __init__(self): """初始化模具服务""" self._api_client = api_http_client self._host = ini_manager.api_base_url def get_task_info(self, task_id: str) -> Optional[TaskInfo]: """ 获取任务单信息 Args: task_id: 任务单编号 Returns: 任务单信息对象，如果失败返回None """ url = f"{self._host}/api/ext/artifact/task?TaskId={task_id}" try: # 调用API获取数据 response_data = self._api_client.get(url, auth=True) # 检查响应状态 if response_data.get('Code') != 200: print(f"获取任务单信息失败: {response_data.get('Message')}") return None # 解析数据 data = response_data.get('Data', {}) if not data: print(f"未获取到任务单 {task_id} 的信息") return None # 转换为任务单信息对象 task_info = TaskInfo(**data) return task_info except Exception as e: print(f"请求任务单信息异常: {e}") return None def get_not_pour_artifacts(self) -> Optional[List[ArtifactInfo]]: """ 获取已入模绑定未浇筑的管片信息 Returns: 未浇筑管片列表，如果失败返回None """ url = f"{self._host}/api/ext/artifact/not_pour" try: # 调用API获取数据 response_data = self._api_client.get(url, auth=True) # 检查响应状态 if response_data.get('Code') != 200: print(f"获取未浇筑管片信息失败: {response_data.get('Message')}") return None # 解析数据 data_list = response_data.get('Data', []) if not data_list: print("当前没有未浇筑的管片") return [] # 转换为管片信息对象列表 artifacts = [ArtifactInfo(**item) for item in data_list] return artifacts except Exception as e: print(f"请求未浇筑管片信息异常: {e}") return None def get_pouring_led(self) -> Optional[LEDInfo]: """ 获取生产动态信息 Returns: LEDInfo对象，如果失败返回None """ url = f"{self._host}/api/ext/produce/pouring_led" try: # 调用API获取数据 response_data = self._api_client.get(url, auth=True) # 检查响应状态 if response_data.get('Code') != 200: print(f"获取生产动态信息失败: {response_data.get('Message')}") return None # 解析数据 data = response_data.get('Data', {}) if not data: print(f"未获取到 pouring_led 信息") return None # 转换为管片信息对象列表 - 使用更安全的字段过滤方式 # 只提取 LEDInfo 类中定义的字段，忽略多余字段和缺失字段 led_info = LEDInfo( TaskID=data.get('TaskID', ''), PlateVolume=data.get('PlateVolume', ''), MouldCode=data.get('MouldCode', ''), ProduceStartTime=data.get('ProduceStartTime', ''), ArtifactID=data.get('ArtifactID', ''), RingTypeCode=data.get('RingTypeCode', ''), PlateIDSerial=data.get('PlateIDSerial', ''), CheckResult=data.get('CheckResult', ''), UpperWeight=data.get('UpperWeight', 0.0), Temper=data.get('Temper', ''), WorkshopTemperature=data.get('WorkshopTemperature', ''), LowBucketWeighingValue=data.get('LowBucketWeighingValue', ''), VibrationFrequency=data.get('VibrationFrequency', ''), TotMete=data.get('TotMete', ''), BetonVolumeAlready=data.get('BetonVolumeAlready', ''), WaterTemperature=data.get('WaterTemperature', ''), FormulaProportion=data.get('FormulaProportion', '') ) return led_info except Exception as e: print(f"请求 pouring_led 信息异常: {e}") return None app_web_service = MouldService() # if __name__ == "__main__": # # 创建模具服务实例 # mould_service = MouldService() # led_info = mould_service.get_pouring_led() # if led_info: # print(led_info) # db = SQLiteHandler.get_instance("db/three.db", max_readers=50, busy_timeout=4000) # 测试获取未浇筑管片信息 # not_poured = mould_service.get_not_pour_artifacts() # if not_poured: # for item in not_poured: # artifact = db.fetch_one("SELECT * FROM ArtifactTask WHERE ArtifactID = ?", (item.ArtifactID,)) # if not artifact: # dict={ # "ArtifactID": item.ArtifactID, # "ArtifactActionID": item.ArtifactActionID, # "ArtifactIDVice1": item.ArtifactIDVice1, # "ProduceRingNumber": item.ProduceRingNumber, # "MouldCode": item.MouldCode, # "SkeletonID": item.SkeletonID, # "RingTypeCode": item.RingTypeCode, # "SizeSpecification": item.SizeSpecification, # "BuriedDepth": item.BuriedDepth, # "BlockNumber": item.BlockNumber, # "BetonVolume": item.BetonVolume, # "BetonTaskID": item.BetonTaskID, # "HoleRingMarking": item.HoleRingMarking, # "GroutingPipeMarking": item.GroutingPipeMarking, # "PolypropyleneFiberMarking": item.PolypropyleneFiberMarking, # "Status": 1, # "Source": 1 # } # db.insert("ArtifactTask", dict) # dict={ # "TaskID": item.BetonTaskID, # "ProjectName": "上海市轨道交通19号线工程盾构区间管片生产2标", # "ProduceMixID": "20251030-02", # "VinNo": "", # "BetonVolume": item.BetonVolume, # "MouldCode": item.MouldCode, # "SkeletonID": item.SkeletonID, # "RingTypeCode": item.RingTypeCode, # "SizeSpecification": item.SizeSpecification, # "BuriedDepth": item.BuriedDepth, # "BlockNumber": item.BlockNumber, # "Mode": 1, # "Status": 1, # "Source": 1, # "OptTime": str(datetime.now() - timedelta(minutes=5)), # "CreateTime": str(datetime.now()) # } # db.insert("PDRecord", dict) # for i in range(2, 5): # row = db.fetch_one("SELECT * FROM ArtifactTask WHERE ID = ?", (i,)) # if row: # dict={ # "TaskID": row["BetonTaskID"], # "ProjectName": "上海市轨道交通19号线工程盾构区间管片生产2标", # "ProduceMixID": "20251030-02", # "VinNo": "", # "BetonVolume": row["BetonVolume"], # "MouldCode": row["MouldCode"], # "SkeletonID": row["SkeletonID"], # "RingTypeCode": row["RingTypeCode"], # "SizeSpecification": row["SizeSpecification"], # "BuriedDepth": row["BuriedDepth"], # "BlockNumber": row["BlockNumber"], # "Mode": 1, # "Status": 1, # "Source": 1, # "OptTime": str(datetime.now() - timedelta(minutes=5)), # "CreateTime": str(datetime.now()) # } # db.insert("PDRecord", dict)