""" 数据库操作模块 使用SQLite存储地磅单、公司名称、货物名称等数据 同时支持MySQL数据库存储（双数据库写入） """ import sqlite3 import os import sys from datetime import datetime from typing import List, Dict, Optional, Tuple import logging # 尝试导入MySQL相关模块 try: import pymysql PYMySQL_AVAILABLE = True except ImportError: PYMySQL_AVAILABLE = False # 尝试导入配置模块 try: import config MYSQL_ENABLED = getattr(config, 'ENABLE_MYSQL', False) and PYMySQL_AVAILABLE MYSQL_CONFIG = getattr(config, 'MYSQL_CONFIG', {}) CONTINUE_ON_MYSQL_ERROR = getattr(config, 'CONTINUE_ON_MYSQL_ERROR', True) except ImportError: MYSQL_ENABLED = False MYSQL_CONFIG = {} CONTINUE_ON_MYSQL_ERROR = True # 配置日志 - 设置为INFO级别以便看到MySQL操作信息 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) def get_base_path(): """ 获取程序运行的基础路径 在开发环境中返回脚本所在目录，在打包后返回exe所在目录 """ if getattr(sys, 'frozen', False): # 打包后的exe环境 return os.path.dirname(sys.executable) else: # 开发环境 return os.path.dirname(os.path.abspath(__file__)) class Database: """数据库操作类，支持SQLite和MySQL双数据库存储""" def __init__(self, db_path: str = None): """ 初始化数据库连接 :param db_path: 数据库文件路径，如果为None则使用默认路径（exe同目录下的dibang.db） """ if db_path is None: # 使用exe同目录下的dibang.db base_path = get_base_path() db_path = os.path.join(base_path, "dibang.db") self.db_path = db_path self.mysql_enabled = MYSQL_ENABLED self.mysql_connection = None self.init_database() def get_connection(self) -> sqlite3.Connection: """获取SQLite数据库连接""" conn = sqlite3.connect(self.db_path) conn.row_factory = sqlite3.Row # 使查询结果可以通过列名访问 return conn def get_mysql_connection(self, force_reconnect: bool = False): """ 获取MySQL数据库连接 :param force_reconnect: 是否强制重新连接（用于确保读取最新数据） :return: MySQL连接对象 """ if not self.mysql_enabled: return None try: # 如果强制重新连接，先关闭旧连接 if force_reconnect and self.mysql_connection is not None: try: self.mysql_connection.close() logger.info("强制重新建立MySQL连接以获取最新数据") except: pass self.mysql_connection = None # 检查连接是否存在且有效 if self.mysql_connection is None: self.mysql_connection = pymysql.connect( host=MYSQL_CONFIG.get('host', 'localhost'), port=MYSQL_CONFIG.get('port', 3306), user=MYSQL_CONFIG.get('user', 'root'), password=MYSQL_CONFIG.get('password', ''), database=MYSQL_CONFIG.get('database', 'dibang'), charset=MYSQL_CONFIG.get('charset', 'utf8mb4'), cursorclass=pymysql.cursors.DictCursor, autocommit=False ) else: # 检查连接是否有效 try: self.mysql_connection.ping(reconnect=True) except Exception as ping_error: # 连接失效，重新连接 logger.info(f"MySQL连接失效，重新连接: {str(ping_error)}") try: self.mysql_connection.close() except: pass self.mysql_connection = pymysql.connect( host=MYSQL_CONFIG.get('host', 'localhost'), port=MYSQL_CONFIG.get('port', 3306), user=MYSQL_CONFIG.get('user', 'root'), password=MYSQL_CONFIG.get('password', ''), database=MYSQL_CONFIG.get('database', 'dibang'), charset=MYSQL_CONFIG.get('charset', 'utf8mb4'), cursorclass=pymysql.cursors.DictCursor, autocommit=False ) return self.mysql_connection except Exception as e: logger.warning(f"MySQL连接失败: {str(e)}") if not CONTINUE_ON_MYSQL_ERROR: raise return None def _execute_mysql(self, sql: str, params: tuple = None): """执行MySQL语句（内部方法）""" if not self.mysql_enabled: return conn = self.get_mysql_connection() if conn is None: return try: with conn.cursor() as cursor: cursor.execute(sql, params) conn.commit() except Exception as e: logger.warning(f"MySQL执行失败: {str(e)}, SQL: {sql}") if not CONTINUE_ON_MYSQL_ERROR: raise try: conn.rollback() except: pass def init_database(self): """初始化数据库表结构""" conn = self.get_connection() cursor = conn.cursor() # 创建地磅单表 cursor.execute(""" CREATE TABLE IF NOT EXISTS weighing_slips ( id INTEGER PRIMARY KEY AUTOINCREMENT, vehicle_no TEXT NOT NULL, plan_no TEXT, cargo_name TEXT NOT NULL, business_type TEXT NOT NULL, batch_no TEXT, supplier TEXT NOT NULL, consignee TEXT NOT NULL, gross_weight REAL NOT NULL, gross_weight_time TEXT NOT NULL, tare_weight REAL NOT NULL, tare_weight_time TEXT NOT NULL, deduction REAL DEFAULT 0, net_weight REAL NOT NULL, planned_quantity REAL, executed_quantity REAL, weighing_point TEXT DEFAULT '山峡地磅', remarks TEXT, is_voided INTEGER DEFAULT 0, created_at TEXT NOT NULL ) """) # 创建公司名称表 cursor.execute(""" CREATE TABLE IF NOT EXISTS companies ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, created_at TEXT NOT NULL ) """) # 创建货物名称表 cursor.execute(""" CREATE TABLE IF NOT EXISTS cargos ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, created_at TEXT NOT NULL ) """) # 创建采购计划表 cursor.execute(""" CREATE TABLE IF NOT EXISTS purchase_plans ( id INTEGER PRIMARY KEY AUTOINCREMENT, plan_date TEXT NOT NULL, business_type TEXT NOT NULL, product TEXT NOT NULL, specification TEXT, planned_quantity REAL NOT NULL, vehicle_no TEXT NOT NULL, customer_name TEXT, remarks TEXT, created_at TEXT NOT NULL ) """) # 如果表已存在但没有customer_name字段，则添加该字段 try: cursor.execute("ALTER TABLE purchase_plans ADD COLUMN customer_name TEXT") except sqlite3.OperationalError: # 字段已存在，忽略错误 pass # 如果表已存在但没有remarks字段，则添加该字段 try: cursor.execute("ALTER TABLE purchase_plans ADD COLUMN remarks TEXT") except sqlite3.OperationalError: # 字段已存在，忽略错误 pass conn.commit() conn.close() # 初始化MySQL数据库表结构 if self.mysql_enabled: self._init_mysql_database() def _init_mysql_database(self): """初始化MySQL数据库表结构""" conn = self.get_mysql_connection() if conn is None: return try: with conn.cursor() as cursor: # 创建地磅单表 cursor.execute(""" CREATE TABLE IF NOT EXISTS weighing_slips ( id INT PRIMARY KEY AUTO_INCREMENT, vehicle_no VARCHAR(255) NOT NULL, plan_no VARCHAR(255), cargo_name VARCHAR(255) NOT NULL, business_type VARCHAR(255) NOT NULL, batch_no VARCHAR(255), supplier VARCHAR(255) NOT NULL, consignee VARCHAR(255) NOT NULL, gross_weight DOUBLE NOT NULL, gross_weight_time VARCHAR(255) NOT NULL, tare_weight DOUBLE NOT NULL, tare_weight_time VARCHAR(255) NOT NULL, deduction DOUBLE DEFAULT 0, net_weight DOUBLE NOT NULL, planned_quantity DOUBLE, executed_quantity DOUBLE, weighing_point VARCHAR(255) DEFAULT '山峡地磅', remarks TEXT, is_voided INT DEFAULT 0, created_at VARCHAR(255) NOT NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) # 创建公司名称表 cursor.execute(""" CREATE TABLE IF NOT EXISTS companies ( id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(255) NOT NULL UNIQUE, created_at VARCHAR(255) NOT NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) # 创建货物名称表 cursor.execute(""" CREATE TABLE IF NOT EXISTS cargos ( id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(255) NOT NULL UNIQUE, created_at VARCHAR(255) NOT NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) # 创建采购计划表 cursor.execute(""" CREATE TABLE IF NOT EXISTS purchase_plans ( id INT PRIMARY KEY AUTO_INCREMENT, plan_date VARCHAR(255) NOT NULL, business_type VARCHAR(255) NOT NULL, product VARCHAR(255) NOT NULL, specification VARCHAR(255), planned_quantity DOUBLE NOT NULL, vehicle_no VARCHAR(255) NOT NULL, customer_name VARCHAR(255), remarks TEXT, created_at VARCHAR(255) NOT NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) # 检查并添加customer_name字段（如果不存在） try: cursor.execute("ALTER TABLE purchase_plans ADD COLUMN customer_name VARCHAR(255)") except: pass # 字段已存在，忽略错误 # 检查并添加remarks字段（如果不存在） try: cursor.execute("ALTER TABLE purchase_plans ADD COLUMN remarks TEXT") except: pass # 字段已存在，忽略错误 conn.commit() except Exception as e: logger.warning(f"MySQL表初始化失败: {str(e)}") if not CONTINUE_ON_MYSQL_ERROR: raise try: conn.rollback() except: pass # ========== 地磅单相关操作 ========== def insert_weighing_slip(self, data: Dict) -> int: """ 插入地磅单记录 :param data: 地磅单数据字典 :return: 插入记录的ID """ # 计算净重 net_weight = data['gross_weight'] - data['tare_weight'] - data.get('deduction', 0) created_at = datetime.now().strftime('%Y-%m-%d %H:%M:%S') record_id = None # 先操作MySQL，获取MySQL生成的ID if self.mysql_enabled: try: mysql_conn = self.get_mysql_connection() if mysql_conn: with mysql_conn.cursor() as mysql_cursor: mysql_cursor.execute(""" INSERT INTO weighing_slips ( vehicle_no, plan_no, cargo_name, business_type, batch_no, supplier, consignee, gross_weight, gross_weight_time, tare_weight, tare_weight_time, deduction, net_weight, planned_quantity, executed_quantity, weighing_point, remarks, is_voided, created_at ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( data['vehicle_no'], data.get('plan_no', ''), data['cargo_name'], data['business_type'], data.get('batch_no', ''), data['supplier'], data['consignee'], data['gross_weight'], data['gross_weight_time'], data['tare_weight'], data['tare_weight_time'], data.get('deduction', 0), net_weight, data.get('planned_quantity'), data.get('executed_quantity'), data.get('weighing_point', '山峡地磅'), data.get('remarks', ''), 0, # is_voided created_at )) record_id = mysql_cursor.lastrowid mysql_conn.commit() except Exception as e: logger.warning(f"MySQL插入地磅单失败: {str(e)}") if not CONTINUE_ON_MYSQL_ERROR: raise # 如果MySQL失败或未启用，使用SQLite生成ID if record_id is None: conn = self.get_connection() cursor = conn.cursor() cursor.execute(""" INSERT INTO weighing_slips ( vehicle_no, plan_no, cargo_name, business_type, batch_no, supplier, consignee, gross_weight, gross_weight_time, tare_weight, tare_weight_time, deduction, net_weight, planned_quantity, executed_quantity, weighing_point, remarks, is_voided, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( data['vehicle_no'], data.get('plan_no', ''), data['cargo_name'], data['business_type'], data.get('batch_no', ''), data['supplier'], data['consignee'], data['gross_weight'], data['gross_weight_time'], data['tare_weight'], data['tare_weight_time'], data.get('deduction', 0), net_weight, data.get('planned_quantity'), data.get('executed_quantity'), data.get('weighing_point', '山峡地磅'), data.get('remarks', ''), 0, # is_voided created_at )) record_id = cursor.lastrowid conn.commit() conn.close() else: # 使用MySQL的ID同步到SQLite conn = self.get_connection() cursor = conn.cursor() cursor.execute(""" INSERT INTO weighing_slips ( id, vehicle_no, plan_no, cargo_name, business_type, batch_no, supplier, consignee, gross_weight, gross_weight_time, tare_weight, tare_weight_time, deduction, net_weight, planned_quantity, executed_quantity, weighing_point, remarks, is_voided, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( record_id, # 使用MySQL返回的ID data['vehicle_no'], data.get('plan_no', ''), data['cargo_name'], data['business_type'], data.get('batch_no', ''), data['supplier'], data['consignee'], data['gross_weight'], data['gross_weight_time'], data['tare_weight'], data['tare_weight_time'], data.get('deduction', 0), net_weight, data.get('planned_quantity'), data.get('executed_quantity'), data.get('weighing_point', '山峡地磅'), data.get('remarks', ''), 0, # is_voided created_at )) conn.commit() conn.close() return record_id def get_all_weighing_slips(self, include_voided: bool = False) -> List[Dict]: """ 获取所有地磅单记录 :param include_voided: 是否包含作废的记录 :return: 地磅单记录列表 """ # 先尝试从MySQL查询（强制重新连接以确保获取最新数据） if self.mysql_enabled: try: # 强制重新连接以确保读取最新数据 mysql_conn = self.get_mysql_connection(force_reconnect=True) if mysql_conn: try: # 确保事务已提交 mysql_conn.commit() # 设置隔离级别为READ COMMITTED with mysql_conn.cursor() as cursor: cursor.execute("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED") with mysql_conn.cursor() as mysql_cursor: if include_voided: mysql_cursor.execute(""" SELECT * FROM weighing_slips ORDER BY id DESC """) else: mysql_cursor.execute(""" SELECT * FROM weighing_slips WHERE is_voided = 0 ORDER BY id DESC """) rows = mysql_cursor.fetchall() result = [dict(row) for row in rows] logger.info(f"✅ 从MySQL获取地磅单列表，共{len(result)}条记录") return result except Exception as query_error: logger.error(f"MySQL查询地磅单执行失败: {str(query_error)}", exc_info=True) raise else: logger.warning("⚠️ MySQL连接不可用，将从SQLite查询地磅单列表") except Exception as e: logger.warning(f"⚠️ MySQL查询地磅单失败，将从SQLite查询: {str(e)}", exc_info=True) # MySQL失败或未启用，从SQLite查询 logger.info("从SQLite获取地磅单列表") conn = self.get_connection() cursor = conn.cursor() if include_voided: cursor.execute(""" SELECT * FROM weighing_slips ORDER BY id DESC """) else: cursor.execute(""" SELECT * FROM weighing_slips WHERE is_voided = 0 ORDER BY id DESC """) rows = cursor.fetchall() conn.close() result = [dict(row) for row in rows] logger.info(f"从SQLite获取地磅单列表，共{len(result)}条记录") return result def get_weighing_slip_by_id(self, slip_id: int) -> Optional[Dict]: """ 根据ID获取地磅单记录 :param slip_id: 地磅单ID :return: 地磅单记录字典，如果不存在返回None """ # 先尝试从MySQL查询 if self.mysql_enabled: try: mysql_conn = self.get_mysql_connection() if mysql_conn: with mysql_conn.cursor() as mysql_cursor: mysql_cursor.execute("SELECT * FROM weighing_slips WHERE id = %s", (slip_id,)) row = mysql_cursor.fetchone() if row: return dict(row) except Exception as e: logger.warning(f"MySQL查询地磅单失败，将从SQLite查询: {str(e)}") # MySQL失败或未启用，从SQLite查询 conn = self.get_connection() cursor = conn.cursor() cursor.execute("SELECT * FROM weighing_slips WHERE id = ?", (slip_id,)) row = cursor.fetchone() conn.close() return dict(row) if row else None def void_weighing_slip(self, slip_id: int) -> bool: """ 作废地磅单 :param slip_id: 地磅单ID :return: 是否成功 """ success = False # 先操作MySQL if self.mysql_enabled: try: mysql_conn = self.get_mysql_connection() if mysql_conn: with mysql_conn.cursor() as mysql_cursor: mysql_cursor.execute(""" UPDATE weighing_slips SET is_voided = 1 WHERE id = %s """, (slip_id,)) affected_rows = mysql_cursor.rowcount mysql_conn.commit() if affected_rows > 0: success = True logger.info(f"MySQL作废地磅单成功: ID {slip_id}") else: logger.warning(f"MySQL作废地磅单: ID {slip_id} 未找到记录") else: logger.warning(f"MySQL连接失败，无法作废地磅单 ID {slip_id}") except Exception as e: logger.error(f"MySQL作废地磅单失败: ID {slip_id}, 错误: {str(e)}", exc_info=True) if not CONTINUE_ON_MYSQL_ERROR: raise # 同步到SQLite if success or not self.mysql_enabled: try: conn = self.get_connection() cursor = conn.cursor() cursor.execute(""" UPDATE weighing_slips SET is_voided = 1 WHERE id = ? """, (slip_id,)) sqlite_success = cursor.rowcount > 0 conn.commit() conn.close() if self.mysql_enabled and not sqlite_success: logger.warning(f"SQLite作废地磅单: ID {slip_id} 未找到记录，可能ID不匹配") elif not self.mysql_enabled: success = sqlite_success except Exception as e: logger.error(f"SQLite作废地磅单失败: ID {slip_id}, 错误: {str(e)}") return success # ========== 公司名称相关操作 ========== def insert_company(self, name: str) -> int: """ 插入公司名称 :param name: 公司名称 :return: 插入记录的ID """ created_at = datetime.now().strftime('%Y-%m-%d %H:%M:%S') record_id = None # 先操作MySQL，获取MySQL生成的ID if self.mysql_enabled: try: mysql_conn = self.get_mysql_connection() if mysql_conn: with mysql_conn.cursor() as mysql_cursor: mysql_cursor.execute(""" INSERT INTO companies (name, created_at) VALUES (%s, %s) """, (name, created_at)) record_id = mysql_cursor.lastrowid mysql_conn.commit() except Exception as e: logger.warning(f"MySQL插入公司失败: {str(e)}") if not CONTINUE_ON_MYSQL_ERROR: raise # 如果MySQL失败或未启用，使用SQLite生成ID if record_id is None: conn = self.get_connection() cursor = conn.cursor() cursor.execute(""" INSERT INTO companies (name, created_at) VALUES (?, ?) """, (name, created_at)) record_id = cursor.lastrowid conn.commit() conn.close() else: # 使用MySQL的ID同步到SQLite conn = self.get_connection() cursor = conn.cursor() cursor.execute(""" INSERT INTO companies (id, name, created_at) VALUES (?, ?, ?) """, (record_id, name, created_at)) conn.commit() conn.close() return record_id def get_all_companies(self) -> List[Dict]: """ 获取所有公司名称 :return: 公司名称列表 """ # 先尝试从MySQL查询（强制重新连接以确保获取最新数据） if self.mysql_enabled: try: # 强制重新连接以确保读取最新数据 mysql_conn = self.get_mysql_connection(force_reconnect=True) if mysql_conn: try: # 确保事务已提交 mysql_conn.commit() # 设置隔离级别为READ COMMITTED with mysql_conn.cursor() as cursor: cursor.execute("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED") with mysql_conn.cursor() as mysql_cursor: mysql_cursor.execute("SELECT * FROM companies ORDER BY name") rows = mysql_cursor.fetchall() result = [dict(row) for row in rows] logger.info(f"✅ 从MySQL获取公司列表，共{len(result)}条记录") return result except Exception as query_error: logger.error(f"MySQL查询公司执行失败: {str(query_error)}", exc_info=True) raise else: logger.warning("⚠️ MySQL连接不可用，将从SQLite查询公司列表") except Exception as e: logger.warning(f"⚠️ MySQL查询公司失败，将从SQLite查询: {str(e)}", exc_info=True) # MySQL失败或未启用，从SQLite查询 logger.info("从SQLite获取公司列表") conn = self.get_connection() cursor = conn.cursor() cursor.execute("SELECT * FROM companies ORDER BY name") rows = cursor.fetchall() conn.close() result = [dict(row) for row in rows] logger.info(f"从SQLite获取公司列表，共{len(result)}条记录") return result def update_company(self, company_id: int, name: str) -> bool: """ 更新公司名称 :param company_id: 公司ID :param name: 新名称 :return: 是否成功 """ success = False # 先操作MySQL if self.mysql_enabled: try: mysql_conn = self.get_mysql_connection() if mysql_conn: with mysql_conn.cursor() as mysql_cursor: mysql_cursor.execute(""" UPDATE companies SET name = %s WHERE id = %s """, (name, company_id)) affected_rows = mysql_cursor.rowcount mysql_conn.commit() if affected_rows > 0: success = True logger.info(f"MySQL更新公司成功: ID {company_id}, 新名称: {name}") else: logger.warning(f"MySQL更新公司: ID {company_id} 未找到记录") else: logger.warning(f"MySQL连接失败，无法更新公司 ID {company_id}") except Exception as e: logger.error(f"MySQL更新公司失败: ID {company_id}, 错误: {str(e)}", exc_info=True) if not CONTINUE_ON_MYSQL_ERROR: raise # 同步到SQLite if success or not self.mysql_enabled: try: conn = self.get_connection() cursor = conn.cursor() cursor.execute(""" UPDATE companies SET name = ? WHERE id = ? """, (name, company_id)) sqlite_success = cursor.rowcount > 0 conn.commit() conn.close() if self.mysql_enabled and not sqlite_success: logger.warning(f"SQLite更新公司: ID {company_id} 未找到记录，可能ID不匹配") elif not self.mysql_enabled: success = sqlite_success except Exception as e: logger.error(f"SQLite更新公司失败: ID {company_id}, 错误: {str(e)}") return success def delete_company(self, company_id: int) -> bool: """ 删除公司名称 :param company_id: 公司ID :return: 是否成功 """ success = False # 先操作MySQL if self.mysql_enabled: try: mysql_conn = self.get_mysql_connection() if mysql_conn: with mysql_conn.cursor() as mysql_cursor: mysql_cursor.execute("DELETE FROM companies WHERE id = %s", (company_id,)) affected_rows = mysql_cursor.rowcount mysql_conn.commit() if affected_rows > 0: success = True logger.info(f"MySQL删除公司成功: ID {company_id}") else: logger.warning(f"MySQL删除公司: ID {company_id} 未找到记录") else: logger.warning(f"MySQL连接失败，无法删除公司 ID {company_id}") except Exception as e: logger.error(f"MySQL删除公司失败: ID {company_id}, 错误: {str(e)}", exc_info=True) if not CONTINUE_ON_MYSQL_ERROR: raise # 同步到SQLite if success or not self.mysql_enabled: try: conn = self.get_connection() cursor = conn.cursor() cursor.execute("DELETE FROM companies WHERE id = ?", (company_id,)) sqlite_success = cursor.rowcount > 0 conn.commit() conn.close() if self.mysql_enabled and not sqlite_success: logger.warning(f"SQLite删除公司: ID {company_id} 未找到记录，可能ID不匹配") elif not self.mysql_enabled: success = sqlite_success except Exception as e: logger.error(f"SQLite删除公司失败: ID {company_id}, 错误: {str(e)}") return success def is_company_used(self, company_id: int) -> bool: """ 检查公司是否被地磅单引用 :param company_id: 公司ID :return: 是否被引用 """ # 先尝试从MySQL查询 if self.mysql_enabled: try: mysql_conn = self.get_mysql_connection() if mysql_conn: with mysql_conn.cursor() as mysql_cursor: # 获取公司名称 mysql_cursor.execute("SELECT name FROM companies WHERE id = %s", (company_id,)) row = mysql_cursor.fetchone() if not row: return False company_name = row['name'] # 检查是否在供货单位或收货单位中被使用 mysql_cursor.execute(""" SELECT COUNT(*) as count FROM weighing_slips WHERE supplier = %s OR consignee = %s """, (company_name, company_name)) result = mysql_cursor.fetchone() return result['count'] > 0 if result else False except Exception as e: logger.warning(f"MySQL检查公司使用情况失败，将从SQLite查询: {str(e)}") # MySQL失败或未启用，从SQLite查询 conn = self.get_connection() cursor = conn.cursor() # 获取公司名称 cursor.execute("SELECT name FROM companies WHERE id = ?", (company_id,)) row = cursor.fetchone() if not row: conn.close() return False company_name = row['name'] # 检查是否在供货单位或收货单位中被使用 cursor.execute(""" SELECT COUNT(*) as count FROM weighing_slips WHERE supplier = ? OR consignee = ? """, (company_name, company_name)) result = cursor.fetchone() conn.close() return result['count'] > 0 def company_exists(self, name: str, exclude_id: Optional[int] = None) -> bool: """ 检查公司名称是否存在 :param name: 公司名称 :param exclude_id: 排除的ID（用于修改时检查） :return: 是否存在 """ # 先尝试从MySQL查询 if self.mysql_enabled: try: mysql_conn = self.get_mysql_connection() if mysql_conn: with mysql_conn.cursor() as mysql_cursor: if exclude_id: mysql_cursor.execute(""" SELECT COUNT(*) as count FROM companies WHERE name = %s AND id != %s """, (name, exclude_id)) else: mysql_cursor.execute(""" SELECT COUNT(*) as count FROM companies WHERE name = %s """, (name,)) result = mysql_cursor.fetchone() return result['count'] > 0 if result else False except Exception as e: logger.warning(f"MySQL检查公司是否存在失败，将从SQLite查询: {str(e)}") # MySQL失败或未启用，从SQLite查询 conn = self.get_connection() cursor = conn.cursor() if exclude_id: cursor.execute(""" SELECT COUNT(*) as count FROM companies WHERE name = ? AND id != ? """, (name, exclude_id)) else: cursor.execute(""" SELECT COUNT(*) as count FROM companies WHERE name = ? """, (name,)) result = cursor.fetchone() conn.close() return result['count'] > 0 # ========== 货物名称相关操作 ========== def insert_cargo(self, name: str) -> int: """ 插入货物名称 :param name: 货物名称 :return: 插入记录的ID """ created_at = datetime.now().strftime('%Y-%m-%d %H:%M:%S') record_id = None # 先操作MySQL，获取MySQL生成的ID if self.mysql_enabled: try: mysql_conn = self.get_mysql_connection() if mysql_conn: with mysql_conn.cursor() as mysql_cursor: mysql_cursor.execute(""" INSERT INTO cargos (name, created_at) VALUES (%s, %s) """, (name, created_at)) record_id = mysql_cursor.lastrowid mysql_conn.commit() except Exception as e: logger.warning(f"MySQL插入货物失败: {str(e)}") if not CONTINUE_ON_MYSQL_ERROR: raise # 如果MySQL失败或未启用，使用SQLite生成ID if record_id is None: conn = self.get_connection() cursor = conn.cursor() cursor.execute(""" INSERT INTO cargos (name, created_at) VALUES (?, ?) """, (name, created_at)) record_id = cursor.lastrowid conn.commit() conn.close() else: # 使用MySQL的ID同步到SQLite conn = self.get_connection() cursor = conn.cursor() cursor.execute(""" INSERT INTO cargos (id, name, created_at) VALUES (?, ?, ?) """, (record_id, name, created_at)) conn.commit() conn.close() return record_id def get_all_cargos(self) -> List[Dict]: """ 获取所有货物名称 :return: 货物名称列表 """ # 先尝试从MySQL查询（强制重新连接以确保获取最新数据） if self.mysql_enabled: try: # 强制重新连接以确保读取最新数据 mysql_conn = self.get_mysql_connection(force_reconnect=True) if mysql_conn: try: # 确保事务已提交，并设置隔离级别为READ COMMITTED以读取最新数据 mysql_conn.commit() # 提交任何未提交的事务 # 设置隔离级别为READ COMMITTED，确保读取已提交的最新数据 with mysql_conn.cursor() as cursor: cursor.execute("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED") with mysql_conn.cursor() as mysql_cursor: mysql_cursor.execute("SELECT * FROM cargos ORDER BY name") rows = mysql_cursor.fetchall() result = [dict(row) for row in rows] logger.info(f"✅ 从MySQL获取货物列表，共{len(result)}条记录") # 打印前几条记录用于调试 if result: logger.info(f"示例：id={result[0].get('id')}, name={result[0].get('name')}") return result except Exception as query_error: logger.error(f"MySQL查询货物执行失败: {str(query_error)}", exc_info=True) raise else: logger.warning("⚠️ MySQL连接不可用（返回None），将从SQLite查询货物列表") except Exception as e: logger.warning(f"⚠️ MySQL查询货物失败，将从SQLite查询: {str(e)}", exc_info=True) # MySQL失败或未启用，从SQLite查询 logger.info("ℹ️ 从SQLite获取货物列表") conn = self.get_connection() cursor = conn.cursor() cursor.execute("SELECT * FROM cargos ORDER BY name") rows = cursor.fetchall() conn.close() result = [dict(row) for row in rows] logger.info(f"从SQLite获取货物列表，共{len(result)}条记录") if result: logger.info(f"示例：id={result[0].get('id')}, name={result[0].get('name')}") return result def update_cargo(self, cargo_id: int, name: str) -> bool: """ 更新货物名称 :param cargo_id: 货物ID :param name: 新名称 :return: 是否成功 """ success = False # 先操作MySQL if self.mysql_enabled: try: mysql_conn = self.get_mysql_connection() if mysql_conn: with mysql_conn.cursor() as mysql_cursor: mysql_cursor.execute(""" UPDATE cargos SET name = %s WHERE id = %s """, (name, cargo_id)) affected_rows = mysql_cursor.rowcount mysql_conn.commit() if affected_rows > 0: success = True logger.info(f"MySQL更新货物成功: ID {cargo_id}, 新名称: {name}") else: logger.warning(f"MySQL更新货物: ID {cargo_id} 未找到记录") else: logger.warning(f"MySQL连接失败，无法更新货物 ID {cargo_id}") except Exception as e: logger.error(f"MySQL更新货物失败: ID {cargo_id}, 错误: {str(e)}", exc_info=True) if not CONTINUE_ON_MYSQL_ERROR: raise # 同步到SQLite if success or not self.mysql_enabled: try: conn = self.get_connection() cursor = conn.cursor() cursor.execute(""" UPDATE cargos SET name = ? WHERE id = ? """, (name, cargo_id)) sqlite_success = cursor.rowcount > 0 conn.commit() conn.close() if self.mysql_enabled and not sqlite_success: logger.warning(f"SQLite更新货物: ID {cargo_id} 未找到记录，可能ID不匹配") elif not self.mysql_enabled: success = sqlite_success except Exception as e: logger.error(f"SQLite更新货物失败: ID {cargo_id}, 错误: {str(e)}") return success def delete_cargo(self, cargo_id: int) -> bool: """ 删除货物名称 :param cargo_id: 货物ID :return: 是否成功 """ success = False # 先操作MySQL if self.mysql_enabled: try: mysql_conn = self.get_mysql_connection() if mysql_conn: with mysql_conn.cursor() as mysql_cursor: mysql_cursor.execute("DELETE FROM cargos WHERE id = %s", (cargo_id,)) affected_rows = mysql_cursor.rowcount mysql_conn.commit() if affected_rows > 0: success = True logger.info(f"MySQL删除货物成功: ID {cargo_id}") else: logger.warning(f"MySQL删除货物: ID {cargo_id} 未找到记录") else: logger.warning(f"MySQL连接失败，无法删除货物 ID {cargo_id}") except Exception as e: logger.error(f"MySQL删除货物失败: ID {cargo_id}, 错误: {str(e)}", exc_info=True) if not CONTINUE_ON_MYSQL_ERROR: raise # 同步到SQLite if success or not self.mysql_enabled: try: conn = self.get_connection() cursor = conn.cursor() cursor.execute("DELETE FROM cargos WHERE id = ?", (cargo_id,)) sqlite_success = cursor.rowcount > 0 conn.commit() conn.close() if self.mysql_enabled and not sqlite_success: logger.warning(f"SQLite删除货物: ID {cargo_id} 未找到记录，可能ID不匹配") elif not self.mysql_enabled: success = sqlite_success except Exception as e: logger.error(f"SQLite删除货物失败: ID {cargo_id}, 错误: {str(e)}") return success def is_cargo_used(self, cargo_id: int) -> bool: """ 检查货物是否被地磅单引用 :param cargo_id: 货物ID :return: 是否被引用 """ # 先尝试从MySQL查询 if self.mysql_enabled: try: mysql_conn = self.get_mysql_connection() if mysql_conn: with mysql_conn.cursor() as mysql_cursor: # 获取货物名称 mysql_cursor.execute("SELECT name FROM cargos WHERE id = %s", (cargo_id,)) row = mysql_cursor.fetchone() if not row: return False cargo_name = row['name'] # 检查是否在地磅单中被使用 mysql_cursor.execute(""" SELECT COUNT(*) as count FROM weighing_slips WHERE cargo_name = %s """, (cargo_name,)) result = mysql_cursor.fetchone() return result['count'] > 0 if result else False except Exception as e: logger.warning(f"MySQL检查货物使用情况失败，将从SQLite查询: {str(e)}") # MySQL失败或未启用，从SQLite查询 conn = self.get_connection() cursor = conn.cursor() # 获取货物名称 cursor.execute("SELECT name FROM cargos WHERE id = ?", (cargo_id,)) row = cursor.fetchone() if not row: conn.close() return False cargo_name = row['name'] # 检查是否在地磅单中被使用 cursor.execute(""" SELECT COUNT(*) as count FROM weighing_slips WHERE cargo_name = ? """, (cargo_name,)) result = cursor.fetchone() conn.close() return result['count'] > 0 def cargo_exists(self, name: str, exclude_id: Optional[int] = None) -> bool: """ 检查货物名称是否存在 :param name: 货物名称 :param exclude_id: 排除的ID（用于修改时检查） :return: 是否存在 """ # 先尝试从MySQL查询 if self.mysql_enabled: try: mysql_conn = self.get_mysql_connection() if mysql_conn: with mysql_conn.cursor() as mysql_cursor: if exclude_id: mysql_cursor.execute(""" SELECT COUNT(*) as count FROM cargos WHERE name = %s AND id != %s """, (name, exclude_id)) else: mysql_cursor.execute(""" SELECT COUNT(*) as count FROM cargos WHERE name = %s """, (name,)) result = mysql_cursor.fetchone() return result['count'] > 0 if result else False except Exception as e: logger.warning(f"MySQL检查货物是否存在失败，将从SQLite查询: {str(e)}") # MySQL失败或未启用，从SQLite查询 conn = self.get_connection() cursor = conn.cursor() if exclude_id: cursor.execute(""" SELECT COUNT(*) as count FROM cargos WHERE name = ? AND id != ? """, (name, exclude_id)) else: cursor.execute(""" SELECT COUNT(*) as count FROM cargos WHERE name = ? """, (name,)) result = cursor.fetchone() conn.close() return result['count'] > 0 # ========== 采购计划相关操作 ========== def insert_purchase_plan(self, data: Dict) -> int: """ 插入采购计划记录 :param data: 采购计划数据字典 :return: 插入记录的ID """ created_at = datetime.now().strftime('%Y-%m-%d %H:%M:%S') record_id = None # 先操作MySQL，获取MySQL生成的ID if self.mysql_enabled: try: mysql_conn = self.get_mysql_connection() if mysql_conn: with mysql_conn.cursor() as mysql_cursor: mysql_cursor.execute(""" INSERT INTO purchase_plans ( plan_date, business_type, product, specification, planned_quantity, vehicle_no, customer_name, remarks, created_at ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( data['plan_date'], data['business_type'], data['product'], data.get('specification', ''), data['planned_quantity'], data['vehicle_no'], data.get('customer_name', ''), data.get('remarks', ''), created_at )) record_id = mysql_cursor.lastrowid mysql_conn.commit() except Exception as e: logger.warning(f"MySQL插入采购计划失败: {str(e)}") if not CONTINUE_ON_MYSQL_ERROR: raise # 如果MySQL失败或未启用，使用SQLite生成ID if record_id is None: conn = self.get_connection() cursor = conn.cursor() cursor.execute(""" INSERT INTO purchase_plans ( plan_date, business_type, product, specification, planned_quantity, vehicle_no, customer_name, remarks, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( data['plan_date'], data['business_type'], data['product'], data.get('specification', ''), data['planned_quantity'], data['vehicle_no'], data.get('customer_name', ''), data.get('remarks', ''), created_at )) record_id = cursor.lastrowid conn.commit() conn.close() else: # 使用MySQL的ID同步到SQLite conn = self.get_connection() cursor = conn.cursor() cursor.execute(""" INSERT INTO purchase_plans ( id, plan_date, business_type, product, specification, planned_quantity, vehicle_no, customer_name, remarks, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( record_id, # 使用MySQL返回的ID data['plan_date'], data['business_type'], data['product'], data.get('specification', ''), data['planned_quantity'], data['vehicle_no'], data.get('customer_name', ''), data.get('remarks', ''), created_at )) conn.commit() conn.close() return record_id def get_all_purchase_plans(self) -> List[Dict]: """ 获取所有采购计划记录 :return: 采购计划记录列表 """ # 先尝试从MySQL查询（强制重新连接以确保获取最新数据） if self.mysql_enabled: try: # 强制重新连接以确保读取最新数据 mysql_conn = self.get_mysql_connection(force_reconnect=True) if mysql_conn: try: # 确保事务已提交 mysql_conn.commit() # 设置隔离级别为READ COMMITTED with mysql_conn.cursor() as cursor: cursor.execute("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED") with mysql_conn.cursor() as mysql_cursor: mysql_cursor.execute(""" SELECT * FROM purchase_plans ORDER BY plan_date DESC, created_at DESC """) rows = mysql_cursor.fetchall() result = [dict(row) for row in rows] logger.info(f"✅ 从MySQL获取采购计划列表，共{len(result)}条记录") return result except Exception as query_error: logger.error(f"MySQL查询采购计划执行失败: {str(query_error)}", exc_info=True) raise else: logger.warning("⚠️ MySQL连接不可用，将从SQLite查询采购计划列表") except Exception as e: logger.warning(f"⚠️ MySQL查询采购计划失败，将从SQLite查询: {str(e)}", exc_info=True) # MySQL失败或未启用，从SQLite查询 logger.info("从SQLite获取采购计划列表") conn = self.get_connection() cursor = conn.cursor() cursor.execute(""" SELECT * FROM purchase_plans ORDER BY plan_date DESC, created_at DESC """) rows = cursor.fetchall() conn.close() result = [dict(row) for row in rows] logger.info(f"从SQLite获取采购计划列表，共{len(result)}条记录") return result def get_purchase_plan_by_id(self, plan_id: int) -> Optional[Dict]: """ 根据ID获取采购计划记录 :param plan_id: 采购计划ID :return: 采购计划记录字典，如果不存在返回None """ # 先尝试从MySQL查询 if self.mysql_enabled: try: mysql_conn = self.get_mysql_connection() if mysql_conn: with mysql_conn.cursor() as mysql_cursor: mysql_cursor.execute("SELECT * FROM purchase_plans WHERE id = %s", (plan_id,)) row = mysql_cursor.fetchone() if row: return dict(row) except Exception as e: logger.warning(f"MySQL查询采购计划失败，将从SQLite查询: {str(e)}") # MySQL失败或未启用，从SQLite查询 conn = self.get_connection() cursor = conn.cursor() cursor.execute("SELECT * FROM purchase_plans WHERE id = ?", (plan_id,)) row = cursor.fetchone() conn.close() return dict(row) if row else None def update_purchase_plan(self, plan_id: int, data: Dict) -> bool: """ 更新采购计划记录 :param plan_id: 采购计划ID :param data: 采购计划数据字典 :return: 是否成功 """ success = False # 先操作MySQL if self.mysql_enabled: try: mysql_conn = self.get_mysql_connection() if mysql_conn: with mysql_conn.cursor() as mysql_cursor: mysql_cursor.execute(""" UPDATE purchase_plans SET plan_date = %s, business_type = %s, product = %s, specification = %s, planned_quantity = %s, vehicle_no = %s, customer_name = %s, remarks = %s WHERE id = %s """, ( data['plan_date'], data['business_type'], data['product'], data.get('specification', ''), data['planned_quantity'], data['vehicle_no'], data.get('customer_name', ''), data.get('remarks', ''), plan_id )) affected_rows = mysql_cursor.rowcount mysql_conn.commit() if affected_rows > 0: success = True logger.info(f"MySQL更新采购计划成功: ID {plan_id}") else: logger.warning(f"MySQL更新采购计划: ID {plan_id} 未找到记录") else: logger.warning(f"MySQL连接失败，无法更新采购计划 ID {plan_id}") except Exception as e: logger.error(f"MySQL更新采购计划失败: ID {plan_id}, 错误: {str(e)}", exc_info=True) if not CONTINUE_ON_MYSQL_ERROR: raise # 同步到SQLite if success or not self.mysql_enabled: try: conn = self.get_connection() cursor = conn.cursor() cursor.execute(""" UPDATE purchase_plans SET plan_date = ?, business_type = ?, product = ?, specification = ?, planned_quantity = ?, vehicle_no = ?, customer_name = ?, remarks = ? WHERE id = ? """, ( data['plan_date'], data['business_type'], data['product'], data.get('specification', ''), data['planned_quantity'], data['vehicle_no'], data.get('customer_name', ''), data.get('remarks', ''), plan_id )) sqlite_success = cursor.rowcount > 0 conn.commit() conn.close() if self.mysql_enabled and not sqlite_success: logger.warning(f"SQLite更新采购计划: ID {plan_id} 未找到记录，可能ID不匹配") elif not self.mysql_enabled: success = sqlite_success except Exception as e: logger.error(f"SQLite更新采购计划失败: ID {plan_id}, 错误: {str(e)}") return success def delete_purchase_plan(self, plan_id: int) -> bool: """ 删除采购计划记录 :param plan_id: 采购计划ID :return: 是否成功 """ success = False # 先操作MySQL if self.mysql_enabled: try: mysql_conn = self.get_mysql_connection() if mysql_conn: with mysql_conn.cursor() as mysql_cursor: mysql_cursor.execute("DELETE FROM purchase_plans WHERE id = %s", (plan_id,)) affected_rows = mysql_cursor.rowcount mysql_conn.commit() if affected_rows > 0: success = True logger.info(f"MySQL删除采购计划成功: ID {plan_id}") else: logger.warning(f"MySQL删除采购计划: ID {plan_id} 未找到记录") else: logger.warning(f"MySQL连接失败，无法删除采购计划 ID {plan_id}") except Exception as e: logger.error(f"MySQL删除采购计划失败: ID {plan_id}, 错误: {str(e)}", exc_info=True) if not CONTINUE_ON_MYSQL_ERROR: raise # 同步到SQLite if success or not self.mysql_enabled: try: conn = self.get_connection() cursor = conn.cursor() cursor.execute("DELETE FROM purchase_plans WHERE id = ?", (plan_id,)) sqlite_success = cursor.rowcount > 0 conn.commit() conn.close() if self.mysql_enabled and not sqlite_success: logger.warning(f"SQLite删除采购计划: ID {plan_id} 未找到记录，可能ID不匹配") elif not self.mysql_enabled: success = sqlite_success except Exception as e: logger.error(f"SQLite删除采购计划失败: ID {plan_id}, 错误: {str(e)}") return success