268 lines
11 KiB
Python
268 lines
11 KiB
Python
"""
|
||
解析结果缓存管理器
|
||
"""
|
||
import os
|
||
import json
|
||
import hashlib
|
||
from datetime import datetime
|
||
from typing import Dict, List, Optional, Tuple, Any
|
||
from pathlib import Path
|
||
from loguru import logger
|
||
import sqlite3
|
||
from db_utils import get_db_connection
|
||
|
||
|
||
class ParseCacheManager:
|
||
"""解析结果缓存管理器"""
|
||
|
||
def __init__(self):
|
||
"""初始化缓存管理器"""
|
||
self._init_cache_tables()
|
||
|
||
def _init_cache_tables(self):
|
||
"""
|
||
初始化解析缓存相关的数据库表结构
|
||
|
||
该方法创建两个核心表：
|
||
1. git_parse_cache - 存储文件解析结果的缓存表
|
||
2. git_repo_parse_status - 记录仓库整体解析状态的状态表
|
||
"""
|
||
# 获取数据库连接
|
||
conn, cursor = get_db_connection()
|
||
try:
|
||
# 创建解析缓存表 - 存储每个文件的解析结果
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS git_parse_cache (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT, -- 自增主键
|
||
repo_id TEXT NOT NULL, -- 仓库ID
|
||
branch TEXT NOT NULL, -- 分支名
|
||
file_path TEXT NOT NULL, -- 文件相对路径
|
||
file_hash TEXT NOT NULL, -- 文件内容SHA-256哈希值
|
||
commit_id TEXT NOT NULL, -- 解析时的commit ID
|
||
parse_result TEXT NOT NULL, -- 解析结果(JSON格式)
|
||
created_at TEXT NOT NULL, -- 创建时间
|
||
updated_at TEXT NOT NULL, -- 更新时间
|
||
UNIQUE(repo_id, branch, file_path) -- 复合唯一约束
|
||
)
|
||
''')
|
||
|
||
# 创建仓库解析状态表 - 记录仓库级别的解析状态
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS git_repo_parse_status (
|
||
repo_id TEXT PRIMARY KEY, -- 仓库ID(主键)
|
||
branch TEXT NOT NULL, -- 分支名
|
||
commit_id TEXT NOT NULL, -- 当前解析的commit ID
|
||
last_parse_time TEXT NOT NULL, -- 最后解析时间
|
||
file_count INTEGER DEFAULT 0, -- 已解析文件数量
|
||
parse_status TEXT DEFAULT 'pending', -- 解析状态(pending/parsing/completed/failed)
|
||
error_message TEXT, -- 错误信息(如果有)
|
||
UNIQUE(repo_id, branch) -- 复合唯一约束
|
||
)
|
||
''')
|
||
|
||
# 创建索引 - 优化按仓库和分支查询的性能
|
||
cursor.execute('''
|
||
CREATE INDEX IF NOT EXISTS idx_git_parse_cache_repo_branch
|
||
ON git_parse_cache(repo_id, branch)
|
||
''')
|
||
|
||
# 提交事务
|
||
conn.commit()
|
||
logger.info("解析缓存表初始化成功")
|
||
except Exception as e:
|
||
# 记录错误日志
|
||
logger.error(f"初始化解析缓存表失败: {e}")
|
||
# 确保在异常情况下关闭连接
|
||
conn.close()
|
||
# 重新抛出异常
|
||
raise
|
||
finally:
|
||
# 确保数据库连接被关闭
|
||
conn.close()
|
||
|
||
def calculate_file_hash(self, file_path: str) -> str:
|
||
"""计算文件内容的哈希值"""
|
||
try:
|
||
with open(file_path, 'rb') as f:
|
||
return hashlib.md5(f.read()).hexdigest()
|
||
except Exception as e:
|
||
logger.error(f"计算文件哈希失败 {file_path}: {e}")
|
||
return ""
|
||
|
||
def get_cached_parse_result(self, repo_id: str, branch: str, file_path: str,
|
||
file_hash: str) -> Optional[Dict[str, Any]]:
|
||
"""获取缓存的解析结果"""
|
||
conn, cursor = get_db_connection()
|
||
try:
|
||
cursor.execute('''
|
||
SELECT parse_result FROM git_parse_cache
|
||
WHERE repo_id = ? AND branch = ? AND file_path = ? AND file_hash = ?
|
||
''', (repo_id, branch, file_path, file_hash))
|
||
|
||
row = cursor.fetchone()
|
||
if row:
|
||
return json.loads(row[0])
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"获取缓存解析结果失败: {e}")
|
||
return None
|
||
finally:
|
||
conn.close()
|
||
|
||
def save_parse_result(self, repo_id: str, branch: str, file_path: str,
|
||
file_hash: str, commit_id: str, parse_result: Dict[str, Any]):
|
||
"""保存解析结果到缓存"""
|
||
conn, cursor = get_db_connection()
|
||
try:
|
||
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
result_json = json.dumps(parse_result, ensure_ascii=False)
|
||
|
||
cursor.execute('''
|
||
INSERT OR REPLACE INTO git_parse_cache
|
||
(repo_id, branch, file_path, file_hash, commit_id, parse_result, created_at, updated_at)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||
''', (repo_id, branch, file_path, file_hash, commit_id, result_json, now, now))
|
||
|
||
conn.commit()
|
||
logger.debug(f"保存解析结果缓存: {repo_id}/{branch}:{file_path}")
|
||
except Exception as e:
|
||
logger.error(f"保存解析结果缓存失败: {e}")
|
||
finally:
|
||
conn.close()
|
||
|
||
def get_repo_parse_status(self, repo_id: str, branch: str) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
获取指定仓库和分支的解析状态信息
|
||
|
||
Args:
|
||
repo_id (str): 仓库唯一标识符
|
||
branch (str): 分支名称
|
||
|
||
Returns:
|
||
Optional[Dict[str, Any]]: 仓库解析状态字典，包含以下字段：
|
||
- repo_id (str): 仓库ID
|
||
- branch (str): 分支名
|
||
- commit_id (str): 当前解析的commit ID
|
||
- last_parse_time (str): 最后解析时间
|
||
- file_count (int): 已解析文件数量
|
||
- parse_status (str): 解析状态(pending/parsing/completed/failed)
|
||
- error_message (str): 错误信息(如果有)
|
||
如果仓库不存在解析记录，则返回None
|
||
"""
|
||
# 获取数据库连接
|
||
conn, cursor = get_db_connection()
|
||
try:
|
||
# 查询指定仓库和分支的解析状态
|
||
cursor.execute('''
|
||
SELECT repo_id, branch, commit_id, last_parse_time, file_count, parse_status, error_message
|
||
FROM git_repo_parse_status WHERE repo_id = ? AND branch = ?
|
||
''', (repo_id, branch))
|
||
|
||
# 获取查询结果
|
||
row = cursor.fetchone()
|
||
if row:
|
||
# 将查询结果转换为字典格式返回
|
||
return {
|
||
"repo_id": row[0],
|
||
"branch": row[1],
|
||
"commit_id": row[2],
|
||
"last_parse_time": row[3],
|
||
"file_count": row[4],
|
||
"parse_status": row[5],
|
||
"error_message": row[6]
|
||
}
|
||
# 如果没有找到记录，返回None
|
||
return None
|
||
except Exception as e:
|
||
# 记录错误日志
|
||
logger.error(f"获取仓库解析状态失败: {e}")
|
||
return None
|
||
finally:
|
||
# 确保数据库连接被关闭
|
||
conn.close()
|
||
|
||
def update_repo_parse_status(self, repo_id: str, branch: str, commit_id: str,
|
||
file_count: int, status: str, error_message: str = None):
|
||
"""更新仓库解析状态"""
|
||
conn, cursor = get_db_connection()
|
||
try:
|
||
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
cursor.execute('''
|
||
INSERT OR REPLACE INTO git_repo_parse_status
|
||
(repo_id, branch, commit_id, last_parse_time, file_count, parse_status, error_message)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||
''', (repo_id, branch, commit_id, now, file_count, status, error_message))
|
||
|
||
conn.commit()
|
||
logger.debug(f"更新仓库解析状态: {repo_id}/{branch} -> {status}")
|
||
except Exception as e:
|
||
logger.error(f"更新仓库解析状态失败: {e}")
|
||
finally:
|
||
conn.close()
|
||
|
||
def invalidate_cache_for_file(self, repo_id: str, branch: str, file_path: str):
|
||
"""使特定文件的缓存失效"""
|
||
conn, cursor = get_db_connection()
|
||
try:
|
||
cursor.execute('''
|
||
DELETE FROM git_parse_cache
|
||
WHERE repo_id = ? AND branch = ? AND file_path = ?
|
||
''', (repo_id, branch, file_path))
|
||
|
||
conn.commit()
|
||
logger.debug(f"使文件缓存失效: {repo_id}/{branch}:{file_path}")
|
||
except Exception as e:
|
||
logger.error(f"使文件缓存失效失败: {e}")
|
||
finally:
|
||
conn.close()
|
||
|
||
def invalidate_cache_for_repo(self, repo_id: str, branch: str):
|
||
"""使整个仓库的缓存失效"""
|
||
conn, cursor = get_db_connection()
|
||
try:
|
||
cursor.execute('''
|
||
DELETE FROM git_parse_cache
|
||
WHERE repo_id = ? AND branch = ?
|
||
''', (repo_id, branch))
|
||
|
||
cursor.execute('''
|
||
DELETE FROM git_repo_parse_status
|
||
WHERE repo_id = ? AND branch = ?
|
||
''', (repo_id, branch))
|
||
|
||
conn.commit()
|
||
logger.info(f"使仓库缓存失效: {repo_id}/{branch}")
|
||
except Exception as e:
|
||
logger.error(f"使仓库缓存失效失败: {e}")
|
||
finally:
|
||
conn.close()
|
||
|
||
def get_cache_statistics(self, repo_id: str = None, branch: str = None) -> Dict[str, Any]:
|
||
"""获取缓存统计信息"""
|
||
conn, cursor = get_db_connection()
|
||
try:
|
||
params = []
|
||
query = "SELECT COUNT(*) as total_files FROM git_parse_cache WHERE 1=1"
|
||
|
||
if repo_id:
|
||
query += " AND repo_id = ?"
|
||
params.append(repo_id)
|
||
|
||
if branch:
|
||
query += " AND branch = ?"
|
||
params.append(branch)
|
||
|
||
cursor.execute(query, params)
|
||
total_files = cursor.fetchone()[0]
|
||
|
||
return {
|
||
"total_cached_files": total_files,
|
||
"repo_id": repo_id,
|
||
"branch": branch
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"获取缓存统计信息失败: {e}")
|
||
return {"total_cached_files": 0}
|
||
finally:
|
||
conn.close() |