252 lines
8.3 KiB
Python
252 lines
8.3 KiB
Python
"""
|
||
处理历史记录管理模块
|
||
"""
|
||
import json
|
||
import logging
|
||
import shutil
|
||
from pathlib import Path
|
||
from datetime import datetime
|
||
from typing import Any, Dict, List, Optional
|
||
from dataclasses import dataclass, field, asdict
|
||
|
||
from app.config import CACHE_DIR
|
||
|
||
|
||
HISTORY_FILE = CACHE_DIR / "history.json"
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@dataclass
|
||
class HistoryRecord:
|
||
"""历史记录"""
|
||
id: str
|
||
timestamp: str
|
||
status: str # pending, processing, completed, failed
|
||
work_dir: str
|
||
file_count: int = 0
|
||
elapsed_time: float = 0.0
|
||
error: Optional[str] = None
|
||
result_tables: List[str] = field(default_factory=list)
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
return asdict(self)
|
||
|
||
def get_log_file(self) -> Path:
|
||
"""获取日志文件路径"""
|
||
return Path(self.work_dir) / "log.txt"
|
||
|
||
|
||
class HistoryManager:
|
||
"""历史记录管理器"""
|
||
|
||
def __init__(self):
|
||
self._ensure_file()
|
||
|
||
def _ensure_file(self):
|
||
"""确保历史文件存在"""
|
||
if not HISTORY_FILE.exists():
|
||
HISTORY_FILE.write_text("[]", encoding='utf-8')
|
||
|
||
def _load(self) -> List[Dict[str, Any]]:
|
||
"""加载历史记录"""
|
||
try:
|
||
return json.loads(HISTORY_FILE.read_text(encoding='utf-8'))
|
||
except (OSError, json.JSONDecodeError):
|
||
return []
|
||
|
||
def _save(self, records: List[Dict[str, Any]]):
|
||
"""保存历史记录"""
|
||
HISTORY_FILE.write_text(json.dumps(records, ensure_ascii=False, indent=2), encoding='utf-8')
|
||
|
||
def _delete_work_dir(self, work_dir: str | None) -> None:
|
||
if not work_dir:
|
||
return
|
||
|
||
try:
|
||
work_path = Path(work_dir).resolve()
|
||
cache_path = CACHE_DIR.resolve()
|
||
|
||
if work_path.exists() and work_path.is_dir():
|
||
try:
|
||
work_path.relative_to(cache_path)
|
||
shutil.rmtree(work_path)
|
||
except ValueError:
|
||
logger.warning("Skip deleting path outside cache directory: %s", work_dir)
|
||
except Exception as e:
|
||
logger.warning("Failed to delete work directory %s: %s", work_dir, e)
|
||
|
||
def create(self, work_dir: Path, file_count: int, record_id: Optional[str] = None) -> HistoryRecord:
|
||
"""创建新的历史记录"""
|
||
# 如果提供了 record_id，使用它；否则生成新的
|
||
if record_id is None:
|
||
record_id = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
|
||
record = HistoryRecord(
|
||
id=record_id,
|
||
timestamp=datetime.now().isoformat(),
|
||
status="pending",
|
||
work_dir=str(work_dir),
|
||
file_count=file_count
|
||
)
|
||
|
||
records = self._load()
|
||
records.insert(0, record.to_dict())
|
||
|
||
# 只保留最近 100 条记录
|
||
records = records[:100]
|
||
self._save(records)
|
||
|
||
return record
|
||
|
||
def _clean_record(self, rec: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""清理记录，移除不存在的字段（如旧的 logs 字段）"""
|
||
cleaned = {
|
||
'id': rec.get('id'),
|
||
'timestamp': rec.get('timestamp'),
|
||
'status': rec.get('status'),
|
||
'work_dir': rec.get('work_dir'),
|
||
'file_count': rec.get('file_count', 0),
|
||
'elapsed_time': rec.get('elapsed_time', 0.0),
|
||
'error': rec.get('error'),
|
||
'result_tables': rec.get('result_tables', [])
|
||
}
|
||
return cleaned
|
||
|
||
def update(self, record_id: str, **kwargs) -> Optional[HistoryRecord]:
|
||
"""更新历史记录"""
|
||
records = self._load()
|
||
|
||
for i, rec in enumerate(records):
|
||
if rec['id'] == record_id:
|
||
rec.update(kwargs)
|
||
self._save(records)
|
||
cleaned = self._clean_record(rec)
|
||
return HistoryRecord(**cleaned)
|
||
|
||
return None
|
||
|
||
def get(self, record_id: str) -> Optional[HistoryRecord]:
|
||
"""获取单条记录"""
|
||
records = self._load()
|
||
for rec in records:
|
||
if rec['id'] == record_id:
|
||
cleaned = self._clean_record(rec)
|
||
return HistoryRecord(**cleaned)
|
||
return None
|
||
|
||
def list(self, limit: int = 50) -> List[Dict[str, Any]]:
|
||
"""获取历史记录列表"""
|
||
records = self._load()
|
||
# 返回简化的列表（不包含日志）
|
||
result = []
|
||
for rec in records[:limit]:
|
||
item = rec.copy()
|
||
item.pop('logs', None) # 列表不返回日志（兼容旧数据）
|
||
result.append(item)
|
||
return result
|
||
|
||
def get_logs(self, record_id: str) -> List[str]:
|
||
"""从 log.txt 文件读取日志"""
|
||
record = self.get(record_id)
|
||
if not record:
|
||
return []
|
||
|
||
log_file = record.get_log_file()
|
||
if not log_file.exists():
|
||
return []
|
||
|
||
try:
|
||
content = log_file.read_text(encoding='utf-8')
|
||
# 按行分割，过滤空行
|
||
logs = [line.strip() for line in content.split('\n') if line.strip()]
|
||
return logs
|
||
except Exception as e:
|
||
logger.warning("Failed to read history log file %s: %s", log_file, e)
|
||
return []
|
||
|
||
def delete(self, record_id: str) -> bool:
|
||
"""删除历史记录，同时删除对应的文件目录"""
|
||
records = self._load()
|
||
|
||
# 找到要删除的记录，获取其工作目录
|
||
work_dir = None
|
||
for rec in records:
|
||
if rec['id'] == record_id:
|
||
work_dir = rec.get('work_dir')
|
||
break
|
||
|
||
# 删除记录
|
||
new_records = [r for r in records if r['id'] != record_id]
|
||
|
||
if len(new_records) < len(records):
|
||
self._save(new_records)
|
||
|
||
self._delete_work_dir(work_dir)
|
||
|
||
return True
|
||
return False
|
||
|
||
def prune_finished(self, keep_count: int) -> int:
|
||
"""按保留数量清理已结束的处理历史，处理中和等待中的记录不会被删除"""
|
||
keep_count = max(int(keep_count), 0)
|
||
records = self._load()
|
||
kept_finished = 0
|
||
pruned: List[Dict[str, Any]] = []
|
||
remaining: List[Dict[str, Any]] = []
|
||
|
||
for rec in records:
|
||
status = rec.get("status")
|
||
if status not in {"completed", "failed"}:
|
||
remaining.append(rec)
|
||
continue
|
||
|
||
if kept_finished < keep_count:
|
||
remaining.append(rec)
|
||
kept_finished += 1
|
||
else:
|
||
pruned.append(rec)
|
||
|
||
if not pruned:
|
||
return 0
|
||
|
||
self._save(remaining)
|
||
for rec in pruned:
|
||
self._delete_work_dir(rec.get("work_dir"))
|
||
|
||
return len(pruned)
|
||
|
||
def clear(self) -> int:
|
||
"""清空所有历史记录，同时清空cache目录（保留history.json）"""
|
||
records = self._load()
|
||
count = len(records)
|
||
|
||
# 清空历史记录
|
||
self._save([])
|
||
|
||
# 清空cache目录（保留history.json文件）
|
||
try:
|
||
cache_path = CACHE_DIR.resolve()
|
||
for item in CACHE_DIR.iterdir():
|
||
try:
|
||
item_path = item.resolve()
|
||
# 安全检查：确保路径在cache目录内
|
||
item_path.relative_to(cache_path)
|
||
|
||
if item.is_dir():
|
||
# 删除所有子目录
|
||
shutil.rmtree(item_path)
|
||
elif item.is_file() and item.name != "history.json":
|
||
# 删除所有文件，但保留history.json
|
||
item_path.unlink()
|
||
except ValueError:
|
||
# 路径不在cache目录内，跳过（安全保护）
|
||
logger.warning("Skip path outside cache directory: %s", item)
|
||
except Exception as e:
|
||
# 单个文件/目录删除失败，继续处理其他文件
|
||
logger.warning("Failed to delete cache item %s: %s", item, e)
|
||
except Exception as e:
|
||
# 记录错误但不影响清空历史记录的操作
|
||
logger.warning("Failed to clear cache directory: %s", e)
|
||
|
||
return count
|