303 lines
10 KiB
Python
303 lines
10 KiB
Python
"""
|
|
爽文书籍 CRUD 与全局资源读取。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
import shutil
|
|
import uuid
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from core.config import settings
|
|
from models.fiction_models import (
|
|
CreateFictionBookRequest,
|
|
EmotionFlowCatalog,
|
|
FictionBookMeta,
|
|
FictionBookSettings,
|
|
FictionBookSummary,
|
|
FictionChapter,
|
|
FictionChapterSummary,
|
|
FictionGuideWorldbook,
|
|
FictionPipelineSettings,
|
|
FictionPrompts,
|
|
FictionReaderSettings,
|
|
GuideGlobalEntries,
|
|
UpdateFictionBookSettingsRequest,
|
|
)
|
|
from services.fiction_prompt_utils import USER_DEFAULT_PROMPTS
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DEFAULT_PROMPTS = FictionPrompts(
|
|
openBook=USER_DEFAULT_PROMPTS["openBook"],
|
|
coarseOutline=USER_DEFAULT_PROMPTS["coarseOutline"],
|
|
eventPlan=USER_DEFAULT_PROMPTS["eventPlan"],
|
|
chapter=USER_DEFAULT_PROMPTS["chapter"],
|
|
nudge=USER_DEFAULT_PROMPTS["nudge"],
|
|
)
|
|
|
|
DEFAULT_READER = FictionReaderSettings()
|
|
DEFAULT_PIPELINE = FictionPipelineSettings()
|
|
|
|
DEFAULT_METADATA: Dict[str, Any] = {
|
|
"version": 2,
|
|
"volumes": [],
|
|
"eventChains": {},
|
|
"chapterPlans": {},
|
|
"progress": {
|
|
"currentChapterSeq": 0,
|
|
"charOffset": 0,
|
|
"ttsPaused": False,
|
|
"genPaused": False,
|
|
},
|
|
}
|
|
|
|
DEFAULT_RUN: Dict[str, Any] = {
|
|
"status": "idle",
|
|
"pipelineStage": None,
|
|
"stage": "idle",
|
|
"message": None,
|
|
"updatedAt": "",
|
|
}
|
|
|
|
|
|
def _read_json(path: Path) -> Any:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def _write_json(path: Path, data: Any) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
|
|
|
|
|
def _slugify(text: str) -> str:
|
|
text = (text or "").strip()
|
|
text = re.sub(r"[^\w\u4e00-\u9fff\-]+", "-", text, flags=re.UNICODE)
|
|
text = re.sub(r"-+", "-", text).strip("-")
|
|
return text[:48] or "book"
|
|
|
|
|
|
class FictionService:
|
|
@property
|
|
def books_root(self) -> Path:
|
|
return settings.FICTION_BOOKS_PATH
|
|
|
|
def _book_dir(self, book_id: str) -> Path:
|
|
return self.books_root / book_id
|
|
|
|
def _meta_path(self, book_id: str) -> Path:
|
|
return self._book_dir(book_id) / "meta.json"
|
|
|
|
def _settings_path(self, book_id: str) -> Path:
|
|
return self._book_dir(book_id) / "settings.json"
|
|
|
|
def _guide_path(self, book_id: str) -> Path:
|
|
return self._book_dir(book_id) / "guide.worldbook.json"
|
|
|
|
def _metadata_path(self, book_id: str) -> Path:
|
|
return self._book_dir(book_id) / "metadata.json"
|
|
|
|
def _run_path(self, book_id: str) -> Path:
|
|
return self._book_dir(book_id) / "run.json"
|
|
|
|
def _chapters_dir(self, book_id: str) -> Path:
|
|
return self._book_dir(book_id) / "chapters"
|
|
|
|
def _chapter_path(self, book_id: str, seq: int) -> Path:
|
|
return self._chapters_dir(book_id) / f"{seq:04d}.json"
|
|
|
|
def chapter_exists(self, book_id: str, seq: int) -> bool:
|
|
return self._chapter_path(book_id, seq).exists()
|
|
|
|
def get_chapter(self, book_id: str, seq: int) -> FictionChapter:
|
|
path = self._chapter_path(book_id, seq)
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"Chapter not found: {book_id}/{seq}")
|
|
return FictionChapter(**_read_json(path))
|
|
|
|
def save_chapter(self, book_id: str, chapter: FictionChapter) -> FictionChapter:
|
|
self._ensure_book(book_id)
|
|
_write_json(self._chapter_path(book_id, chapter.seq), chapter.model_dump())
|
|
self._touch_book_meta(book_id)
|
|
return chapter
|
|
|
|
def list_written_chapter_seqs(self, book_id: str) -> List[int]:
|
|
chapters_dir = self._chapters_dir(book_id)
|
|
if not chapters_dir.exists():
|
|
return []
|
|
seqs: List[int] = []
|
|
for path in chapters_dir.glob("*.json"):
|
|
try:
|
|
seqs.append(int(path.stem))
|
|
except ValueError:
|
|
continue
|
|
return sorted(seqs)
|
|
|
|
def list_chapter_summaries(self, book_id: str) -> List[FictionChapterSummary]:
|
|
summaries: List[FictionChapterSummary] = []
|
|
for seq in self.list_written_chapter_seqs(book_id):
|
|
ch = self.get_chapter(book_id, seq)
|
|
summaries.append(
|
|
FictionChapterSummary(
|
|
seq=ch.seq,
|
|
title=ch.title,
|
|
charCount=ch.charCount,
|
|
eventId=ch.eventId,
|
|
phaseKey=ch.phaseKey,
|
|
)
|
|
)
|
|
return summaries
|
|
|
|
def _ensure_book(self, book_id: str) -> None:
|
|
if not self._meta_path(book_id).exists():
|
|
raise FileNotFoundError(f"Book not found: {book_id}")
|
|
|
|
def _touch_book_meta(self, book_id: str) -> None:
|
|
meta_path = self._meta_path(book_id)
|
|
meta = _read_json(meta_path)
|
|
meta["updatedAt"] = datetime.now().isoformat()
|
|
_write_json(meta_path, meta)
|
|
|
|
def get_default_settings(self) -> FictionBookSettings:
|
|
return FictionBookSettings(
|
|
prompts=DEFAULT_PROMPTS,
|
|
reader=DEFAULT_READER,
|
|
pipeline=DEFAULT_PIPELINE,
|
|
)
|
|
|
|
def get_emotion_catalog(self) -> EmotionFlowCatalog:
|
|
path = settings.FICTION_EMOTION_CATALOG_FILE
|
|
if not path.exists():
|
|
return EmotionFlowCatalog(flows=[])
|
|
return EmotionFlowCatalog(**_read_json(path))
|
|
|
|
def get_guide_global_entries(self) -> GuideGlobalEntries:
|
|
path = settings.FICTION_GUIDE_GLOBAL_ENTRIES_FILE
|
|
if not path.exists():
|
|
return GuideGlobalEntries(entries=[])
|
|
return GuideGlobalEntries(**_read_json(path))
|
|
|
|
def list_books(self) -> List[FictionBookSummary]:
|
|
root = self.books_root
|
|
if not root.exists():
|
|
return []
|
|
summaries: List[FictionBookSummary] = []
|
|
for child in sorted(root.iterdir()):
|
|
if not child.is_dir():
|
|
continue
|
|
meta_path = child / "meta.json"
|
|
if not meta_path.exists():
|
|
continue
|
|
meta = _read_json(meta_path)
|
|
summaries.append(
|
|
FictionBookSummary(
|
|
id=meta.get("id", child.name),
|
|
title=meta.get("title", child.name),
|
|
allowedFlowIds=meta.get("allowedFlowIds", []),
|
|
updatedAt=meta.get("updatedAt", ""),
|
|
)
|
|
)
|
|
summaries.sort(key=lambda x: x.updatedAt or "", reverse=True)
|
|
return summaries
|
|
|
|
def get_book_meta(self, book_id: str) -> FictionBookMeta:
|
|
meta_path = self._meta_path(book_id)
|
|
if not meta_path.exists():
|
|
raise FileNotFoundError(f"Book not found: {book_id}")
|
|
return FictionBookMeta(**_read_json(meta_path))
|
|
|
|
def get_book_settings(self, book_id: str) -> FictionBookSettings:
|
|
settings_path = self._settings_path(book_id)
|
|
if not settings_path.exists():
|
|
raise FileNotFoundError(f"Book not found: {book_id}")
|
|
return FictionBookSettings(**_read_json(settings_path))
|
|
|
|
def update_book_settings(
|
|
self, book_id: str, req: UpdateFictionBookSettingsRequest
|
|
) -> FictionBookSettings:
|
|
meta_path = self._meta_path(book_id)
|
|
settings_path = self._settings_path(book_id)
|
|
if not meta_path.exists():
|
|
raise FileNotFoundError(f"Book not found: {book_id}")
|
|
current = FictionBookSettings(**_read_json(settings_path))
|
|
data = current.model_dump()
|
|
if req.prompts is not None:
|
|
data["prompts"] = req.prompts.model_dump()
|
|
if req.reader is not None:
|
|
data["reader"] = req.reader.model_dump()
|
|
if req.pipeline is not None:
|
|
data["pipeline"] = req.pipeline.model_dump()
|
|
_write_json(settings_path, data)
|
|
meta = _read_json(meta_path)
|
|
meta["updatedAt"] = datetime.now().isoformat()
|
|
_write_json(meta_path, meta)
|
|
return FictionBookSettings(**data)
|
|
|
|
def get_book_guide(self, book_id: str) -> FictionGuideWorldbook:
|
|
guide_path = self._guide_path(book_id)
|
|
if not guide_path.exists():
|
|
raise FileNotFoundError(f"Book not found: {book_id}")
|
|
return FictionGuideWorldbook(**_read_json(guide_path))
|
|
|
|
def _unique_book_id(self, base_id: str) -> str:
|
|
candidate = base_id
|
|
n = 1
|
|
while self._book_dir(candidate).exists():
|
|
candidate = f"{base_id}-{n}"
|
|
n += 1
|
|
return candidate
|
|
|
|
def create_book(self, req: CreateFictionBookRequest) -> FictionBookMeta:
|
|
title = (req.title or "").strip()
|
|
if not title:
|
|
raise ValueError("书名不能为空")
|
|
|
|
base_id = _slugify(title)
|
|
if not base_id or base_id == "book":
|
|
base_id = str(uuid.uuid4())[:8]
|
|
book_id = self._unique_book_id(base_id)
|
|
|
|
dest = self._book_dir(book_id)
|
|
dest.mkdir(parents=True, exist_ok=False)
|
|
self._chapters_dir(book_id).mkdir(parents=True, exist_ok=True)
|
|
|
|
now = datetime.now().isoformat()
|
|
meta = {
|
|
"id": book_id,
|
|
"title": title,
|
|
"allowedFlowIds": list(req.allowedFlowIds or []),
|
|
"createdAt": now,
|
|
"updatedAt": now,
|
|
}
|
|
_write_json(self._meta_path(book_id), meta)
|
|
|
|
default_settings = self.get_default_settings()
|
|
_write_json(self._settings_path(book_id), default_settings.model_dump())
|
|
|
|
guide = req.guide.model_dump() if req.guide else FictionGuideWorldbook().model_dump()
|
|
_write_json(self._guide_path(book_id), guide)
|
|
|
|
metadata = dict(DEFAULT_METADATA)
|
|
_write_json(self._metadata_path(book_id), metadata)
|
|
|
|
run_data = dict(DEFAULT_RUN)
|
|
run_data["updatedAt"] = now
|
|
_write_json(self._run_path(book_id), run_data)
|
|
|
|
return FictionBookMeta(**meta)
|
|
|
|
def delete_book(self, book_id: str) -> None:
|
|
book_dir = self._book_dir(book_id)
|
|
if not book_dir.exists():
|
|
raise FileNotFoundError(f"Book not found: {book_id}")
|
|
shutil.rmtree(book_dir)
|
|
|
|
|
|
fiction_service = FictionService()
|