"""Story business logic service.""" import asyncio from dataclasses import dataclass from typing import Literal from fastapi import HTTPException from sqlalchemy import desc, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import joinedload from app.core.logging import get_logger from app.db.models import ChildProfile, Story, StoryUniverse from app.schemas.story_schemas import ( AchievementItem, FullStoryResponse, GenerateRequest, GenerationRequest, GenerationResponse, StorybookPageResponse, StorybookRequest, StorybookResponse, ) from app.services.adapters.storybook.primary import Storybook from app.services.adapters.text.models import StoryOutput from app.services.audio_storage import ( audio_cache_exists, delete_audio_cache, get_audio_cache_metadata, read_audio_cache, write_story_audio_cache, ) from app.services.generation_jobs import ( create_generation_job, finish_generation_job, record_generation_event, ) from app.services.memory_service import build_enhanced_memory_context from app.services.provider_router import ( generate_image, generate_story_content, generate_storybook, ) from app.services.story_status import ( StoryAssetStatus, sync_story_status, ) from app.tasks.achievements import extract_story_achievements logger = get_logger(__name__) AssetCompletionKind = Literal["cover_image", "storybook_images", "audio"] @dataclass(frozen=True) class AssetCompletionResult: """Service-level result for a generated asset completion attempt.""" asset: AssetCompletionKind status: StoryAssetStatus value: str | bytes | None = None error: str | None = None blocks_main_result: bool = False @property def succeeded(self) -> bool: """Whether the asset reached a usable ready state.""" return self.status == StoryAssetStatus.READY and self.error is None async def _record_job_event_if_present( db: AsyncSession, *, job, event_type: str, status: str, story_id: int | None = None, message: str | None = None, metadata: dict | None = None, ) -> None: """Append a workflow event when the caller is running under a tracked job.""" if job is None: return await record_generation_event( db, job=job, story_id=story_id, event_type=event_type, status=status, message=message, metadata=metadata, ) def _asset_result_metadata(result: AssetCompletionResult) -> dict: """Build JSON-safe metadata for asset workflow events.""" return { "asset": result.asset, "status": result.status.value, "error": result.error, "blocks_main_result": result.blocks_main_result, } def _build_storybook_error_message( *, cover_failed: bool, failed_pages: list[int], ) -> str | None: """Summarize storybook image generation errors for the latest attempt.""" parts: list[str] = [] if cover_failed: parts.append("封面生成失败") if failed_pages: pages = "、".join(str(page) for page in sorted(failed_pages)) parts.append(f"第 {pages} 页插图生成失败") return "；".join(parts) if parts else None def _resolve_storybook_image_status( *, generate_images: bool, cover_prompt: str | None, cover_url: str | None, pages_data: list[dict], ) -> StoryAssetStatus: """Resolve the persisted image status for a storybook.""" if not generate_images: return StoryAssetStatus.NOT_REQUESTED expected_assets = 0 ready_assets = 0 if cover_prompt or cover_url: expected_assets += 1 if cover_url: ready_assets += 1 for page in pages_data: if not page.get("image_prompt") and not page.get("image_url"): continue expected_assets += 1 if page.get("image_url"): ready_assets += 1 if expected_assets == 0: return StoryAssetStatus.NOT_REQUESTED if ready_assets == expected_assets: return StoryAssetStatus.READY return StoryAssetStatus.FAILED async def _prepare_generation_context( *, profile_id: str | None, universe_id: str | None, user_id: str, db: AsyncSession, job=None, ) -> tuple[str | None, str | None, str]: """Validate ownership and build the shared generation context.""" resolved_profile_id, resolved_universe_id = await validate_profile_and_universe( profile_id, universe_id, user_id, db ) memory_context = await build_enhanced_memory_context( resolved_profile_id, resolved_universe_id, db, ) await _record_job_event_if_present( db, job=job, event_type="context_prepared", status="succeeded", message="Profile, universe, and memory context were prepared.", metadata={ "profile_id": resolved_profile_id, "universe_id": resolved_universe_id, "has_memory_context": bool(memory_context), }, ) return resolved_profile_id, resolved_universe_id, memory_context def _trigger_story_postprocessing(story: Story) -> None: """Trigger non-blocking post-processing for a persisted story.""" if story.universe_id: extract_story_achievements.delay(story.id, story.universe_id) async def _record_postprocessing_event_if_needed( db: AsyncSession, *, job, story: Story, ) -> None: if not story.universe_id: return await record_generation_event( db, job=job, story_id=story.id, event_type="postprocessing_queued", status="queued", message="Achievement extraction queued after the main story record was saved.", metadata={"universe_id": story.universe_id}, ) async def _persist_text_story_result( *, result: StoryOutput, user_id: str, profile_id: str | None, universe_id: str | None, db: AsyncSession, job=None, ) -> Story: """Persist generated text content as the unified story record.""" story = Story( user_id=user_id, child_profile_id=profile_id, universe_id=universe_id, title=result.title, story_text=result.story_text, cover_prompt=result.cover_prompt_suggestion, mode=result.mode, ) sync_story_status( story, image_status=StoryAssetStatus.NOT_REQUESTED, audio_status=StoryAssetStatus.NOT_REQUESTED, last_error=None, ) db.add(story) await db.commit() await db.refresh(story) _trigger_story_postprocessing(story) await _record_job_event_if_present( db, job=job, story_id=story.id, event_type="story_saved", status="succeeded", message="Readable story record was saved.", metadata={ "mode": story.mode, "generation_status": story.generation_status, "image_status": story.image_status, "audio_status": story.audio_status, }, ) return story def _storybook_pages_to_data(storybook: Storybook) -> list[dict]: """Convert generated storybook pages to the persisted JSON shape.""" return [ { "page_number": page.page_number, "text": page.text, "image_prompt": page.image_prompt, "image_url": page.image_url, } for page in storybook.pages ] def _storybook_pages_to_response(pages_data: list[dict]) -> list[StorybookPageResponse]: """Convert persisted storybook page JSON to API response models.""" return [ StorybookPageResponse( page_number=page["page_number"], text=page["text"], image_prompt=page["image_prompt"], image_url=page.get("image_url"), ) for page in pages_data ] async def _generate_storybook_image_assets( storybook: Storybook, db: AsyncSession, *, user_id: str, job=None, ) -> tuple[str | None, bool, list[int]]: """Generate storybook cover and page images before persistence.""" final_cover_url = storybook.cover_url cover_failed = False failed_pages: list[int] = [] completed_pages: list[int] = [] attempted_cover = bool(storybook.cover_prompt and not storybook.cover_url) attempted_pages = [ page.page_number for page in storybook.pages if page.image_prompt and not page.image_url ] logger.info("storybook_parallel_generation_start", page_count=len(storybook.pages)) await _record_job_event_if_present( db, job=job, event_type="storybook_images_started", status="running", message="Storybook cover and page image generation started.", metadata={ "attempted_cover": attempted_cover, "attempted_pages": attempted_pages, }, ) async def _gen_cover() -> str | None: nonlocal cover_failed if storybook.cover_prompt and not storybook.cover_url: try: return await generate_image( storybook.cover_prompt, db=db, user_id=user_id, generation_job=job, ) except Exception as exc: cover_failed = True logger.warning("cover_gen_failed", error=str(exc)) return storybook.cover_url async def _gen_page(page) -> None: if not page.image_prompt or page.image_url: return try: page.image_url = await generate_image( page.image_prompt, db=db, user_id=user_id, generation_job=job, ) completed_pages.append(page.page_number) except Exception as exc: failed_pages.append(page.page_number) logger.warning("page_gen_failed", page=page.page_number, error=str(exc)) results = await asyncio.gather( _gen_cover(), *(_gen_page(page) for page in storybook.pages), return_exceptions=True, ) cover_result = results[0] if isinstance(cover_result, str): final_cover_url = cover_result logger.info("storybook_parallel_generation_complete") if attempted_cover: await _record_job_event_if_present( db, job=job, event_type=( "storybook_cover_image_failed" if cover_failed else "storybook_cover_image_succeeded" ), status="failed" if cover_failed else "succeeded", message=( "Storybook cover image generation failed." if cover_failed else "Storybook cover image was generated." ), metadata={"asset": "image", "scope": "cover"}, ) for page_number in sorted(completed_pages): await _record_job_event_if_present( db, job=job, event_type="storybook_page_image_succeeded", status="succeeded", message="Storybook page image was generated.", metadata={"asset": "image", "scope": "page", "page_number": page_number}, ) for page_number in sorted(failed_pages): await _record_job_event_if_present( db, job=job, event_type="storybook_page_image_failed", status="failed", message="Storybook page image generation failed.", metadata={"asset": "image", "scope": "page", "page_number": page_number}, ) await _record_job_event_if_present( db, job=job, event_type="storybook_images_completed", status="failed" if cover_failed or failed_pages else "succeeded", message="Storybook image generation finished.", metadata={ "asset": "image", "attempted_cover": attempted_cover, "completed_pages": sorted(completed_pages), "failed_pages": sorted(failed_pages), }, ) return final_cover_url, cover_failed, failed_pages async def _persist_storybook_result( *, storybook: Storybook, user_id: str, profile_id: str | None, universe_id: str | None, final_cover_url: str | None, generate_images: bool, cover_failed: bool, failed_pages: list[int], db: AsyncSession, job=None, ) -> tuple[Story, list[dict]]: """Persist generated storybook content as the unified story record.""" pages_data = _storybook_pages_to_data(storybook) story = Story( user_id=user_id, child_profile_id=profile_id, universe_id=universe_id, title=storybook.title, mode="storybook", pages=pages_data, story_text=None, cover_prompt=storybook.cover_prompt, image_url=final_cover_url, ) sync_story_status( story, image_status=_resolve_storybook_image_status( generate_images=generate_images, cover_prompt=storybook.cover_prompt, cover_url=final_cover_url, pages_data=pages_data, ), audio_status=StoryAssetStatus.NOT_REQUESTED, last_error=_build_storybook_error_message( cover_failed=cover_failed, failed_pages=failed_pages, ), ) db.add(story) await db.commit() await db.refresh(story) _trigger_story_postprocessing(story) await _record_job_event_if_present( db, job=job, story_id=story.id, event_type="story_saved", status="succeeded", message="Storybook record was saved.", metadata={ "mode": story.mode, "page_count": len(pages_data), "generation_status": story.generation_status, "image_status": story.image_status, "audio_status": story.audio_status, }, ) return story, pages_data async def _complete_cover_image_asset( story: Story, db: AsyncSession, *, raise_on_failure: bool = False, last_error_prefix: str | None = None, log_event: str = "cover_asset_generation_failed", job=None, ) -> AssetCompletionResult: """Generate or retry a text story cover through one asset workflow.""" if not story.cover_prompt: raise HTTPException(status_code=400, detail="Story has no cover prompt") sync_story_status(story, image_status=StoryAssetStatus.GENERATING) await db.commit() await _record_job_event_if_present( db, job=job, story_id=story.id, event_type="cover_image_started", status="running", message="Cover image generation started.", metadata={"asset": "image", "cover_prompt_present": True}, ) try: image_url = await generate_image( story.cover_prompt, db=db, user_id=story.user_id, generation_job=job, story_id=story.id, ) story.image_url = image_url sync_story_status(story, image_status=StoryAssetStatus.READY) await db.commit() result = AssetCompletionResult( asset="cover_image", status=StoryAssetStatus.READY, value=image_url, blocks_main_result=raise_on_failure, ) await _record_job_event_if_present( db, job=job, story_id=story.id, event_type="cover_image_succeeded", status="succeeded", message="Cover image was generated.", metadata=_asset_result_metadata(result), ) return result except Exception as exc: provider_error = str(exc) last_error = ( f"{last_error_prefix}: {provider_error}" if last_error_prefix else provider_error ) sync_story_status( story, image_status=StoryAssetStatus.FAILED, last_error=last_error, ) await db.commit() logger.warning(log_event, story_id=story.id, error=provider_error) result = AssetCompletionResult( asset="cover_image", status=StoryAssetStatus.FAILED, error=provider_error, blocks_main_result=raise_on_failure, ) await _record_job_event_if_present( db, job=job, story_id=story.id, event_type="cover_image_failed", status="failed", message="Cover image generation failed.", metadata=_asset_result_metadata(result), ) if raise_on_failure: raise HTTPException( status_code=500, detail=f"Image generation failed: {provider_error}", ) from exc return result def _get_storybook_pages_data(story: Story) -> list[dict]: """Return mutable storybook page data from the persisted JSON field.""" return [dict(page) for page in story.pages or [] if isinstance(page, dict)] async def _complete_storybook_image_assets( story: Story, db: AsyncSession, *, job=None, ) -> AssetCompletionResult: """Complete missing cover/page images for a persisted storybook.""" pages_data = _get_storybook_pages_data(story) has_image_prompt = bool(story.cover_prompt) or any( page.get("image_prompt") for page in pages_data ) if not has_image_prompt: raise HTTPException(status_code=400, detail="Storybook has no image prompts") sync_story_status(story, image_status=StoryAssetStatus.GENERATING) await db.commit() await _record_job_event_if_present( db, job=job, story_id=story.id, event_type="storybook_images_started", status="running", message="Storybook missing image completion started.", metadata={"asset": "image"}, ) cover_failed = False failed_pages: list[int] = [] completed_pages: list[int] = [] if story.cover_prompt and not story.image_url: try: story.image_url = await generate_image( story.cover_prompt, db=db, user_id=story.user_id, generation_job=job, story_id=story.id, ) await _record_job_event_if_present( db, job=job, story_id=story.id, event_type="storybook_cover_image_succeeded", status="succeeded", message="Storybook cover image was generated.", metadata={"asset": "image", "scope": "cover"}, ) except Exception as exc: cover_failed = True logger.warning( "storybook_cover_asset_completion_failed", story_id=story.id, error=str(exc), ) await _record_job_event_if_present( db, job=job, story_id=story.id, event_type="storybook_cover_image_failed", status="failed", message="Storybook cover image generation failed.", metadata={"asset": "image", "scope": "cover", "error": str(exc)}, ) for page in pages_data: if not page.get("image_prompt") or page.get("image_url"): continue try: page["image_url"] = await generate_image( page["image_prompt"], db=db, user_id=story.user_id, generation_job=job, story_id=story.id, ) page_number = page.get("page_number") if isinstance(page_number, int): completed_pages.append(page_number) await _record_job_event_if_present( db, job=job, story_id=story.id, event_type="storybook_page_image_succeeded", status="succeeded", message="Storybook page image was generated.", metadata={"asset": "image", "scope": "page", "page_number": page_number}, ) except Exception as exc: page_number = page.get("page_number") if isinstance(page_number, int): failed_pages.append(page_number) logger.warning( "storybook_page_asset_completion_failed", story_id=story.id, page=page_number, error=str(exc), ) await _record_job_event_if_present( db, job=job, story_id=story.id, event_type="storybook_page_image_failed", status="failed", message="Storybook page image generation failed.", metadata={ "asset": "image", "scope": "page", "page_number": page_number, "error": str(exc), }, ) story.pages = pages_data error_message = _build_storybook_error_message( cover_failed=cover_failed, failed_pages=failed_pages, ) image_status = _resolve_storybook_image_status( generate_images=True, cover_prompt=story.cover_prompt, cover_url=story.image_url, pages_data=pages_data, ) sync_story_status( story, image_status=image_status, last_error=error_message, ) await db.commit() result = AssetCompletionResult( asset="storybook_images", status=image_status, value=story.image_url, error=error_message, ) await _record_job_event_if_present( db, job=job, story_id=story.id, event_type="storybook_images_completed", status="failed" if error_message else "succeeded", message="Storybook image completion finished.", metadata={ **_asset_result_metadata(result), "completed_pages": sorted(completed_pages), "failed_pages": sorted(failed_pages), }, ) return result async def _read_cached_audio_asset(story: Story, db: AsyncSession) -> bytes | None: """Read cached audio or repair stale audio cache metadata.""" if story.audio_path and audio_cache_exists(story.audio_path): if story.audio_status != StoryAssetStatus.READY.value: sync_story_status(story, audio_status=StoryAssetStatus.READY) await db.commit() return read_audio_cache(story.audio_path) if story.audio_path and not audio_cache_exists(story.audio_path): logger.warning( "story_audio_cache_missing", story_id=story.id, audio_path=story.audio_path, ) story.audio_path = None if story.audio_status == StoryAssetStatus.READY.value: sync_story_status(story, audio_status=StoryAssetStatus.NOT_REQUESTED) await db.commit() return None async def _complete_audio_asset( story: Story, db: AsyncSession, *, raise_on_failure: bool = True, job=None, ) -> AssetCompletionResult: """Complete TTS audio generation through one asset workflow.""" if not story.story_text: raise HTTPException(status_code=400, detail="Story has no text") cached_audio = await _read_cached_audio_asset(story, db) if cached_audio is not None: result = AssetCompletionResult( asset="audio", status=StoryAssetStatus.READY, value=cached_audio, blocks_main_result=raise_on_failure, ) await _record_job_event_if_present( db, job=job, story_id=story.id, event_type="audio_cache_hit", status="succeeded", message="Cached story audio was reused.", metadata=_asset_result_metadata(result), ) return result from app.services.provider_router import text_to_speech sync_story_status(story, audio_status=StoryAssetStatus.GENERATING) await db.commit() await _record_job_event_if_present( db, job=job, story_id=story.id, event_type="audio_started", status="running", message="Story audio generation started.", metadata={"asset": "audio"}, ) try: audio_data = await text_to_speech( story.story_text, db=db, user_id=story.user_id, generation_job=job, story_id=story.id, ) story.audio_path = write_story_audio_cache(story.id, audio_data) sync_story_status( story, audio_status=StoryAssetStatus.READY, ) await db.commit() result = AssetCompletionResult( asset="audio", status=StoryAssetStatus.READY, value=audio_data, blocks_main_result=raise_on_failure, ) await _record_job_event_if_present( db, job=job, story_id=story.id, event_type="audio_succeeded", status="succeeded", message="Story audio was generated and cached.", metadata=_asset_result_metadata(result), ) return result except Exception as exc: provider_error = str(exc) story.audio_path = None sync_story_status( story, audio_status=StoryAssetStatus.FAILED, last_error=provider_error, ) await db.commit() logger.error("audio_generation_failed", story_id=story.id, error=provider_error) result = AssetCompletionResult( asset="audio", status=StoryAssetStatus.FAILED, error=provider_error, blocks_main_result=raise_on_failure, ) await _record_job_event_if_present( db, job=job, story_id=story.id, event_type="audio_failed", status="failed", message="Story audio generation failed.", metadata=_asset_result_metadata(result), ) if raise_on_failure: raise HTTPException( status_code=500, detail=f"Audio generation failed: {provider_error}", ) from exc return result async def validate_profile_and_universe( profile_id: str | None, universe_id: str | None, user_id: str, db: AsyncSession, ) -> tuple[str | None, str | None]: """Validate child profile and universe ownership/relationship.""" if not profile_id and not universe_id: return None, None if profile_id: result = await db.execute( select(ChildProfile).where( ChildProfile.id == profile_id, ChildProfile.user_id == user_id, ) ) profile = result.scalar_one_or_none() if not profile: raise HTTPException(status_code=404, detail="孩子档案不存在") if universe_id: result = await db.execute( select(StoryUniverse) .join(ChildProfile, StoryUniverse.child_profile_id == ChildProfile.id) .where( StoryUniverse.id == universe_id, ChildProfile.user_id == user_id, ) ) universe = result.scalar_one_or_none() if not universe: raise HTTPException(status_code=404, detail="故事宇宙不存在") if profile_id and universe.child_profile_id != profile_id: raise HTTPException(status_code=400, detail="故事宇宙与孩子档案不匹配") if not profile_id: profile_id = universe.child_profile_id return profile_id, universe_id async def generate_and_save_story( request: GenerateRequest, user_id: str, db: AsyncSession, *, job=None, ) -> Story: """Generate generic story content and save to DB.""" profile_id, universe_id, memory_context = await _prepare_generation_context( profile_id=request.child_profile_id, universe_id=request.universe_id, user_id=user_id, db=db, job=job, ) try: result = await generate_story_content( input_type=request.type, data=request.data, education_theme=request.education_theme, memory_context=memory_context, db=db, user_id=user_id, generation_job=job, ) except Exception as exc: raise HTTPException( status_code=502, detail="Story generation failed, please try again.", ) from exc await _record_job_event_if_present( db, job=job, event_type="narrative_generated", status="succeeded", message="Story narrative was generated.", metadata={"mode": result.mode, "title": result.title}, ) return await _persist_text_story_result( result=result, user_id=user_id, profile_id=profile_id, universe_id=universe_id, db=db, job=job, ) async def generate_full_story_service( request: GenerateRequest, user_id: str, db: AsyncSession, *, job=None, ) -> FullStoryResponse: """Generate story with parallel image generation.""" story = await generate_and_save_story(request, user_id, db, job=job) image_url: str | None = None errors: dict[str, str | None] = {} if story.cover_prompt: image_result = await _complete_cover_image_asset( story, db, log_event="image_generation_failed", job=job, ) if image_result.succeeded and isinstance(image_result.value, str): image_url = image_result.value if image_result.error: errors["image"] = image_result.error return FullStoryResponse( id=story.id, title=story.title, story_text=story.story_text, cover_prompt=story.cover_prompt, image_url=image_url, audio_ready=False, mode=story.mode, errors=errors, child_profile_id=story.child_profile_id, universe_id=story.universe_id, generation_status=story.generation_status, text_status=story.text_status, image_status=story.image_status, audio_status=story.audio_status, last_error=story.last_error, retryable_assets=story.retryable_assets, ) async def generate_storybook_service( request: StorybookRequest, user_id: str, db: AsyncSession, *, job=None, ) -> StorybookResponse: """Generate storybook with parallel image generation for pages.""" profile_id, universe_id, memory_context = await _prepare_generation_context( profile_id=request.child_profile_id, universe_id=request.universe_id, user_id=user_id, db=db, job=job, ) logger.info( "storybook_request", user_id=user_id, keywords=request.keywords, page_count=request.page_count, profile_id=profile_id, universe_id=universe_id, ) try: storybook = await generate_storybook( keywords=request.keywords, page_count=request.page_count, education_theme=request.education_theme, memory_context=memory_context, db=db, user_id=user_id, generation_job=job, ) except Exception as e: logger.error("storybook_generation_failed", error=str(e)) raise HTTPException(status_code=500, detail=f"故事书生成失败: {e}") await _record_job_event_if_present( db, job=job, event_type="narrative_generated", status="succeeded", message="Storybook narrative and page plan were generated.", metadata={ "mode": "storybook", "title": storybook.title, "page_count": len(storybook.pages), }, ) final_cover_url = storybook.cover_url cover_failed = False failed_pages: list[int] = [] if request.generate_images: ( final_cover_url, cover_failed, failed_pages, ) = await _generate_storybook_image_assets( storybook, db, user_id=user_id, job=job, ) story, pages_data = await _persist_storybook_result( storybook=storybook, user_id=user_id, profile_id=profile_id, universe_id=universe_id, final_cover_url=final_cover_url, generate_images=request.generate_images, cover_failed=cover_failed, failed_pages=failed_pages, db=db, job=job, ) response_pages = _storybook_pages_to_response(pages_data) return StorybookResponse( id=story.id, title=storybook.title, main_character=storybook.main_character, art_style=storybook.art_style, pages=response_pages, cover_prompt=storybook.cover_prompt, cover_url=final_cover_url, generation_status=story.generation_status, text_status=story.text_status, image_status=story.image_status, audio_status=story.audio_status, last_error=story.last_error, retryable_assets=story.retryable_assets, ) async def generate_generation_service( request: GenerationRequest, user_id: str, db: AsyncSession, ) -> GenerationResponse: """Unified generation workflow entry point for stories and storybooks.""" job = await create_generation_job( db, user_id=user_id, output_mode=request.output_mode, input_type=request.type, request_payload=request.model_dump(mode="json"), ) try: response = await _generate_generation_service_with_job(request, user_id, db, job=job) except HTTPException as exc: await finish_generation_job( db, job=job, story=None, status="failed", current_step="generation_failed", error_message=str(exc.detail), message="Generation failed before a durable story result was available.", ) raise except Exception as exc: await finish_generation_job( db, job=job, story=None, status="failed", current_step="generation_failed", error_message=str(exc), message="Generation failed before a durable story result was available.", ) raise return response async def _generate_generation_service_with_job( request: GenerationRequest, user_id: str, db: AsyncSession, *, job, ) -> GenerationResponse: """Run the unified generation workflow after the tracking job has been created.""" if request.output_mode == "storybook": storybook = await generate_storybook_service( StorybookRequest( keywords=request.data, page_count=request.page_count, education_theme=request.education_theme, generate_images=request.generate_images, child_profile_id=request.child_profile_id, universe_id=request.universe_id, ), user_id, db, job=job, ) if storybook.id is None: raise HTTPException(status_code=500, detail="Storybook generation did not persist.") saved_story = await get_story_detail(storybook.id, user_id, db) await _record_postprocessing_event_if_needed(db, job=job, story=saved_story) await finish_generation_job( db, job=job, story=saved_story, current_step="generation_completed", message="Storybook generation completed with persisted text and current asset states.", ) return GenerationResponse( id=storybook.id, generation_job_id=job.id, title=storybook.title, mode="storybook", pages=storybook.pages, cover_prompt=storybook.cover_prompt, image_url=storybook.cover_url, cover_url=storybook.cover_url, main_character=storybook.main_character, art_style=storybook.art_style, generation_status=storybook.generation_status, text_status=saved_story.text_status, image_status=storybook.image_status, audio_status=storybook.audio_status, last_error=storybook.last_error, child_profile_id=saved_story.child_profile_id, universe_id=saved_story.universe_id, retryable_assets=saved_story.retryable_assets, ) generate_request = GenerateRequest( type=request.type, data=request.data, education_theme=request.education_theme, child_profile_id=request.child_profile_id, universe_id=request.universe_id, ) if request.generate_images: story = await generate_full_story_service(generate_request, user_id, db, job=job) saved_story = await get_story_detail(story.id, user_id, db) await _record_postprocessing_event_if_needed(db, job=job, story=saved_story) await finish_generation_job( db, job=job, story=saved_story, current_step="generation_completed", message="Story generation completed with persisted text and current asset states.", ) return GenerationResponse( id=story.id, generation_job_id=job.id, title=story.title, mode=story.mode, story_text=story.story_text, cover_prompt=story.cover_prompt, image_url=story.image_url, cover_url=story.image_url, audio_ready=story.audio_ready, errors=story.errors, generation_status=story.generation_status, text_status=saved_story.text_status, image_status=story.image_status, audio_status=story.audio_status, last_error=story.last_error, child_profile_id=story.child_profile_id, universe_id=story.universe_id, retryable_assets=saved_story.retryable_assets, ) story = await generate_and_save_story(generate_request, user_id, db, job=job) await _record_postprocessing_event_if_needed(db, job=job, story=story) await finish_generation_job( db, job=job, story=story, current_step="generation_completed", message="Story generation completed with a persisted readable narrative.", ) return GenerationResponse( id=story.id, generation_job_id=job.id, title=story.title, mode=story.mode, story_text=story.story_text, cover_prompt=story.cover_prompt, image_url=story.image_url, cover_url=story.image_url, generation_status=story.generation_status, text_status=story.text_status, image_status=story.image_status, audio_status=story.audio_status, last_error=story.last_error, child_profile_id=story.child_profile_id, universe_id=story.universe_id, retryable_assets=story.retryable_assets, ) async def list_stories( user_id: str, limit: int, offset: int, db: AsyncSession, ) -> list[Story]: """List stories for user.""" result = await db.execute( select(Story) .where(Story.user_id == user_id) .order_by(desc(Story.created_at)) .offset(offset) .limit(limit) ) return result.scalars().all() async def get_story_detail( story_id: int, user_id: str, db: AsyncSession, ) -> Story: """Get story detail.""" result = await db.execute( select(Story).where(Story.id == story_id, Story.user_id == user_id) ) story = result.scalar_one_or_none() if not story: raise HTTPException(status_code=404, detail="Story not found") return story async def delete_story( story_id: int, user_id: str, db: AsyncSession, ) -> None: """Delete a story.""" story = await get_story_detail(story_id, user_id, db) await db.delete(story) await db.commit() async def create_story_from_result( result: StoryOutput, user_id: str, profile_id: str | None, universe_id: str | None, db: AsyncSession, ) -> Story: """Save a generated story to DB (helper for stream endpoint).""" return await _persist_text_story_result( result=result, user_id=user_id, profile_id=profile_id, universe_id=universe_id, db=db, ) async def _retry_cover_image_asset(story: Story, db: AsyncSession, *, job=None) -> None: """Retry cover generation for a text story.""" await _complete_cover_image_asset( story, db, last_error_prefix="封面生成失败", log_event="cover_asset_retry_failed", job=job, ) async def _retry_storybook_image_assets( story: Story, db: AsyncSession, *, job=None, ) -> None: """Retry missing storybook cover/page images.""" await _complete_storybook_image_assets(story, db, job=job) async def _retry_audio_asset(story: Story, db: AsyncSession, *, job=None) -> None: """Retry audio generation while preserving persisted status on provider failure.""" await _complete_audio_asset(story, db, raise_on_failure=False, job=job) async def retry_story_assets( story_id: int, user_id: str, assets: list[str], db: AsyncSession, ) -> Story: """Retry selected assets through one workflow-level endpoint.""" requested_assets = list(dict.fromkeys(assets)) job = await create_generation_job( db, user_id=user_id, output_mode="asset_retry", input_type=",".join(requested_assets), request_payload={"story_id": story_id, "assets": requested_assets}, story_id=story_id, ) story: Story | None = None try: story = await get_story_detail(story_id, user_id, db) await record_generation_event( db, job=job, story_id=story.id, event_type="asset_retry_started", status="running", message="Asset retry started.", metadata={"assets": requested_assets}, ) if "image" in requested_assets: if story.mode == "storybook": await _retry_storybook_image_assets(story, db, job=job) else: await _retry_cover_image_asset(story, db, job=job) if "audio" in requested_assets: await _retry_audio_asset(story, db, job=job) story = await get_story_detail(story_id, user_id, db) await finish_generation_job( db, job=job, story=story, current_step="asset_retry_completed", message="Asset retry completed with persisted status updates.", metadata={"assets": requested_assets}, ) return story except HTTPException as exc: await finish_generation_job( db, job=job, story=story, status="failed", current_step="asset_retry_failed", error_message=str(exc.detail), message="Asset retry failed.", metadata={"assets": requested_assets}, ) raise except Exception as exc: await finish_generation_job( db, job=job, story=story, status="failed", current_step="asset_retry_failed", error_message=str(exc), message="Asset retry failed.", metadata={"assets": requested_assets}, ) raise async def generate_story_cover( story_id: int, user_id: str, db: AsyncSession, ) -> str: """Generate cover image for an existing story.""" job = await create_generation_job( db, user_id=user_id, output_mode="asset_generation", input_type="image", request_payload={"story_id": story_id, "assets": ["image"]}, story_id=story_id, ) story: Story | None = None try: story = await get_story_detail(story_id, user_id, db) image_result = await _complete_cover_image_asset( story, db, raise_on_failure=True, log_event="cover_generation_failed", job=job, ) story = await get_story_detail(story_id, user_id, db) await finish_generation_job( db, job=job, story=story, current_step="asset_generation_completed", message="Cover image generation completed.", metadata={"assets": ["image"]}, ) if image_result.succeeded and isinstance(image_result.value, str): return image_result.value except HTTPException as exc: await finish_generation_job( db, job=job, story=story, status="failed", current_step="asset_generation_failed", error_message=str(exc.detail), message="Cover image generation failed.", metadata={"assets": ["image"]}, ) raise raise HTTPException(status_code=500, detail="Image generation failed") async def generate_story_audio( story_id: int, user_id: str, db: AsyncSession, ) -> bytes: """Generate audio for a story.""" job = await create_generation_job( db, user_id=user_id, output_mode="asset_generation", input_type="audio", request_payload={"story_id": story_id, "assets": ["audio"]}, story_id=story_id, ) story: Story | None = None try: story = await get_story_detail(story_id, user_id, db) audio_result = await _complete_audio_asset( story, db, raise_on_failure=True, job=job, ) story = await get_story_detail(story_id, user_id, db) await finish_generation_job( db, job=job, story=story, current_step="asset_generation_completed", message="Story audio generation completed.", metadata={"assets": ["audio"]}, ) if audio_result.succeeded and isinstance(audio_result.value, bytes): return audio_result.value except HTTPException as exc: await finish_generation_job( db, job=job, story=story, status="failed", current_step="asset_generation_failed", error_message=str(exc.detail), message="Story audio generation failed.", metadata={"assets": ["audio"]}, ) raise raise HTTPException(status_code=500, detail="Audio generation failed") async def get_story_audio_status( story_id: int, user_id: str, db: AsyncSession, ) -> dict: """Return audio cache status without generating audio.""" story = await get_story_detail(story_id, user_id, db) metadata = get_audio_cache_metadata(story.audio_path) if story.audio_path and not metadata.exists: logger.warning( "story_audio_cache_missing_on_status", story_id=story.id, audio_path=story.audio_path, ) story.audio_path = None if story.audio_status == StoryAssetStatus.READY.value: sync_story_status(story, audio_status=StoryAssetStatus.NOT_REQUESTED) await db.commit() metadata = get_audio_cache_metadata(story.audio_path) return { "story_id": story.id, "audio_ready": story.audio_status == StoryAssetStatus.READY.value and metadata.exists, "cache_exists": metadata.exists, "cache_size_bytes": metadata.size_bytes, "cache_updated_at": metadata.updated_at, "generation_status": story.generation_status, "text_status": story.text_status, "image_status": story.image_status, "audio_status": story.audio_status, "last_error": story.last_error, "retryable_assets": story.retryable_assets, } async def clear_story_audio_cache( story_id: int, user_id: str, db: AsyncSession, ) -> dict: """Delete cached audio and mark audio as retryable again.""" story = await get_story_detail(story_id, user_id, db) delete_audio_cache(story.audio_path) story.audio_path = None sync_story_status( story, audio_status=StoryAssetStatus.NOT_REQUESTED, last_error=None, ) await db.commit() return await get_story_audio_status(story_id, user_id, db) async def get_story_achievements( story_id: int, user_id: str, db: AsyncSession, ) -> list[AchievementItem]: """Get achievements unlocked by a specific story.""" result = await db.execute( select(Story) .options(joinedload(Story.story_universe)) .where(Story.id == story_id, Story.user_id == user_id) ) story = result.scalar_one_or_none() if not story: raise HTTPException(status_code=404, detail="Story not found") if not story.universe_id or not story.story_universe: return [] universe = story.story_universe if not universe.achievements: return [] results = [] for ach in universe.achievements: if isinstance(ach, dict) and ach.get("source_story_id") == story_id: results.append(AchievementItem( type=ach.get("type", "Unknown"), description=ach.get("description", ""), obtained_at=ach.get("obtained_at") )) return results