Files
admin 46af290db3
Build and Push Docker Images / build-and-push (push) Failing after 3m33s
initial commit
2026-11-07 16:39:43 +03:00

303 lines
11 KiB
Python

"""
Сервис для работы с документами (иерархическая структура: Книга → Глава → Сцена)
Документы хранятся в папке проекта: data/projects/{project_id}/documents.json
"""
import json
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional
from uuid import uuid4
from models.document import DocumentNode
from services.project_service import project_service
class DocumentService:
def __init__(self):
self.project_service = project_service
def _get_documents_file(self, project_id: str, user_id: str) -> Path:
project_dir = self.project_service.get_project_dir(project_id, user_id)
return project_dir / "documents.json"
def _load_documents(self, project_id: str, user_id: str) -> List[DocumentNode]:
documents_file = self._get_documents_file(project_id, user_id)
if documents_file.exists():
with open(documents_file, "r", encoding="utf-8") as f:
data = json.load(f)
# Старый формат: dict, где ключи — ID документов
if isinstance(data, dict):
flat_list = list(data.values())
if flat_list and len(flat_list) > 0:
return self._migrate_flat_to_hierarchical(
flat_list, project_id, user_id
)
return []
# Новый формат: list
if isinstance(data, list) and len(data) > 0:
# Миграция: если старые данные без parent_id
first = data[0]
if isinstance(first, dict) and "parent_id" not in first:
return self._migrate_flat_to_hierarchical(
data, project_id, user_id
)
return [DocumentNode(**doc) for doc in data]
return []
def _migrate_flat_to_hierarchical(
self, old_data: List[Dict], project_id: str, user_id: str
) -> List[DocumentNode]:
"""Миграция плоских документов в иерархию"""
book = DocumentNode(
project_id=project_id,
title="Книга 1",
type="book",
order=0,
)
chapter = DocumentNode(
project_id=project_id,
parent_id=book.id,
title="Глава 1",
type="chapter",
order=0,
)
scenes = []
for i, old_doc in enumerate(old_data):
scene = DocumentNode(
id=old_doc.get("id", str(uuid4())),
project_id=project_id,
parent_id=chapter.id,
title=old_doc.get("title", f"Сцена {i + 1}"),
content=old_doc.get("content", ""),
type="scene",
order=i,
)
scenes.append(scene)
result = [book, chapter] + scenes
self._save_documents(project_id, result, user_id)
print(f"Migrated {len(old_data)} flat documents to hierarchical structure")
return result
def _save_documents(
self, project_id: str, documents: List[DocumentNode], user_id: str
):
documents_file = self._get_documents_file(project_id, user_id)
with open(documents_file, "w", encoding="utf-8") as f:
json.dump(
[doc.model_dump() for doc in documents],
f,
ensure_ascii=False,
indent=2,
)
def get_documents_tree(self, project_id: str, user_id: str) -> List[Dict[str, Any]]:
"""Получить документы в виде дерева"""
documents = self._load_documents(project_id, user_id)
return self._build_tree(documents)
def _build_tree(self, documents: List[DocumentNode]) -> List[Dict[str, Any]]:
"""Строит дерево из плоского списка"""
doc_map = {doc.id: {**doc.model_dump(), "children": []} for doc in documents}
roots = []
for doc in documents:
node = doc_map[doc.id]
if doc.parent_id is None:
roots.append(node)
elif doc.parent_id in doc_map:
doc_map[doc.parent_id]["children"].append(node)
def sort_tree(nodes):
nodes.sort(key=lambda x: x["order"])
for node in nodes:
sort_tree(node["children"])
sort_tree(roots)
return roots
def get_all_documents(self, project_id: str, user_id: str) -> List[DocumentNode]:
"""Получить все документы (плоский список)"""
return self._load_documents(project_id, user_id)
def get_document(
self, project_id: str, document_id: str, user_id: str
) -> Optional[DocumentNode]:
documents = self._load_documents(project_id, user_id)
for doc in documents:
if doc.id == document_id:
return doc
return None
def create_document(
self,
project_id: str,
title: str,
doc_type: str,
user_id: str,
parent_id: Optional[str] = None,
content: str = "",
) -> DocumentNode:
"""Создать новый документ с валидацией иерархии"""
documents = self._load_documents(project_id, user_id)
# Валидация иерархии
if doc_type == "chapter" and parent_id:
parent = next((d for d in documents if d.id == parent_id), None)
if not parent or parent.type != "book":
raise ValueError("Глава может быть создана только внутри книги")
if doc_type == "scene" and parent_id:
parent = next((d for d in documents if d.id == parent_id), None)
if not parent or parent.type != "chapter":
raise ValueError("Сцена может быть создана только внутри главы")
if doc_type == "book" and parent_id:
raise ValueError("Книга не может иметь родителя")
# Вычисляем order (последний среди братьев + 1)
siblings = [d for d in documents if d.parent_id == parent_id]
max_order = max([s.order for s in siblings], default=-1)
document = DocumentNode(
project_id=project_id,
parent_id=parent_id,
title=title,
content=content,
type=doc_type,
order=max_order + 1,
)
documents.append(document)
self._save_documents(project_id, documents, user_id)
return document
def update_document(
self,
project_id: str,
document_id: str,
user_id: str,
title: Optional[str] = None,
content: Optional[str] = None,
parent_id: Optional[str] = None,
order: Optional[int] = None,
) -> Optional[DocumentNode]:
"""Обновить документ (включая перенос)"""
documents = self._load_documents(project_id, user_id)
for i, doc in enumerate(documents):
if doc.id == document_id:
# Валидация при переносе
if parent_id is not None and parent_id != doc.parent_id:
new_parent = next((d for d in documents if d.id == parent_id), None)
if doc.type == "book":
raise ValueError("Книга не может быть перемещена")
if doc.type == "chapter" and (
not new_parent or new_parent.type != "book"
):
raise ValueError("Глава может находиться только внутри книги")
if doc.type == "scene" and (
not new_parent or new_parent.type != "chapter"
):
raise ValueError("Сцена может находиться только внутри главы")
documents[i].parent_id = parent_id
if title is not None:
documents[i].title = title
if content is not None:
documents[i].content = content
if order is not None:
documents[i].order = order
documents[i].updated_at = datetime.now().isoformat()
self._save_documents(project_id, documents, user_id)
return documents[i]
return None
def delete_document(self, project_id: str, document_id: str, user_id: str) -> bool:
"""Удалить документ с каскадом (дети тоже удаляются)"""
documents = self._load_documents(project_id, user_id)
target = next((d for d in documents if d.id == document_id), None)
if not target:
return False
ids_to_delete: set[str] = set()
def collect_descendants(parent_id: str):
ids_to_delete.add(parent_id)
children = [d for d in documents if d.parent_id == parent_id]
for child in children:
collect_descendants(child.id)
collect_descendants(document_id)
documents = [d for d in documents if d.id not in ids_to_delete]
self._save_documents(project_id, documents, user_id)
return True
def get_children(
self, project_id: str, parent_id: str, user_id: str
) -> List[DocumentNode]:
"""Получить прямых детей узла"""
documents = self._load_documents(project_id, user_id)
children = [d for d in documents if d.parent_id == parent_id]
children.sort(key=lambda x: x.order)
return children
def initialize_test_document(
self, project_id: str, user_id: str
) -> List[DocumentNode]:
"""
Создаёт тестовую иерархию документов при первом запуске,
если файл документов пуст.
"""
documents = self._load_documents(project_id, user_id)
if documents:
return documents
book = DocumentNode(
project_id=project_id,
title="Книга 1",
type="book",
order=0,
)
chapter = DocumentNode(
project_id=project_id,
parent_id=book.id,
title="Глава 1",
type="chapter",
order=0,
)
scene = DocumentNode(
project_id=project_id,
parent_id=chapter.id,
title="Сцена 1",
content="Это тестовый документ для разработки Writer IDE. Здесь будет располагаться контент ваших сценариев.",
type="scene",
order=0,
)
result = [book, chapter, scene]
self._save_documents(project_id, result, user_id)
print(f"Created test hierarchical documents for project: {project_id}")
return result
# Singleton instance
document_service = DocumentService()