db-api
This commit is contained in:
53
busisness/blls.py
Normal file
53
busisness/blls.py
Normal file
@ -0,0 +1,53 @@
|
||||
from dataclasses import fields
|
||||
from typing import List, Optional, Dict, Any
|
||||
from datetime import datetime
|
||||
from models import ArtifactInfoModel,PDRecordModel
|
||||
from dals import ArtifactDal,PDRecordDal
|
||||
|
||||
class ArtifactBll:
|
||||
def __init__(self):
|
||||
"""初始化数据访问层，创建数据库连接"""
|
||||
# 假设数据库文件在db目录下
|
||||
self.dal = ArtifactDal()
|
||||
pass
|
||||
|
||||
def get_artifact_task(self) -> List[ArtifactInfoModel]:
|
||||
"""获取官片任务数据"""
|
||||
return self.dal.get_top_artifact(5,"ArtifactID asc")
|
||||
|
||||
class PDRecordBll:
|
||||
def __init__(self):
|
||||
"""初始化数据访问层，创建数据库连接"""
|
||||
# 假设数据库文件在db目录下
|
||||
self.dal = PDRecordDal()
|
||||
pass
|
||||
|
||||
def get_PD_record(self) -> List[PDRecordModel]:
|
||||
"""获取PD官片任务数据"""
|
||||
return self.dal.get_top_pd(5,"ID desc")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
artifact_dal = ArtifactBll()
|
||||
|
||||
artifacts = artifact_dal.get_artifact_task()
|
||||
print("\n打印artifacts数据:")
|
||||
for i, artifact in enumerate(artifacts):
|
||||
# 如果是数据类对象，转换为字典输出
|
||||
if hasattr(artifact, '__dataclass_fields__'):
|
||||
print(f"第{i+1}条: {artifact.__dict__}")
|
||||
else:
|
||||
print(f"第{i+1}条: {artifact}")
|
||||
|
||||
pdrecord_dal=PDRecordBll()
|
||||
pdrecords = pdrecord_dal.get_PD_record()
|
||||
print("\n打印pdrecords数据:")
|
||||
for i, record in enumerate(pdrecords):
|
||||
# 如果是数据类对象，转换为字典输出
|
||||
if hasattr(record, '__dataclass_fields__'):
|
||||
print(f"第{i+1}条: {record.__dict__}")
|
||||
else:
|
||||
print(f"第{i+1}条: {record}")
|
||||
|
||||
|
||||
print('\ntest success')
|
||||
204
busisness/dals.py
Normal file
204
busisness/dals.py
Normal file
@ -0,0 +1,204 @@
|
||||
from dataclasses import fields
|
||||
from typing import List, Optional, Dict, Any
|
||||
from datetime import datetime
|
||||
from models import ArtifactInfoModel,PDRecordModel
|
||||
from common.sqlite_handler import SQLiteHandler
|
||||
|
||||
|
||||
def filter_dict_for_model(data_dict: Dict[str, Any], model_class) -> Dict[str, Any]:
|
||||
"""过滤字典，只保留模型中定义的字段"""
|
||||
# 获取模型中定义的所有字段名称
|
||||
model_fields = {field.name for field in fields(model_class)}
|
||||
# 过滤字典，只保留模型中存在的字段
|
||||
return {k: v for k, v in data_dict.items() if k in model_fields}
|
||||
|
||||
class BaseDal:
|
||||
def __init__(self) -> None:
|
||||
"""初始化数据访问层，创建数据库连接"""
|
||||
# 假设数据库文件在db目录下
|
||||
self.db_dao = SQLiteHandler.get_instance("db/three.db", max_readers=50, busy_timeout=4000)
|
||||
|
||||
class ArtifactDal(BaseDal):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
||||
def get_all(self) -> List[ArtifactInfoModel]:
|
||||
"""获取所有ArtifactTask记录"""
|
||||
try:
|
||||
# 查询所有记录
|
||||
sql = "SELECT * FROM ArtifactTask"
|
||||
results = self.db_dao.execute_read(sql)
|
||||
|
||||
# 将查询结果转换为ArtifactInfo对象列表
|
||||
artifacts = []
|
||||
for row in results:
|
||||
# 过滤字典，只保留模型中定义的字段
|
||||
filtered_data = filter_dict_for_model(dict(row), ArtifactInfoModel)
|
||||
artifact = ArtifactInfoModel(**filtered_data)
|
||||
artifacts.append(artifact)
|
||||
|
||||
return artifacts
|
||||
except Exception as e:
|
||||
print(f"获取所有构件任务失败: {e}")
|
||||
return []
|
||||
|
||||
def get_top_artifact(self, top: int,desc:str="ArtifactID asc",where:str="1=1") -> List[ArtifactInfoModel]:
|
||||
"""获取top条数数据，根据ArtifactID升序"""
|
||||
try:
|
||||
# 确保top为正整数
|
||||
if not isinstance(top, int) or top <= 0:
|
||||
raise ValueError("top参数必须是正整数")
|
||||
|
||||
# 查询指定数量的记录，按ArtifactID升序排列
|
||||
sql = f"SELECT * FROM ArtifactTask WHERE {where} ORDER BY {desc} LIMIT ?"
|
||||
results = self.db_dao.execute_read(sql, (top,))
|
||||
|
||||
# 将所有查询结果转换为ArtifactInfoModel对象列表
|
||||
artifacts = []
|
||||
for row in results:
|
||||
# 保证row的变量和模板变量一致
|
||||
artifact = ArtifactInfoModel()
|
||||
artifact.ArtifactID=row["ArtifactID"]
|
||||
artifact.ProduceRingNumber=row["ProduceRingNumber"]
|
||||
artifact.MouldCode=row["MouldCode"]
|
||||
artifact.SkeletonID=row["SkeletonID"]
|
||||
artifact.RingTypeCode=row["RingTypeCode"]
|
||||
artifact.SizeSpecification=row["SizeSpecification"]
|
||||
artifact.BuriedDepth=row["BuriedDepth"]
|
||||
artifact.BlockNumber=row["BlockNumber"]
|
||||
artifact.BetonVolume=row["BetonVolume"]
|
||||
artifact.BetonTaskID=row["BetonTaskID"]
|
||||
artifact.Status=row["Status"]
|
||||
artifacts.append(artifact)
|
||||
|
||||
return artifacts
|
||||
except Exception as e:
|
||||
print(f"获取top构件任务失败: {e}")
|
||||
return []
|
||||
|
||||
|
||||
def get_by_id(self, artifact_id: int) -> Optional[ArtifactInfoModel]:
|
||||
"""根据构件ID获取构件任务"""
|
||||
try:
|
||||
sql = "SELECT * FROM ArtifactTask WHERE ArtifactID = ?"
|
||||
results = self.db_dao.execute_read(sql, (artifact_id,))
|
||||
|
||||
rows = list(results)
|
||||
if not rows:
|
||||
return None
|
||||
|
||||
return ArtifactInfoModel(**dict(rows[0]))
|
||||
except Exception as e:
|
||||
print(f"根据ID获取构件任务失败: {e}")
|
||||
return None
|
||||
|
||||
def insert_artifact(self, artifact_data: dict) -> Optional[int]:
|
||||
"""插入一条构件任务记录"""
|
||||
try:
|
||||
# 使用insert方法插入数据
|
||||
row_id = self.db_dao.insert("ArtifactTask", artifact_data)
|
||||
return row_id
|
||||
except Exception as e:
|
||||
print(f"插入构件任务失败: {e}")
|
||||
return None
|
||||
|
||||
def update_artifact(self, artifact_id: int, update_data: dict) -> bool:
|
||||
"""更新构件任务记录"""
|
||||
try:
|
||||
# 构建WHERE条件
|
||||
where_condition = {"ArtifactID": artifact_id}
|
||||
# 使用update方法更新数据
|
||||
affected_rows = self.db_dao.update("ArtifactTask", update_data, where_condition)
|
||||
return affected_rows > 0
|
||||
except Exception as e:
|
||||
print(f"更新构件任务失败: {e}")
|
||||
return False
|
||||
|
||||
def validate_artifact(self, artifact_info: ArtifactInfoModel) -> bool:
|
||||
"""验证构件信息是否符合业务规则"""
|
||||
try:
|
||||
# 检查必要字段
|
||||
if not hasattr(artifact_info, 'Name') or not artifact_info.Name:
|
||||
return False
|
||||
|
||||
# 可以添加更多业务规则验证
|
||||
# 例如：检查构件编号格式、验证日期等
|
||||
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"验证构件信息失败: {e}")
|
||||
return False
|
||||
|
||||
class PDRecordDal(BaseDal):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
||||
def get_top_pd(self, top: int,desc:str="ID desc",where:str="1=1") -> List[PDRecordModel]:
|
||||
"""获取top条数数据，根据ID降序"""
|
||||
try:
|
||||
# 确保top为正整数
|
||||
if not isinstance(top, int) or top <= 0:
|
||||
raise ValueError("top参数必须是正整数")
|
||||
|
||||
# 查询指定数量的记录，按ID降序排列
|
||||
sql = f"SELECT * FROM PDRecord WHERE {where} ORDER BY {desc} LIMIT ?"
|
||||
results = self.db_dao.execute_read(sql, (top,))
|
||||
|
||||
pdrecords = []
|
||||
for row in results:
|
||||
#
|
||||
pdrecord = PDRecordModel()
|
||||
pdrecord.ID=row["ID"]
|
||||
pdrecord.TaskID=row["TaskID"]
|
||||
pdrecord.ProjectName=row["ProjectName"]
|
||||
pdrecord.ProduceMixID=row["ProduceMixID"]
|
||||
pdrecord.VinNo=row["VinNo"]
|
||||
pdrecord.BetonVolume=row["BetonVolume"]
|
||||
pdrecord.Status=row["Status"]
|
||||
pdrecord.OptTime=row["OptTime"]
|
||||
pdrecords.append(pdrecord)
|
||||
|
||||
return pdrecords
|
||||
except Exception as e:
|
||||
print(f"获取top PD官片任务失败: {e}")
|
||||
return []
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
artifact_dal = ArtifactDal()
|
||||
|
||||
artifacts = artifact_dal.get_artifact_task()
|
||||
|
||||
# 显示获取到的数据
|
||||
print(f"获取到 {len(artifacts)} 条构件任务数据：")
|
||||
for i, artifact in enumerate(artifacts, 1):
|
||||
print(f"\n记录 {i}:")
|
||||
print(f"ID: {artifact.ID}")
|
||||
print(f"ArtifactID: {artifact.ArtifactID}")
|
||||
print(f"ProduceRingNumber: {artifact.ProduceRingNumber}")
|
||||
print(f"MouldCode: {artifact.MouldCode}")
|
||||
print(f"SkeletonID: {artifact.SkeletonID}")
|
||||
print(f"RingTypeCode: {artifact.RingTypeCode}")
|
||||
print(f"SizeSpecification: {artifact.SizeSpecification}")
|
||||
print(f"BuriedDepth: {artifact.BuriedDepth}")
|
||||
print(f"BlockNumber: {artifact.BlockNumber}")
|
||||
print(f"HoleRingMarking: {artifact.HoleRingMarking}")
|
||||
print(f"GroutingPipeMarking: {artifact.GroutingPipeMarking}")
|
||||
print(f"PolypropyleneFiberMarking: {artifact.PolypropyleneFiberMarking}")
|
||||
print(f"BetonVolume: {artifact.BetonVolume}")
|
||||
print(f"BetonTaskID: {artifact.BetonTaskID}")
|
||||
print(f"FK_PDID: {artifact.FK_PDID}")
|
||||
print(f"Status: {artifact.Status}")
|
||||
print(f"BeginTime: {artifact.BeginTime}")
|
||||
print(f"EndTime: {artifact.EndTime}")
|
||||
print(f"PStatus: {artifact.PStatus}")
|
||||
print(f"Source: {artifact.Source}")
|
||||
print(f"FBetonVolume: {artifact.FBetonVolume}")
|
||||
print(f"OptTime: {artifact.OptTime}")
|
||||
|
||||
# 可以打印更多字段...
|
||||
|
||||
print('\ntest success')
|
||||
310
busisness/models.py
Normal file
310
busisness/models.py
Normal file
@ -0,0 +1,310 @@
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional, Dict, Any, List
|
||||
|
||||
|
||||
@dataclass
|
||||
class LoginRequest:
|
||||
"""登录请求模型"""
|
||||
Program: int
|
||||
SC: str
|
||||
loginName: str
|
||||
password: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class LoginResponse:
|
||||
"""登录响应模型"""
|
||||
Code: int
|
||||
Message: Optional[str]
|
||||
Data: Optional[Dict[str, Any]]
|
||||
|
||||
@property
|
||||
def app_id(self) -> Optional[str]:
|
||||
"""获取AppID"""
|
||||
return self.Data.get('AppID') if self.Data else None
|
||||
|
||||
@property
|
||||
def expire_time(self) -> Optional[str]:
|
||||
"""获取过期时间"""
|
||||
return self.Data.get('ExpireTime') if self.Data else None
|
||||
|
||||
@property
|
||||
def sign_token(self) -> Optional[str]:
|
||||
"""获取SignToken"""
|
||||
return self.Data.get('SignToken') if self.Data else None
|
||||
|
||||
@property
|
||||
def zr_jwt(self) -> Optional[str]:
|
||||
"""获取ZrJwt"""
|
||||
return self.Data.get('ZrJwt') if self.Data else None
|
||||
|
||||
|
||||
@dataclass
|
||||
class ArtifactInfo:
|
||||
"""管片信息模型"""
|
||||
#管片编号
|
||||
ArtifactID: str
|
||||
#管片ID
|
||||
ArtifactActionID: int
|
||||
#管片副标识1
|
||||
ArtifactIDVice1: str
|
||||
#产品环号
|
||||
ProduceRingNumber: int
|
||||
#模具编号
|
||||
MouldCode: str
|
||||
#骨架ID
|
||||
SkeletonID: str
|
||||
#环类型编码
|
||||
RingTypeCode: str
|
||||
#尺寸规格
|
||||
SizeSpecification: str
|
||||
#埋深
|
||||
BuriedDepth: str
|
||||
#块号
|
||||
BlockNumber: str
|
||||
#环号标记
|
||||
HoleRingMarking: str
|
||||
#出管标记
|
||||
GroutingPipeMarking: str
|
||||
#聚丙烯纤维标记
|
||||
PolypropyleneFiberMarking: str
|
||||
# 浇筑方量
|
||||
BetonVolume: float
|
||||
# 任务单号（混凝土）
|
||||
BetonTaskID: str
|
||||
|
||||
@dataclass
|
||||
class ArtifactInfoModel:
|
||||
def __init__(self):
|
||||
pass
|
||||
"""管片表模型"""
|
||||
ID: int
|
||||
#管片编号
|
||||
ArtifactID: str
|
||||
#管片ID
|
||||
ArtifactActionID: int
|
||||
#管片副标识1
|
||||
ArtifactIDVice1: str
|
||||
#产品环号
|
||||
ProduceRingNumber: int
|
||||
#模具编号
|
||||
MouldCode: str
|
||||
#骨架ID
|
||||
SkeletonID: str
|
||||
#环类型编码
|
||||
RingTypeCode: str
|
||||
#尺寸规格
|
||||
SizeSpecification: str
|
||||
#埋深
|
||||
BuriedDepth: str
|
||||
#块号
|
||||
BlockNumber: str
|
||||
#环号标记
|
||||
HoleRingMarking: str
|
||||
#出管标记
|
||||
GroutingPipeMarking: str
|
||||
#聚丙烯纤维标记
|
||||
PolypropyleneFiberMarking: str
|
||||
# 浇筑方量
|
||||
BetonVolume: float
|
||||
# 任务单号
|
||||
BetonTaskID: str
|
||||
#FK_PDID
|
||||
FK_PDID: int=0
|
||||
#状态
|
||||
Status: int=1
|
||||
#开始时间
|
||||
BeginTime: str=""
|
||||
#结束时间
|
||||
EndTime: str=""
|
||||
#PStatus
|
||||
PStatus: int=0
|
||||
#Source
|
||||
Source: int=1
|
||||
#FBetonVolume
|
||||
FBetonVolume: float=0.0
|
||||
#OptTime
|
||||
OptTime: str=""
|
||||
|
||||
|
||||
@dataclass
|
||||
class ArtifactResponse:
|
||||
"""管片信息响应模型"""
|
||||
Code: int
|
||||
Message: Optional[str]
|
||||
Data: Optional[ArtifactInfo]
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Dict[str, Any]) -> 'ArtifactResponse':
|
||||
"""
|
||||
从字典创建响应对象
|
||||
|
||||
Args:
|
||||
data: 响应数据字典
|
||||
|
||||
Returns:
|
||||
ArtifactResponse: 响应对象
|
||||
"""
|
||||
response_data = data.get('Data')
|
||||
artifact_info = None
|
||||
if response_data:
|
||||
artifact_info = ArtifactInfo(**response_data)
|
||||
|
||||
return cls(
|
||||
Code=data.get('Code'),
|
||||
Message=data.get('Message'),
|
||||
Data=artifact_info
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class TaskInfo:
|
||||
"""任务单信息模型"""
|
||||
#任务单ID
|
||||
TaskID: str
|
||||
#任务单状态
|
||||
TaskStatus: int
|
||||
#任务单状态文本
|
||||
TaskStatusText: str
|
||||
#计划生产数量
|
||||
TaskCount: int
|
||||
#已生产数量
|
||||
AlreadyProduceCount: int
|
||||
#生产进度
|
||||
Progress: str
|
||||
#工程名称
|
||||
ProjectName: str
|
||||
#计划生产日期
|
||||
TaskPlanDateText: str
|
||||
#强度等级
|
||||
BetonGrade: str
|
||||
#设计配合比编号
|
||||
MixID: str
|
||||
#生产配合比编号
|
||||
ProduceMixID: str
|
||||
#计划方量
|
||||
PlannedVolume: float
|
||||
#已供方量
|
||||
ProducedVolume: float
|
||||
#出洞环标记
|
||||
HoleRingMarking: str
|
||||
#注浆管标记
|
||||
GroutingPipeMarking: str
|
||||
#聚丙烯纤维标记
|
||||
PolypropyleneFiberMarking: str
|
||||
#生产日期
|
||||
TaskDateText: str
|
||||
#盘数
|
||||
PlateCount: int
|
||||
#任务单下发状态
|
||||
SendStatus: int
|
||||
#任务单下发状态
|
||||
SendStatusText: str
|
||||
#配合比下发状态
|
||||
MixSendStatus: int
|
||||
#配合比下发状态
|
||||
MixSendStatusText: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class TaskResponse:
|
||||
"""任务单响应模型"""
|
||||
Code: int
|
||||
Message: Optional[str]
|
||||
Data: Optional[TaskInfo]
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Dict[str, Any]) -> 'TaskResponse':
|
||||
"""
|
||||
从字典创建响应对象
|
||||
|
||||
Args:
|
||||
data: 响应数据字典
|
||||
|
||||
Returns:
|
||||
TaskResponse: 响应对象
|
||||
"""
|
||||
response_data = data.get('Data')
|
||||
task_info = None
|
||||
if response_data:
|
||||
task_info = TaskInfo(**response_data)
|
||||
|
||||
return cls(
|
||||
Code=data.get('Code'),
|
||||
Message=data.get('Message'),
|
||||
Data=task_info
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class NotPourArtifactResponse:
|
||||
"""未浇筑管片列表响应模型"""
|
||||
Code: int
|
||||
Message: Optional[str]
|
||||
Data: Optional[List[ArtifactInfo]]
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Dict[str, Any]) -> 'NotPourArtifactResponse':
|
||||
"""
|
||||
从字典创建响应对象
|
||||
|
||||
Args:
|
||||
data: 响应数据字典
|
||||
|
||||
Returns:
|
||||
NotPourArtifactResponse: 响应对象
|
||||
"""
|
||||
response_data = data.get('Data')
|
||||
artifacts = None
|
||||
if response_data:
|
||||
artifacts = [ArtifactInfo(**item) for item in response_data]
|
||||
|
||||
return cls(
|
||||
Code=data.get('Code'),
|
||||
Message=data.get('Message'),
|
||||
Data=artifacts
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class PDRecordModel:
|
||||
def __init__(self):
|
||||
pass
|
||||
"""管片表模型"""
|
||||
ID: int
|
||||
#派单编号
|
||||
PDCode: str
|
||||
#任务单号
|
||||
TaskID: int
|
||||
#工程名称
|
||||
ProjectName: str
|
||||
#生产配合比编号
|
||||
ProduceMixID: str
|
||||
#车架号
|
||||
VinNo: str
|
||||
#派单方量
|
||||
BetonVolume: float
|
||||
#模具编号
|
||||
MouldCode: str
|
||||
#骨架编号
|
||||
SkeletonID: str
|
||||
#环类型编码
|
||||
RingTypeCode: str
|
||||
#尺寸规格
|
||||
SizeSpecification: str
|
||||
#埋深
|
||||
BuriedDepth: str
|
||||
#块号
|
||||
BlockNumber: str
|
||||
# 派单模式(1自动派单 2手动派单0未知 ）
|
||||
Mode: int=0
|
||||
# 派单状态（1计划中2已下发0未知)，默认1
|
||||
Status: int=1
|
||||
#搅拌生产状态()
|
||||
GStatus: int=0
|
||||
#数据来源(1 api 2离线)
|
||||
Source: int=1
|
||||
#创建时间
|
||||
CreateTime: str=""
|
||||
#派单时间（下发)
|
||||
OptTime: str=""
|
||||
Reference in New Issue
Block a user