""" Mô-đun Quản lý dự án - Hỗ trợ lưu, tải, xuất dự án """ import json import os import re import tempfile import logging from typing import List, Dict, Optional, Tuple from datetime import datetime from pathlib import Path from services.novel_generator import NovelProject, Chapter from locales.i18n import t from core.database import get_db logger = logging.getLogger(__name__) class ProjectManager: """quản lý dự án""" @staticmethod def _slugify(name: str) -> str: s = str(name or "").lower() s = re.sub(r'[^\w\s-]', '', s) s = re.sub(r'[\s_]+', '-', s) s = re.sub(r'-+', '-', s).strip('-') return s or "untitled" @staticmethod def create_project( title: str, genre: str, sub_genres: List[str], character_setting: str, world_setting: str, plot_idea: str ) -> Tuple[Optional[NovelProject], str]: """ Tạo dự án mới Returns: (Đối tượng dự án, Thông tin trạng thái) """ try: if not title or not title.strip(): return None, "Title cannot be empty" project_id = ProjectManager._slugify(title) now = datetime.now().isoformat() project = NovelProject( title=title.strip(), genre=genre.strip() if genre else "", sub_genres=sub_genres if sub_genres else [], character_setting=character_setting.strip() if character_setting else "", world_setting=world_setting.strip() if world_setting else "", plot_idea=plot_idea.strip() if plot_idea else "", id=project_id, created_at=now, updated_at=now ) logger.info(f"Project created: {project_id}") return project, t("project_manager.create_success", title=title) except Exception as e: logger.error(f"Project create failed: {e}") return None, t("project_manager.create_failed", error=str(e)) @staticmethod def save_project(project: NovelProject) -> Tuple[bool, str]: """ Lưu dự án vào SQLite Returns: (Cờ thành công (boolean), Thông tin trạng thái) """ try: if not project or not project.title: return False, "Project data incomplete" # Sử dụng project.id hiện có hoặc tạo mới if getattr(project, 'id', None): project_id = project.id else: project_id = ProjectManager._slugify(project.title) project.id = project_id conn = get_db() now = datetime.now().isoformat() project.updated_at = now # Lưu project conn.execute(""" INSERT OR REPLACE INTO projects (id, title, genre, sub_genres, character_setting, world_setting, plot_idea, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( project_id, project.title, project.genre, json.dumps(project.sub_genres if isinstance(project.sub_genres, list) else [], ensure_ascii=False), project.character_setting, project.world_setting, project.plot_idea, project.created_at, now )) # Xóa chapters cũ và insert lại conn.execute("DELETE FROM chapters WHERE project_id = ?", (project_id,)) for ch in project.chapters: conn.execute(""" INSERT INTO chapters (project_id, num, title, desc, content, word_count, generated_at) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( project_id, ch.num, ch.title, ch.desc, ch.content, ch.word_count, ch.generated_at )) conn.commit() logger.info(f"Project saved to database: {project_id}") return True, t("project_manager.save_success", title=project_id) except Exception as e: logger.error(f"Project save failed: {e}") return False, t("project_manager.save_failed", error=str(e)) @staticmethod def load_project(project_id: str) -> Tuple[Optional[NovelProject], str]: """ Tải dự án từ SQLite Returns: (Đối tượng dự án, Thông tin trạng thái) """ try: conn = get_db() row = conn.execute( "SELECT * FROM projects WHERE id = ?", (project_id,) ).fetchone() if not row: return None, t("project_manager.load_not_found", id=project_id) # Lấy list json string try: sg_str = row["sub_genres"] except (IndexError, KeyError): sg_str = "[]" try: sg_list = json.loads(sg_str) if sg_str else [] except: sg_list = [] # Xây dựng lại dự án project = NovelProject( title=row["title"], genre=row["genre"], sub_genres=sg_list, character_setting=row["character_setting"], world_setting=row["world_setting"], plot_idea=row["plot_idea"], created_at=row["created_at"], updated_at=row["updated_at"] ) project.id = row["id"] # Tải chapters ch_rows = conn.execute( "SELECT * FROM chapters WHERE project_id = ? ORDER BY num", (project_id,) ).fetchall() for ch_row in ch_rows: chapter = Chapter( num=ch_row["num"], title=ch_row["title"], desc=ch_row["desc"], content=ch_row["content"], word_count=ch_row["word_count"], generated_at=ch_row["generated_at"] ) project.chapters.append(chapter) logger.info(f"Project loaded from core.database: {project_id}") return project, t("project_manager.load_success") except Exception as e: logger.error(f"Project load failed: {e}") return None, t("project_manager.load_failed", error=str(e)) @staticmethod def list_projects() -> List[Dict]: """ Liệt kê tất cả dự án từ SQLite Returns: Danh sách thông tin dự án """ try: conn = get_db() rows = conn.execute( "SELECT id, title, genre, created_at, updated_at FROM projects ORDER BY updated_at DESC" ).fetchall() projects = [] for row in rows: # Đếm chapters ch_count = conn.execute( "SELECT COUNT(*) as total, SUM(CASE WHEN content != '' THEN 1 ELSE 0 END) as completed FROM chapters WHERE project_id = ?", (row["id"],) ).fetchone() projects.append({ "id": row["id"], "title": row["title"], "genre": row["genre"], "created_at": row["created_at"], "updated_at": row["updated_at"], "chapter_count": ch_count["total"] if ch_count else 0, "completed_chapters": ch_count["completed"] if ch_count else 0 }) logger.info(f"Found {len(projects)} projects in database") return projects except Exception as e: logger.error(f"List projects failed: {e}") return [] @staticmethod def get_project_by_title(project_title: str) -> Optional[Dict]: """ Lấy thông tin dự án theo tiêu đề Returns: Từ điển dự án hoặc None """ projects = ProjectManager.list_projects() for project in projects: if project.get("title") == project_title: return project return None @staticmethod def delete_project(project_id: str) -> Tuple[bool, str]: """ Xóa dự án từ SQLite Returns: (Cờ thành công (boolean), Thông tin trạng thái) """ try: conn = get_db() cursor = conn.execute("DELETE FROM projects WHERE id = ?", (project_id,)) conn.commit() if cursor.rowcount == 0: return False, t("project_manager.delete_not_found", id=project_id) logger.info(f"Project deleted from core.database: {project_id}") return True, t("project_manager.delete_success") except Exception as e: logger.error(f"Project delete failed: {e}") return False, t("project_manager.delete_failed", error=str(e)) @staticmethod def export_project(project: NovelProject, export_format: str = "json") -> Tuple[Optional[str], str]: """ Xuất cấu hình dự án (để chia sẻ hoặc sao lưu) Args: project: Đối tượng dự án export_format: Định dạng xuất (json/zip) Returns: (Đường dẫn tệp, Thông tin trạng thái) """ try: if not project: return None, "" export_dir = os.path.join("exports", "project_backups") os.makedirs(export_dir, exist_ok=True) if export_format == "json": data = { "title": project.title, "genre": project.genre, "sub_genres": project.sub_genres, "character_setting": project.character_setting, "world_setting": project.world_setting, "plot_idea": project.plot_idea, "created_at": project.created_at, "chapters": [ch.to_dict() for ch in project.chapters] } filename = f"{project.title}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" filepath = os.path.join(export_dir, filename) with open(filepath, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=4) logger.info(f"Project exported: {filename}") return filepath, t("project_manager.export_success", filepath=filename) else: return None, f"Unsupported format: {export_format}" except Exception as e: logger.error(f"Project export failed: {e}") return None, t("project_manager.export_failed", error=str(e)) def get_project_manager() -> ProjectManager: """Nhận phiên bản quản lý dự án""" return ProjectManager() def list_project_titles(): """Lấy danh sách tiêu đề dự án""" try: projects = ProjectManager.list_projects() return [p["title"] for p in projects] except Exception: return []