diff --git a/busisness/blls.py b/busisness/blls.py new file mode 100644 index 0000000..d35259c --- /dev/null +++ b/busisness/blls.py @@ -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') \ No newline at end of file diff --git a/busisness/dals.py b/busisness/dals.py new file mode 100644 index 0000000..3d8b45e --- /dev/null +++ b/busisness/dals.py @@ -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') \ No newline at end of file diff --git a/busisness/models.py b/busisness/models.py new file mode 100644 index 0000000..6b9b36f --- /dev/null +++ b/busisness/models.py @@ -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="" diff --git a/common/__init__.py b/common/__init__.py new file mode 100644 index 0000000..0cf7fa7 --- /dev/null +++ b/common/__init__.py @@ -0,0 +1,5 @@ +# Common utilities package + +from .sqlite_handler import SQLiteHandler + +__all__ = ['SQLiteHandler'] \ No newline at end of file diff --git a/common/ini_handler.py b/common/ini_handler.py new file mode 100644 index 0000000..6f19baf --- /dev/null +++ b/common/ini_handler.py @@ -0,0 +1,278 @@ +import os +import configparser +from typing import Dict, Any, Optional, List + + +class IniHandlerError(Exception): + """INI处理器异常基类""" + def __init__(self, message: str, file_path: str = None): + self.file_path = file_path + if file_path: + message = f"{message} (文件: {file_path})" + super().__init__(message) + +class IniHandler: + """ + INI文件操作处理器（基于文件路径的缓存） + """ + + _instances = {} # 按文件路径缓存的实例 + + def __new__(cls, file_path: str): + if file_path not in cls._instances: + instance = super().__new__(cls) + instance._file_path = file_path + instance._config = None # 单个文件的配置对象 + cls._instances[file_path] = instance + return cls._instances[file_path] + + def __init__(self, file_path: str): + """ + 初始化INI处理器 + + Args: + file_path: INI文件路径 + """ + # 文件路径已经在__new__中设置，这里不需要重复设置 + pass + + def load_config(self) -> configparser.ConfigParser: + """ + 加载并缓存INI配置文件 + """ + if self._config is not None: + return self._config + + if not os.path.exists(self._file_path): + raise IniHandlerError("INI文件不存在", self._file_path) + + try: + config = configparser.ConfigParser() + config.read(self._file_path, encoding='utf-8') + self._config = config + return config + except Exception as e: + raise IniHandlerError(f"读取INI文件失败: {e}", self._file_path) + + + def get_value(self, section: str, option: str) -> Any: + """ + 获取INI文件中的值 + + Args: + section: 节名 + option: 选项名 + default: 默认值 + + Returns: + 获取的值，如果不存在则返回默认值 + """ + try: + config = self.load_config() + if config.has_section(section) and config.has_option(section, option): + return config.get(section, option) + except IniHandlerError: + # 文件不存在或读取失败时返回默认值 + return default + except Exception as e: + raise IniHandlerError(f"获取INI值失败: {e}", self._file_path) + + def get_int_value(self, section: str, option: str) -> int: + """ + 获取INI文件中的整数值 + + Args: + section: 节名 + option: 选项名 + default: 默认值 + + Returns: + 获取的整数值，如果不存在或转换失败则返回默认值 + """ + try: + config = self.load_config() + if config.has_section(section) and config.has_option(section, option): + return config.getint(section, option) + except IniHandlerError: + raise + except Exception as e: + raise IniHandlerError(f"获取INI整数值失败: {e}", self._file_path) + + def get_float_value(self, section: str, option: str) -> float: + """ + 获取INI文件中的浮点数值 + Args: + section: 节名 + option: 选项名 + default: 默认值 + + Returns: + 获取的浮点数值，如果不存在或转换失败则返回默认值 + """ + try: + config = self.load_config() + if config.has_section(section) and config.has_option(section, option): + return config.getfloat(section, option) + except IniHandlerError: + raise + except Exception as e: + raise IniHandlerError(f"获取INI浮点数值失败: {e}", self._file_path) + + def get_boolean_value(self, section: str, option: str) -> bool: + """ + 获取INI文件中的布尔值 + Args: + section: 节名 + option: 选项名 + default: 默认值 + + Returns: + 获取的布尔值，如果不存在或转换失败则返回默认值 + """ + try: + config = self.load_config() + if config.has_section(section) and config.has_option(section, option): + return config.getboolean(section, option) + except IniHandlerError: + raise + except Exception as e: + raise IniHandlerError(f"获取INI布尔值失败: {e}", self._file_path) + + + def get_section(self, section: str) -> Dict[str, str]: + """ + 获取INI文件中的整个节 + Args: + section: 节名 + Returns: + 节中所有键值对的字典，如果节不存在则返回空字典 + """ + try: + config = self.load_config() + if config.has_section(section): + return dict(config[section]) + return {} + except IniHandlerError: + raise + except Exception as e: + raise IniHandlerError(f"获取INI节失败: {e}", self._file_path) + + def get_sections(self) -> List[str]: + """ + 获取INI文件中的所有节名 + Returns: + 节名列表 + """ + try: + config = self.load_config() + return config.sections() + except IniHandlerError: + raise + except Exception as e: + raise IniHandlerError(f"获取INI节名列表失败: {e}", self._file_path) + + def remove_section(self, section: str) -> None: + """ + 移除INI文件中的节 + Args: + section: 节名 + """ + try: + config = self.load_config() + if config.has_section(section): + config.remove_section(section) + except IniHandlerError: + raise + except Exception as e: + raise IniHandlerError(f"移除INI节失败: {e}", self._file_path) + + def remove_option(self, section: str, option: str) -> None: + """ + 移除INI文件中的选项 + Args: + section: 节名 + option: 选项名 + """ + try: + config = self.load_config() + if config.has_section(section) and config.has_option(section, option): + config.remove_option(section, option) + except IniHandlerError: + raise + except Exception as e: + raise IniHandlerError(f"移除INI选项失败: {e}", self._file_path) + + def has_section(self, section: str) -> bool: + """ + 检查INI文件中是否存在指定的节 + + Args: + section: 节名 + + Returns: + bool: 是否存在 + """ + try: + config = self.load_config() + return config.has_section(section) + except IniHandlerError: + raise + except Exception as e: + raise IniHandlerError(f"检查INI节失败: {e}", self._file_path) + + def has_option(self, section: str, option: str) -> bool: + """ + 检查INI文件中是否存在指定的选项 + + Args: + section: 节名 + option: 选项名 + Returns: + bool: 是否存在 + """ + try: + config = self.load_config() + return config.has_option(section, option) + except IniHandlerError: + raise + except Exception as e: + raise IniHandlerError(f"检查INI选项失败: {e}", self._file_path) + + def write_ini_file(self) -> bool: + """ + 写入INI文件（持久化,写入前先set_value,set_value设置1或多个值后调用此方法） + Returns: + bool: 是否写入成功 + """ + try: + # 确保目录存在 + dir_path = os.path.dirname(self._file_path) + if dir_path and not os.path.exists(dir_path): + os.makedirs(dir_path) + # 写入文件 + with open(self._file_path, 'w', encoding='utf-8') as f: + self._config.write(f) + return True + except Exception as e: + raise IniHandlerError(f"写入INI文件失败: {e}", self._file_path) + + def set_value(self, section: str, option: str, value: Any) -> None: + """ + 设置INI文件中的值 + + Args: + section: 节名 + option: 选项名 + value: 要设置的值 + """ + try: + config = self.load_config() + if not config.has_section(section): + config.add_section(section) + config.set(section, option, str(value)) + except IniHandlerError: + raise + except Exception as e: + raise IniHandlerError(f"设置INI值失败: {e}", self._file_path) + + \ No newline at end of file diff --git a/common/logging_service.py b/common/logging_service.py new file mode 100644 index 0000000..10a244a --- /dev/null +++ b/common/logging_service.py @@ -0,0 +1,84 @@ +import logging +import queue +import threading +import time +from logging.handlers import TimedRotatingFileHandler +from PySide6.QtCore import Signal, QObject +import os + + + +# class QTextLogger(logging.Handler): +# def __init__(self): +# super().__init__() + +# def emit(self, record): +# pass + + +class LoggingService(QObject): + # log_info_signal = Signal(str) + # log_warning_signal = Signal(str) + # log_error_signal = Signal(str) + def __init__(self): + super().__init__() + # self.logger_textEdit_info = logging.getLogger('info_logger') + # self.logger_textEdit_info.setLevel(logging.INFO) + # self.logger_textEdit_warning = logging.getLogger('warning_logger') + # self.logger_textEdit_warning.setLevel(logging.WARNING) + self.logger_file_info = logging.getLogger('file_logger') + self.logger_file_info.setLevel(logging.INFO) + self.log_queue = queue.Queue() + self.logger_thread = threading.Thread(target=self._process_logs, daemon=True,name="util_log") + self.logger_thread.start() + + def init_log(self,file_path): + # text_edit_handler = QTextLogger() + # formatter = logging.Formatter('%(asctime)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S') + # text_edit_handler.setFormatter(formatter) + # self.logger_textEdit_info.addHandler(text_edit_handler) + + # text_edit_handler_warning = QTextLogger() + # formatter_warning = logging.Formatter('%(asctime)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S') + # text_edit_handler_warning.setFormatter(formatter_warning) + # self.logger_textEdit_warning.addHandler(text_edit_handler_warning) + # 确保日志目录存在 + log_dir = os.path.dirname(file_path) + if log_dir and not os.path.exists(log_dir): + os.makedirs(log_dir, exist_ok=True) + + handler = TimedRotatingFileHandler(file_path, when='D', interval=1, backupCount=30,encoding='utf-8') + handler.suffix = "%Y-%m-%d" + formatter = logging.Formatter('%(asctime)s - %(levelname)s- %(message)s', datefmt='%Y-%m-%d %H:%M:%S') + handler.setLevel(logging.INFO) + handler.setFormatter(formatter) + self.logger_file_info.addHandler(handler) + + # 添加错误专用日志文件 + error_handler = TimedRotatingFileHandler( + file_path.replace('.log', '_error.log'), + when='D', interval=1, backupCount=30, encoding='utf-8') + error_handler.suffix = "%Y-%m-%d" + error_handler.setLevel(logging.ERROR) # 只处理错误级别 + error_handler.setFormatter(formatter) + self.logger_file_info.addHandler(error_handler) + + + def _process_logs(self): + while True: + time.sleep(0.1) + level, message = self.log_queue.get() + if level == logging.INFO: + # self.log_info_signal.emit(message) + self.logger_file_info.info(message) + elif level == logging.ERROR: + self.logger_file_info.error(message) + # self.log_error_signal.emit(message) + elif level == logging.WARNING: + self.logger_file_info.warning(message) + # self.log_warning_signal.emit(message) + def log_message(self,level,message): + self.log_queue.put((level, message)) + + +logging_service = LoggingService() diff --git a/common/sqlite_handler.py b/common/sqlite_handler.py new file mode 100644 index 0000000..3c71444 --- /dev/null +++ b/common/sqlite_handler.py @@ -0,0 +1,838 @@ +""" +SQLite数据库操作通用公共类 +提供数据库连接、表管理、CRUD操作等通用功能 +""" +import sqlite3 +import threading +import time +from typing import List, Dict, Any, Optional, Union +from pathlib import Path +from datetime import datetime + + +class SQLiteError(Exception): + """SQLite操作基础异常类""" + def __init__(self, message: str, sqlite_error_code: int = None, sql: str = None, + params: tuple = None, db_path: str = None): + super().__init__(message) + self.message = message + self.sqlite_error_code = sqlite_error_code + self.sql = sql + self.params = params + self.db_path = db_path + + def __str__(self): + error_info = f"SQLiteError: {self.message}" + if self.sqlite_error_code: + error_info += f" (错误代码: {self.sqlite_error_code})" + if self.sql: + error_info += f"\nSQL: {self.sql}" + if self.params: + error_info += f"\n参数: {self.params}" + if self.db_path: + error_info += f"\n数据库: {self.db_path}" + return error_info + + +class SQLiteBusyError(SQLiteError): + """SQLite BUSY错误异常（用于并发控制）""" + def __init__(self, message: str, sqlite_error_code: int = None, sql: str = None, + params: tuple = None, db_path: str = None, retry_attempt: int = None, max_retries: int = None): + super().__init__(message, sqlite_error_code, sql, params, db_path) + self.retry_attempt = retry_attempt + self.max_retries = max_retries + + def __str__(self): + error_info = f"SQLiteBusyError: {self.message}" + if self.sqlite_error_code: + error_info += f" (错误代码: {self.sqlite_error_code})" + if self.retry_attempt is not None and self.max_retries is not None: + error_info += f" (重试 {self.retry_attempt}/{self.max_retries})" + if self.sql: + error_info += f"\nSQL: {self.sql}" + if self.params: + error_info += f"\n参数: {self.params}" + if self.db_path: + error_info += f"\n数据库: {self.db_path}" + return error_info + + +def _handle_sqlite_error(error: sqlite3.Error, operation: str, sql: str = None, + params: tuple = None, db_path: str = None, + retry_attempt: int = None, max_retries: int = None) -> SQLiteError: + """ + 将SQLite错误转换为自定义异常 + + Args: + error: 原始SQLite错误 + operation: 操作描述 + sql: 执行的SQL语句 + params: SQL参数 + db_path: 数据库路径 + retry_attempt: 重试次数（用于BUSY错误） + max_retries: 最大重试次数 + + Returns: + SQLiteError: 对应的自定义异常 + """ + error_code = getattr(error, 'sqlite_errorcode', None) + error_name = getattr(error, 'sqlite_errorname', None) + + error_message = f"{operation}失败: {str(error)}" + if error_name: + error_message += f" ({error_name})" + + # 如果是SQLITE_BUSY错误，返回SQLiteBusyError（用于并发控制） + if error_name == 'SQLITE_BUSY': + return SQLiteBusyError( + message=error_message, + sqlite_error_code=error_code, + sql=sql, + params=params, + db_path=db_path, + retry_attempt=retry_attempt, + max_retries=max_retries + ) + + # 其他所有错误都返回通用的SQLiteError + return SQLiteError( + message=error_message, + sqlite_error_code=error_code, + sql=sql, + params=params, + db_path=db_path + ) + + +class SQLiteHandler: + """SQLite数据库操作通用类（单例模式）""" + + _lock = threading.Lock() # 单例锁 + _instance = None + @classmethod + def get_instance(cls, *args, **kwargs): + if cls._instance is None: + with cls._lock: + if cls._instance is None: # 双重检查 + cls._instance = cls(*args, **kwargs) + return cls._instance + + def __init__(self, db_path: str = "three.db", max_readers: int = 10, busy_timeout: int = 5000): + """ + 初始化SQLite处理器 + + Args: + db_path: 数据库文件路径，默认为app.db + max_readers: 最大并发读取数，默认为10 + busy_timeout: SQLITE_BUSY超时时间（毫秒），默认为5000毫秒 + """ + self.db_path = db_path + self.max_readers = max_readers + self.busy_timeout = busy_timeout + + # 读写分离锁 + self._read_lock = threading.RLock() # 读锁（允许多个读） + self._write_lock = threading.Lock() # 写锁 + self._active_readers = 0 # 当前活跃读取数 + + # 连接配置 + self._connection_params = { + # 是否检测相同的线程（False允许在不同线程间共享连接） + "check_same_thread": False, + # 设置数据库操作超时时间（秒） + "timeout": 10 + } + + # SQLITE_BUSY重试配置 + self.busy_retry_attempts = 3 # 重试次数 + self.busy_retry_delay = 1 # 重试延迟（秒） + + # 确保数据库目录存在 + db_dir = Path(db_path).parent + if not db_dir.exists(): + db_dir.mkdir(parents=True, exist_ok=True) + + # 初始化数据库参数 + self._setup_database_params() + + def _setup_database_params(self): + """设置数据库连接参数""" + try: + # 创建临时连接来设置参数 + conn = sqlite3.connect(self.db_path, **self._connection_params) + + # 启用WAL模式（Write-Ahead Logging） + cursor = conn.execute("PRAGMA journal_mode = WAL") + journal_mode = cursor.fetchone()[0] + print(f"WAL模式设置: {journal_mode}") + + # 启用外键约束 + conn.execute("PRAGMA foreign_keys = ON") + + # 设置SQLITE_BUSY超时 + conn.execute(f"PRAGMA busy_timeout = {self.busy_timeout}") + + # 设置行工厂为字典形式 + conn.row_factory = sqlite3.Row + + conn.close() + print("数据库参数设置完成") + except sqlite3.Error as e: + print(f"设置数据库参数失败: {e}") + raise + + def _create_connection(self) -> sqlite3.Connection: + """创建新的数据库连接""" + conn = sqlite3.connect(self.db_path, **self._connection_params) + conn.row_factory = sqlite3.Row + return conn + + def _acquire_read_lock(self): + """获取读锁""" + with self._read_lock: + if self._active_readers >= self.max_readers: + raise SQLiteBusyError(f"已达到最大并发读取数: {self.max_readers}") + self._active_readers += 1 + + def _release_read_lock(self): + """释放读锁""" + with self._read_lock: + self._active_readers -= 1 + + def _acquire_write_lock(self): + """获取写锁，超时时间为30秒""" + if not self._write_lock.acquire(timeout=30): + raise SQLiteBusyError("获取写锁超时（30秒），可能发生死锁") + + + + def execute_read(self, sql: str, params: tuple = None) -> list: + """ + 执行读操作SQL语句（使用读写锁） + + Args: + sql: SQL语句 + params: 参数元组 + + Returns: + list: 查询结果列表 + + Raises: + SQLiteError: SQL执行错误 + SQLiteBusyError: 并发读取数超过限制 + """ + start_time = time.time() + + # 获取读锁 + self._acquire_read_lock() + + conn = self._create_connection() + try: + if params: + cursor = conn.execute(sql, params) + else: + cursor = conn.execute(sql) + # 获取所有结果并返回，避免连接关闭后游标失效 + results = cursor.fetchall() + return results + except sqlite3.Error as e: + # 抛出自定义异常 + raise _handle_sqlite_error(e, "执行读操作", sql, params, self.db_path) + finally: + # 关闭连接 + if conn: + conn.close() + # 释放读锁 + self._release_read_lock() + + # 记录执行时间 + execution_time = time.time() - start_time + if execution_time > 1.0: + print(f"⚠️ 读操作执行时间过长: {execution_time:.3f}秒 - SQL: {sql[:100]}{'...' if len(sql) > 100 else ''}") + + def execute_write(self, sql: str, params: tuple = None) -> sqlite3.Cursor: + """ + 执行写操作SQL语句（使用读写锁和BEGIN IMMEDIATE事务） + + Args: + sql: SQL语句 + params: 参数元组 + + Returns: + sqlite3.Cursor: 执行结果游标 + + Raises: + SQLiteError: SQL执行错误 + SQLiteBusyError: SQLITE_BUSY错误，达到最大重试次数 + """ + start_time = time.time() + + # 获取写锁 + self._acquire_write_lock() + + conn = self._create_connection() + cursor = None + + try: + # 使用BEGIN IMMEDIATE事务 + for attempt in range(self.busy_retry_attempts): + try: + # 开始IMMEDIATE事务 + conn.execute("BEGIN IMMEDIATE") + + # 执行SQL + if params: + cursor = conn.execute(sql, params) + else: + cursor = conn.execute(sql) + + # 提交事务 + conn.commit() + return cursor + + except sqlite3.OperationalError as e: + # 如果是SQLITE_BUSY错误，进行重试 + if "database is locked" in str(e) or "SQLITE_BUSY" in str(e): + if attempt < self.busy_retry_attempts - 1: + conn.rollback() + print(f"SQLITE_BUSY错误，第{attempt + 1}次重试...") + time.sleep(self.busy_retry_delay) + continue + else: + # 最后一次尝试失败 + conn.rollback() + raise _handle_sqlite_error( + e, "执行写操作", sql, params, self.db_path, + attempt, self.busy_retry_attempts + ) + else: + # 其他操作错误，直接抛出 + conn.rollback() + raise _handle_sqlite_error(e, "执行写操作", sql, params, self.db_path) + except sqlite3.Error as e: + # 其他SQLite错误，回滚并抛出 + conn.rollback() + raise _handle_sqlite_error(e, "执行写操作", sql, params, self.db_path) + + except sqlite3.Error as e: + raise + finally: + # 关闭连接 + if conn: + conn.close() + # 释放写锁 + self._write_lock.release() + + # 记录执行时间 + execution_time = time.time() - start_time + if execution_time > 1.0: + print(f"⚠️ 写操作执行时间过长: {execution_time:.3f}秒 - SQL: {sql[:100]}{'...' if len(sql) > 100 else ''}") + + def executemany_write(self, sql: str, params_list: List[tuple]) -> sqlite3.Cursor: + """ + 执行批量写操作 + + Args: + sql: SQL语句 + params_list: 参数列表 + + Returns: + sqlite3.Cursor: 游标对象 + + Raises: + SQLiteError: SQL执行错误 + SQLiteBusyError: SQLITE_BUSY错误，达到最大重试次数 + """ + start_time = time.time() + + # 获取写锁 + self._acquire_write_lock() + + conn = self._create_connection() + cursor = None + + try: + # 使用BEGIN IMMEDIATE事务 + for attempt in range(self.busy_retry_attempts): + try: + # 开始IMMEDIATE事务 + conn.execute("BEGIN IMMEDIATE") + + # 执行批量SQL + cursor = conn.cursor() + cursor.executemany(sql, params_list) + + # 提交事务 + conn.commit() + return cursor + + except sqlite3.OperationalError as e: + # 如果是SQLITE_BUSY错误，进行重试 + if "database is locked" in str(e) or "SQLITE_BUSY" in str(e): + if attempt < self.busy_retry_attempts - 1: + conn.rollback() + print(f"SQLITE_BUSY错误，第{attempt + 1}次重试...") + time.sleep(self.busy_retry_delay) + continue + else: + # 最后一次尝试失败 + conn.rollback() + raise _handle_sqlite_error( + e, "执行批量写操作", sql, params_list, self.db_path, + attempt, self.busy_retry_attempts + ) + else: + # 其他操作错误，直接抛出 + conn.rollback() + raise _handle_sqlite_error(e, "执行批量写操作", sql, params_list, self.db_path) + except sqlite3.Error as e: + # 其他SQLite错误，回滚并抛出 + conn.rollback() + raise _handle_sqlite_error(e, "执行批量写操作", sql, params_list, self.db_path) + + except sqlite3.Error as e: + raise + finally: + # 关闭连接 + if conn: + conn.close() + # 释放写锁 + self._write_lock.release() + + # 记录执行时间 + execution_time = time.time() - start_time + if execution_time > 1.0: + print(f"⚠️ 批量写操作执行时间过长: {execution_time:.3f}秒 - SQL: {sql[:100]}{'...' if len(sql) > 100 else ''} (批量数量: {len(params_list)})") + + def fetch_all(self, sql: str, params: tuple = None) -> List[Dict[str, Any]]: + """ + 查询所有记录（使用读连接） + + Args: + sql: 查询SQL + params: 查询参数 + + Returns: + List[Dict[str, Any]]: 查询结果列表 + """ + start_time = time.time() + + conn = None + try: + # 获取读锁 + self._acquire_read_lock() + + conn = self._create_connection() + cursor = conn.cursor() + + # 执行查询 + if params: + cursor.execute(sql, params) + else: + cursor.execute(sql) + + rows = cursor.fetchall() + return [dict(row) for row in rows] + except sqlite3.Error as e: + # 抛出自定义异常 + raise _handle_sqlite_error(e, "执行读操作", sql, params, self.db_path) + finally: + # 关闭连接 + if conn: + conn.close() + # 释放读锁 + self._release_read_lock() + + # 记录执行时间 + execution_time = time.time() - start_time + if execution_time > 1.0: + print(f"⚠️ fetch_all执行时间过长: {execution_time:.3f}秒 - SQL: {sql[:100]}{'...' if len(sql) > 100 else ''}") + + def fetch_one(self, sql: str, params: tuple = None) -> Optional[Dict[str, Any]]: + """ + 查询单条记录（使用读连接） + + Args: + sql: 查询SQL + params: 查询参数 + + Returns: + Optional[Dict[str, Any]]: 查询结果，如果没有记录返回None + """ + start_time = time.time() + + conn = None + try: + # 获取读锁 + self._acquire_read_lock() + + conn = self._create_connection() + cursor = conn.cursor() + + # 执行查询 + if params: + cursor.execute(sql, params) + else: + cursor.execute(sql) + + row = cursor.fetchone() + return dict(row) if row else None + except sqlite3.Error as e: + # 抛出自定义异常 + raise _handle_sqlite_error(e, "执行读操作", sql, params, self.db_path) + finally: + # 关闭连接 + if conn: + conn.close() + # 释放读锁 + self._release_read_lock() + + # 记录执行时间 + execution_time = time.time() - start_time + if execution_time > 1.0: + print(f"⚠️ fetch_one执行时间过长: {execution_time:.3f}秒 - SQL: {sql[:100]}{'...' if len(sql) > 100 else ''}") + + def insert(self, table: str, data: Dict[str, Any]) -> int: + """ + 插入单条记录（使用写连接） + + Args: + table: 表名 + data: 插入数据字典 + + Returns: + int: 插入记录的主键ID + """ + columns = ', '.join(data.keys()) + placeholders = ', '.join(['?'] * len(data)) + sql = f"INSERT INTO {table} ({columns}) VALUES ({placeholders})" + + cursor = self.execute_write(sql, tuple(data.values())) + return cursor.lastrowid + + def insert_select(self, target_table: str, target_columns: List[str], + source_table: str, source_columns: List[str], + where_condition: str = None, where_params: tuple = None) -> int: + """ + 使用 INSERT INTO ... SELECT ... WHERE 语法从源表向目标表插入数据 + Args: + target_table: 目标表名 + target_columns: 目标表的列名列表 + source_table: 源表名 + source_columns: 源表的列名列表，与目标表列一一对应 + where_condition: WHERE条件语句（不含WHERE关键字） + where_params: WHERE条件参数 + + Returns: + int: 插入的记录数量 + """ + # 构建目标列和源列的字符串 + target_cols_str = ', '.join(target_columns) + source_cols_str = ', '.join(source_columns) + + # 构建基础SQL + sql = f"INSERT INTO {target_table} ({target_cols_str}) SELECT {source_cols_str} FROM {source_table}" + + # 添加WHERE条件（如果有） + if where_condition: + sql += f" WHERE {where_condition}" + + # 执行SQL语句 + cursor = self.execute_write(sql, where_params) + return cursor.rowcount + + def insert_many(self, table: str, data_list: List[Dict[str, Any]]) -> int: + """ + 批量插入记录（使用写连接） + + Args: + table: 表名 + data_list: 数据字典列表 + + Returns: + int: 插入的记录数量 + """ + if not data_list: + return 0 + + columns = ', '.join(data_list[0].keys()) + placeholders = ', '.join(['?'] * len(data_list[0])) + sql = f"INSERT INTO {table} ({columns}) VALUES ({placeholders})" + + params_list = [tuple(data.values()) for data in data_list] + cursor = self.executemany_write(sql, params_list) + return cursor.rowcount + + def update(self, table: str, data: Dict[str, Any], where: str, where_params: tuple = None) -> int: + """ + 更新记录（使用写连接） + + Args: + table: 表名 + data: 更新数据字典 + where: WHERE条件 + where_params: WHERE条件参数 + + Returns: + int: 更新的记录数量 + """ + set_clause = ', '.join([f"{key} = ?" for key in data.keys()]) + sql = f"UPDATE {table} SET {set_clause} WHERE {where}" + + params = tuple(data.values()) + if where_params: + params += where_params + + cursor = self.execute_write(sql, params) + return cursor.rowcount + + def delete(self, table: str, where: str, where_params: tuple = None) -> int: + """ + 删除记录（使用写连接） + + Args: + table: 表名 + where: WHERE条件 + where_params: WHERE条件参数 + + Returns: + int: 删除的记录数量 + """ + sql = f"DELETE FROM {table} WHERE {where}" + cursor = self.execute_write(sql, where_params) + return cursor.rowcount + + def add_column(self, table_name: str, column_name: str, column_type: str) -> bool: + """ + 添加列到表（使用写连接） + + Args: + table_name: 表名 + column_name: 列名 + column_type: 列数据类型 + + Returns: + bool: 是否添加成功 + """ + if not self.table_exists(table_name): + print(f"表 {table_name} 不存在") + return False + + sql = f"ALTER TABLE {table_name} ADD COLUMN {column_name} {column_type}" + + try: + self.execute_write(sql) + print(f"列 {column_name} 添加到表 {table_name} 成功") + return True + except sqlite3.Error as e: + print(f"添加列 {column_name} 到表 {table_name} 失败: {e}") + return False + + def table_exists(self, table_name: str) -> bool: + """ + 检查表是否存在（使用读连接） + + Args: + table_name: 表名 + + Returns: + bool: 表是否存在 + """ + sql = "SELECT name FROM sqlite_master WHERE type='table' AND name=?" + result = self.fetch_one(sql, (table_name,)) + return result is not None + + def create_table(self, table_name: str, columns: Dict[str, str]) -> bool: + """ + 创建表（使用写连接） + + Args: + table_name: 表名 + columns: 列定义字典，格式为 {列名: 数据类型} + + Returns: + bool: 是否创建成功 + """ + if self.table_exists(table_name): + print(f"表 {table_name} 已存在") + return True + + # 构建CREATE TABLE语句 + column_definitions = ', '.join([f"{col_name} {col_type}" for col_name, col_type in columns.items()]) + sql = f"CREATE TABLE {table_name} ({column_definitions})" + + try: + self.execute_write(sql) + print(f"表 {table_name} 创建成功") + return True + except sqlite3.Error as e: + print(f"创建表 {table_name} 失败: {e}") + return False + + def get_table_info(self, table_name: str) -> List[Dict[str, Any]]: + """ + 获取表结构信息（使用读连接） + + Args: + table_name: 表名 + + Returns: + List[Dict[str, Any]]: 表结构信息列表 + """ + if not self.table_exists(table_name): + print(f"表 {table_name} 不存在") + return [] + + sql = f"PRAGMA table_info({table_name})" + return self.fetch_all(sql) + + def get_table_count(self, table_name: str, where: str = None, where_params: tuple = None) -> int: + """ + 获取表记录数量（使用读连接） + + Args: + table_name: 表名 + where: WHERE条件 + where_params: WHERE条件参数 + + Returns: + int: 记录数量 + """ + sql = f"SELECT COUNT(*) as count FROM {table_name}" + if where: + sql += f" WHERE {where}" + + result = self.fetch_one(sql, where_params) + return result['count'] if result else 0 + + def vacuum(self,max_retries=3): + """执行VACUUM操作，优化数据库（使用写连接）""" + for attempt in range(max_retries): + try: + self.execute_write("VACUUM") + return True + except SQLiteBusyError: + if attempt < max_retries - 1: + time.sleep(2 ** attempt) # 指数退避 + else: + raise + + def backup_database(self, backup_path: str) -> bool: + """ + 备份数据库到指定文件（使用写连接） + + Args: + backup_path: 备份文件路径 + + Returns: + bool: 备份是否成功 + """ + try: + # 确保备份目录存在 + backup_dir = Path(backup_path).parent + if not backup_dir.exists(): + backup_dir.mkdir(parents=True, exist_ok=True) + + # 使用SQLite的备份API + source_conn = self._create_connection() + backup_conn = sqlite3.connect(backup_path) + + # 执行备份 + source_conn.backup(backup_conn) + backup_conn.close() + + print(f"数据库备份成功: {backup_path}") + return True + except sqlite3.Error as e: + print(f"数据库备份失败: {e}") + return False + + def create_incremental_backup(self, backup_dir: str, max_backups: int = 10) -> bool: + """ + 创建增量备份，自动管理备份文件数量 + + Args: + backup_dir: 备份目录 + max_backups: 最大备份文件数量 + + Returns: + bool: 备份是否成功 + """ + try: + # 确保备份目录存在 + backup_path = Path(backup_dir) + if not backup_path.exists(): + backup_path.mkdir(parents=True, exist_ok=True) + + # 生成备份文件名（包含时间戳） + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + db_name = Path(self.db_path).stem + backup_file = backup_path / f"{db_name}_backup_{timestamp}.db" + + # 执行备份 + success = self.backup_database(str(backup_file)) + + if success: + # 清理旧的备份文件 + backup_files = list(backup_path.glob(f"{db_name}_backup_*.db")) + backup_files.sort(key=lambda x: x.stat().st_mtime, reverse=True) + + # 删除超出数量的旧备份 + for old_backup in backup_files[max_backups:]: + old_backup.unlink() + print(f"删除旧备份: {old_backup.name}") + + print(f"增量备份完成，当前备份数量: {min(len(backup_files), max_backups)}") + + return success + except Exception as e: + print(f"增量备份失败: {e}") + return False + +# 使用示例 +if __name__ == "__main__": + # 创建数据库处理器 + db = SQLiteHandler.get_instance("db/three.db", max_readers=50, busy_timeout=4000) + + # # 创建表 + # columns = { + # "id": "INTEGER PRIMARY KEY AUTOINCREMENT", + # "name": "TEXT NOT NULL", + # "age": "INTEGER", + # "created_at": "TIMESTAMP DEFAULT CURRENT_TIMESTAMP" + # } + # db.create_table("users", columns) + + # # 插入数据 + # user_data = {"name": "张三", "age": 25} + # user_id = db.insert("users", user_data) + # print(f"插入用户ID: {user_id}") + + # # 批量插入 + # users_data = [ + # {"name": "李四", "age": 30}, + # {"name": "王五", "age": 28} + # ] + # count = db.insert_many("users", users_data) + # print(f"批量插入数量: {count}") + + # # 查询数据 + # users = db.fetch_all("SELECT * FROM users") + # print("所有用户:", users) + + # # 更新数据 + # update_count = db.update("users", {"age": 26}, "id = ?", (user_id,)) + # print(f"更新记录数: {update_count}") + + # # 查询单条 + # user = db.fetch_one("SELECT * FROM users WHERE id = ?", (user_id,)) + # print("单个用户:", user) + + # # 删除数据 + # delete_count = db.delete("users", "age > ?", (25,)) + # print(f"删除记录数: {delete_count}") + + # # 获取表信息 + # table_info = db.get_table_info("users") + # print("表结构:", table_info) + \ No newline at end of file diff --git a/config/__pycache__/__init__.cpython-39.pyc b/config/__pycache__/__init__.cpython-39.pyc deleted file mode 100644 index b40288b..0000000 Binary files a/config/__pycache__/__init__.cpython-39.pyc and /dev/null differ diff --git a/config/__pycache__/settings.cpython-39.pyc b/config/__pycache__/settings.cpython-39.pyc deleted file mode 100644 index 182437c..0000000 Binary files a/config/__pycache__/settings.cpython-39.pyc and /dev/null differ diff --git a/config/ini_manager.py b/config/ini_manager.py new file mode 100644 index 0000000..f58499c --- /dev/null +++ b/config/ini_manager.py @@ -0,0 +1,157 @@ +# config/config_manager.py +import logging +import os +import json +import configparser +import time +import threading +from common.logging_service import logging_service +from common.ini_handler import IniHandler + + +class IniManager: + """ + settings.init配置管理类，读取API相关配置 + 暂不支持多个配置文件 + """ + + def __init__(self): + """初始化实例变量""" + self._DEFAULT_CONFIG_FILE = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + 'settings.ini' + ) + self.ini_handler = IniHandler(self._DEFAULT_CONFIG_FILE) + # 默认配置值 + self._DEFAULT_API_MAX_RETRIES = 3 + self._DEFAULT_API_RETRY_INTERVAL = 1.0 + self._DEFAULT_API_TIMEOUT = 30.0 + self._DEFAULT_API_AUTH_TIMEOUT = 12 * 60 * 60 # 12小时 + self._DEFAULT_API_BASE_URL = "https://www.shnthy.com:9154" + self._DEFAULT_API_LOGIN_MODEL = { + "Program": 11, + "SC": "1000000001", + "loginName": "leduser", + "password": "bfcda35cf4eba92d4583931bbe4ff72ffdfa8b5c9c4b72611bd33f5babee069d" + } + + # 参数缓存字典，格式：{参数名: {'value': 值, 'mtime': 文件修改时间}} + self._param_cache = {} + # 缓存锁，确保多线程安全 + self._cache_lock = threading.RLock() + # 文件修改时间缓存 + self._file_mtime_cache = None + + def _get_file_mtime(self): + """获取文件修改时间，带缓存""" + try: + current_mtime = os.path.getmtime(self._DEFAULT_CONFIG_FILE) + return current_mtime + except (FileNotFoundError, OSError): + raise FileNotFoundError(f"配置文件不存在或无法访问: settings.ini") + + def _read_config_value(self, section, key, default_value, value_type=str): + """读取配置值，带参数级缓存 + + Args: + section: 配置段 + key: 配置键 + default_value: 默认值 + value_type: 值类型 + + Returns: + 配置值 + """ + cache_key = f"{section}.{key}" + current_mtime = self._get_file_mtime() + # 检查缓存 + if cache_key in self._param_cache: + cached_data = self._param_cache[cache_key] + # 如果文件未修改，直接返回缓存值 + if cached_data['value'] is not None and cached_data['mtime'] == current_mtime: + return cached_data['value'] + + try: + # 获取值 + if value_type == int: + value = self.ini_handler.get_int_value(section, key) + elif value_type == float: + value = self.ini_handler.get_float_value(section, key) + else: + value = self.ini_handler.get_value(section, key) + + except Exception as e: + value = default_value + logging_service.log_message(logging.ERROR,str(e)) + + # 使用锁确保缓存操作的线程安全 + with self._cache_lock: + self._param_cache[cache_key] = { + 'value': value, + 'mtime': current_mtime + } + return value + + @property + def api_max_retries(self): + """获取API最大重试次数""" + return self._read_config_value('api', 'max_retries', self._DEFAULT_API_MAX_RETRIES, int) + + @property + def api_retry_interval(self): + """获取API重试间隔(秒)""" + return self._read_config_value('api', 'retry_interval', self._DEFAULT_API_RETRY_INTERVAL, float) + + @property + def api_timeout(self): + """获取API请求超时时间(秒)""" + return self._read_config_value('api', 'timeout', self._DEFAULT_API_TIMEOUT, float) + + @property + def api_auth_timeout(self): + """获取API授权失效前最大时间(秒)""" + return self._read_config_value('api', 'auth_timeout', self._DEFAULT_API_AUTH_TIMEOUT, int) + + @property + def api_base_url(self): + """获取API基础URL""" + return self._read_config_value('api', 'base_url', self._DEFAULT_API_BASE_URL, str) + + @property + def api_login_url(self): + """获取API登录URL""" + return f"{self.api_base_url}/api/user/perlogin" + + @property + def api_login_model(self): + """获取API登录模型(字典格式)""" + login_model_str = self._read_config_value('api', 'login_model', None, str) + if login_model_str: + try: + value = json.loads(login_model_str) + except Exception as e: + print(f"解析API登录模型失败: {e}") + value = self._DEFAULT_API_LOGIN_MODEL + else: + value = self._DEFAULT_API_LOGIN_MODEL + + return value + + @property + def log_path(self): + """获取日志文件路径""" + return self._read_config_value('app', 'LOG_PATH', 'logs/app.log', str) + +ini_manager = IniManager() + + +# if __name__ == '__main__': +# logging_service.init_log(ini_manager.log_path) +# try: + +# logging_service.log_message(logging.INFO, "测试INI配置管理类") +# logging_service.log_message(logging.ERROR, f"api_max_retries") +# except Exception as e: +# logging_service.log_message(logging.ERROR,str(e)) + +# time.sleep(1000) \ No newline at end of file diff --git a/config/settings.py b/config/settings.py index 9587a08..6ee5883 100644 --- a/config/settings.py +++ b/config/settings.py @@ -49,3 +49,7 @@ class Settings: self.visual_check_interval = 1.0 # 视觉检查间隔(秒) self.alignment_check_interval = 0.5 # 对齐检查间隔(秒) self.max_error_count = 3 # 最大错误计数 + + #是否在线生产 + self.is_online_control = True # 是否API在线 + diff --git a/controller/main_controller.py b/controller/main_controller.py index 1c0767c..051dc75 100644 --- a/controller/main_controller.py +++ b/controller/main_controller.py @@ -18,4 +18,4 @@ class MainController: self.main_window.show() def _initSubViews(self): - pass \ No newline at end of file + pass diff --git a/db/three.db b/db/three.db new file mode 100644 index 0000000..cb2afc3 Binary files /dev/null and b/db/three.db differ diff --git a/db/three.db-shm b/db/three.db-shm new file mode 100644 index 0000000..6199341 Binary files /dev/null and b/db/three.db-shm differ diff --git a/db/three.db-wal b/db/three.db-wal new file mode 100644 index 0000000..08e8bfd Binary files /dev/null and b/db/three.db-wal differ diff --git a/doc/table表设计.doc b/doc/table表设计.doc new file mode 100644 index 0000000..9d8a23d Binary files /dev/null and b/doc/table表设计.doc differ diff --git a/doc/~$ble表设计.doc b/doc/~$ble表设计.doc new file mode 100644 index 0000000..f0bcfdb Binary files /dev/null and b/doc/~$ble表设计.doc differ diff --git a/logs/app.log b/logs/app.log index e0187ca..c907599 100644 --- a/logs/app.log +++ b/logs/app.log @@ -1,26 +1,3 @@ -2025-09-20 15:47:36,993 - FeedingControl.FeedingController - INFO - 豸ʼ -2025-09-20 15:47:37,011 - FeedingControl.MainSystem - INFO - ιϵͳʼ -2025-09-20 15:47:37,011 - FeedingControl.MainSystem - INFO - ϵͳ -2025-09-20 15:47:37,011 - FeedingControl.MainSystem - ERROR - ϵͳд: 'FeedingController' object has no attribute 'start' -2025-09-20 15:47:37,011 - FeedingControl.MainSystem - INFO - ֹͣϵͳ -2025-09-20 15:47:37,012 - FeedingControl - ERROR - ϵͳʧ: 'FeedingController' object has no attribute 'stop' -2025-09-20 15:53:09,420 - FeedingControl.FeedingController - INFO - 豸ʼ -2025-09-20 15:53:09,420 - FeedingControl.MainSystem - INFO - ιϵͳʼ -2025-09-20 15:53:09,420 - FeedingControl.MainSystem - INFO - ϵͳ -2025-09-20 15:53:09,429 - FeedingControl.MainSystem - ERROR - ϵͳд: 'FeedingController' object has no attribute 'start' -2025-09-20 15:53:09,429 - FeedingControl.MainSystem - INFO - ֹͣϵͳ -2025-09-20 15:53:09,429 - FeedingControl - ERROR - ϵͳʧ: 'FeedingController' object has no attribute 'stop' -2025-09-20 15:53:49,321 - FeedingControl.FeedingController - INFO - 豸ʼ -2025-09-20 15:53:49,322 - FeedingControl.MainSystem - INFO - ιϵͳʼ -2025-09-20 15:53:49,322 - FeedingControl.MainSystem - WARNING - ϵͳδ -2025-09-20 15:53:49,322 - FeedingControl.MainSystem - INFO - ϵͳ -2025-09-20 15:53:49,322 - FeedingControl.MainSystem - ERROR - ϵͳд: 'FeedingController' object has no attribute 'start' -2025-09-20 15:53:49,322 - FeedingControl.MainSystem - INFO - ֹͣϵͳ -2025-09-20 15:53:49,322 - FeedingControl - ERROR - ϵͳʧ: 'FeedingController' object has no attribute 'stop' -2025-09-20 16:00:19,400 - FeedingControl.FeedingController - INFO - 豸ʼ -2025-09-20 16:00:19,400 - FeedingControl.MainSystem - INFO - ιϵͳʼ -2025-09-20 16:00:19,400 - FeedingControl.MainSystem - WARNING - ϵͳδ -2025-09-20 16:00:19,400 - FeedingControl.MainSystem - INFO - ϵͳ -2025-09-20 16:00:19,400 - FeedingControl.MainSystem - ERROR - ϵͳд: 'FeedingController' object has no attribute 'start' -2025-09-20 16:00:19,400 - FeedingControl.MainSystem - INFO - ֹͣϵͳ -2025-09-20 16:00:19,400 - FeedingControl - ERROR - ϵͳʧ: 'FeedingController' object has no attribute 'stop' +2025-10-29 17:18:22 - ERROR- INI文件不存在 (文件: settings2.ini) +2025-10-29 17:34:40 - INFO- 测试INI配置管理类 +2025-10-29 17:34:41 - ERROR- api_max_retries diff --git a/resources/resources_rc.py b/resources/resources_rc.py index cd54c61..18dda4d 100644 --- a/resources/resources_rc.py +++ b/resources/resources_rc.py @@ -1,6 +1,6 @@ # Resource object code (Python 3) # Created by: object code -# Created by: The Resource Compiler for Qt version 6.9.1 +# Created by: The Resource Compiler for Qt version 6.7.2 # WARNING! All changes made in this file will be lost! from PySide6 import QtCore diff --git a/service/__init__.py b/service/__init__.py new file mode 100644 index 0000000..d472482 --- /dev/null +++ b/service/__init__.py @@ -0,0 +1 @@ +# Service package initialization \ No newline at end of file diff --git a/service/api_http_client.py b/service/api_http_client.py new file mode 100644 index 0000000..d935f3d --- /dev/null +++ b/service/api_http_client.py @@ -0,0 +1,285 @@ +from logging import exception +from typing import Optional, Dict, Any +import time +import threading +import requests +from models import LoginResponse +from config.ini_manager import ini_manager + + +class BaseHttpClient: + """基础HTTP客户端 - 纯网络请求功能""" + + def __init__(self): + """初始化基础HTTP客户端""" + self._session = requests.Session() + self._settings = None + + def request(self, method: str, url: str, data: Dict[str, Any] = None, + headers: Dict[str, str] = None, timeout: int = None, + retries: int = 0, retry_interval: float = 1.0, + **kwargs) -> Dict[str, Any]: + """ + 发送HTTP请求，支持重试机制 + + Args: + method: HTTP方法 + url: 请求URL + data: 请求数据 + headers: 请求头 + timeout: 超时时间 + retries: 重试次数（可选，默认使用配置中的重试次数） + retry_interval: 重试间隔（秒，默认1秒） + **kwargs: 其他参数 + + Returns: + Dict[str, Any]: 响应数据 + """ + + if headers is None: + headers = {} + # 重试逻辑 + for attempt in range(retries + 1): + try: + if method.upper() == 'GET': + response = self._session.get( + url, headers=headers, timeout=timeout, **kwargs + ) + elif method.upper() == 'POST': + response = self._session.post( + url, json=data, headers=headers, timeout=timeout, **kwargs + ) + else: + raise ValueError(f"不支持的HTTP方法: {method}") + + # 检查响应状态 + response.raise_for_status() + + # 解析JSON响应 + return response.json() + + except exception as e: + # 如果是最后一次尝试，直接抛出异常 + if attempt == retries: + print(f"请求失败（第{attempt + 1}次尝试）: {e}") + raise + # 打印重试信息 + print(f"请求失败（第{attempt + 1}次尝试），{retry_interval}秒后重试: {e}") + # 等待重试间隔 + time.sleep(retry_interval) + + def get(self, url: str, headers: Dict[str, str] = None, + timeout: int = None, retries: int = 0, retry_interval: float = 1.0, + **kwargs) -> Dict[str, Any]: + """ + GET请求 + + Args: + url: 请求URL + headers: 请求头 + timeout: 超时时间 + **kwargs: 其他参数 + + Returns: + Dict[str, Any]: 响应数据 + """ + return self.request('GET', url, headers=headers, timeout=timeout, retries=retries, retry_interval=retry_interval, **kwargs) + + def post(self, url: str, data: Dict[str, Any] = None, + headers: Dict[str, str] = None, timeout: int = None, + retries: int = 0, retry_interval: float = 1.0, + **kwargs) -> Dict[str, Any]: + """ + POST请求 + + Args: + url: 请求URL + data: 请求数据 + headers: 请求头 + timeout: 超时时间 + retries: 重试次数（可选，默认使用配置中的重试次数） + retry_interval: 重试间隔（秒，默认1秒） + **kwargs: 其他参数 + + Returns: + Dict[str, Any]: 响应数据 + """ + return self.request('POST', url, data=data, headers=headers, timeout=timeout, retries=retries, retry_interval=retry_interval, **kwargs) + + +class ApiHttpClient(BaseHttpClient): + """API客户端 - 业务API调用，整合认证和单例功能""" + + def __init__(self): + """初始化API客户端""" + """初始化API客户端""" + super().__init__() + # 认证缓存 + self._auth_cache = { + 'app_id': None, + 'expire_time': None, + 'sign_token': None, + 'zr_jwt': None + } + self._cache_lock = threading.RLock() + + @property + def settings(self): + """获取配置对象，由子类实现""" + if self._settings is None: + self._settings = ini_manager + return self._settings + + + def login(self) -> bool: + """ + 用户登录获取AppID + + Args: + url: 登录URL（可选，默认使用配置中的URL） + login_model: 登录请求模型（可选，默认使用配置中的模型） + + Returns: + bool: 登录是否成功 + """ + url = self.settings.api_login_url + login_model = self.settings.api_login_model + + print("开始登录...") + try: + # 发送登录请求 + response_data = self.request( + method='POST', + url=url, + data=login_model, + timeout=self.settings.api_timeout + ) + + # 解析登录响应 + login_response = LoginResponse(**response_data) + + if login_response.Code != 200: + error_msg = login_response.Message or "登录失败" + print(f"获取AppID失败: {error_msg}") + return False + + # 更新认证缓存 + with self._cache_lock: + self._auth_cache.update({ + 'app_id': login_response.app_id, + 'expire_time': login_response.expire_time, + 'sign_token': login_response.sign_token, + 'zr_jwt': login_response.zr_jwt + }) + + print(f"成功获取AppID: {self._auth_cache['app_id']}") + print(f"过期时间: {self._auth_cache['expire_time']}") + return True + + except Exception as e: + print(f"登录过程中出现异常: {e}") + self._clear_auth_cache() + return False + + def is_app_id_valid(self) -> bool: + """检查AppID是否有效""" + with self._cache_lock: + expire_time = self._auth_cache.get('expire_time') + + if not expire_time: + return False + + # 检查是否过期（提前12小时过期，避免临界情况） + try: + expire_timestamp = time.mktime(time.strptime(expire_time, '%Y-%m-%d %H:%M:%S')) + is_valid = time.time() < expire_timestamp - self.settings.api_auth_timeout + if not is_valid: + print("认证信息已过期") + return is_valid + except (ValueError, TypeError): + print("日期格式不正确") + return False + + def _get_auth_headers(self) -> Dict[str, str]: + """获取认证头信息""" + with self._cache_lock: + app_id = self._auth_cache.get('app_id') + + headers = { + 'AppID': app_id, + 'Content-Type': 'application/json' + } + + return headers + + + def get(self, url: str, auth: bool = True, **kwargs) -> Dict[str, Any]: + """ + GET请求（支持认证检查） + + Args: + url: 请求URL + auth: 是否需要认证 + timeout: 超时时间 + retries: 重试次数 + retry_interval: 重试间隔 + **kwargs: 其他参数 + + Returns: + Dict[str, Any]: 响应数据 + """ + if auth: + if not self.is_app_id_valid(): + self.login() + if not self.is_app_id_valid(): + raise Exception("登录失败，无法获取有效AppID") + auth_headers = self._get_auth_headers() + return self.request(method='GET', url=url, headers=auth_headers, timeout=self.settings.api_timeout, + retries=self.settings.api_max_retries, retry_interval=self.settings.api_retry_interval, **kwargs) + else: + return self.request(method='GET', url=url, timeout=self.settings.api_timeout, + retries=self.settings.api_max_retries, retry_interval=self.settings.api_retry_interval, **kwargs) + + def post(self, url: str, data: Dict[str, Any] = None, auth: bool = True,**kwargs) -> Dict[str, Any]: + """ + POST请求（支持认证检查） + + Args: + url: 请求URL + data: 请求数据 + auth: 是否需要认证 + timeout: 超时时间 + retries: 重试次数 + retry_interval: 重试间隔 + **kwargs: 其他参数 + + Returns: + Dict[str, Any]: 响应数据 + """ + if auth: + if not self.is_app_id_valid(): + self.login() + if not self.is_app_id_valid(): + raise Exception("登录失败，无法获取有效AppID") + auth_headers = self._get_auth_headers() + return self.request(method='POST', url=url, data=data, headers=auth_headers, timeout=self.settings.api_timeout, + retries=self.settings.api_max_retries, retry_interval=self.settings.api_retry_interval, **kwargs) + else: + return self.request(method='POST', url=url, data=data, timeout=self.settings.api_timeout, + retries=self.settings.api_max_retries, retry_interval=self.settings.api_retry_interval, **kwargs) + + def _clear_auth_cache(self): + """清除认证缓存""" + with self._cache_lock: + self._auth_cache.clear() + # 重新初始化必要的键 + self._auth_cache.update({ + 'app_id': None, + 'expire_time': None, + 'sign_token': None, + 'zr_jwt': None + }) + print("认证缓存已清除") + + +api_http_client = ApiHttpClient() \ No newline at end of file diff --git a/service/mould_service.py b/service/mould_service.py new file mode 100644 index 0000000..115f7ee --- /dev/null +++ b/service/mould_service.py @@ -0,0 +1,158 @@ +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 +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 + + + +if __name__ == "__main__": + # 创建模具服务实例 + mould_service = MouldService() + 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) + diff --git a/settings.ini b/settings.ini new file mode 100644 index 0000000..c491aa3 --- /dev/null +++ b/settings.ini @@ -0,0 +1,12 @@ +[api] +max_retries = 3 +retry_interval = 1.0 +timeout = 30.0 +auth_timeout = 43200 +base_url = https://www.shnthy.com:9154 +login_model = {"Program": 11, "SC": "1000000001", "loginName": "leduser", "password": "bfcda35cf4eba92d4583931bbe4ff72ffdfa8b5c9c4b72611bd33f5babee069d"} + +[app] +log_path = logs/app.log +db_path = db/three.db +