324 lines
10 KiB
Python
324 lines
10 KiB
Python
"""
|
|
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
|