""" SQLite下载历史管理器 用于记录图片下载历史，不修改生产MySQL数据库 """ import os import sqlite3 from datetime import datetime from typing import List, Dict, Optional import sys # 添加父目录到路径 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from config import SQLITE_DB_PATH class SQLiteManager: """SQLite下载历史管理器""" def __init__(self, db_path: str = SQLITE_DB_PATH): """初始化SQLite连接""" self.db_path = db_path self._init_database() def _init_database(self): """初始化数据库表结构""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # 创建下载历史表 cursor.execute(''' CREATE TABLE IF NOT EXISTS download_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, datasource TEXT NOT NULL, goods_id INTEGER, goods_name TEXT, image_url TEXT NOT NULL, local_path TEXT NOT NULL, download_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, file_size INTEGER, UNIQUE(datasource, image_url) ) ''') # 创建索引 cursor.execute(''' CREATE INDEX IF NOT EXISTS idx_datasource_goods_id ON download_history(datasource, goods_id) ''') cursor.execute(''' CREATE INDEX IF NOT EXISTS idx_download_time ON download_history(download_time) ''') conn.commit() conn.close() print(f"SQLite数据库初始化成功: {self.db_path}") except Exception as e: print(f"SQLite数据库初始化失败: {e}") raise def add_download_record( self, datasource: str, goods_id: int, goods_name: str, image_url: str, local_path: str, file_size: int ) -> bool: """ 添加下载记录 Returns: 是否成功 """ try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' INSERT OR REPLACE INTO download_history (datasource, goods_id, goods_name, image_url, local_path, file_size, download_time) VALUES (?, ?, ?, ?, ?, ?, ?) ''', ( datasource, goods_id, goods_name, image_url, local_path, file_size, datetime.now() )) conn.commit() conn.close() return True except Exception as e: print(f"添加下载记录失败: {e}") return False def is_downloaded(self, datasource: str, image_url: str) -> bool: """ 检查图片是否已下载（检查数据库记录 + 文件是否存在） Args: datasource: 数据源 image_url: 图片URL Returns: 是否已下载 """ try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' SELECT local_path FROM download_history WHERE datasource = ? AND image_url = ? ''', (datasource, image_url)) result = cursor.fetchone() conn.close() if result: local_path = result[0] # 检查文件是否真实存在 return os.path.exists(local_path) return False except Exception as e: print(f"检查下载状态失败: {e}") return False def get_download_history( self, datasource: Optional[str] = None, start_date: Optional[datetime] = None, end_date: Optional[datetime] = None, limit: int = 100 ) -> List[Dict]: """ 获取下载历史 Returns: 下载记录列表 """ try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() query = 'SELECT * FROM download_history WHERE 1=1' params = [] if datasource: query += ' AND datasource = ?' params.append(datasource) if start_date: query += ' AND download_time >= ?' params.append(start_date.strftime('%Y-%m-%d %H:%M:%S')) if end_date: query += ' AND download_time <= ?' params.append(end_date.strftime('%Y-%m-%d %H:%M:%S')) query += ' ORDER BY download_time DESC LIMIT ?' params.append(limit) cursor.execute(query, params) columns = [desc[0] for desc in cursor.description] results = [] for row in cursor.fetchall(): results.append(dict(zip(columns, row))) conn.close() return results except Exception as e: print(f"获取下载历史失败: {e}") return [] def get_download_statistics(self) -> Dict: """获取下载统计信息""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # 总下载数 cursor.execute('SELECT COUNT(*) FROM download_history') total_count = cursor.fetchone()[0] # 各数据源下载数 cursor.execute(''' SELECT datasource, COUNT(*) as count FROM download_history GROUP BY datasource ''') datasource_stats = {row[0]: row[1] for row in cursor.fetchall()} # 今日下载数 today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) cursor.execute(''' SELECT COUNT(*) FROM download_history WHERE download_time >= ? ''', (today.strftime('%Y-%m-%d %H:%M:%S'),)) today_count = cursor.fetchone()[0] conn.close() return { 'total_count': total_count, 'datasource_stats': datasource_stats, 'today_count': today_count } except Exception as e: print(f"获取下载统计失败: {e}") return {} def get_local_path(self, datasource: str, image_url: str) -> Optional[str]: """ 获取图片的本地文件路径 Args: datasource: 数据源 image_url: 图片URL Returns: 本地文件路径，如果未下载则返回None """ try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' SELECT local_path FROM download_history WHERE datasource = ? AND image_url = ? ''', (datasource, image_url)) result = cursor.fetchone() conn.close() return result[0] if result else None except Exception as e: print(f"获取本地路径失败: {e}") return None def delete_download_record(self, datasource: str, image_url: str, delete_file: bool = True) -> bool: """ 删除下载记录（可选删除物理文件） Args: datasource: 数据源 image_url: 图片URL delete_file: 是否删除物理文件 Returns: 是否成功 """ try: # 如果需要删除文件，先获取文件路径 local_path = None if delete_file: local_path = self.get_local_path(datasource, image_url) # 删除数据库记录 conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' DELETE FROM download_history WHERE datasource = ? AND image_url = ? ''', (datasource, image_url)) conn.commit() conn.close() # 删除物理文件 if delete_file and local_path and os.path.exists(local_path): try: os.remove(local_path) print(f"已删除本地文件: {local_path}") except Exception as e: print(f"删除本地文件失败: {e}") return True except Exception as e: print(f"删除下载记录失败: {e}") return False def clear_invalid_records(self) -> int: """清理文件不存在的下载记录""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute('SELECT id, local_path FROM download_history') records = cursor.fetchall() deleted_count = 0 for record_id, local_path in records: if not os.path.exists(local_path): cursor.execute('DELETE FROM download_history WHERE id = ?', (record_id,)) deleted_count += 1 conn.commit() conn.close() print(f"已清理 {deleted_count} 条无效记录") return deleted_count except Exception as e: print(f"清理无效记录失败: {e}") return 0