diff --git a/mongodb_to_sqlite.bat b/mongodb_to_sqlite.bat
deleted file mode 100644
index f960e508..00000000
--- a/mongodb_to_sqlite.bat
+++ /dev/null
@@ -1,72 +0,0 @@
-@echo off
-CHCP 65001 > nul
-setlocal enabledelayedexpansion
-
-echo 你需要选择启动方式，输入字母来选择:
-echo V = 不知道什么意思就输入 V
-echo C = 输入 C 使用 Conda 环境
-echo.
-choice /C CV /N /M "不知道什么意思就输入 V (C/V)?" /T 10 /D V
-
-set "ENV_TYPE="
-if %ERRORLEVEL% == 1 set "ENV_TYPE=CONDA"
-if %ERRORLEVEL% == 2 set "ENV_TYPE=VENV"
-
-if "%ENV_TYPE%" == "CONDA" goto activate_conda
-if "%ENV_TYPE%" == "VENV" goto activate_venv
-
-REM 如果 choice 超时或返回意外值，默认使用 venv
-echo WARN: Invalid selection or timeout from choice. Defaulting to VENV.
-set "ENV_TYPE=VENV"
-goto activate_venv
-
-:activate_conda
- set /p CONDA_ENV_NAME="请输入要使用的 Conda 环境名称: "
- if not defined CONDA_ENV_NAME (
- echo 错误: 未输入 Conda 环境名称.
- pause
- exit /b 1
- )
- echo 选择: Conda '!CONDA_ENV_NAME!'
- REM 激活Conda环境
- call conda activate !CONDA_ENV_NAME!
- if !ERRORLEVEL! neq 0 (
- echo 错误: Conda环境 '!CONDA_ENV_NAME!' 激活失败. 请确保Conda已安装并正确配置, 且 '!CONDA_ENV_NAME!' 环境存在.
- pause
- exit /b 1
- )
- goto env_activated
-
-:activate_venv
- echo Selected: venv (default or selected)
- REM 查找venv虚拟环境
- set "venv_path=%~dp0venv\Scripts\activate.bat"
- if not exist "%venv_path%" (
- echo Error: venv not found. Ensure the venv directory exists alongside the script.
- pause
- exit /b 1
- )
- REM 激活虚拟环境
- call "%venv_path%"
- if %ERRORLEVEL% neq 0 (
- echo Error: Failed to activate venv virtual environment.
- pause
- exit /b 1
- )
- goto env_activated
-
-:env_activated
-echo Environment activated successfully!
-
-REM --- 后续脚本执行 ---
-
-REM 运行预处理脚本
-python "%~dp0scripts\mongodb_to_sqlite.py"
-if %ERRORLEVEL% neq 0 (
- echo Error: mongodb_to_sqlite.py execution failed.
- pause
- exit /b 1
-)
-
-echo All processing steps completed!
-pause
\ No newline at end of file
diff --git a/requirements.txt b/requirements.txt
index 721cf95f..d4dd2339 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -15,7 +15,6 @@ matplotlib
networkx
numpy
openai
-google-genai
pandas
peewee
pyarrow
diff --git a/scripts/manifest_tool.py b/scripts/manifest_tool.py
deleted file mode 100644
index 8312dc3e..00000000
--- a/scripts/manifest_tool.py
+++ /dev/null
@@ -1,237 +0,0 @@
-"""
-插件Manifest管理命令行工具
-
-提供插件manifest文件的创建、验证和管理功能
-"""
-
-import os
-import sys
-import argparse
-import json
-from pathlib import Path
-from src.common.logger import get_logger
-from src.plugin_system.utils.manifest_utils import (
- ManifestValidator,
-)
-
-# 添加项目根目录到Python路径
-project_root = Path(__file__).parent.parent.parent.parent
-sys.path.insert(0, str(project_root))
-
-
-logger = get_logger("manifest_tool")
-
-
-def create_minimal_manifest(plugin_dir: str, plugin_name: str, description: str = "", author: str = "") -> bool:
- """创建最小化的manifest文件
-
- Args:
- plugin_dir: 插件目录
- plugin_name: 插件名称
- description: 插件描述
- author: 插件作者
-
- Returns:
- bool: 是否创建成功
- """
- manifest_path = os.path.join(plugin_dir, "_manifest.json")
-
- if os.path.exists(manifest_path):
- print(f"❌ Manifest文件已存在: {manifest_path}")
- return False
-
- # 创建最小化manifest
- minimal_manifest = {
- "manifest_version": 1,
- "name": plugin_name,
- "version": "1.0.0",
- "description": description or f"{plugin_name}插件",
- "author": {"name": author or "Unknown"},
- }
-
- try:
- with open(manifest_path, "w", encoding="utf-8") as f:
- json.dump(minimal_manifest, f, ensure_ascii=False, indent=2)
- print(f"✅ 已创建最小化manifest文件: {manifest_path}")
- return True
- except Exception as e:
- print(f"❌ 创建manifest文件失败: {e}")
- return False
-
-
-def create_complete_manifest(plugin_dir: str, plugin_name: str) -> bool:
- """创建完整的manifest模板文件
-
- Args:
- plugin_dir: 插件目录
- plugin_name: 插件名称
-
- Returns:
- bool: 是否创建成功
- """
- manifest_path = os.path.join(plugin_dir, "_manifest.json")
-
- if os.path.exists(manifest_path):
- print(f"❌ Manifest文件已存在: {manifest_path}")
- return False
-
- # 创建完整模板
- complete_manifest = {
- "manifest_version": 1,
- "name": plugin_name,
- "version": "1.0.0",
- "description": f"{plugin_name}插件描述",
- "author": {"name": "插件作者", "url": "https://github.com/your-username"},
- "license": "MIT",
- "host_application": {"min_version": "1.0.0", "max_version": "4.0.0"},
- "homepage_url": "https://github.com/your-repo",
- "repository_url": "https://github.com/your-repo",
- "keywords": ["keyword1", "keyword2"],
- "categories": ["Category1"],
- "default_locale": "zh-CN",
- "locales_path": "_locales",
- "plugin_info": {
- "is_built_in": False,
- "plugin_type": "general",
- "components": [{"type": "action", "name": "sample_action", "description": "示例动作组件"}],
- },
- }
-
- try:
- with open(manifest_path, "w", encoding="utf-8") as f:
- json.dump(complete_manifest, f, ensure_ascii=False, indent=2)
- print(f"✅ 已创建完整manifest模板: {manifest_path}")
- print("💡 请根据实际情况修改manifest文件中的内容")
- return True
- except Exception as e:
- print(f"❌ 创建manifest文件失败: {e}")
- return False
-
-
-def validate_manifest_file(plugin_dir: str) -> bool:
- """验证manifest文件
-
- Args:
- plugin_dir: 插件目录
-
- Returns:
- bool: 是否验证通过
- """
- manifest_path = os.path.join(plugin_dir, "_manifest.json")
-
- if not os.path.exists(manifest_path):
- print(f"❌ 未找到manifest文件: {manifest_path}")
- return False
-
- try:
- with open(manifest_path, "r", encoding="utf-8") as f:
- manifest_data = json.load(f)
-
- validator = ManifestValidator()
- is_valid = validator.validate_manifest(manifest_data)
-
- # 显示验证结果
- print("📋 Manifest验证结果:")
- print(validator.get_validation_report())
-
- if is_valid:
- print("✅ Manifest文件验证通过")
- else:
- print("❌ Manifest文件验证失败")
-
- return is_valid
-
- except json.JSONDecodeError as e:
- print(f"❌ Manifest文件格式错误: {e}")
- return False
- except Exception as e:
- print(f"❌ 验证过程中发生错误: {e}")
- return False
-
-
-def scan_plugins_without_manifest(root_dir: str) -> None:
- """扫描缺少manifest文件的插件
-
- Args:
- root_dir: 扫描的根目录
- """
- print(f"🔍 扫描目录: {root_dir}")
-
- plugins_without_manifest = []
-
- for root, dirs, files in os.walk(root_dir):
- # 跳过隐藏目录和__pycache__
- dirs[:] = [d for d in dirs if not d.startswith(".") and d != "__pycache__"]
-
- # 检查是否包含plugin.py文件（标识为插件目录）
- if "plugin.py" in files:
- manifest_path = os.path.join(root, "_manifest.json")
- if not os.path.exists(manifest_path):
- plugins_without_manifest.append(root)
-
- if plugins_without_manifest:
- print(f"❌ 发现 {len(plugins_without_manifest)} 个插件缺少manifest文件:")
- for plugin_dir in plugins_without_manifest:
- plugin_name = os.path.basename(plugin_dir)
- print(f" - {plugin_name}: {plugin_dir}")
- print("💡 使用 'python manifest_tool.py create-minimal <插件目录>' 创建manifest文件")
- else:
- print("✅ 所有插件都有manifest文件")
-
-
-def main():
- """主函数"""
- parser = argparse.ArgumentParser(description="插件Manifest管理工具")
- subparsers = parser.add_subparsers(dest="command", help="可用命令")
-
- # 创建最小化manifest命令
- create_minimal_parser = subparsers.add_parser("create-minimal", help="创建最小化manifest文件")
- create_minimal_parser.add_argument("plugin_dir", help="插件目录路径")
- create_minimal_parser.add_argument("--name", help="插件名称")
- create_minimal_parser.add_argument("--description", help="插件描述")
- create_minimal_parser.add_argument("--author", help="插件作者")
-
- # 创建完整manifest命令
- create_complete_parser = subparsers.add_parser("create-complete", help="创建完整manifest模板")
- create_complete_parser.add_argument("plugin_dir", help="插件目录路径")
- create_complete_parser.add_argument("--name", help="插件名称")
-
- # 验证manifest命令
- validate_parser = subparsers.add_parser("validate", help="验证manifest文件")
- validate_parser.add_argument("plugin_dir", help="插件目录路径")
-
- # 扫描插件命令
- scan_parser = subparsers.add_parser("scan", help="扫描缺少manifest的插件")
- scan_parser.add_argument("root_dir", help="扫描的根目录路径")
-
- args = parser.parse_args()
-
- if not args.command:
- parser.print_help()
- return
-
- try:
- if args.command == "create-minimal":
- plugin_name = args.name or os.path.basename(os.path.abspath(args.plugin_dir))
- success = create_minimal_manifest(args.plugin_dir, plugin_name, args.description or "", args.author or "")
- sys.exit(0 if success else 1)
-
- elif args.command == "create-complete":
- plugin_name = args.name or os.path.basename(os.path.abspath(args.plugin_dir))
- success = create_complete_manifest(args.plugin_dir, plugin_name)
- sys.exit(0 if success else 1)
-
- elif args.command == "validate":
- success = validate_manifest_file(args.plugin_dir)
- sys.exit(0 if success else 1)
-
- elif args.command == "scan":
- scan_plugins_without_manifest(args.root_dir)
-
- except Exception as e:
- print(f"❌ 执行命令时发生错误: {e}")
- sys.exit(1)
-
-
-if __name__ == "__main__":
- main()
diff --git a/scripts/mongodb_to_sqlite.py b/scripts/mongodb_to_sqlite.py
deleted file mode 100644
index 0c15ee83..00000000
--- a/scripts/mongodb_to_sqlite.py
+++ /dev/null
@@ -1,920 +0,0 @@
-import os
-import json
-import sys # 新增系统模块导入
-
-# import time
-import pickle
-from pathlib import Path
-
-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
-from typing import Dict, Any, List, Optional, Type
-from dataclasses import dataclass, field
-from datetime import datetime
-from pymongo import MongoClient
-from pymongo.errors import ConnectionFailure
-from peewee import Model, Field, IntegrityError
-
-# Rich 进度条和显示组件
-from rich.console import Console
-from rich.progress import (
- Progress,
- TextColumn,
- BarColumn,
- TaskProgressColumn,
- TimeRemainingColumn,
- TimeElapsedColumn,
- SpinnerColumn,
-)
-from rich.table import Table
-from rich.panel import Panel
-# from rich.text import Text
-
-from src.common.database.database import db
-from src.common.database.database_model import (
- ChatStreams,
- Emoji,
- Messages,
- Images,
- ImageDescriptions,
- PersonInfo,
- Knowledges,
- ThinkingLog,
- GraphNodes,
- GraphEdges,
-)
-from src.common.logger import get_logger
-
-logger = get_logger("mongodb_to_sqlite")
-
-ROOT_PATH = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
-
-
-@dataclass
-class MigrationConfig:
- """迁移配置类"""
-
- mongo_collection: str
- target_model: Type[Model]
- field_mapping: Dict[str, str]
- batch_size: int = 500
- enable_validation: bool = True
- skip_duplicates: bool = True
- unique_fields: List[str] = field(default_factory=list) # 用于重复检查的字段
-
-
-# 数据验证相关类已移除 - 用户要求不要数据验证
-
-
-@dataclass
-class MigrationCheckpoint:
- """迁移断点数据"""
-
- collection_name: str
- processed_count: int
- last_processed_id: Any
- timestamp: datetime
- batch_errors: List[Dict[str, Any]] = field(default_factory=list)
-
-
-@dataclass
-class MigrationStats:
- """迁移统计信息"""
-
- total_documents: int = 0
- processed_count: int = 0
- success_count: int = 0
- error_count: int = 0
- skipped_count: int = 0
- duplicate_count: int = 0
- validation_errors: int = 0
- batch_insert_count: int = 0
- errors: List[Dict[str, Any]] = field(default_factory=list)
- start_time: Optional[datetime] = None
- end_time: Optional[datetime] = None
-
- def add_error(self, doc_id: Any, error: str, doc_data: Optional[Dict] = None):
- """添加错误记录"""
- self.errors.append(
- {"doc_id": str(doc_id), "error": error, "timestamp": datetime.now().isoformat(), "doc_data": doc_data}
- )
- self.error_count += 1
-
- def add_validation_error(self, doc_id: Any, field: str, error: str):
- """添加验证错误"""
- self.add_error(doc_id, f"验证失败 - {field}: {error}")
- self.validation_errors += 1
-
-
-class MongoToSQLiteMigrator:
- """MongoDB到SQLite数据迁移器 - 使用Peewee ORM"""
-
- def __init__(self, mongo_uri: Optional[str] = None, database_name: Optional[str] = None):
- self.database_name = database_name or os.getenv("DATABASE_NAME", "MegBot")
- self.mongo_uri = mongo_uri or self._build_mongo_uri()
- self.mongo_client: Optional[MongoClient] = None
- self.mongo_db = None
-
- # 迁移配置
- self.migration_configs = self._initialize_migration_configs()
-
- # 进度条控制台
- self.console = Console()
- # 检查点目录
- self.checkpoint_dir = Path(os.path.join(ROOT_PATH, "data", "checkpoints"))
- self.checkpoint_dir.mkdir(exist_ok=True)
-
- # 验证规则已禁用
- self.validation_rules = self._initialize_validation_rules()
-
- def _build_mongo_uri(self) -> str:
- """构建MongoDB连接URI"""
- if mongo_uri := os.getenv("MONGODB_URI"):
- return mongo_uri
-
- user = os.getenv("MONGODB_USER")
- password = os.getenv("MONGODB_PASS")
- host = os.getenv("MONGODB_HOST", "localhost")
- port = os.getenv("MONGODB_PORT", "27017")
- auth_source = os.getenv("MONGODB_AUTH_SOURCE", "admin")
-
- if user and password:
- return f"mongodb://{user}:{password}@{host}:{port}/{self.database_name}?authSource={auth_source}"
- else:
- return f"mongodb://{host}:{port}/{self.database_name}"
-
- def _initialize_migration_configs(self) -> List[MigrationConfig]:
- """初始化迁移配置"""
- return [ # 表情包迁移配置
- MigrationConfig(
- mongo_collection="emoji",
- target_model=Emoji,
- field_mapping={
- "full_path": "full_path",
- "format": "format",
- "hash": "emoji_hash",
- "description": "description",
- "emotion": "emotion",
- "usage_count": "usage_count",
- "last_used_time": "last_used_time",
- # record_time字段将在转换时自动设置为当前时间
- },
- enable_validation=False, # 禁用数据验证
- unique_fields=["full_path", "emoji_hash"],
- ),
- # 聊天流迁移配置
- MigrationConfig(
- mongo_collection="chat_streams",
- target_model=ChatStreams,
- field_mapping={
- "stream_id": "stream_id",
- "create_time": "create_time",
- "group_info.platform": "group_platform", # 由于Mongodb处理私聊时会让group_info值为null，而新的数据库不允许为null，所以私聊聊天流是没法迁移的，等更新吧。
- "group_info.group_id": "group_id", # 同上
- "group_info.group_name": "group_name", # 同上
- "last_active_time": "last_active_time",
- "platform": "platform",
- "user_info.platform": "user_platform",
- "user_info.user_id": "user_id",
- "user_info.user_nickname": "user_nickname",
- "user_info.user_cardname": "user_cardname",
- },
- enable_validation=False, # 禁用数据验证
- unique_fields=["stream_id"],
- ),
- # 消息迁移配置
- MigrationConfig(
- mongo_collection="messages",
- target_model=Messages,
- field_mapping={
- "message_id": "message_id",
- "time": "time",
- "chat_id": "chat_id",
- "chat_info.stream_id": "chat_info_stream_id",
- "chat_info.platform": "chat_info_platform",
- "chat_info.user_info.platform": "chat_info_user_platform",
- "chat_info.user_info.user_id": "chat_info_user_id",
- "chat_info.user_info.user_nickname": "chat_info_user_nickname",
- "chat_info.user_info.user_cardname": "chat_info_user_cardname",
- "chat_info.group_info.platform": "chat_info_group_platform",
- "chat_info.group_info.group_id": "chat_info_group_id",
- "chat_info.group_info.group_name": "chat_info_group_name",
- "chat_info.create_time": "chat_info_create_time",
- "chat_info.last_active_time": "chat_info_last_active_time",
- "user_info.platform": "user_platform",
- "user_info.user_id": "user_id",
- "user_info.user_nickname": "user_nickname",
- "user_info.user_cardname": "user_cardname",
- "processed_plain_text": "processed_plain_text",
- "memorized_times": "memorized_times",
- },
- enable_validation=False, # 禁用数据验证
- unique_fields=["message_id"],
- ),
- # 图片迁移配置
- MigrationConfig(
- mongo_collection="images",
- target_model=Images,
- field_mapping={
- "hash": "emoji_hash",
- "description": "description",
- "path": "path",
- "timestamp": "timestamp",
- "type": "type",
- },
- unique_fields=["path"],
- ),
- # 图片描述迁移配置
- MigrationConfig(
- mongo_collection="image_descriptions",
- target_model=ImageDescriptions,
- field_mapping={
- "type": "type",
- "hash": "image_description_hash",
- "description": "description",
- "timestamp": "timestamp",
- },
- unique_fields=["image_description_hash", "type"],
- ),
- # 个人信息迁移配置
- MigrationConfig(
- mongo_collection="person_info",
- target_model=PersonInfo,
- field_mapping={
- "person_id": "person_id",
- "person_name": "person_name",
- "name_reason": "name_reason",
- "platform": "platform",
- "user_id": "user_id",
- "nickname": "nickname",
- "relationship_value": "relationship_value",
- "konw_time": "know_time",
- },
- unique_fields=["person_id"],
- ),
- # 知识库迁移配置
- MigrationConfig(
- mongo_collection="knowledges",
- target_model=Knowledges,
- field_mapping={"content": "content", "embedding": "embedding"},
- unique_fields=["content"], # 假设内容唯一
- ),
- # 思考日志迁移配置
- MigrationConfig(
- mongo_collection="thinking_log",
- target_model=ThinkingLog,
- field_mapping={
- "chat_id": "chat_id",
- "trigger_text": "trigger_text",
- "response_text": "response_text",
- "trigger_info": "trigger_info_json",
- "response_info": "response_info_json",
- "timing_results": "timing_results_json",
- "chat_history": "chat_history_json",
- "chat_history_in_thinking": "chat_history_in_thinking_json",
- "chat_history_after_response": "chat_history_after_response_json",
- "heartflow_data": "heartflow_data_json",
- "reasoning_data": "reasoning_data_json",
- },
- unique_fields=["chat_id", "trigger_text"],
- ),
- # 图节点迁移配置
- MigrationConfig(
- mongo_collection="graph_data.nodes",
- target_model=GraphNodes,
- field_mapping={
- "concept": "concept",
- "memory_items": "memory_items",
- "hash": "hash",
- "created_time": "created_time",
- "last_modified": "last_modified",
- },
- unique_fields=["concept"],
- ),
- # 图边迁移配置
- MigrationConfig(
- mongo_collection="graph_data.edges",
- target_model=GraphEdges,
- field_mapping={
- "source": "source",
- "target": "target",
- "strength": "strength",
- "hash": "hash",
- "created_time": "created_time",
- "last_modified": "last_modified",
- },
- unique_fields=["source", "target"], # 组合唯一性
- ),
- ]
-
- def _initialize_validation_rules(self) -> Dict[str, Any]:
- """数据验证已禁用 - 返回空字典"""
- return {}
-
- def connect_mongodb(self) -> bool:
- """连接到MongoDB"""
- try:
- self.mongo_client = MongoClient(
- self.mongo_uri, serverSelectionTimeoutMS=5000, connectTimeoutMS=10000, maxPoolSize=10
- )
-
- # 测试连接
- self.mongo_client.admin.command("ping")
- self.mongo_db = self.mongo_client[self.database_name]
-
- logger.info(f"成功连接到MongoDB: {self.database_name}")
- return True
-
- except ConnectionFailure as e:
- logger.error(f"MongoDB连接失败: {e}")
- return False
- except Exception as e:
- logger.error(f"MongoDB连接异常: {e}")
- return False
-
- def disconnect_mongodb(self):
- """断开MongoDB连接"""
- if self.mongo_client:
- self.mongo_client.close()
- logger.info("MongoDB连接已关闭")
-
- def _get_nested_value(self, document: Dict[str, Any], field_path: str) -> Any:
- """获取嵌套字段的值"""
- if "." not in field_path:
- return document.get(field_path)
-
- parts = field_path.split(".")
- value = document
-
- for part in parts:
- if isinstance(value, dict):
- value = value.get(part)
- else:
- return None
-
- if value is None:
- break
-
- return value
-
- def _convert_field_value(self, value: Any, target_field: Field) -> Any:
- """根据目标字段类型转换值"""
- if value is None:
- return None
-
- field_type = target_field.__class__.__name__
-
- try:
- if target_field.name == "record_time" and field_type == "DateTimeField":
- return datetime.now()
-
- if field_type in ["CharField", "TextField"]:
- if isinstance(value, (list, dict)):
- return json.dumps(value, ensure_ascii=False)
- return str(value) if value is not None else ""
-
- elif field_type == "IntegerField":
- if isinstance(value, str):
- # 处理字符串数字
- clean_value = value.strip()
- if clean_value.replace(".", "").replace("-", "").isdigit():
- return int(float(clean_value))
- return 0
- return int(value) if value is not None else 0
-
- elif field_type in ["FloatField", "DoubleField"]:
- return float(value) if value is not None else 0.0
-
- elif field_type == "BooleanField":
- if isinstance(value, str):
- return value.lower() in ("true", "1", "yes", "on")
- return bool(value)
-
- elif field_type == "DateTimeField":
- if isinstance(value, (int, float)):
- return datetime.fromtimestamp(value)
- elif isinstance(value, str):
- try:
- # 尝试解析ISO格式日期
- return datetime.fromisoformat(value.replace("Z", "+00:00"))
- except ValueError:
- try:
- # 尝试解析时间戳字符串
- return datetime.fromtimestamp(float(value))
- except ValueError:
- return datetime.now()
- return datetime.now()
-
- return value
-
- except (ValueError, TypeError) as e:
- logger.warning(f"字段值转换失败 ({field_type}): {value} -> {e}")
- return self._get_default_value_for_field(target_field)
-
- def _get_default_value_for_field(self, field: Field) -> Any:
- """获取字段的默认值"""
- field_type = field.__class__.__name__
-
- if hasattr(field, "default") and field.default is not None:
- return field.default
-
- if field.null:
- return None
-
- # 根据字段类型返回默认值
- if field_type in ["CharField", "TextField"]:
- return ""
- elif field_type == "IntegerField":
- return 0
- elif field_type in ["FloatField", "DoubleField"]:
- return 0.0
- elif field_type == "BooleanField":
- return False
- elif field_type == "DateTimeField":
- return datetime.now()
-
- return None
-
- def _validate_data(self, collection_name: str, data: Dict[str, Any], doc_id: Any, stats: MigrationStats) -> bool:
- """数据验证已禁用 - 始终返回True"""
- return True
-
- def _save_checkpoint(self, collection_name: str, processed_count: int, last_id: Any):
- """保存迁移断点"""
- checkpoint = MigrationCheckpoint(
- collection_name=collection_name,
- processed_count=processed_count,
- last_processed_id=last_id,
- timestamp=datetime.now(),
- )
-
- checkpoint_file = self.checkpoint_dir / f"{collection_name}_checkpoint.pkl"
- try:
- with open(checkpoint_file, "wb") as f:
- pickle.dump(checkpoint, f)
- except Exception as e:
- logger.warning(f"保存断点失败: {e}")
-
- def _load_checkpoint(self, collection_name: str) -> Optional[MigrationCheckpoint]:
- """加载迁移断点"""
- checkpoint_file = self.checkpoint_dir / f"{collection_name}_checkpoint.pkl"
- if not checkpoint_file.exists():
- return None
-
- try:
- with open(checkpoint_file, "rb") as f:
- return pickle.load(f)
- except Exception as e:
- logger.warning(f"加载断点失败: {e}")
- return None
-
- def _batch_insert(self, model: Type[Model], data_list: List[Dict[str, Any]]) -> int:
- """批量插入数据"""
- if not data_list:
- return 0
-
- success_count = 0
- try:
- with db.atomic():
- # 分批插入，避免SQL语句过长
- batch_size = 100
- for i in range(0, len(data_list), batch_size):
- batch = data_list[i : i + batch_size]
- model.insert_many(batch).execute()
- success_count += len(batch)
- except Exception as e:
- logger.error(f"批量插入失败: {e}")
- # 如果批量插入失败，尝试逐个插入
- for data in data_list:
- try:
- model.create(**data)
- success_count += 1
- except Exception:
- pass # 忽略单个插入失败
-
- return success_count
-
- def _check_duplicate_by_unique_fields(
- self, model: Type[Model], data: Dict[str, Any], unique_fields: List[str]
- ) -> bool:
- """根据唯一字段检查重复"""
- if not unique_fields:
- return False
-
- try:
- query = model.select()
- for field_name in unique_fields:
- if field_name in data and data[field_name] is not None:
- field_obj = getattr(model, field_name)
- query = query.where(field_obj == data[field_name])
-
- return query.exists()
- except Exception as e:
- logger.debug(f"重复检查失败: {e}")
- return False
-
- def _create_model_instance(self, model: Type[Model], data: Dict[str, Any]) -> Optional[Model]:
- """使用ORM创建模型实例"""
- try:
- # 过滤掉不存在的字段
- valid_data = {}
- for field_name, value in data.items():
- if hasattr(model, field_name):
- valid_data[field_name] = value
- else:
- logger.debug(f"跳过未知字段: {field_name}")
-
- # 创建实例
- instance = model.create(**valid_data)
- return instance
-
- except IntegrityError as e:
- # 处理唯一约束冲突等完整性错误
- logger.debug(f"完整性约束冲突: {e}")
- return None
- except Exception as e:
- logger.error(f"创建模型实例失败: {e}")
- return None
-
- def migrate_collection(self, config: MigrationConfig) -> MigrationStats:
- """迁移单个集合 - 使用优化的批量插入和进度条"""
- stats = MigrationStats()
- stats.start_time = datetime.now()
-
- # 检查是否有断点
- checkpoint = self._load_checkpoint(config.mongo_collection)
- start_from_id = checkpoint.last_processed_id if checkpoint else None
- if checkpoint:
- stats.processed_count = checkpoint.processed_count
- logger.info(f"从断点恢复: 已处理 {checkpoint.processed_count} 条记录")
-
- logger.info(f"开始迁移: {config.mongo_collection} -> {config.target_model._meta.table_name}")
-
- try:
- # 获取MongoDB集合
- mongo_collection = self.mongo_db[config.mongo_collection]
-
- # 构建查询条件（用于断点恢复）
- query = {}
- if start_from_id:
- query = {"_id": {"$gt": start_from_id}}
-
- stats.total_documents = mongo_collection.count_documents(query)
-
- if stats.total_documents == 0:
- logger.warning(f"集合 {config.mongo_collection} 为空，跳过迁移")
- return stats
-
- logger.info(f"待迁移文档数量: {stats.total_documents}")
-
- # 创建Rich进度条
- with Progress(
- SpinnerColumn(),
- TextColumn("[progress.description]{task.description}"),
- BarColumn(),
- TaskProgressColumn(),
- TimeElapsedColumn(),
- TimeRemainingColumn(),
- console=self.console,
- refresh_per_second=10,
- ) as progress:
- task = progress.add_task(f"迁移 {config.mongo_collection}", total=stats.total_documents)
- # 批量处理数据
- batch_data = []
- batch_count = 0
- last_processed_id = None
-
- for mongo_doc in mongo_collection.find(query).batch_size(config.batch_size):
- try:
- doc_id = mongo_doc.get("_id", "unknown")
- last_processed_id = doc_id
-
- # 构建目标数据
- target_data = {}
- for mongo_field, sqlite_field in config.field_mapping.items():
- value = self._get_nested_value(mongo_doc, mongo_field)
-
- # 获取目标字段对象并转换类型
- if hasattr(config.target_model, sqlite_field):
- field_obj = getattr(config.target_model, sqlite_field)
- converted_value = self._convert_field_value(value, field_obj)
- target_data[sqlite_field] = converted_value
-
- # 数据验证已禁用
- # if config.enable_validation:
- # if not self._validate_data(config.mongo_collection, target_data, doc_id, stats):
- # stats.skipped_count += 1
- # continue
-
- # 重复检查
- if config.skip_duplicates and self._check_duplicate_by_unique_fields(
- config.target_model, target_data, config.unique_fields
- ):
- stats.duplicate_count += 1
- stats.skipped_count += 1
- logger.debug(f"跳过重复记录: {doc_id}")
- continue
-
- # 添加到批量数据
- batch_data.append(target_data)
- stats.processed_count += 1
-
- # 执行批量插入
- if len(batch_data) >= config.batch_size:
- success_count = self._batch_insert(config.target_model, batch_data)
- stats.success_count += success_count
- stats.batch_insert_count += 1
-
- # 保存断点
- self._save_checkpoint(config.mongo_collection, stats.processed_count, last_processed_id)
-
- batch_data.clear()
- batch_count += 1
-
- # 更新进度条
- progress.update(task, advance=config.batch_size)
-
- except Exception as e:
- doc_id = mongo_doc.get("_id", "unknown")
- stats.add_error(doc_id, f"处理文档异常: {e}", mongo_doc)
- logger.error(f"处理文档失败 (ID: {doc_id}): {e}")
-
- # 处理剩余的批量数据
- if batch_data:
- success_count = self._batch_insert(config.target_model, batch_data)
- stats.success_count += success_count
- stats.batch_insert_count += 1
- progress.update(task, advance=len(batch_data))
-
- # 完成进度条
- progress.update(task, completed=stats.total_documents)
-
- stats.end_time = datetime.now()
- duration = stats.end_time - stats.start_time
-
- logger.info(
- f"迁移完成: {config.mongo_collection} -> {config.target_model._meta.table_name}\n"
- f"总计: {stats.total_documents}, 成功: {stats.success_count}, "
- f"错误: {stats.error_count}, 跳过: {stats.skipped_count}, 重复: {stats.duplicate_count}\n"
- f"耗时: {duration.total_seconds():.2f}秒, 批量插入次数: {stats.batch_insert_count}"
- )
-
- # 清理断点文件
- checkpoint_file = self.checkpoint_dir / f"{config.mongo_collection}_checkpoint.pkl"
- if checkpoint_file.exists():
- checkpoint_file.unlink()
-
- except Exception as e:
- logger.error(f"迁移集合 {config.mongo_collection} 时发生异常: {e}")
- stats.add_error("collection_error", str(e))
-
- return stats
-
- def migrate_all(self) -> Dict[str, MigrationStats]:
- """执行所有迁移任务"""
- logger.info("开始执行数据库迁移...")
-
- if not self.connect_mongodb():
- logger.error("无法连接到MongoDB，迁移终止")
- return {}
-
- all_stats = {}
-
- try:
- # 创建总体进度表格
- total_collections = len(self.migration_configs)
- self.console.print(
- Panel(
- f"[bold blue]MongoDB 到 SQLite 数据迁移[/bold blue]\n"
- f"[yellow]总集合数: {total_collections}[/yellow]",
- title="迁移开始",
- expand=False,
- )
- )
- for idx, config in enumerate(self.migration_configs, 1):
- self.console.print(
- f"\n[bold green]正在处理集合 {idx}/{total_collections}: {config.mongo_collection}[/bold green]"
- )
- stats = self.migrate_collection(config)
- all_stats[config.mongo_collection] = stats
-
- # 显示单个集合的快速统计
- if stats.processed_count > 0:
- success_rate = stats.success_count / stats.processed_count * 100
- if success_rate >= 95:
- status_emoji = "✅"
- status_color = "bright_green"
- elif success_rate >= 80:
- status_emoji = "⚠️"
- status_color = "yellow"
- else:
- status_emoji = "❌"
- status_color = "red"
-
- self.console.print(
- f" {status_emoji} [{status_color}]完成: {stats.success_count}/{stats.processed_count} "
- f"({success_rate:.1f}%) 错误: {stats.error_count}[/{status_color}]"
- )
-
- # 错误率检查
- if stats.processed_count > 0:
- error_rate = stats.error_count / stats.processed_count
- if error_rate > 0.1: # 错误率超过10%
- self.console.print(
- f" [red]⚠️ 警告: 错误率较高 {error_rate:.1%} "
- f"({stats.error_count}/{stats.processed_count})[/red]"
- )
-
- finally:
- self.disconnect_mongodb()
-
- self._print_migration_summary(all_stats)
- return all_stats
-
- def _print_migration_summary(self, all_stats: Dict[str, MigrationStats]):
- """使用Rich打印美观的迁移汇总信息"""
- # 计算总体统计
- total_processed = sum(stats.processed_count for stats in all_stats.values())
- total_success = sum(stats.success_count for stats in all_stats.values())
- total_errors = sum(stats.error_count for stats in all_stats.values())
- total_skipped = sum(stats.skipped_count for stats in all_stats.values())
- total_duplicates = sum(stats.duplicate_count for stats in all_stats.values())
- total_validation_errors = sum(stats.validation_errors for stats in all_stats.values())
- total_batch_inserts = sum(stats.batch_insert_count for stats in all_stats.values())
-
- # 计算总耗时
- total_duration_seconds = 0
- for stats in all_stats.values():
- if stats.start_time and stats.end_time:
- duration = stats.end_time - stats.start_time
- total_duration_seconds += duration.total_seconds()
-
- # 创建详细统计表格
- table = Table(title="[bold blue]数据迁移汇总报告[/bold blue]", show_header=True, header_style="bold magenta")
- table.add_column("集合名称", style="cyan", width=20)
- table.add_column("文档总数", justify="right", style="blue")
- table.add_column("处理数量", justify="right", style="green")
- table.add_column("成功数量", justify="right", style="green")
- table.add_column("错误数量", justify="right", style="red")
- table.add_column("跳过数量", justify="right", style="yellow")
- table.add_column("重复数量", justify="right", style="bright_yellow")
- table.add_column("验证错误", justify="right", style="red")
- table.add_column("批次数", justify="right", style="purple")
- table.add_column("成功率", justify="right", style="bright_green")
- table.add_column("耗时(秒)", justify="right", style="blue")
-
- for collection_name, stats in all_stats.items():
- success_rate = (stats.success_count / stats.processed_count * 100) if stats.processed_count > 0 else 0
- duration = 0
- if stats.start_time and stats.end_time:
- duration = (stats.end_time - stats.start_time).total_seconds()
-
- # 根据成功率设置颜色
- if success_rate >= 95:
- success_rate_style = "[bright_green]"
- elif success_rate >= 80:
- success_rate_style = "[yellow]"
- else:
- success_rate_style = "[red]"
-
- table.add_row(
- collection_name,
- str(stats.total_documents),
- str(stats.processed_count),
- str(stats.success_count),
- f"[red]{stats.error_count}[/red]" if stats.error_count > 0 else "0",
- f"[yellow]{stats.skipped_count}[/yellow]" if stats.skipped_count > 0 else "0",
- f"[bright_yellow]{stats.duplicate_count}[/bright_yellow]" if stats.duplicate_count > 0 else "0",
- f"[red]{stats.validation_errors}[/red]" if stats.validation_errors > 0 else "0",
- str(stats.batch_insert_count),
- f"{success_rate_style}{success_rate:.1f}%[/{success_rate_style[1:]}",
- f"{duration:.2f}",
- )
-
- # 添加总计行
- total_success_rate = (total_success / total_processed * 100) if total_processed > 0 else 0
- if total_success_rate >= 95:
- total_rate_style = "[bright_green]"
- elif total_success_rate >= 80:
- total_rate_style = "[yellow]"
- else:
- total_rate_style = "[red]"
-
- table.add_section()
- table.add_row(
- "[bold]总计[/bold]",
- f"[bold]{sum(stats.total_documents for stats in all_stats.values())}[/bold]",
- f"[bold]{total_processed}[/bold]",
- f"[bold]{total_success}[/bold]",
- f"[bold red]{total_errors}[/bold red]" if total_errors > 0 else "[bold]0[/bold]",
- f"[bold yellow]{total_skipped}[/bold yellow]" if total_skipped > 0 else "[bold]0[/bold]",
- f"[bold bright_yellow]{total_duplicates}[/bold bright_yellow]"
- if total_duplicates > 0
- else "[bold]0[/bold]",
- f"[bold red]{total_validation_errors}[/bold red]" if total_validation_errors > 0 else "[bold]0[/bold]",
- f"[bold]{total_batch_inserts}[/bold]",
- f"[bold]{total_rate_style}{total_success_rate:.1f}%[/{total_rate_style[1:]}[/bold]",
- f"[bold]{total_duration_seconds:.2f}[/bold]",
- )
-
- self.console.print(table)
-
- # 创建状态面板
- status_items = []
- if total_errors > 0:
- status_items.append(f"[red]⚠️ 发现 {total_errors} 个错误，请检查日志详情[/red]")
-
- if total_validation_errors > 0:
- status_items.append(f"[red]🔍 数据验证失败: {total_validation_errors} 条记录[/red]")
-
- if total_duplicates > 0:
- status_items.append(f"[yellow]📋 跳过重复记录: {total_duplicates} 条[/yellow]")
-
- if total_success_rate >= 95:
- status_items.append(f"[bright_green]✅ 迁移成功率优秀: {total_success_rate:.1f}%[/bright_green]")
- elif total_success_rate >= 80:
- status_items.append(f"[yellow]⚡ 迁移成功率良好: {total_success_rate:.1f}%[/yellow]")
- else:
- status_items.append(f"[red]❌ 迁移成功率较低: {total_success_rate:.1f}%，需要检查[/red]")
-
- if status_items:
- status_panel = Panel(
- "\n".join(status_items), title="[bold yellow]迁移状态总结[/bold yellow]", border_style="yellow"
- )
- self.console.print(status_panel)
-
- # 性能统计面板
- avg_speed = total_processed / total_duration_seconds if total_duration_seconds > 0 else 0
- performance_info = (
- f"[cyan]总处理时间:[/cyan] {total_duration_seconds:.2f} 秒\n"
- f"[cyan]平均处理速度:[/cyan] {avg_speed:.1f} 条记录/秒\n"
- f"[cyan]批量插入优化:[/cyan] 执行了 {total_batch_inserts} 次批量操作"
- )
-
- performance_panel = Panel(performance_info, title="[bold green]性能统计[/bold green]", border_style="green")
- self.console.print(performance_panel)
-
- def add_migration_config(self, config: MigrationConfig):
- """添加新的迁移配置"""
- self.migration_configs.append(config)
-
- def migrate_single_collection(self, collection_name: str) -> Optional[MigrationStats]:
- """迁移单个指定的集合"""
- config = next((c for c in self.migration_configs if c.mongo_collection == collection_name), None)
- if not config:
- logger.error(f"未找到集合 {collection_name} 的迁移配置")
- return None
-
- if not self.connect_mongodb():
- logger.error("无法连接到MongoDB")
- return None
-
- try:
- stats = self.migrate_collection(config)
- self._print_migration_summary({collection_name: stats})
- return stats
- finally:
- self.disconnect_mongodb()
-
- def export_error_report(self, all_stats: Dict[str, MigrationStats], filepath: str):
- """导出错误报告"""
- error_report = {
- "timestamp": datetime.now().isoformat(),
- "summary": {
- collection: {
- "total": stats.total_documents,
- "processed": stats.processed_count,
- "success": stats.success_count,
- "errors": stats.error_count,
- "skipped": stats.skipped_count,
- "duplicates": stats.duplicate_count,
- }
- for collection, stats in all_stats.items()
- },
- "errors": {collection: stats.errors for collection, stats in all_stats.items() if stats.errors},
- }
-
- try:
- with open(filepath, "w", encoding="utf-8") as f:
- json.dump(error_report, f, ensure_ascii=False, indent=2)
- logger.info(f"错误报告已导出到: {filepath}")
- except Exception as e:
- logger.error(f"导出错误报告失败: {e}")
-
-
-def main():
- """主程序入口"""
- migrator = MongoToSQLiteMigrator()
-
- # 执行迁移
- migration_results = migrator.migrate_all()
-
- # 导出错误报告（如果有错误）
- if any(stats.error_count > 0 for stats in migration_results.values()):
- error_report_path = f"migration_errors_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
- migrator.export_error_report(migration_results, error_report_path)
-
- logger.info("数据迁移完成！")
-
-
-if __name__ == "__main__":
- main()
diff --git a/src/chat/heart_flow/heartFC_chat.py b/src/chat/heart_flow/heartFC_chat.py
index 7cb5bf85..8680392a 100644
--- a/src/chat/heart_flow/heartFC_chat.py
+++ b/src/chat/heart_flow/heartFC_chat.py
@@ -21,7 +21,6 @@ from src.chat.heart_flow.hfc_utils import send_typing, stop_typing
from src.chat.frequency_control.talk_frequency_control import talk_frequency_control
from src.chat.frequency_control.focus_value_control import focus_value_control
from src.chat.express.expression_learner import expression_learner_manager
-from src.person_info.relationship_builder_manager import relationship_builder_manager
from src.person_info.person_info import Person
from src.plugin_system.base.component_types import ChatMode, EventType, ActionInfo
from src.plugin_system.core import events_manager
@@ -84,7 +83,6 @@ class HeartFChatting:
raise ValueError(f"无法找到聊天流: {self.stream_id}")
self.log_prefix = f"[{get_chat_manager().get_stream_name(self.stream_id) or self.stream_id}]"
- self.relationship_builder = relationship_builder_manager.get_or_create_builder(self.stream_id)
self.expression_learner = expression_learner_manager.get_expression_learner(self.stream_id)
self.talk_frequency_control = talk_frequency_control.get_talk_frequency_control(self.stream_id)
@@ -385,7 +383,6 @@ class HeartFChatting:
await send_typing()
async with global_prompt_manager.async_message_scope(self.chat_stream.context.get_template_name()):
- await self.relationship_builder.build_relation()
await self.expression_learner.trigger_learning_for_chat()
# # 记忆构建：为当前chat_id构建记忆
diff --git a/src/chat/planner_actions/planner.py b/src/chat/planner_actions/planner.py
index 55473c0d..8ef47874 100644
--- a/src/chat/planner_actions/planner.py
+++ b/src/chat/planner_actions/planner.py
@@ -96,7 +96,10 @@ def init_prompt():
- {mentioned_bonus}
- 如果你刚刚进行了回复，不要对同一个话题重复回应
-请你选中一条需要回复的消息并输出其id,输出格式如下：
+你之前的动作记录：
+{actions_before_now_block}
+
+请你从新消息中选出一条需要回复的消息并输出其id,输出格式如下：
{{
"action": "reply",
"target_message_id":"想要回复的消息id，消息id格式:m+数字",
@@ -845,6 +848,7 @@ class ActionPlanner:
mentioned_bonus=mentioned_bonus,
moderation_prompt=moderation_prompt_block,
name_block=name_block,
+ actions_before_now_block=actions_before_now_block,
)
return prompt, message_id_list
except Exception as e:
diff --git a/src/chat/utils/statistic.py b/src/chat/utils/statistic.py
index d272a300..5cec59f6 100644
--- a/src/chat/utils/statistic.py
+++ b/src/chat/utils/statistic.py
@@ -166,6 +166,7 @@ class StatisticOutputTask(AsyncTask):
self.stat_period: List[Tuple[str, timedelta, str]] = [
("all_time", now - deploy_time, "自部署以来"), # 必须保留"all_time"
("last_7_days", timedelta(days=7), "最近7天"),
+ ("last_3_days", timedelta(days=3), "最近3天"),
("last_24_hours", timedelta(days=1), "最近24小时"),
("last_3_hours", timedelta(hours=3), "最近3小时"),
("last_hour", timedelta(hours=1), "最近1小时"),
@@ -781,45 +782,216 @@ class StatisticOutputTask(AsyncTask):
总请求数: {stat_data[TOTAL_REQ_CNT]}
总花费: {stat_data[TOTAL_COST]:.4f} ¥
- 按模型分类统计
-
- | 模型名称 | 调用次数 | 输入Token | 输出Token | Token总量 | 累计花费 | 平均耗时(秒) | 标准差(秒) |
|---|
-
- {model_rows}
-
-
+
+
+
按模型分类统计
+
+ | 模型名称 | 调用次数 | 输入Token | 输出Token | Token总量 | 累计花费 | 平均耗时(秒) | 标准差(秒) |
|---|
+
+ {model_rows}
+
+
+
+
+
模型调用次数分布
+
+
+
按模块分类统计
-
-
- | 模块名称 | 调用次数 | 输入Token | 输出Token | Token总量 | 累计花费 | 平均耗时(秒) | 标准差(秒) |
|---|
-
-
- {module_rows}
-
-
+
+
+
按模块分类统计
+
+
+ | 模块名称 | 调用次数 | 输入Token | 输出Token | Token总量 | 累计花费 | 平均耗时(秒) | 标准差(秒) |
|---|
+
+
+ {module_rows}
+
+
+
+
+
模块调用次数分布
+
+
+
按请求类型分类统计
-
-
- | 请求类型 | 调用次数 | 输入Token | 输出Token | Token总量 | 累计花费 | 平均耗时(秒) | 标准差(秒) |
|---|
-
-
- {type_rows}
-
-
+
+
+
按请求类型分类统计
+
+
+ | 请求类型 | 调用次数 | 输入Token | 输出Token | Token总量 | 累计花费 | 平均耗时(秒) | 标准差(秒) |
|---|
+
+
+ {type_rows}
+
+
+
+
+
请求类型分布
+
+
+
聊天消息统计
-
-
- | 联系人/群组名称 | 消息数量 |
|---|
-
-
- {chat_rows}
-
-
+
+
+
聊天消息统计
+
+
+ | 联系人/群组名称 | 消息数量 |
|---|
+
+
+ {chat_rows}
+
+
+
+
+
消息分布
+
+
+