torin
a97a2fe005
Some checks failed
Build and Push Docker Images / changes (push) Has been cancelled
Build and Push Docker Images / build-backend (push) Has been cancelled
Build and Push Docker Images / build-frontend (push) Has been cancelled
Build and Push Docker Images / build-admin-frontend (push) Has been cancelled
113 lines
3.1 KiB
Python
113 lines
3.1 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"
|
|
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
|
|
generation_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 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 (
|
|
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
|
|
|
|
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
|