import sqlite3 import datetime import json import csv from typing import List, Dict, Optional, Tuple, Any from contextlib import contextmanager import logging # 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class BatteryDatabase: def __init__(self, db_path="battery_data.db"): self.db_path = db_path # self.init_database() @contextmanager def get_connection(self): """数据库连接上下文管理器""" conn = sqlite3.connect(self.db_path) conn.row_factory = sqlite3.Row try: yield conn conn.commit() except Exception as e: conn.rollback() raise e finally: conn.close() def init_database(self): """初始化数据库和表结构""" with self.get_connection() as conn: cursor = conn.cursor() # 设置数据库使用本地时间 cursor.execute("PRAGMA foreign_keys = ON") # 创建批次记录表 cursor.execute(''' CREATE TABLE IF NOT EXISTS batch_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, batch_name TEXT NOT NULL, created_time TIMESTAMP DEFAULT (datetime('now', 'localtime')), description TEXT, status TEXT DEFAULT 'active' ) ''') # 创建弹夹表 cursor.execute(''' CREATE TABLE IF NOT EXISTS magazines ( id INTEGER PRIMARY KEY AUTOINCREMENT, magazine_code TEXT NOT NULL, created_time TIMESTAMP DEFAULT (datetime('now', 'localtime')), description TEXT ) ''') # 创建箱表 cursor.execute(''' CREATE TABLE IF NOT EXISTS boxes ( id INTEGER PRIMARY KEY AUTOINCREMENT, magazine_id INTEGER NOT NULL, box_code TEXT NOT NULL, created_time TIMESTAMP DEFAULT (datetime('now', 'localtime')), description TEXT, FOREIGN KEY (magazine_id) REFERENCES magazines(id) ) ''') # 创建电池数据表 cursor.execute(''' CREATE TABLE IF NOT EXISTS battery_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, box_id INTEGER NOT NULL, time TIMESTAMP NOT NULL, code TEXT NOT NULL, top_qr_code TEXT, bottom_qr_code TEXT, h2_value REAL CHECK (h2_value >= 0), o2_value REAL CHECK (o2_value >= 0), symbol TEXT, thickness REAL CHECK (thickness >= 0), status TEXT DEFAULT 'normal', notes TEXT, box_index INTEGER, magazine_index INTEGER, FOREIGN KEY (box_id) REFERENCES boxes(id) ) ''') # 创建保存电池数据表 cursor.execute(''' CREATE TABLE IF NOT EXISTS save_battery_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, batch_id INTEGER NOT NULL, time TIMESTAMP NOT NULL, code TEXT NOT NULL, top_qr_code TEXT, bottom_qr_code TEXT, h2_value REAL CHECK (h2_value >= 0), o2_value REAL CHECK (o2_value >= 0), symbol TEXT, thickness REAL CHECK (thickness >= 0), status TEXT DEFAULT 'normal', notes TEXT, group_id INTEGER, data_index INTEGER, box_code INTEGER, box_index INTEGER, magazine_code INTEGER, magazine_index INTEGER ) ''') # 创建索引以提高查询性能 cursor.execute('CREATE INDEX IF NOT EXISTS idx_box_id ON battery_data(box_id)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_time ON battery_data(time)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_code ON battery_data(code)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_box_magazine ON boxes(magazine_id)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_batch_status ON batch_records(status)') # 为save_battery_data表也创建索引 cursor.execute('CREATE INDEX IF NOT EXISTS idx_save_time ON save_battery_data(time)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_save_code ON save_battery_data(code)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_save_group ON save_battery_data(group_id)') logger.info("数据库初始化完成") # ========== 增 ========== def create_batch(self, batch_name: str, description: str = "") -> int: """创建新的批次记录""" try: with self.get_connection() as conn: cursor = conn.cursor() cursor.execute(''' INSERT INTO batch_records (batch_name, description) VALUES (?, ?) ''', (batch_name, description)) batch_id = cursor.lastrowid logger.info(f"创建批次成功: {batch_name} (ID: {batch_id})") return batch_id except sqlite3.IntegrityError: logger.error(f"批次名称已存在: {batch_name}") raise ValueError(f"批次名称 '{batch_name}' 已存在") except Exception as e: logger.error(f"创建批次失败: {e}") raise def create_magazine(self, magazine_code: str, description: str = "") -> int: """创建弹夹""" try: with self.get_connection() as conn: cursor = conn.cursor() cursor.execute(''' INSERT INTO magazines (magazine_code, description) VALUES (?, ?) ''', (magazine_code, description)) magazine_id = cursor.lastrowid logger.info(f"创建弹夹成功: {magazine_code} (ID: {magazine_id})") return magazine_id except sqlite3.IntegrityError: logger.error(f"弹夹代码已存在: {magazine_code}") raise ValueError(f"弹夹代码 '{magazine_code}' 已存在") except Exception as e: logger.error(f"创建弹夹失败: {e}") raise def create_box(self, magazine_id: int, box_code: str, description: str = "") -> int: """创建箱""" try: with self.get_connection() as conn: cursor = conn.cursor() cursor.execute(''' INSERT INTO boxes (magazine_id, box_code, description) VALUES (?, ?, ?) ''', (magazine_id, box_code, description)) box_id = cursor.lastrowid logger.info(f"创建箱成功: {box_code} (ID: {box_id})") return box_id except sqlite3.IntegrityError: logger.error(f"箱代码已存在: {box_code}") raise ValueError(f"箱代码 '{box_code}' 已存在") except Exception as e: logger.error(f"创建箱失败: {e}") raise def _validate_battery_data(self, battery_data: Dict): """验证电池数据""" required_fields = ['code'] for field in required_fields: if field not in battery_data or not battery_data[field]: raise ValueError(f"缺少必要字段: {field}") # 验证数值范围 if battery_data.get('h2_value') is not None and battery_data['h2_value'] < 0: raise ValueError("H2值不能为负数") if battery_data.get('o2_value') is not None and battery_data['o2_value'] < 0: raise ValueError("O2值不能为负数") if battery_data.get('thickness') is not None and battery_data['thickness'] < 0: raise ValueError("厚度不能为负数") def add_batch_batteries(self, box_id: int, batteries: List[Dict]) -> bool: """批量添加电池数据""" try: with self.get_connection() as conn: cursor = conn.cursor() for battery in batteries: cursor.execute(''' INSERT INTO battery_data (box_id, time, code, top_qr_code, bottom_qr_code, h2_value, o2_value, symbol, thickness, status, notes, box_index, magazine_index) VALUES (?, ?, ?, ?, ?, ?, ?, ?,?, ?, ?, ?, ?) ''', ( box_id, battery.get('time', datetime.datetime.now()), battery['serial_number'], battery.get('Below_QR'), battery.get('Up_QR'), battery.get('H2'), battery.get('02'), battery.get('Cell_ran'), battery.get('thick'), battery.get('status', 'normal'), battery.get('source_code'), battery.get('box_index'), battery.get('magazine_index') )) logger.info(f"批量添加电池数据成功，共 {len(batteries)} 条记录") return True except Exception as e: logger.error(f"批量添加电池数据失败: {e}") return False def add_save_battery_data_batch_with_groups(self, batch_id, battery_data_with_groups: List[Tuple[Dict, int]]) -> \ List[int]: """批量添加保存电池数据（带分组信息）""" try: record_ids = [] with self.get_connection() as conn: cursor = conn.cursor() for battery_data, group_id in battery_data_with_groups: self._validate_battery_data(battery_data) cursor.execute(''' INSERT INTO save_battery_data (batch_id, time, code, top_qr_code, bottom_qr_code, h2_value, o2_value, symbol, thickness, status, notes, group_id, data_index, box_code, box_index, magazine_code, magazine_index) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''', ( batch_id, battery_data.get('time', datetime.datetime.now()), battery_data['code'], battery_data.get('top_qr_code'), battery_data.get('bottom_qr_code'), battery_data.get('h2_value'), battery_data.get('o2_value'), battery_data.get('symbol'), battery_data.get('thickness'), battery_data.get('status', 'normal'), battery_data.get('notes'), group_id, battery_data.get('data_index'), battery_data.get('box_code'), battery_data.get('box_index'), battery_data.get('magazine_code'), battery_data.get('magazine_index'), )) record_ids.append(cursor.lastrowid) conn.commit() logger.info(f"批量添加保存电池数据成功: 共 {len(record_ids)} 条记录") return record_ids except sqlite3.IntegrityError as e: logger.error(f"电池代码已存在: {e}") raise ValueError(f"电池代码已存在") except Exception as e: logger.error(f"批量添加保存电池数据失败: {e}") raise # ========== 查 ========== def get_latest_battery_data(self, limit: int = 2000, status: Optional[str] = None) -> List[Dict]: """ 获取电池数据表的最新数据 Args: limit: 返回的记录数量，默认为2000 status: 筛选状态（如'normal'），默认为None表示不过滤状态 Returns: 包含电池数据的字典列表 """ try: with self.get_connection() as conn: cursor = conn.cursor() # 构建SQL查询 query = ''' SELECT bd.id, bd.time, bd.code, bd.top_qr_code, bd.bottom_qr_code, bd.h2_value, bd.o2_value, bd.symbol, bd.thickness, bd.status, bd.notes, bd.box_id, bx.box_code, m.magazine_code, bx.box_code, bd.box_index, m.magazine_code, bd.magazine_index FROM battery_data bd LEFT JOIN boxes bx ON bd.box_id = bx.id LEFT JOIN magazines m ON bx.magazine_id = m.id ''' params = [] # 添加状态筛选条件 if status: query += ' WHERE bd.status = ?' params.append(status) # 添加排序和限制 query += ' ORDER BY bd.time DESC, bd.id DESC LIMIT ?' params.append(limit) # 执行查询 cursor.execute(query, params) # 转换为字典列表 results = [dict(row) for row in cursor.fetchall()] logger.info(f"获取最新电池数据成功，共 {len(results)} 条记录") return results except Exception as e: logger.error(f"获取最新电池数据失败: {e}") return [] def get_batches_by_time_range_with_daily_sort(self, start_time: str, end_time: str) -> List[Dict]: """根据时间范围查询批次记录表所对应的保存电池数据表的批次，并按天排序 Args: start_time: 开始时间 (格式: 'YYYY-MM-DD HH:MM:SS') end_time: 结束时间 (格式: 'YYYY-MM-DD HH:MM:SS') Returns: 包含批次信息的字典列表，按天分组排序 """ try: with self.get_connection() as conn: cursor = conn.cursor() cursor.execute(''' SELECT DISTINCT br.id, br.batch_name, br.created_time, br.description, br.status, DATE(br.created_time) as batch_date, COUNT(sbd.id) as total_data_count, COUNT(DISTINCT sbd.group_id) as group_count, MIN(sbd.time) as earliest_data_time, MAX(sbd.time) as latest_data_time FROM batch_records br INNER JOIN save_battery_data sbd ON br.id = sbd.batch_id WHERE sbd.time BETWEEN ? AND ? GROUP BY br.id, br.batch_name, br.created_time, br.description, br.status, batch_date ORDER BY batch_date DESC, br.created_time DESC ''', (start_time, end_time)) batches = [dict(row) for row in cursor.fetchall()] logger.info(f"根据时间范围 {start_time} 到 {end_time} 查询到 {len(batches)} 个批次") print(batches) return batches except Exception as e: logger.error(f"根据时间范围查询批次失败: {e}") return [] def get_batch_data(self, batch_id: int, group_id: int = None) -> List[Dict]: """根据批次ID和组ID查询数据 Args: batch_id: 批次ID group_id: 组ID（可选），如果提供则返回该组的详细数据，否则返回批次下所有组的信息 Returns: 包含组信息或详细数据的字典列表 """ try: with self.get_connection() as conn: cursor = conn.cursor() if group_id is not None: # 查询指定组的详细数据 cursor.execute(''' SELECT sbd.id, sbd.time, sbd.code, sbd.top_qr_code, sbd.bottom_qr_code, sbd.h2_value, sbd.o2_value, sbd.symbol, sbd.thickness, sbd.status, sbd.notes, sbd.batch_id, sbd.group_id, sbd.data_index, sbd.box_code, sbd.box_index, sbd.magazine_code, sbd.magazine_index, br.batch_name FROM save_battery_data sbd LEFT JOIN batch_records br ON sbd.batch_id = br.id WHERE sbd.batch_id = ? AND sbd.group_id = ? ORDER BY sbd.data_index ASC ''', (batch_id, group_id)) data = [dict(row) for row in cursor.fetchall()] logger.info(f"查询批次 {batch_id} 组 {group_id} 的数据成功，共 {len(data)} 条记录") return data else: # 查询批次下所有组的信息 cursor.execute(''' SELECT group_id, COUNT(*) as data_count, MIN(data_index) as min_index, MAX(data_index) as max_index, MIN(time) as start_time, MAX(time) as end_time, AVG(h2_value) as avg_h2, AVG(o2_value) as avg_o2, AVG(thickness) as avg_thickness, COUNT(CASE WHEN status = 'normal' THEN 1 END) as normal_count, COUNT(CASE WHEN status != 'normal' THEN 1 END) as abnormal_count FROM save_battery_data WHERE batch_id = ? GROUP BY group_id ORDER BY group_id ASC ''', (batch_id,)) groups = [] rows = cursor.fetchall() print(rows) # 调试信息 for row in rows: group_data = dict(row) # 处理平均值，保留4位小数 for key in ['avg_h2', 'avg_o2', 'avg_thickness']: if group_data[key] is not None: group_data[key] = round(group_data[key], 4) groups.append(group_data) logger.info(f"查询批次 {batch_id} 的组成功，共 {len(groups)} 个组") return groups except Exception as e: if group_id is not None: logger.error(f"查询批次 {batch_id} 组 {group_id} 数据失败: {e}") else: logger.error(f"查询批次 {batch_id} 组信息失败: {e}") return [] def get_group_by_qr_codes_exact(self, batch_id: int, top_qr_code: str = None, bottom_qr_code: str = None) -> List[Dict]: """通过上面二维码和/或下面二维码精确查找对应的组信息 Args: batch_id: 批次ID top_qr_code: 上面二维码（可选） bottom_qr_code: 下面二维码（可选） Returns: 包含组信息的字典列表 """ try: with self.get_connection() as conn: cursor = conn.cursor() # 构建查询条件 conditions = ["batch_id = ?"] params = [batch_id] if top_qr_code and top_qr_code.strip(): conditions.append("top_qr_code = ?") params.append(top_qr_code.strip()) if bottom_qr_code and bottom_qr_code.strip(): conditions.append("bottom_qr_code = ?") params.append(bottom_qr_code.strip()) # 如果两个二维码都没有提供，返回空列表 if not top_qr_code and not bottom_qr_code: return [] query = f''' SELECT DISTINCT group_id FROM save_battery_data WHERE {' AND '.join(conditions)} ORDER BY group_id ASC ''' cursor.execute(query, params) results = [dict(row) for row in cursor.fetchall()] logger.info(f"通过二维码精确查询批次 {batch_id} 成功，找到 {len(results)} 个匹配的组") return results except Exception as e: logger.error(f"通过二维码精确查询组信息失败: {e}") return [] # # 使用示例和测试 # def main(): # # 创建数据库实例 # db = BatteryDatabase("../battery_data.db") # full_start_time = "2025-11-06 11:38:51" # full_end_time = "2025-11-06 11:38:51" # batches = db.get_batches_by_time_range_with_daily_sort(full_start_time,full_end_time) # print(batches) # if __name__ == "__main__": # main()