157 lines
4.4 KiB
Python
157 lines
4.4 KiB
Python
"""Story generation status helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from enum import Enum
|
|
from typing import Protocol
|
|
|
|
|
|
class StoryGenerationStatus(str, Enum):
|
|
"""Overall story generation lifecycle."""
|
|
|
|
NARRATIVE_READY = "narrative_ready"
|
|
PARTIAL_READY = "partial_ready"
|
|
ASSETS_GENERATING = "assets_generating"
|
|
COMPLETED = "completed"
|
|
DEGRADED_COMPLETED = "degraded_completed"
|
|
FAILED = "failed"
|
|
|
|
|
|
class StoryAssetStatus(str, Enum):
|
|
"""Asset generation state for image and audio."""
|
|
|
|
NOT_REQUESTED = "not_requested"
|
|
GENERATING = "generating"
|
|
READY = "ready"
|
|
FAILED = "failed"
|
|
|
|
|
|
class StoryLike(Protocol):
|
|
"""Protocol for story-like objects used by status helpers."""
|
|
|
|
story_text: str | None
|
|
pages: list[dict] | None
|
|
cover_prompt: str | None
|
|
image_url: str | None
|
|
generation_status: str
|
|
text_status: str
|
|
image_status: str
|
|
audio_status: str
|
|
last_error: str | None
|
|
|
|
|
|
_ERROR_UNSET = object()
|
|
|
|
|
|
def _normalize_asset_status(value: str | None) -> StoryAssetStatus:
|
|
if not value:
|
|
return StoryAssetStatus.NOT_REQUESTED
|
|
|
|
try:
|
|
return StoryAssetStatus(value)
|
|
except ValueError:
|
|
return StoryAssetStatus.NOT_REQUESTED
|
|
|
|
|
|
def has_narrative_content(story: StoryLike) -> bool:
|
|
"""Whether the story already has readable content."""
|
|
|
|
return bool(story.story_text) or bool(story.pages)
|
|
|
|
|
|
def _has_retryable_image(story: StoryLike, image_status: StoryAssetStatus) -> bool:
|
|
if image_status in {StoryAssetStatus.READY, StoryAssetStatus.GENERATING}:
|
|
return False
|
|
|
|
pages = story.pages or []
|
|
has_missing_page_image = any(
|
|
isinstance(page, dict)
|
|
and page.get("image_prompt")
|
|
and not page.get("image_url")
|
|
for page in pages
|
|
)
|
|
return bool(story.cover_prompt and not story.image_url) or has_missing_page_image
|
|
|
|
|
|
def _has_pending_assets(
|
|
story: StoryLike,
|
|
*,
|
|
image_status: StoryAssetStatus,
|
|
audio_status: StoryAssetStatus,
|
|
) -> bool:
|
|
"""Whether readable content still has optional assets to complete."""
|
|
|
|
if _has_retryable_image(story, image_status):
|
|
return True
|
|
|
|
return bool(story.story_text) and audio_status not in {
|
|
StoryAssetStatus.READY,
|
|
StoryAssetStatus.GENERATING,
|
|
}
|
|
|
|
|
|
def resolve_story_generation_status(story: StoryLike) -> StoryGenerationStatus:
|
|
"""Derive the overall status from narrative and asset states."""
|
|
|
|
if not has_narrative_content(story):
|
|
return StoryGenerationStatus.FAILED
|
|
|
|
image_status = _normalize_asset_status(story.image_status)
|
|
audio_status = _normalize_asset_status(story.audio_status)
|
|
|
|
if StoryAssetStatus.GENERATING in (image_status, audio_status):
|
|
return StoryGenerationStatus.ASSETS_GENERATING
|
|
|
|
if StoryAssetStatus.FAILED in (image_status, audio_status):
|
|
return StoryGenerationStatus.DEGRADED_COMPLETED
|
|
|
|
if _has_pending_assets(story, image_status=image_status, audio_status=audio_status):
|
|
return StoryGenerationStatus.PARTIAL_READY
|
|
|
|
if (
|
|
image_status == StoryAssetStatus.NOT_REQUESTED
|
|
and audio_status == StoryAssetStatus.NOT_REQUESTED
|
|
):
|
|
return StoryGenerationStatus.NARRATIVE_READY
|
|
|
|
return StoryGenerationStatus.COMPLETED
|
|
|
|
|
|
def has_failed_assets(story: StoryLike) -> bool:
|
|
"""Whether any persisted asset is still in a failed state."""
|
|
|
|
image_status = _normalize_asset_status(story.image_status)
|
|
audio_status = _normalize_asset_status(story.audio_status)
|
|
return StoryAssetStatus.FAILED in (image_status, audio_status)
|
|
|
|
|
|
def sync_story_status(
|
|
story: StoryLike,
|
|
*,
|
|
image_status: StoryAssetStatus | None = None,
|
|
audio_status: StoryAssetStatus | None = None,
|
|
last_error: str | None | object = _ERROR_UNSET,
|
|
) -> None:
|
|
"""Update asset statuses and refresh overall generation status."""
|
|
|
|
if image_status is not None:
|
|
story.image_status = image_status.value
|
|
|
|
if audio_status is not None:
|
|
story.audio_status = audio_status.value
|
|
|
|
if last_error is not _ERROR_UNSET:
|
|
story.last_error = last_error
|
|
|
|
story.text_status = (
|
|
StoryAssetStatus.READY.value
|
|
if has_narrative_content(story)
|
|
else StoryAssetStatus.FAILED.value
|
|
)
|
|
|
|
generation_status = resolve_story_generation_status(story)
|
|
story.generation_status = generation_status.value
|
|
|
|
if last_error is _ERROR_UNSET and not has_failed_assets(story):
|
|
story.last_error = None
|