#!/usr/bin/env python3 """ 慢剧工作台数据库初始化脚本 基于PRD v2.0.0要求，创建SQLite数据库 """ import sqlite3 import json import os from datetime import datetime from pathlib import Path class DatabaseInitializer: def __init__(self, config_path="/root/data/disk/system/config/manhua_studio_config.json"): self.config_path = config_path self.load_config() def load_config(self): """加载配置文件""" with open(self.config_path, 'r', encoding='utf-8') as f: self.config = json.load(f) self.db_config = self.config['database_config']['engines'] def init_database(self, db_name): """初始化指定数据库""" db_path = self.db_config[db_name]['path'] db_dir = os.path.dirname(db_path) # 创建数据库目录 os.makedirs(db_dir, exist_ok=True) print(f"初始化数据库: {db_name} -> {db_path}") # 连接到数据库 conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row try: # 设置PRAGMA参数 self.set_pragmas(conn, db_name) # 创建表结构 if db_name == 'projects': self.create_projects_tables(conn) elif db_name == 'agents': self.create_agents_tables(conn) elif db_name == 'analytics': self.create_analytics_tables(conn) # 插入初始数据 self.insert_initial_data(conn, db_name) conn.commit() print(f"✅ 数据库 {db_name} 初始化成功") except Exception as e: print(f"❌ 数据库 {db_name} 初始化失败: {e}") conn.rollback() raise finally: conn.close() def set_pragmas(self, conn, db_name): """设置SQLite PRAGMA参数""" cursor = conn.cursor() # 通用PRAGMA设置 pragmas = [ "PRAGMA journal_mode=WAL", # 启用WAL模式，支持读写并发 "PRAGMA busy_timeout=5000", # 繁忙超时5秒 "PRAGMA foreign_keys=ON", # 启用外键约束 "PRAGMA synchronous=NORMAL", # 平衡安全性和性能 ] # 数据库特定的PRAGMA if db_name == 'projects': pragmas.extend([ "PRAGMA cache_size=-2000", # 2MB缓存 "PRAGMA temp_store=MEMORY", # 临时表使用内存 ]) for pragma in pragmas: cursor.execute(pragma) cursor.close() def create_projects_tables(self, conn): """创建项目数据库表""" cursor = conn.cursor() # 项目表 cursor.execute(''' CREATE TABLE IF NOT EXISTS projects ( project_id TEXT PRIMARY KEY, name TEXT NOT NULL, description TEXT, status TEXT CHECK(status IN ( 'draft', 'collecting', 'creating', 'producing', 'reviewing', 'publishing', 'analyzing', 'completed', 'failed' )), workflow_step TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, created_by TEXT, metadata JSON ) ''') # 创建updated_at的触发器 cursor.execute(''' CREATE TRIGGER IF NOT EXISTS update_projects_timestamp AFTER UPDATE ON projects BEGIN UPDATE projects SET updated_at = CURRENT_TIMESTAMP WHERE project_id = NEW.project_id; END ''') # 项目版本表 cursor.execute(''' CREATE TABLE IF NOT EXISTS project_versions ( version_id TEXT PRIMARY KEY, project_id TEXT, version_number INTEGER, description TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, snapshot_data JSON, FOREIGN KEY (project_id) REFERENCES projects(project_id) ON DELETE CASCADE ) ''') # 项目文件表 cursor.execute(''' CREATE TABLE IF NOT EXISTS project_files ( file_id TEXT PRIMARY KEY, project_id TEXT, file_path TEXT NOT NULL, file_type TEXT, file_size INTEGER, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, metadata JSON, FOREIGN KEY (project_id) REFERENCES projects(project_id) ON DELETE CASCADE ) ''') # 创建索引 cursor.execute('CREATE INDEX IF NOT EXISTS idx_projects_status ON projects(status)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_projects_created ON projects(created_at)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_project_files_type ON project_files(file_type)') cursor.close() def create_agents_tables(self, conn): """创建Agent数据库表""" cursor = conn.cursor() # Agent任务表 cursor.execute(''' CREATE TABLE IF NOT EXISTS agent_tasks ( task_id TEXT PRIMARY KEY, project_id TEXT, agent_name TEXT CHECK(agent_name IN ( 'cto', 'collector', 'writer', 'producer', 'publisher', 'analyzer' )), task_type TEXT, priority INTEGER DEFAULT 1, status TEXT CHECK(status IN ( 'pending', 'running', 'completed', 'failed', 'retrying' )), input_data JSON, output_data JSON, dependencies JSON, retry_count INTEGER DEFAULT 0, max_retries INTEGER DEFAULT 3, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, started_at TIMESTAMP, completed_at TIMESTAMP, error_message TEXT, FOREIGN KEY (project_id) REFERENCES projects(project_id) ON DELETE CASCADE ) ''') # Agent状态表 cursor.execute(''' CREATE TABLE IF NOT EXISTS agent_status ( agent_name TEXT PRIMARY KEY, status TEXT CHECK(status IN ('idle', 'busy', 'error', 'offline')), current_task_id TEXT, last_heartbeat TIMESTAMP DEFAULT CURRENT_TIMESTAMP, performance_metrics JSON, FOREIGN KEY (current_task_id) REFERENCES agent_tasks(task_id) ON DELETE SET NULL ) ''') # Agent配置表 cursor.execute(''' CREATE TABLE IF NOT EXISTS agent_configs ( agent_name TEXT PRIMARY KEY, config_data JSON, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_by TEXT ) ''') # 创建索引 cursor.execute('CREATE INDEX IF NOT EXISTS idx_agent_tasks_status ON agent_tasks(status)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_agent_tasks_agent ON agent_tasks(agent_name)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_agent_tasks_project ON agent_tasks(project_id)') cursor.close() def create_analytics_tables(self, conn): """创建分析数据库表""" cursor = conn.cursor() # 创意表 cursor.execute(''' CREATE TABLE IF NOT EXISTS ideas ( idea_id TEXT PRIMARY KEY, project_id TEXT, title TEXT, description TEXT, source TEXT, hotness_score REAL, potential_score REAL, tags JSON, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (project_id) REFERENCES projects(project_id) ON DELETE CASCADE ) ''') # 剧本表 cursor.execute(''' CREATE TABLE IF NOT EXISTS scripts ( script_id TEXT PRIMARY KEY, project_id TEXT, version INTEGER, title TEXT, outline TEXT, full_content JSON, characters JSON, scenes JSON, estimated_duration INTEGER, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (project_id) REFERENCES projects(project_id) ON DELETE CASCADE ) ''') # 角色表 cursor.execute(''' CREATE TABLE IF NOT EXISTS characters ( character_id TEXT PRIMARY KEY, project_id TEXT, name TEXT, description TEXT, attributes JSON, images JSON, prompts JSON, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (project_id) REFERENCES projects(project_id) ON DELETE CASCADE ) ''') # 场景表 cursor.execute(''' CREATE TABLE IF NOT EXISTS scenes ( scene_id TEXT PRIMARY KEY, project_id TEXT, script_id TEXT, scene_number INTEGER, description TEXT, duration INTEGER, images JSON, video_clips JSON, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (project_id) REFERENCES projects(project_id) ON DELETE CASCADE, FOREIGN KEY (script_id) REFERENCES scripts(script_id) ON DELETE CASCADE ) ''') # 发布记录表 cursor.execute(''' CREATE TABLE IF NOT EXISTS publish_records ( publish_id TEXT PRIMARY KEY, project_id TEXT, platform TEXT, status TEXT CHECK(status IN ('pending', 'published', 'failed', 'removed')), publish_url TEXT, publish_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, view_count INTEGER DEFAULT 0, like_count INTEGER DEFAULT 0, comment_count INTEGER DEFAULT 0, share_count INTEGER DEFAULT 0, feedback_summary TEXT, FOREIGN KEY (project_id) REFERENCES projects(project_id) ON DELETE CASCADE ) ''') # 分析数据表 cursor.execute(''' CREATE TABLE IF NOT EXISTS analytics_data ( analytic_id TEXT PRIMARY KEY, project_id TEXT, metric_type TEXT, metric_value REAL, period_start TIMESTAMP, period_end TIMESTAMP, analysis_result TEXT, recommendations JSON, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (project_id) REFERENCES projects(project_id) ON DELETE CASCADE ) ''') # 创建索引 cursor.execute('CREATE INDEX IF NOT EXISTS idx_ideas_project ON ideas(project_id)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_scripts_project ON scripts(project_id)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_publish_records_project ON publish_records(project_id)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_publish_records_platform ON publish_records(platform)') cursor.close() def insert_initial_data(self, conn, db_name): """插入初始数据""" cursor = conn.cursor() if db_name == 'agents': # 插入Agent状态数据 agents = [ ('cto', 'idle', None, datetime.now().isoformat(), '{}'), ('collector', 'idle', None, datetime.now().isoformat(), '{}'), ('writer', 'idle', None, datetime.now().isoformat(), '{}'), ('producer', 'idle', None, datetime.now().isoformat(), '{}'), ('publisher', 'idle', None, datetime.now().isoformat(), '{}'), ('analyzer', 'idle', None, datetime.now().isoformat(), '{}'), ] cursor.executemany(''' INSERT OR IGNORE INTO agent_status (agent_name, status, current_task_id, last_heartbeat, performance_metrics) VALUES (?, ?, ?, ?, ?) ''', agents) # 插入Agent配置数据 agent_configs = [ ('cto', json.dumps({ "max_concurrent_tasks": 10, "heartbeat_interval": 30, "monitoring_interval": 30 }), datetime.now().isoformat(), 'system'), ('collector', json.dumps({ "collection_schedule": "*/30 * * * *", "data_retention_days": 30 }), datetime.now().isoformat(), 'system'), ('writer', json.dumps({ "max_script_length": 5000, "quality_level": "high" }), datetime.now().isoformat(), 'system'), ('producer', json.dumps({ "video_resolution": "1080x1920", "fps": 30, "aspect_ratio": "9:16" }), datetime.now().isoformat(), 'system'), ('publisher', json.dumps({ "supported_platforms": ["bilibili", "douyin", "kuaishou", "weibo"], "review_standards": ["quality", "compliance", "engagement"] }), datetime.now().isoformat(), 'system'), ('analyzer', json.dumps({ "analysis_schedule": "0 6 * * *", "report_frequency": "daily" }), datetime.now().isoformat(), 'system'), ] cursor.executemany(''' INSERT OR IGNORE INTO agent_configs (agent_name, config_data, updated_at, updated_by) VALUES (?, ?, ?, ?) ''', agent_configs) print(f"✅ 插入了 {len(agents)} 个Agent状态记录") print(f"✅ 插入了 {len(agent_configs)} 个Agent配置记录") cursor.close() def verify_database(self, db_name): """验证数据库结构""" db_path = self.db_config[db_name]['path'] if not os.path.exists(db_path): print(f"❌ 数据库文件不存在: {db_path}") return False conn = sqlite3.connect(db_path) cursor = conn.cursor() try: # 检查表数量 cursor.execute("SELECT name FROM sqlite_master WHERE type='table'") tables = cursor.fetchall() print(f"数据库 {db_name} 包含 {len(tables)} 个表:") for table in tables: print(f" - {table[0]}") return True except Exception as e: print(f"❌ 数据库验证失败: {e}") return False finally: conn.close() def init_all_databases(self): """初始化所有数据库""" print("=" * 60) print("开始初始化慢剧工作台数据库系统") print("=" * 60) databases = ['projects', 'agents', 'analytics'] for db_name in databases: print(f"\n📊 处理数据库: {db_name}") print("-" * 40) try: # 初始化数据库 self.init_database(db_name) # 验证数据库 if self.verify_database(db_name): print(f"✅ 数据库 {db_name} 验证通过") else: print(f"⚠️ 数据库 {db_name} 验证有问题") except Exception as e: print(f"❌ 数据库 {db_name} 初始化失败: {e}") continue print("\n" + "=" * 60) print("数据库初始化完成") print("=" * 60) def main(): """主函数""" try: # 创建初始化器 initializer = DatabaseInitializer() # 初始化所有数据库 initializer.init_all_databases() print("\n🎉 慢剧工作台数据库系统初始化成功！") print("\n已创建的数据库:") print(" - projects.db: 项目管理和工作流数据") print(" - agents.db: Agent状态和任务数据") print(" - analytics.db: 创作内容和分析数据") except Exception as e: print(f"❌ 数据库初始化失败: {e}") return 1 return 0 if __name__ == "__main__": exit(main())