445 lines
14 KiB
Python
445 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
LoreBible目录管理模块
|
||
负责管理LoreBible目录结构、文件操作和现有角色扫描
|
||
"""
|
||
|
||
import os
|
||
import json
|
||
import re
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional, Set, Tuple
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class LoreBibleManager:
|
||
"""LoreBible目录管理器"""
|
||
|
||
# 标准目录结构
|
||
DEFAULT_DIRECTORY_STRUCTURE = {
|
||
"00_Prepare": "临时档案目录",
|
||
"01_Research": "研究资料目录",
|
||
"02_LoreBible": {
|
||
"Characters": "角色档案目录",
|
||
"Locations": "地点设定目录",
|
||
"Organizations": "组织设定目录",
|
||
"Timeline": "时间线目录"
|
||
}
|
||
}
|
||
|
||
def __init__(self, workspace: str):
|
||
"""初始化LoreBible管理器
|
||
|
||
Args:
|
||
workspace: 工作目录路径
|
||
"""
|
||
self.workspace = Path(workspace).resolve()
|
||
self.characters_dir = self.workspace / "02_LoreBible" / "Characters"
|
||
self.prepare_dir = self.workspace / "00_Prepare"
|
||
self._character_index = None # 角色信息索引缓存
|
||
|
||
def validate_directory_structure(self) -> Tuple[bool, List[str]]:
|
||
"""验证目录结构是否符合标准
|
||
|
||
Returns:
|
||
(是否有效, 缺失目录列表)
|
||
"""
|
||
missing_dirs = []
|
||
|
||
# 检查必需目录
|
||
required_dirs = [
|
||
self.workspace,
|
||
self.characters_dir,
|
||
self.prepare_dir
|
||
]
|
||
|
||
for dir_path in required_dirs:
|
||
if not dir_path.exists():
|
||
missing_dirs.append(str(dir_path.relative_to(self.workspace)))
|
||
|
||
return len(missing_dirs) == 0, missing_dirs
|
||
|
||
def create_directory_structure(self) -> bool:
|
||
"""创建缺失的目录结构
|
||
|
||
Returns:
|
||
是否成功创建
|
||
"""
|
||
try:
|
||
# 创建必需目录
|
||
self.characters_dir.mkdir(parents=True, exist_ok=True)
|
||
self.prepare_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 创建可选目录
|
||
optional_dirs = [
|
||
self.workspace / "01_Research",
|
||
self.workspace / "02_LoreBible" / "Locations",
|
||
self.workspace / "02_LoreBible" / "Organizations",
|
||
self.workspace / "02_LoreBible" / "Timeline"
|
||
]
|
||
|
||
for dir_path in optional_dirs:
|
||
dir_path.mkdir(parents=True, exist_ok=True)
|
||
|
||
logger.info(f"目录结构已创建在: {self.workspace}")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"创建目录结构失败: {e}")
|
||
return False
|
||
|
||
def scan_existing_characters(self) -> List[Dict]:
|
||
"""扫描现有角色档案
|
||
|
||
Returns:
|
||
角色信息列表
|
||
"""
|
||
if not self.characters_dir.exists():
|
||
return []
|
||
|
||
characters = []
|
||
pattern = re.compile(r'\.md$', re.IGNORECASE)
|
||
|
||
for file_path in self.characters_dir.glob("*.md"):
|
||
try:
|
||
character_info = self._parse_character_file(file_path)
|
||
if character_info:
|
||
characters.append(character_info)
|
||
except Exception as e:
|
||
logger.warning(f"解析角色文件失败 {file_path}: {e}")
|
||
|
||
return characters
|
||
|
||
def _parse_character_file(self, file_path: Path) -> Optional[Dict]:
|
||
"""解析单个角色档案文件
|
||
|
||
Args:
|
||
file_path: 角色档案文件路径
|
||
|
||
Returns:
|
||
角色信息字典，解析失败返回None
|
||
"""
|
||
try:
|
||
content = file_path.read_text(encoding='utf-8', errors='ignore')
|
||
|
||
# 提取基本信息
|
||
info = {
|
||
"file_path": str(file_path),
|
||
"file_name": file_path.name,
|
||
"file_size": file_path.stat().st_size,
|
||
"last_modified": file_path.stat().st_mtime
|
||
}
|
||
|
||
# 提取角色姓名（从标题行）
|
||
name_match = re.search(r'^#\s+(.+?)\s+-', content, re.MULTILINE)
|
||
if name_match:
|
||
info["name"] = name_match.group(1).strip()
|
||
else:
|
||
# 尝试从文件名提取
|
||
info["name"] = file_path.stem.replace('character_profile_', '').replace('_', ' ')
|
||
|
||
# 提取年龄
|
||
age_match = re.search(r'- \*\*年龄\*\*[：:]\s*(.+?)(?:\n|$)', content)
|
||
if age_match:
|
||
info["age"] = age_match.group(1).strip()
|
||
|
||
# 提取性别
|
||
gender_match = re.search(r'- \*\*性别\*\*[：:]\s*(.+?)(?:\n|$)', content)
|
||
if gender_match:
|
||
info["gender"] = gender_match.group(1).strip()
|
||
|
||
# 提取职业/身份
|
||
occupation_match = re.search(r'- \*\*职业/身份\*\*[：:]\s*(.+?)(?:\n|$)', content)
|
||
if occupation_match:
|
||
info["occupation"] = occupation_match.group(1).strip()
|
||
|
||
# 提取故事中的角色
|
||
role_match = re.search(r'- \*\*故事中的角色\*\*[：:]\s*(.+?)(?:\n|$)', content)
|
||
if role_match:
|
||
info["role"] = role_match.group(1).strip()
|
||
|
||
# 提取角色类型（从元数据）
|
||
type_match = re.search(r'- \*\*角色类型\*\*[：:]\s*(.+?)(?:\n|$)', content)
|
||
if type_match:
|
||
info["character_type"] = type_match.group(1).strip()
|
||
|
||
# 提取创建时间
|
||
created_match = re.search(r'- \*\*创建时间\*\*[：:]\s*(.+?)(?:\n|$)', content)
|
||
if created_match:
|
||
info["created"] = created_match.group(1).strip()
|
||
|
||
# 提取状态
|
||
status_match = re.search(r'- \*\*状态\*\*[：:]\s*(.+?)(?:\n|$)', content)
|
||
if status_match:
|
||
info["status"] = status_match.group(1).strip()
|
||
|
||
# 提取关系信息（简单提取）
|
||
relationships_section = self._extract_section(content, "人物关系")
|
||
if relationships_section:
|
||
info["has_relationships"] = True
|
||
# 可以进一步解析具体关系
|
||
|
||
# 提取背景故事
|
||
background_section = self._extract_section(content, "背景故事")
|
||
if background_section:
|
||
info["has_background"] = True
|
||
|
||
return info
|
||
|
||
except Exception as e:
|
||
logger.error(f"解析角色文件失败 {file_path}: {e}")
|
||
return None
|
||
|
||
def _extract_section(self, content: str, section_title: str) -> Optional[str]:
|
||
"""提取指定章节内容
|
||
|
||
Args:
|
||
content: 文档内容
|
||
section_title: 章节标题
|
||
|
||
Returns:
|
||
章节内容，未找到返回None
|
||
"""
|
||
# 寻找章节标题
|
||
pattern = rf'##\s+{re.escape(section_title)}(.*?)(?=##\s+|---|\Z)'
|
||
match = re.search(pattern, content, re.DOTALL | re.IGNORECASE)
|
||
|
||
if match:
|
||
return match.group(1).strip()
|
||
return None
|
||
|
||
def get_character_index(self, force_refresh: bool = False) -> Dict:
|
||
"""获取角色信息索引（带缓存）
|
||
|
||
Args:
|
||
force_refresh: 是否强制刷新缓存
|
||
|
||
Returns:
|
||
角色索引字典
|
||
"""
|
||
if self._character_index is None or force_refresh:
|
||
characters = self.scan_existing_characters()
|
||
self._character_index = {
|
||
"total_count": len(characters),
|
||
"characters": characters,
|
||
"name_map": {char.get("name", ""): char for char in characters if char.get("name")},
|
||
"by_type": self._group_by_type(characters),
|
||
"last_updated": os.path.getmtime(self.characters_dir) if self.characters_dir.exists() else 0
|
||
}
|
||
|
||
return self._character_index
|
||
|
||
def _group_by_type(self, characters: List[Dict]) -> Dict:
|
||
"""按角色类型分组
|
||
|
||
Args:
|
||
characters: 角色列表
|
||
|
||
Returns:
|
||
按类型分组的字典
|
||
"""
|
||
grouped = {}
|
||
for char in characters:
|
||
char_type = char.get("character_type", "未知")
|
||
if char_type not in grouped:
|
||
grouped[char_type] = []
|
||
grouped[char_type].append(char)
|
||
return grouped
|
||
|
||
def save_temp_profile(self, content: str, character_name: str, session_id: str = None) -> Optional[Path]:
|
||
"""保存临时档案到00_Prepare目录
|
||
|
||
Args:
|
||
content: 档案内容
|
||
character_name: 角色名称
|
||
session_id: 会话ID，为None时自动生成
|
||
|
||
Returns:
|
||
临时文件路径，失败返回None
|
||
"""
|
||
try:
|
||
if not self.prepare_dir.exists():
|
||
self.prepare_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 生成会话ID
|
||
import uuid
|
||
if session_id is None:
|
||
session_id = str(uuid.uuid4())[:8]
|
||
|
||
# 安全文件名
|
||
safe_name = re.sub(r'[^\w\s-]', '', character_name).strip()
|
||
safe_name = re.sub(r'[-\s]+', '_', safe_name)
|
||
|
||
# 临时文件名
|
||
filename = f"temp_{safe_name}_{session_id}.md"
|
||
temp_path = self.prepare_dir / filename
|
||
|
||
# 写入文件
|
||
temp_path.write_text(content, encoding='utf-8')
|
||
logger.info(f"临时档案已保存: {temp_path}")
|
||
|
||
return temp_path
|
||
|
||
except Exception as e:
|
||
logger.error(f"保存临时档案失败: {e}")
|
||
return None
|
||
|
||
def move_to_characters(self, temp_file_path: Path, character_name: str = None) -> Optional[Path]:
|
||
"""将临时档案移动到Characters目录
|
||
|
||
Args:
|
||
temp_file_path: 临时文件路径
|
||
character_name: 角色名称，为None时从文件内容提取
|
||
|
||
Returns:
|
||
最终文件路径，失败返回None
|
||
"""
|
||
try:
|
||
if not temp_file_path.exists():
|
||
logger.error(f"临时文件不存在: {temp_file_path}")
|
||
return None
|
||
|
||
# 确保目标目录存在
|
||
if not self.characters_dir.exists():
|
||
self.characters_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 读取内容提取角色名
|
||
if character_name is None:
|
||
content = temp_file_path.read_text(encoding='utf-8', errors='ignore')
|
||
name_match = re.search(r'^#\s+(.+?)\s+-', content, re.MULTILINE)
|
||
if name_match:
|
||
character_name = name_match.group(1).strip()
|
||
else:
|
||
character_name = temp_file_path.stem.replace('temp_', '').split('_')[0]
|
||
|
||
# 安全文件名
|
||
safe_name = re.sub(r'[^\w\s-]', '', character_name).strip()
|
||
safe_name = re.sub(r'[-\s]+', '_', safe_name)
|
||
|
||
# 最终文件名
|
||
import datetime
|
||
timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
|
||
filename = f"character_profile_{safe_name}_{timestamp}.md"
|
||
final_path = self.characters_dir / filename
|
||
|
||
# 移动文件
|
||
temp_file_path.rename(final_path)
|
||
logger.info(f"档案已移动到: {final_path}")
|
||
|
||
# 刷新缓存
|
||
self._character_index = None
|
||
|
||
return final_path
|
||
|
||
except Exception as e:
|
||
logger.error(f"移动档案失败: {e}")
|
||
return None
|
||
|
||
def cleanup_old_temp_files(self, max_age_hours: int = 24) -> int:
|
||
"""清理旧的临时文件
|
||
|
||
Args:
|
||
max_age_hours: 最大保留时间（小时）
|
||
|
||
Returns:
|
||
清理的文件数量
|
||
"""
|
||
if not self.prepare_dir.exists():
|
||
return 0
|
||
|
||
import time
|
||
current_time = time.time()
|
||
max_age_seconds = max_age_hours * 3600
|
||
|
||
cleaned_count = 0
|
||
|
||
for file_path in self.prepare_dir.glob("temp_*.md"):
|
||
try:
|
||
file_age = current_time - file_path.stat().st_mtime
|
||
if file_age > max_age_seconds:
|
||
file_path.unlink()
|
||
cleaned_count += 1
|
||
logger.debug(f"清理临时文件: {file_path}")
|
||
except Exception as e:
|
||
logger.warning(f"清理文件失败 {file_path}: {e}")
|
||
|
||
if cleaned_count > 0:
|
||
logger.info(f"清理了 {cleaned_count} 个临时文件")
|
||
|
||
return cleaned_count
|
||
|
||
def get_workspace_info(self) -> Dict:
|
||
"""获取工作空间信息
|
||
|
||
Returns:
|
||
工作空间信息字典
|
||
"""
|
||
info = {
|
||
"workspace": str(self.workspace),
|
||
"characters_dir": str(self.characters_dir),
|
||
"prepare_dir": str(self.prepare_dir),
|
||
"characters_count": 0,
|
||
"temp_files_count": 0,
|
||
"directory_exists": self.workspace.exists()
|
||
}
|
||
|
||
if self.characters_dir.exists():
|
||
info["characters_count"] = len(list(self.characters_dir.glob("*.md")))
|
||
|
||
if self.prepare_dir.exists():
|
||
info["temp_files_count"] = len(list(self.prepare_dir.glob("temp_*.md")))
|
||
|
||
return info
|
||
|
||
|
||
def main():
|
||
"""命令行测试"""
|
||
import sys
|
||
|
||
if len(sys.argv) < 2:
|
||
print("用法: python lore_bible_manager.py <工作目录>")
|
||
sys.exit(1)
|
||
|
||
workspace = sys.argv[1]
|
||
manager = LoreBibleManager(workspace)
|
||
|
||
print(f"工作目录: {workspace}")
|
||
|
||
# 验证目录结构
|
||
is_valid, missing_dirs = manager.validate_directory_structure()
|
||
if is_valid:
|
||
print("✓ 目录结构完整")
|
||
else:
|
||
print("✗ 缺失目录:")
|
||
for dir_name in missing_dirs:
|
||
print(f" - {dir_name}")
|
||
|
||
# 询问是否创建
|
||
response = input("是否创建缺失目录? (y/n): ").strip().lower()
|
||
if response == 'y':
|
||
if manager.create_directory_structure():
|
||
print("✓ 目录结构已创建")
|
||
else:
|
||
print("✗ 创建目录失败")
|
||
|
||
# 扫描现有角色
|
||
characters = manager.scan_existing_characters()
|
||
print(f"现有角色数量: {len(characters)}")
|
||
|
||
for char in characters[:5]: # 显示前5个
|
||
print(f" - {char.get('name', '未知')} ({char.get('character_type', '未知类型')})")
|
||
|
||
if len(characters) > 5:
|
||
print(f" ... 还有 {len(characters) - 5} 个角色")
|
||
|
||
# 显示工作空间信息
|
||
info = manager.get_workspace_info()
|
||
print(f"\n工作空间信息:")
|
||
print(f" 角色档案数: {info['characters_count']}")
|
||
print(f" 临时文件数: {info['temp_files_count']}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |