mirror of
https://github.com/MODSetter/SurfSense.git
synced 2026-08-15 11:53:39 +00:00
feat(podcasts): align delivery to the Artifact model
Rename podcasts -> podcast_runs and treat the run row as job state only: delivered audio and its markdown live in the Artifact, so the row drops storage_backend/storage_key/file_location and gains artifact_id. mark_ready commits status and artifact_id atomically, preserving the READY <-> artifact_id invariant that Zero publishes. Public shares stream audio through the artifact content route and get artifact_id + workspace_id injected at snapshot time; the authenticated player and library read artifactId from Zero. Retire the legacy authenticated stream route, the podcast storage helpers, and the REST list endpoint.
This commit is contained in:
CREDO23
parent
15bf2ba20c
commit
0a2c856288
32 changed files with 385 additions and 547 deletions
|
|
@ -0,0 +1,28 @@
|
|||
"""Add podcasts.artifact_id pointing at the delivered Artifact.
|
||||
|
||||
Add-only. NULL means no Artifact yet; the render task and the backfill fill it.
|
||||
|
||||
Revision ID: 182
|
||||
Revises: 181
|
||||
"""
|
||||
|
||||
from collections.abc import Sequence
|
||||
|
||||
from alembic import op
|
||||
|
||||
revision: str = "182"
|
||||
down_revision: str | None = "181"
|
||||
branch_labels: str | Sequence[str] | None = None
|
||||
depends_on: str | Sequence[str] | None = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.execute(
|
||||
"ALTER TABLE podcasts "
|
||||
"ADD COLUMN IF NOT EXISTS artifact_id INTEGER "
|
||||
"REFERENCES artifacts(id) ON DELETE SET NULL"
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.execute("ALTER TABLE podcasts DROP COLUMN IF EXISTS artifact_id")
|
||||
|
|
@ -0,0 +1,72 @@
|
|||
"""Rename podcasts to podcast_runs, drop the row's audio columns, publish to Zero.
|
||||
|
||||
The delivered audio now lives in the Artifact referenced by ``artifact_id``. Run
|
||||
``backfill_podcast_artifacts.py --yes`` before this so no episode is lost.
|
||||
|
||||
Guarded like 180: refuses while any READY row has no Artifact. Renames the
|
||||
table, drops ``storage_backend`` / ``storage_key`` / ``file_location``, and
|
||||
reconciles ``zero_publication`` so runs reach the UI by push instead of polling.
|
||||
|
||||
Revision ID: 183
|
||||
Revises: 182
|
||||
"""
|
||||
|
||||
from collections.abc import Sequence
|
||||
|
||||
from sqlalchemy import text
|
||||
|
||||
from alembic import op
|
||||
from app.zero_publication import apply_publication
|
||||
|
||||
revision: str = "183"
|
||||
down_revision: str | None = "182"
|
||||
branch_labels: str | Sequence[str] | None = None
|
||||
depends_on: str | Sequence[str] | None = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
pending = (
|
||||
op.get_bind()
|
||||
.execute(
|
||||
text(
|
||||
"SELECT count(*) FROM podcasts "
|
||||
"WHERE status = 'ready' AND artifact_id IS NULL"
|
||||
)
|
||||
)
|
||||
.scalar()
|
||||
)
|
||||
if pending:
|
||||
raise RuntimeError(
|
||||
f"{pending} READY podcasts row(s) have no Artifact. "
|
||||
"Run `python -m scripts.backfill_podcast_artifacts --yes` before this "
|
||||
"migration, or their audio will be lost."
|
||||
)
|
||||
|
||||
op.execute("ALTER TABLE podcasts RENAME TO podcast_runs")
|
||||
op.execute("ALTER TABLE podcast_runs DROP COLUMN IF EXISTS storage_backend")
|
||||
op.execute("ALTER TABLE podcast_runs DROP COLUMN IF EXISTS storage_key")
|
||||
op.execute("ALTER TABLE podcast_runs DROP COLUMN IF EXISTS file_location")
|
||||
apply_publication(op.get_bind())
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
# A published table's columns can't be dropped while the publication
|
||||
# depends on them, so release it first.
|
||||
op.execute(
|
||||
"""
|
||||
DO $$
|
||||
BEGIN
|
||||
IF EXISTS (
|
||||
SELECT 1 FROM pg_publication_tables
|
||||
WHERE pubname = 'zero_publication'
|
||||
AND tablename = 'podcast_runs'
|
||||
) THEN
|
||||
ALTER PUBLICATION zero_publication DROP TABLE podcast_runs;
|
||||
END IF;
|
||||
END $$;
|
||||
"""
|
||||
)
|
||||
op.execute("ALTER TABLE podcast_runs ADD COLUMN IF NOT EXISTS storage_backend VARCHAR(32)")
|
||||
op.execute("ALTER TABLE podcast_runs ADD COLUMN IF NOT EXISTS storage_key TEXT")
|
||||
op.execute("ALTER TABLE podcast_runs ADD COLUMN IF NOT EXISTS file_location TEXT")
|
||||
op.execute("ALTER TABLE podcast_runs RENAME TO podcasts")
|
||||
|
|
@ -1,7 +1,7 @@
|
|||
"""Podcast media → Artifact."""
|
||||
"""Podcast media -> Artifact."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from app.artifacts.media.podcast.storage import open_stream, purge, persist
|
||||
from app.artifacts.media.podcast.record import record
|
||||
|
||||
__all__ = ["open_stream", "purge", "persist"]
|
||||
__all__ = ["record"]
|
||||
|
|
|
|||
|
|
@ -1,7 +1,4 @@
|
|||
"""Record a finished podcast as an Artifact.
|
||||
|
||||
Audio blobs live in ``app.artifacts.media.podcast.storage``; this writes the Artifact.
|
||||
"""
|
||||
"""Record a finished podcast as an Artifact; the Artifact owns the audio blob."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
|
|
|||
|
|
@ -1,54 +0,0 @@
|
|||
"""Durable object-store helpers for podcast audio."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from collections.abc import AsyncIterator
|
||||
from typing import Any
|
||||
|
||||
from app.file_storage.factory import get_storage_backend
|
||||
|
||||
_AUDIO_CONTENT_TYPE = "audio/mpeg"
|
||||
|
||||
|
||||
def build_audio_key(*, workspace_id: int, podcast_id: int) -> str:
|
||||
"""Object key: ``podcasts/{workspace_id}/{podcast_id}/{uuid}.mp3``."""
|
||||
return f"podcasts/{workspace_id}/{podcast_id}/{uuid.uuid4().hex}.mp3"
|
||||
|
||||
|
||||
async def persist(
|
||||
*, workspace_id: int, podcast_id: int, data: bytes
|
||||
) -> tuple[str, str]:
|
||||
"""Persist audio bytes; return ``(backend_name, storage_key)``."""
|
||||
backend = get_storage_backend()
|
||||
key = build_audio_key(workspace_id=workspace_id, podcast_id=podcast_id)
|
||||
await backend.put(key, data, content_type=_AUDIO_CONTENT_TYPE)
|
||||
return backend.backend_name, key
|
||||
|
||||
|
||||
def open_stream(storage_key: str) -> AsyncIterator[bytes]:
|
||||
return get_storage_backend().open_stream(storage_key)
|
||||
|
||||
|
||||
def open_podcast_stream(podcast: Any) -> AsyncIterator[bytes]:
|
||||
"""Stream a ready podcast's audio. Raises if it has no ``storage_key``."""
|
||||
if not podcast.storage_key:
|
||||
raise FileNotFoundError(f"podcast {podcast.id} has no stored audio")
|
||||
return open_stream(podcast.storage_key)
|
||||
|
||||
|
||||
async def exists(podcast: Any) -> bool:
|
||||
return bool(podcast.storage_key) and await get_storage_backend().exists(
|
||||
podcast.storage_key
|
||||
)
|
||||
|
||||
|
||||
async def purge(podcast: Any) -> None:
|
||||
"""Delete a podcast's stored audio if present."""
|
||||
await purge_key(podcast.storage_key)
|
||||
|
||||
|
||||
async def purge_key(key: str | None) -> None:
|
||||
"""Delete a stored audio object by key (e.g. superseded on re-render)."""
|
||||
if key:
|
||||
await get_storage_backend().delete(key)
|
||||
|
|
@ -8,37 +8,26 @@ then enqueues the matching Celery task; lifecycle errors map to 409/422.
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from collections.abc import AsyncIterator
|
||||
from contextlib import asynccontextmanager
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Response
|
||||
from fastapi.responses import StreamingResponse
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.auth.context import AuthContext
|
||||
from app.config import config as app_config
|
||||
from app.db import (
|
||||
Permission,
|
||||
Workspace,
|
||||
WorkspaceMembership,
|
||||
get_async_session,
|
||||
)
|
||||
from app.podcasts.generation.brief import propose_brief
|
||||
from app.podcasts.persistence import Podcast, PodcastRepository, PodcastStatus
|
||||
from app.podcasts.persistence import Podcast, PodcastRepository
|
||||
from app.podcasts.service import (
|
||||
InvalidTransitionError,
|
||||
PodcastService,
|
||||
PreconditionFailedError,
|
||||
SpecConflictError,
|
||||
)
|
||||
from app.artifacts.media.podcast.storage import (
|
||||
exists as audio_exists,
|
||||
open_podcast_stream,
|
||||
purge,
|
||||
)
|
||||
from app.podcasts.tasks import draft_transcript_task
|
||||
from app.podcasts.tts import get_text_to_speech
|
||||
from app.podcasts.voices import (
|
||||
|
|
@ -53,7 +42,6 @@ from .schemas import (
|
|||
CreatePodcastRequest,
|
||||
LanguageOptions,
|
||||
PodcastDetail,
|
||||
PodcastSummary,
|
||||
UpdateSpecRequest,
|
||||
VoiceOption,
|
||||
)
|
||||
|
|
@ -61,41 +49,6 @@ from .schemas import (
|
|||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get("/podcasts", response_model=list[PodcastSummary])
|
||||
async def list_podcasts(
|
||||
workspace_id: int | None = None,
|
||||
skip: int = 0,
|
||||
limit: int = 100,
|
||||
session: AsyncSession = Depends(get_async_session),
|
||||
auth: AuthContext = Depends(get_auth_context),
|
||||
):
|
||||
user = auth.user
|
||||
if skip < 0 or limit < 1:
|
||||
raise HTTPException(status_code=400, detail="Invalid pagination parameters")
|
||||
|
||||
if workspace_id is not None:
|
||||
await _require(session, auth, workspace_id, Permission.PODCASTS_READ)
|
||||
query = (
|
||||
select(Podcast)
|
||||
.where(Podcast.workspace_id == workspace_id)
|
||||
.order_by(Podcast.created_at.desc())
|
||||
.offset(skip)
|
||||
.limit(limit)
|
||||
)
|
||||
else:
|
||||
query = (
|
||||
select(Podcast)
|
||||
.join(Workspace)
|
||||
.join(WorkspaceMembership)
|
||||
.where(WorkspaceMembership.user_id == user.id)
|
||||
.order_by(Podcast.created_at.desc())
|
||||
.offset(skip)
|
||||
.limit(limit)
|
||||
)
|
||||
result = await session.execute(query)
|
||||
return list(result.scalars().all())
|
||||
|
||||
|
||||
@router.get("/podcasts/voices", response_model=list[VoiceOption])
|
||||
async def list_voices(language: str | None = None):
|
||||
"""Voices the active TTS provider offers, optionally filtered by language."""
|
||||
|
|
@ -277,55 +230,11 @@ async def delete_podcast(
|
|||
auth: AuthContext = Depends(get_auth_context),
|
||||
):
|
||||
podcast = await _load(session, auth, podcast_id, Permission.PODCASTS_DELETE)
|
||||
await purge(podcast)
|
||||
await session.delete(podcast)
|
||||
await session.commit()
|
||||
return {"message": "Podcast deleted successfully"}
|
||||
|
||||
|
||||
@router.get("/podcasts/{podcast_id}/stream")
|
||||
async def stream_podcast(
|
||||
podcast_id: int,
|
||||
session: AsyncSession = Depends(get_async_session),
|
||||
auth: AuthContext = Depends(get_auth_context),
|
||||
):
|
||||
podcast = await _load(session, auth, podcast_id, Permission.PODCASTS_READ)
|
||||
|
||||
if podcast.storage_key:
|
||||
# Verify first so a missing object is a 404, not a mid-stream crash.
|
||||
if not await audio_exists(podcast):
|
||||
raise HTTPException(
|
||||
status_code=404, detail="Podcast audio is no longer available"
|
||||
)
|
||||
return StreamingResponse(
|
||||
open_podcast_stream(podcast),
|
||||
media_type="audio/mpeg",
|
||||
headers={"Accept-Ranges": "bytes"},
|
||||
)
|
||||
|
||||
# Back-compat: rows rendered before the storage migration kept a local path.
|
||||
if podcast.file_location and os.path.isfile(podcast.file_location):
|
||||
path = podcast.file_location
|
||||
|
||||
def iterfile():
|
||||
with open(path, mode="rb") as handle:
|
||||
yield from handle
|
||||
|
||||
return StreamingResponse(
|
||||
iterfile(),
|
||||
media_type="audio/mpeg",
|
||||
headers={
|
||||
"Accept-Ranges": "bytes",
|
||||
"Content-Disposition": f"inline; filename={Path(path).name}",
|
||||
},
|
||||
)
|
||||
|
||||
# No audio: terminal states never will have any, otherwise it's in flight.
|
||||
if PodcastStatus(podcast.status).is_terminal:
|
||||
raise HTTPException(status_code=404, detail="Podcast audio not found")
|
||||
raise HTTPException(status_code=409, detail="Podcast audio is not ready yet")
|
||||
|
||||
|
||||
async def _require(
|
||||
session: AsyncSession,
|
||||
auth: AuthContext,
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@ from __future__ import annotations
|
|||
|
||||
from datetime import datetime
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from app.podcasts.duration_limits import (
|
||||
DEFAULT_MAX_SECONDS,
|
||||
|
|
@ -74,19 +74,6 @@ class LanguageOptions(BaseModel):
|
|||
allows_custom: bool
|
||||
|
||||
|
||||
class PodcastSummary(BaseModel):
|
||||
"""Lightweight list item."""
|
||||
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
id: int
|
||||
title: str
|
||||
status: PodcastStatus
|
||||
created_at: datetime
|
||||
workspace_id: int
|
||||
thread_id: int | None = None
|
||||
|
||||
|
||||
class PodcastDetail(BaseModel):
|
||||
"""Full podcast state for the detail view and action responses."""
|
||||
|
||||
|
|
@ -124,12 +111,4 @@ class PodcastDetail(BaseModel):
|
|||
|
||||
@classmethod
|
||||
async def resolve(cls, session, podcast: Podcast) -> PodcastDetail:
|
||||
from app.artifacts.media.legacy import existing_legacy_artifact
|
||||
|
||||
art = await existing_legacy_artifact(
|
||||
session,
|
||||
workspace_id=podcast.workspace_id,
|
||||
kind="podcast",
|
||||
legacy_id=podcast.id,
|
||||
)
|
||||
return cls.of(podcast, artifact_id=art.id if art else None)
|
||||
return cls.of(podcast, artifact_id=podcast.artifact_id)
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
"""``podcasts`` table: a generated podcast, its brief, transcript, and state."""
|
||||
"""``podcast_runs`` table: a generated podcast, its brief, transcript, state."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
|
@ -19,16 +19,15 @@ from .enums import PodcastStatus
|
|||
|
||||
|
||||
class Podcast(BaseModel, TimestampMixin):
|
||||
"""A podcast across its whole lifecycle: brief, transcript, audio, status.
|
||||
"""A podcast run: brief, transcript, lifecycle state, and its Artifact link.
|
||||
|
||||
``spec`` (the reviewable brief) and ``podcast_transcript`` are JSONB so the
|
||||
flexible Pydantic shapes can evolve without migrations. ``spec_version``
|
||||
backs optimistic concurrency on brief edits. Rendered audio lives in the
|
||||
object store, addressed by ``storage_backend`` + ``storage_key`` rather than
|
||||
a raw path.
|
||||
backs optimistic concurrency on brief edits. The delivered audio and
|
||||
markdown live in the Artifact referenced by ``artifact_id``.
|
||||
"""
|
||||
|
||||
__tablename__ = "podcasts"
|
||||
__tablename__ = "podcast_runs"
|
||||
|
||||
title = Column(String(500), nullable=False)
|
||||
|
||||
|
|
@ -57,16 +56,17 @@ class Podcast(BaseModel, TimestampMixin):
|
|||
# The drafted dialogue (Transcript); null until drafting completes.
|
||||
podcast_transcript = Column(JSONB, nullable=True)
|
||||
|
||||
# Where the rendered audio lives in the object store; null until READY.
|
||||
storage_backend = Column(String(32), nullable=True)
|
||||
storage_key = Column(Text, nullable=True)
|
||||
duration_seconds = Column(Integer, nullable=True)
|
||||
|
||||
# Human-readable reason when status is FAILED.
|
||||
error = Column(Text, nullable=True)
|
||||
|
||||
# Legacy local audio path; retained for back-compat until cutover.
|
||||
file_location = Column(Text, nullable=True)
|
||||
# The delivered Artifact; NULL until READY. The Artifact owns the audio.
|
||||
artifact_id = Column(
|
||||
Integer,
|
||||
ForeignKey("artifacts.id", ondelete="SET NULL"),
|
||||
nullable=True,
|
||||
)
|
||||
|
||||
workspace_id = Column(
|
||||
Integer,
|
||||
|
|
|
|||
|
|
@ -169,19 +169,13 @@ class PodcastService:
|
|||
await self._session.flush()
|
||||
return podcast
|
||||
|
||||
async def attach_audio(
|
||||
self,
|
||||
podcast: Podcast,
|
||||
*,
|
||||
storage_backend: str,
|
||||
storage_key: str,
|
||||
duration_seconds: int | None = None,
|
||||
async def mark_ready(
|
||||
self, podcast: Podcast, *, duration_seconds: int | None = None
|
||||
) -> Podcast:
|
||||
"""Record rendered audio and mark the podcast ready."""
|
||||
"""Mark a rendered podcast ready. The delivered audio lives in the Artifact."""
|
||||
self._transition(podcast, PodcastStatus.READY)
|
||||
podcast.storage_backend = storage_backend
|
||||
podcast.storage_key = storage_key
|
||||
podcast.duration_seconds = duration_seconds
|
||||
if duration_seconds is not None:
|
||||
podcast.duration_seconds = duration_seconds
|
||||
podcast.error = None
|
||||
await self._session.flush()
|
||||
return podcast
|
||||
|
|
@ -221,8 +215,8 @@ def _status(podcast: Podcast) -> PodcastStatus:
|
|||
|
||||
|
||||
def has_stored_episode(podcast: Podcast) -> bool:
|
||||
"""Whether finished audio is stored (``file_location`` covers legacy rows)."""
|
||||
return bool(podcast.storage_key or podcast.file_location)
|
||||
"""Whether a delivered episode exists; its audio lives in the Artifact."""
|
||||
return podcast.artifact_id is not None
|
||||
|
||||
|
||||
def read_spec(podcast: Podcast) -> PodcastSpec | None:
|
||||
|
|
|
|||
|
|
@ -1,8 +1,8 @@
|
|||
"""Audio-rendering task: RENDERING -> READY.
|
||||
|
||||
Synthesises and merges the approved transcript, stores the MP3 in the object
|
||||
store, and marks the podcast ready. The working directory is stable per podcast
|
||||
so a re-render (e.g. after a voice change) reuses the segment cache.
|
||||
Synthesises and merges the approved transcript, records the delivered Artifact
|
||||
(which owns the MP3 and markdown), and marks the podcast ready. The working
|
||||
directory is stable per podcast so a re-render reuses the segment cache.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -15,15 +15,13 @@ from sqlalchemy import select
|
|||
|
||||
from app.celery_app import celery_app
|
||||
from app.observability import analytics as ph_analytics
|
||||
from app.podcasts.persistence import PodcastRepository
|
||||
from app.podcasts.persistence import PodcastRepository, PodcastStatus
|
||||
from app.podcasts.rendering import PodcastRenderer
|
||||
from app.podcasts.service import (
|
||||
InvalidTransitionError,
|
||||
PodcastService,
|
||||
read_spec,
|
||||
read_transcript,
|
||||
)
|
||||
from app.artifacts.media.podcast.storage import purge_key, persist
|
||||
from app.podcasts.tts import get_text_to_speech
|
||||
from app.podcasts.voices import get_voice_catalog
|
||||
from app.tasks.celery_tasks import get_celery_session_maker, run_async_celery_task
|
||||
|
|
@ -67,58 +65,48 @@ async def _render_audio(podcast_id: int) -> dict:
|
|||
spec=spec, transcript=transcript, workdir=workdir
|
||||
)
|
||||
|
||||
superseded_key = podcast.storage_key
|
||||
|
||||
backend_name, key = await persist(
|
||||
workspace_id=podcast.workspace_id,
|
||||
podcast_id=podcast_id,
|
||||
data=rendered.data,
|
||||
)
|
||||
try:
|
||||
await PodcastService(session).attach_audio(
|
||||
podcast, storage_backend=backend_name, storage_key=key
|
||||
)
|
||||
await session.commit()
|
||||
|
||||
from app.artifacts.media.podcast.record import record as record_podcast
|
||||
|
||||
await record_podcast(
|
||||
session,
|
||||
podcast,
|
||||
audio=rendered.data,
|
||||
transcript=transcript,
|
||||
)
|
||||
|
||||
# Credit-consuming deliverable; the frontend never confirms the
|
||||
# render finished. Owner (workspace.user_id) resolved lazily so
|
||||
# disabled installs pay nothing for the extra query.
|
||||
if ph_analytics.is_enabled():
|
||||
# Local import: app.db <-> app.podcasts.persistence have a
|
||||
# module-init cycle; deferring keeps this task importable.
|
||||
from app.db import Workspace
|
||||
|
||||
owner_id = await session.scalar(
|
||||
select(Workspace.user_id).where(
|
||||
Workspace.id == podcast.workspace_id
|
||||
)
|
||||
)
|
||||
if owner_id:
|
||||
ph_analytics.capture(
|
||||
"podcast_generated",
|
||||
distinct_id=str(owner_id),
|
||||
properties={
|
||||
"workspace_id": podcast.workspace_id,
|
||||
"podcast_id": podcast_id,
|
||||
},
|
||||
groups={"workspace": str(podcast.workspace_id)},
|
||||
)
|
||||
except InvalidTransitionError:
|
||||
# A user back-out won the race (e.g. the regeneration was
|
||||
# reverted): drop the stale render and leave the row alone.
|
||||
await purge_key(key)
|
||||
# A user back-out during the render leaves the row out of RENDERING;
|
||||
# bail before creating an Artifact that would never be linked.
|
||||
if PodcastStatus(podcast.status) is not PodcastStatus.RENDERING:
|
||||
return {"status": "superseded", "podcast_id": podcast_id}
|
||||
|
||||
# Purge only after the new audio is committed, so a failed re-render never
|
||||
# destroys the episode the user can still play.
|
||||
await purge_key(superseded_key)
|
||||
from app.artifacts.media.podcast.record import record as record_podcast
|
||||
|
||||
# Record the Artifact while still RENDERING, then flip to READY and link
|
||||
# it in one commit: a READY row is never committed without its audio.
|
||||
saved = await record_podcast(
|
||||
session,
|
||||
podcast,
|
||||
audio=rendered.data,
|
||||
transcript=transcript,
|
||||
)
|
||||
if saved is None:
|
||||
raise RuntimeError(f"podcast {podcast_id}: recording the Artifact failed")
|
||||
|
||||
await PodcastService(session).mark_ready(podcast)
|
||||
podcast.artifact_id = saved.artifact_id
|
||||
await session.commit()
|
||||
|
||||
# Credit-consuming deliverable; the frontend never confirms the
|
||||
# render finished. Owner (workspace.user_id) resolved lazily so
|
||||
# disabled installs pay nothing for the extra query.
|
||||
if ph_analytics.is_enabled():
|
||||
# Local import: app.db <-> app.podcasts.persistence have a
|
||||
# module-init cycle; deferring keeps this task importable.
|
||||
from app.db import Workspace
|
||||
|
||||
owner_id = await session.scalar(
|
||||
select(Workspace.user_id).where(Workspace.id == podcast.workspace_id)
|
||||
)
|
||||
if owner_id:
|
||||
ph_analytics.capture(
|
||||
"podcast_generated",
|
||||
distinct_id=str(owner_id),
|
||||
properties={
|
||||
"workspace_id": podcast.workspace_id,
|
||||
"podcast_id": podcast_id,
|
||||
},
|
||||
groups={"workspace": str(podcast.workspace_id)},
|
||||
)
|
||||
|
||||
return {"status": "ready", "podcast_id": podcast_id}
|
||||
|
|
|
|||
|
|
@ -232,8 +232,17 @@ async def create_snapshot(
|
|||
if podcast_info:
|
||||
podcasts_data.append(podcast_info)
|
||||
podcast_ids_seen.add(podcast_id)
|
||||
# Update status to "ready" so frontend renders PodcastPlayer
|
||||
part["result"] = {**result_data, "status": "ready"}
|
||||
# The tool result carries podcast_id, not artifact_id,
|
||||
# so the generic allowlist check above misses it.
|
||||
new_result = {**result_data, "status": "ready"}
|
||||
artifact_id = podcast_info.get("artifact_id")
|
||||
if isinstance(artifact_id, int):
|
||||
artifact_ids.add(artifact_id)
|
||||
new_result["artifact_id"] = artifact_id
|
||||
new_result["workspace_id"] = podcast_info.get(
|
||||
"workspace_id"
|
||||
)
|
||||
part["result"] = new_result
|
||||
|
||||
elif tool_name in ("generate_report", "generate_resume"):
|
||||
result_data = part.get("result", {})
|
||||
|
|
@ -328,14 +337,13 @@ async def _get_podcast_for_snapshot(
|
|||
if not podcast or podcast.status != PodcastStatus.READY:
|
||||
return None
|
||||
|
||||
# Audio is served from the Artifact; only the transcript stays embedded.
|
||||
return {
|
||||
"original_id": podcast.id,
|
||||
"title": podcast.title,
|
||||
"transcript": podcast.podcast_transcript,
|
||||
"storage_backend": podcast.storage_backend,
|
||||
"storage_key": podcast.storage_key,
|
||||
# Legacy fallback for rows rendered before the storage migration.
|
||||
"file_path": podcast.file_location,
|
||||
"artifact_id": podcast.artifact_id,
|
||||
"workspace_id": podcast.workspace_id,
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -629,7 +637,6 @@ async def clone_from_snapshot(
|
|||
|
||||
data = snapshot.snapshot_data
|
||||
messages_data = data.get("messages", [])
|
||||
podcasts_lookup = {p.get("original_id"): p for p in data.get("podcasts", [])}
|
||||
reports_lookup = {r.get("original_id"): r for r in data.get("reports", [])}
|
||||
|
||||
new_thread = NewChatThread(
|
||||
|
|
@ -646,7 +653,6 @@ async def clone_from_snapshot(
|
|||
session.add(new_thread)
|
||||
await session.flush()
|
||||
|
||||
podcast_id_mapping: dict[int, int] = {}
|
||||
report_id_mapping: dict[int, int] = {}
|
||||
|
||||
# Check which authors from snapshot still exist in DB
|
||||
|
|
@ -680,37 +686,8 @@ async def clone_from_snapshot(
|
|||
|
||||
if isinstance(content, list):
|
||||
for part in content:
|
||||
if (
|
||||
isinstance(part, dict)
|
||||
and part.get("type") == "tool-call"
|
||||
and part.get("toolName") == "generate_podcast"
|
||||
):
|
||||
result = part.get("result", {})
|
||||
old_podcast_id = result.get("podcast_id")
|
||||
|
||||
if old_podcast_id and old_podcast_id not in podcast_id_mapping:
|
||||
podcast_info = podcasts_lookup.get(old_podcast_id)
|
||||
if podcast_info:
|
||||
new_podcast = Podcast(
|
||||
title=podcast_info.get("title", "Cloned Podcast"),
|
||||
podcast_transcript=podcast_info.get("transcript"),
|
||||
storage_backend=podcast_info.get("storage_backend"),
|
||||
storage_key=podcast_info.get("storage_key"),
|
||||
file_location=podcast_info.get("file_path"),
|
||||
status=PodcastStatus.READY,
|
||||
workspace_id=target_workspace_id,
|
||||
thread_id=new_thread.id,
|
||||
)
|
||||
session.add(new_podcast)
|
||||
await session.flush()
|
||||
podcast_id_mapping[old_podcast_id] = new_podcast.id
|
||||
|
||||
if old_podcast_id and old_podcast_id in podcast_id_mapping:
|
||||
part["result"] = {
|
||||
**result,
|
||||
"podcast_id": podcast_id_mapping[old_podcast_id],
|
||||
}
|
||||
|
||||
# generate_podcast is copied verbatim; the snapshot already
|
||||
# carries its artifact_id + workspace_id (as video does).
|
||||
if (
|
||||
isinstance(part, dict)
|
||||
and part.get("type") == "tool-call"
|
||||
|
|
|
|||
|
|
@ -76,6 +76,7 @@ PODCAST_COLS = [
|
|||
"spec_version",
|
||||
"duration_seconds",
|
||||
"error",
|
||||
"artifact_id",
|
||||
"workspace_id",
|
||||
"thread_id",
|
||||
"created_at",
|
||||
|
|
@ -107,7 +108,7 @@ ZERO_PUBLICATION: Mapping[str, Sequence[str] | None] = {
|
|||
"user": USER_COLS,
|
||||
"automations": AUTOMATION_COLS,
|
||||
"automation_runs": AUTOMATION_RUN_COLS,
|
||||
"podcasts": PODCAST_COLS,
|
||||
"podcast_runs": PODCAST_COLS,
|
||||
"video_presentation_runs": VIDEO_PRESENTATION_RUN_COLS,
|
||||
}
|
||||
|
||||
|
|
@ -136,7 +137,7 @@ def _expected_columns(conn: Connection, table: str) -> list[str] | None:
|
|||
if table in {
|
||||
"documents",
|
||||
"user",
|
||||
"podcasts",
|
||||
"podcast_runs",
|
||||
"video_presentation_runs",
|
||||
} and "_0_version" in _table_columns(conn, table):
|
||||
expected.append("_0_version")
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@ import contextlib
|
|||
import uuid
|
||||
from collections.abc import AsyncGenerator, AsyncIterator
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
|
@ -128,14 +129,25 @@ class FakeStorageBackend:
|
|||
self.deleted.append(key)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_storage(monkeypatch) -> FakeStorageBackend:
|
||||
"""Route audio storage to an in-memory backend for the stream routes."""
|
||||
@pytest.fixture(autouse=True)
|
||||
def fake_storage(monkeypatch, patched_embed_texts) -> FakeStorageBackend:
|
||||
"""In-memory object store + Postgres-only artifact indexing for the suite.
|
||||
|
||||
The delivered audio is written by ``save_artifact``; pinning it to False
|
||||
keeps that on the indexed (non-git) path so the blob lands in this backend.
|
||||
"""
|
||||
del patched_embed_texts
|
||||
backend = FakeStorageBackend()
|
||||
monkeypatch.setattr(
|
||||
"app.artifacts.media.podcast.storage.get_storage_backend", lambda: backend
|
||||
"app.artifacts.service.get_storage_backend", lambda *a, **k: backend
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"app.artifacts.service.knowledge_store_enabled_for",
|
||||
AsyncMock(return_value=False),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"app.file_storage.factory.get_storage_backend", lambda *a, **k: backend
|
||||
)
|
||||
monkeypatch.setattr("app.file_storage.factory.get_storage_backend", lambda: backend)
|
||||
return backend
|
||||
|
||||
|
||||
|
|
@ -272,12 +284,17 @@ def make_podcast(db_session: AsyncSession):
|
|||
elif target is PodcastStatus.RENDERING:
|
||||
await service.attach_transcript(podcast, build_transcript())
|
||||
elif target is PodcastStatus.READY:
|
||||
await service.attach_audio(
|
||||
await service.mark_ready(podcast, duration_seconds=123)
|
||||
from app.artifacts.media.podcast.record import record as record_podcast
|
||||
|
||||
saved = await record_podcast(
|
||||
db_session,
|
||||
podcast,
|
||||
storage_backend="memory",
|
||||
storage_key="podcasts/audio.mp3",
|
||||
duration_seconds=123,
|
||||
audio=b"merged-audio",
|
||||
transcript=build_transcript(),
|
||||
)
|
||||
assert saved is not None
|
||||
podcast.artifact_id = saved.artifact_id
|
||||
await db_session.flush()
|
||||
return podcast
|
||||
|
||||
|
|
|
|||
|
|
@ -191,7 +191,6 @@ async def test_regenerate_without_a_brief_is_rejected(
|
|||
workspace_id=db_workspace.id,
|
||||
status=PodcastStatus.READY,
|
||||
spec_version=1,
|
||||
file_location="/var/old/podcast.mp3",
|
||||
)
|
||||
db_session.add(podcast)
|
||||
await db_session.flush()
|
||||
|
|
|
|||
|
|
@ -1,15 +1,18 @@
|
|||
"""The audio-rendering task against a real database.
|
||||
|
||||
From RENDERING, the task synthesises and merges the approved transcript, stores
|
||||
the bytes, and marks the podcast READY with the storage location recorded. The
|
||||
DB, service, renderer orchestration, and storage wrapper run for real; the true
|
||||
externals are faked — the TTS provider, the FFmpeg merge, and the object store.
|
||||
From RENDERING the task synthesises and merges the approved transcript, records
|
||||
the delivered Artifact (which owns the audio), stamps ``artifact_id``, and marks
|
||||
the podcast READY. The DB, service, renderer orchestration, and artifact service
|
||||
run for real; the true externals are faked — the TTS provider, the FFmpeg merge,
|
||||
and the object store.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.artifacts.persistence import ArtifactFile, ArtifactFileRole
|
||||
from app.podcasts.persistence import PodcastStatus
|
||||
from app.podcasts.service import PodcastService
|
||||
from app.podcasts.tasks import render
|
||||
|
|
@ -19,8 +22,19 @@ from .conftest import build_transcript
|
|||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
async def test_render_marks_ready_and_stores_audio(
|
||||
db_workspace, make_podcast, bind_task_session, fake_tts, fake_merge, fake_storage
|
||||
async def _primary_key(db_session, artifact_id: int) -> str:
|
||||
row = await db_session.scalar(
|
||||
select(ArtifactFile).where(
|
||||
ArtifactFile.artifact_id == artifact_id,
|
||||
ArtifactFile.role == ArtifactFileRole.PRIMARY,
|
||||
)
|
||||
)
|
||||
assert row is not None
|
||||
return row.storage_key
|
||||
|
||||
|
||||
async def test_render_marks_ready_and_records_the_artifact(
|
||||
db_session, db_workspace, make_podcast, bind_task_session, fake_tts, fake_merge, fake_storage
|
||||
):
|
||||
podcast = await make_podcast(
|
||||
workspace_id=db_workspace.id, status=PodcastStatus.RENDERING
|
||||
|
|
@ -30,12 +44,12 @@ async def test_render_marks_ready_and_stores_audio(
|
|||
|
||||
assert result["status"] == "ready"
|
||||
assert podcast.status == PodcastStatus.READY
|
||||
assert podcast.storage_backend == "memory"
|
||||
assert podcast.storage_key
|
||||
assert fake_storage.objects[podcast.storage_key] == b"merged-audio"
|
||||
assert podcast.artifact_id is not None
|
||||
key = await _primary_key(db_session, podcast.artifact_id)
|
||||
assert fake_storage.objects[key] == b"merged-audio"
|
||||
|
||||
|
||||
async def test_rerender_replaces_audio_and_purges_the_old_object(
|
||||
async def test_rerender_reuses_the_artifact_and_purges_the_old_object(
|
||||
db_session,
|
||||
db_workspace,
|
||||
make_podcast,
|
||||
|
|
@ -44,13 +58,11 @@ async def test_rerender_replaces_audio_and_purges_the_old_object(
|
|||
fake_merge,
|
||||
fake_storage,
|
||||
):
|
||||
# A regenerated episode keeps exactly one stored object: the new render
|
||||
# must not leak the superseded audio in the object store.
|
||||
podcast = await make_podcast(
|
||||
workspace_id=db_workspace.id, status=PodcastStatus.READY
|
||||
)
|
||||
old_key = podcast.storage_key
|
||||
fake_storage.objects[old_key] = b"old-audio"
|
||||
original_artifact_id = podcast.artifact_id
|
||||
old_key = await _primary_key(db_session, original_artifact_id)
|
||||
|
||||
service = PodcastService(db_session)
|
||||
await service.regenerate(podcast)
|
||||
|
|
@ -61,12 +73,15 @@ async def test_rerender_replaces_audio_and_purges_the_old_object(
|
|||
|
||||
assert result["status"] == "ready"
|
||||
assert podcast.status == PodcastStatus.READY
|
||||
assert podcast.storage_key != old_key
|
||||
assert fake_storage.objects[podcast.storage_key] == b"merged-audio"
|
||||
# The Artifact is revised in place, not replaced.
|
||||
assert podcast.artifact_id == original_artifact_id
|
||||
new_key = await _primary_key(db_session, podcast.artifact_id)
|
||||
assert new_key != old_key
|
||||
assert fake_storage.objects[new_key] == b"merged-audio"
|
||||
assert old_key in fake_storage.deleted
|
||||
|
||||
|
||||
async def test_render_losing_to_a_user_revert_keeps_the_episode_and_leaks_nothing(
|
||||
async def test_render_losing_to_a_user_revert_keeps_the_episode(
|
||||
db_session,
|
||||
db_workspace,
|
||||
make_podcast,
|
||||
|
|
@ -76,13 +91,11 @@ async def test_render_losing_to_a_user_revert_keeps_the_episode_and_leaks_nothin
|
|||
fake_storage,
|
||||
):
|
||||
# The user reverts the regeneration while the render is in flight: the
|
||||
# stale render must neither resurrect the redo nor leak the object it
|
||||
# already stored.
|
||||
# stale render must not finalize a new take.
|
||||
podcast = await make_podcast(
|
||||
workspace_id=db_workspace.id, status=PodcastStatus.READY
|
||||
)
|
||||
old_key = podcast.storage_key
|
||||
fake_storage.objects[old_key] = b"old-audio"
|
||||
original_artifact_id = podcast.artifact_id
|
||||
|
||||
service = PodcastService(db_session)
|
||||
await service.regenerate(podcast)
|
||||
|
|
@ -94,7 +107,4 @@ async def test_render_losing_to_a_user_revert_keeps_the_episode_and_leaks_nothin
|
|||
|
||||
assert result["status"] == "superseded"
|
||||
assert podcast.status == PodcastStatus.READY
|
||||
assert podcast.storage_key == old_key
|
||||
assert old_key not in fake_storage.deleted
|
||||
stale_keys = [key for key in fake_storage.objects if key != old_key]
|
||||
assert all(key in fake_storage.deleted for key in stale_keys)
|
||||
assert podcast.artifact_id == original_artifact_id
|
||||
|
|
|
|||
|
|
@ -1,8 +1,6 @@
|
|||
"""Podcasts are scoped to workspace membership.
|
||||
|
||||
A user can only create or read podcasts in spaces they belong to, and an
|
||||
unscoped listing returns only the caller's own podcasts — never another
|
||||
member's.
|
||||
A user can only create or read podcasts in spaces they belong to.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
|
@ -38,16 +36,3 @@ async def test_creating_in_a_nonmember_space_is_forbidden(
|
|||
)
|
||||
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
async def test_listing_returns_only_the_callers_podcasts(
|
||||
client, db_workspace, make_podcast, foreign_podcast
|
||||
):
|
||||
mine = await make_podcast(workspace_id=db_workspace.id, title="Mine")
|
||||
|
||||
resp = await client.get(BASE)
|
||||
|
||||
assert resp.status_code == 200
|
||||
ids = {p["id"] for p in resp.json()}
|
||||
assert mine.id in ids
|
||||
assert foreign_podcast.id not in ids
|
||||
|
|
|
|||
|
|
@ -1,52 +0,0 @@
|
|||
"""Streaming a podcast's rendered audio over HTTP.
|
||||
|
||||
A ready podcast streams its bytes; an in-flight one is 409, a stored-but-missing
|
||||
object is 404. Storage is an in-memory backend (the object store is a boundary).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from app.podcasts.persistence import PodcastStatus
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
BASE = "/api/v1/podcasts"
|
||||
|
||||
|
||||
async def test_stream_serves_stored_audio(
|
||||
client, db_workspace, make_podcast, fake_storage
|
||||
):
|
||||
podcast = await make_podcast(
|
||||
workspace_id=db_workspace.id, status=PodcastStatus.READY
|
||||
)
|
||||
fake_storage.objects["podcasts/audio.mp3"] = b"the-audio"
|
||||
|
||||
resp = await client.get(f"{BASE}/{podcast.id}/stream")
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert resp.headers["content-type"] == "audio/mpeg"
|
||||
assert resp.content == b"the-audio"
|
||||
|
||||
|
||||
async def test_stream_409_while_in_flight(client, db_workspace, make_podcast):
|
||||
podcast = await make_podcast(
|
||||
workspace_id=db_workspace.id, status=PodcastStatus.DRAFTING
|
||||
)
|
||||
|
||||
resp = await client.get(f"{BASE}/{podcast.id}/stream")
|
||||
|
||||
assert resp.status_code == 409
|
||||
|
||||
|
||||
async def test_stream_404_when_object_missing(
|
||||
client, db_workspace, make_podcast, fake_storage
|
||||
):
|
||||
podcast = await make_podcast(
|
||||
workspace_id=db_workspace.id, status=PodcastStatus.READY
|
||||
)
|
||||
|
||||
resp = await client.get(f"{BASE}/{podcast.id}/stream")
|
||||
|
||||
assert resp.status_code == 404
|
||||
|
|
@ -9,7 +9,6 @@ test builds a row in one lifecycle shape and asserts the mapping reflects it.
|
|||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
|
|
@ -56,15 +55,15 @@ def test_an_awaiting_brief_podcast_exposes_the_deserialized_brief(make_spec):
|
|||
|
||||
|
||||
def test_a_legacy_episode_still_exposes_its_transcript_and_audio():
|
||||
# Pre-rework rows stored [{speaker_id, dialog}] and a local file path;
|
||||
# they must keep flowing through the new read model, not fail validation.
|
||||
# Pre-rework rows stored [{speaker_id, dialog}]; they must keep flowing
|
||||
# through the new read model. Audio now lives in the linked Artifact.
|
||||
podcast = _podcast(
|
||||
status=PodcastStatus.READY,
|
||||
podcast_transcript=[
|
||||
{"speaker_id": 0, "dialog": "Welcome back."},
|
||||
{"speaker_id": 1, "dialog": "Glad to be here."},
|
||||
],
|
||||
file_location="/var/old/podcast.mp3",
|
||||
artifact_id=99,
|
||||
)
|
||||
|
||||
detail = PodcastDetail.of(podcast)
|
||||
|
|
@ -82,8 +81,7 @@ def test_a_ready_podcast_reports_available_audio(make_spec, make_transcript):
|
|||
status=PodcastStatus.READY,
|
||||
spec=make_spec().model_dump(mode="json"),
|
||||
podcast_transcript=make_transcript().model_dump(mode="json"),
|
||||
storage_backend="local",
|
||||
storage_key="k",
|
||||
artifact_id=77,
|
||||
duration_seconds=120,
|
||||
)
|
||||
|
||||
|
|
@ -98,14 +96,7 @@ def test_a_ready_podcast_reports_available_audio(make_spec, make_transcript):
|
|||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resolve_attaches_dual_written_artifact_id(monkeypatch):
|
||||
import app.artifacts.media.legacy as legacy
|
||||
|
||||
podcast = _podcast(status=PodcastStatus.READY)
|
||||
monkeypatch.setattr(
|
||||
legacy,
|
||||
"existing_legacy_artifact",
|
||||
AsyncMock(return_value=SimpleNamespace(id=55)),
|
||||
)
|
||||
async def test_resolve_reads_the_stamped_artifact_id():
|
||||
podcast = _podcast(status=PodcastStatus.READY, artifact_id=55)
|
||||
detail = await PodcastDetail.resolve(AsyncMock(), podcast)
|
||||
assert detail.artifact_id == 55
|
||||
|
|
|
|||
|
|
@ -85,3 +85,25 @@ async def test_unreferenced_video_artifact_is_refused_without_a_query(
|
|||
|
||||
assert served is None
|
||||
session.execute.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_podcast_snapshot_carries_artifact_id_and_no_storage_key():
|
||||
from app.podcasts.persistence import PodcastStatus
|
||||
|
||||
podcast = SimpleNamespace(
|
||||
id=7,
|
||||
title="Ep",
|
||||
podcast_transcript=None,
|
||||
artifact_id=42,
|
||||
workspace_id=3,
|
||||
status=PodcastStatus.READY,
|
||||
)
|
||||
info = await public_chat_service._get_podcast_for_snapshot(
|
||||
_session_returning(podcast), 7
|
||||
)
|
||||
|
||||
assert info["artifact_id"] == 42
|
||||
assert info["workspace_id"] == 3
|
||||
assert "storage_key" not in info
|
||||
assert "storage_backend" not in info
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@
|
|||
import type { ToolCallMessagePartProps } from "@assistant-ui/react";
|
||||
import { Loader2, RotateCcw, Undo2, X } from "lucide-react";
|
||||
import { usePathname } from "next/navigation";
|
||||
import { type ReactNode, useEffect, useState } from "react";
|
||||
import { type ReactNode, useState } from "react";
|
||||
import { toast } from "sonner";
|
||||
import { TextShimmerLoader } from "@/components/prompt-kit/loader";
|
||||
import {
|
||||
|
|
@ -195,8 +195,6 @@ function BackOutButton({ podcastId, hasEpisode }: { podcastId: number; hasEpisod
|
|||
);
|
||||
}
|
||||
|
||||
const BACK_OUT_STATUSES = new Set(["awaiting_brief", "drafting", "rendering"]);
|
||||
|
||||
/** Status-driven card for an authenticated viewer, fed by Zero push. */
|
||||
function LivePodcastCard({
|
||||
podcastId,
|
||||
|
|
@ -207,33 +205,6 @@ function LivePodcastCard({
|
|||
}) {
|
||||
const { podcast, isLoading } = usePodcastLive(podcastId);
|
||||
|
||||
// Whether a finished episode exists decides revert-vs-cancel, and Zero
|
||||
// doesn't publish audio fields — so the in-flight states check over REST,
|
||||
// re-checking on each status change (a fresh podcast gains its episode,
|
||||
// a regeneration starts with one).
|
||||
const status = podcast?.status;
|
||||
const [hasEpisode, setHasEpisode] = useState(false);
|
||||
const [artifactId, setArtifactId] = useState<number | undefined>();
|
||||
useEffect(() => {
|
||||
if (!status) return;
|
||||
if (status === "ready" || BACK_OUT_STATUSES.has(status)) {
|
||||
let stale = false;
|
||||
podcastsApiService
|
||||
.getDetail(podcastId)
|
||||
.then((detail) => {
|
||||
if (stale) return;
|
||||
if (BACK_OUT_STATUSES.has(status)) setHasEpisode(detail.has_audio);
|
||||
if (status === "ready") {
|
||||
setArtifactId(detail.artifact_id ?? undefined);
|
||||
}
|
||||
})
|
||||
.catch(() => {});
|
||||
return () => {
|
||||
stale = true;
|
||||
};
|
||||
}
|
||||
}, [podcastId, status]);
|
||||
|
||||
if (!podcast) {
|
||||
if (isLoading) {
|
||||
return <WorkingState title={fallbackTitle} label="Loading podcast" />;
|
||||
|
|
@ -247,8 +218,10 @@ function LivePodcastCard({
|
|||
}
|
||||
|
||||
const title = podcast.title || fallbackTitle;
|
||||
|
||||
const backOut = <BackOutButton podcastId={podcast.id} hasEpisode={hasEpisode} />;
|
||||
// A finished episode is exactly a stamped Artifact; it decides
|
||||
// revert-vs-cancel and feeds the player, both straight from the Zero row.
|
||||
const artifactId = podcast.artifactId ?? undefined;
|
||||
const backOut = <BackOutButton podcastId={podcast.id} hasEpisode={artifactId != null} />;
|
||||
|
||||
switch (podcast.status) {
|
||||
case "pending":
|
||||
|
|
|
|||
|
|
@ -157,14 +157,26 @@ export function PodcastPlayer({
|
|||
let lines: TranscriptLine[] = [];
|
||||
|
||||
if (shareToken) {
|
||||
if (podcastId == null) throw new Error("Podcast id required for shared chat");
|
||||
// Artifact route when available; legacy per-podcast stream for
|
||||
// snapshots predating the backfill.
|
||||
const audioUrl =
|
||||
artifactId != null
|
||||
? `/api/v1/public/${shareToken}/artifacts/${artifactId}/content`
|
||||
: podcastId != null
|
||||
? `/api/v1/public/${shareToken}/podcasts/${podcastId}/stream`
|
||||
: null;
|
||||
if (!audioUrl) throw new Error("Podcast identity missing for shared chat");
|
||||
const [blob, details] = await Promise.all([
|
||||
baseApiService.getBlob(`/api/v1/public/${shareToken}/podcasts/${podcastId}/stream`),
|
||||
baseApiService.get(`/api/v1/public/${shareToken}/podcasts/${podcastId}`),
|
||||
baseApiService.getBlob(audioUrl),
|
||||
podcastId != null
|
||||
? baseApiService.get(`/api/v1/public/${shareToken}/podcasts/${podcastId}`)
|
||||
: Promise.resolve(null),
|
||||
]);
|
||||
audioBlob = blob;
|
||||
const parsed = publicPodcastDetailsSchema.safeParse(details);
|
||||
lines = (parsed.success ? (parsed.data.podcast_transcript ?? []) : []).map(
|
||||
const parsed = details
|
||||
? publicPodcastDetailsSchema.safeParse(details)
|
||||
: null;
|
||||
lines = (parsed?.success ? (parsed.data.podcast_transcript ?? []) : []).map(
|
||||
(entry, turn) => ({
|
||||
key: `turn-${turn}`,
|
||||
label: `Speaker ${entry.speaker_id + 1}`,
|
||||
|
|
@ -178,25 +190,6 @@ export function PodcastPlayer({
|
|||
podcastId,
|
||||
controller.signal
|
||||
));
|
||||
} else if (podcastId != null) {
|
||||
const [audioResponse, detail] = await Promise.all([
|
||||
authenticatedFetch(buildBackendUrl(`/api/v1/podcasts/${podcastId}/stream`), {
|
||||
method: "GET",
|
||||
signal: controller.signal,
|
||||
}),
|
||||
podcastsApiService.getDetail(podcastId),
|
||||
]);
|
||||
|
||||
if (!audioResponse.ok) {
|
||||
throw new Error(`Failed to load audio: ${audioResponse.status}`);
|
||||
}
|
||||
|
||||
audioBlob = await audioResponse.blob();
|
||||
lines = (detail.transcript?.turns ?? []).map((entry, turn) => ({
|
||||
key: `turn-${turn}`,
|
||||
label: speakerLabel(detail.spec, entry.speaker),
|
||||
text: entry.text,
|
||||
}));
|
||||
} else {
|
||||
throw new Error("Podcast identity missing");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -156,16 +156,3 @@ export const podcastDetail = z.object({
|
|||
artifact_id: z.number().int().positive().nullish(),
|
||||
});
|
||||
export type PodcastDetail = z.infer<typeof podcastDetail>;
|
||||
|
||||
// Lightweight list item — mirror app/podcasts/api/schemas.py PodcastSummary.
|
||||
export const podcastSummary = z.object({
|
||||
id: z.number(),
|
||||
title: z.string(),
|
||||
status: podcastStatus,
|
||||
created_at: z.string(),
|
||||
workspace_id: z.number(),
|
||||
thread_id: z.number().nullish(),
|
||||
});
|
||||
export type PodcastSummary = z.infer<typeof podcastSummary>;
|
||||
|
||||
export const podcastSummaryList = z.array(podcastSummary);
|
||||
|
|
|
|||
|
|
@ -1,7 +1,6 @@
|
|||
import { useQuery } from "@tanstack/react-query";
|
||||
import { fetchArtifacts } from "@/features/artifacts/artifact-query";
|
||||
import type { ArtifactListItem } from "@/features/artifacts/model";
|
||||
import { podcastsApiService } from "@/lib/apis/podcasts-api.service";
|
||||
import { reportsApiService } from "@/lib/apis/reports-api.service";
|
||||
import type {
|
||||
LibraryArtifact,
|
||||
|
|
@ -9,12 +8,6 @@ import type {
|
|||
LibraryArtifactStatus,
|
||||
} from "../model/artifact";
|
||||
|
||||
function podcastStatus(status: string): LibraryArtifactStatus {
|
||||
if (status === "ready") return "ready";
|
||||
if (status === "failed" || status === "cancelled") return "error";
|
||||
return "running";
|
||||
}
|
||||
|
||||
function indexingStatus(status: string): LibraryArtifactStatus {
|
||||
if (status === "failed") return "error";
|
||||
if (status === "ready") return "ready";
|
||||
|
|
@ -44,23 +37,18 @@ function fromArtifactRow(row: ArtifactListItem): LibraryArtifact {
|
|||
};
|
||||
}
|
||||
|
||||
// Podcast rows still list separately: a podcast has no Artifact until delivered.
|
||||
// Delivered podcasts arrive as Artifact rows; in-flight/failed runs stream from
|
||||
// Zero (see useLibraryPodcastRuns), matching how videos are handled.
|
||||
async function fetchLibraryArtifacts(workspaceId: number): Promise<LibraryArtifact[]> {
|
||||
const [rows, reports, podcasts] = await Promise.all([
|
||||
const [rows, reports] = await Promise.all([
|
||||
fetchArtifacts(workspaceId).catch(() => []),
|
||||
reportsApiService.list(workspaceId).catch(() => []),
|
||||
podcastsApiService.list(workspaceId).catch(() => []),
|
||||
]);
|
||||
|
||||
const artifacts: LibraryArtifact[] = [];
|
||||
const coveredPodcasts = new Set<number>();
|
||||
|
||||
for (const row of rows) {
|
||||
const item = fromArtifactRow(row);
|
||||
artifacts.push(item);
|
||||
if (item.kind === "podcast" && row.legacy?.kind === "podcast") {
|
||||
coveredPodcasts.add(row.legacy.id);
|
||||
}
|
||||
artifacts.push(fromArtifactRow(row));
|
||||
}
|
||||
|
||||
for (const report of reports) {
|
||||
|
|
@ -77,20 +65,6 @@ async function fetchLibraryArtifacts(workspaceId: number): Promise<LibraryArtifa
|
|||
});
|
||||
}
|
||||
|
||||
for (const podcast of podcasts) {
|
||||
if (coveredPodcasts.has(podcast.id)) continue;
|
||||
artifacts.push({
|
||||
key: `podcast-${podcast.id}`,
|
||||
kind: "podcast",
|
||||
entityId: podcast.id,
|
||||
title: podcast.title,
|
||||
status: podcastStatus(podcast.status),
|
||||
createdAt: podcast.created_at,
|
||||
contentType: "markdown",
|
||||
sourceThreadId: podcast.thread_id,
|
||||
});
|
||||
}
|
||||
|
||||
return artifacts.sort(
|
||||
(a, b) => new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime()
|
||||
);
|
||||
|
|
|
|||
|
|
@ -0,0 +1,48 @@
|
|||
"use client";
|
||||
|
||||
import { useQuery as useZeroQuery } from "@rocicorp/zero/react";
|
||||
import { useMemo } from "react";
|
||||
import { queries } from "@/zero/queries";
|
||||
import type { LibraryArtifact, LibraryArtifactStatus } from "../model/artifact";
|
||||
|
||||
function runStatus(status: string): LibraryArtifactStatus {
|
||||
if (status === "failed" || status === "cancelled") return "error";
|
||||
return "running";
|
||||
}
|
||||
|
||||
interface ZeroPodcastRunRow {
|
||||
id: number;
|
||||
title: string;
|
||||
status: string;
|
||||
artifactId?: number | null;
|
||||
workspaceId: number;
|
||||
threadId?: number | null;
|
||||
createdAt: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* In-flight and failed podcast runs, sourced from Zero by push. A delivered run
|
||||
* ("ready") is already represented by its Artifact row, so it is filtered out
|
||||
* here to avoid a duplicate card.
|
||||
*/
|
||||
export function useLibraryPodcastRuns(workspaceId: number): LibraryArtifact[] {
|
||||
const [rows] = useZeroQuery(queries.podcastRuns.bySpace({ workspaceId }));
|
||||
|
||||
return useMemo(
|
||||
() =>
|
||||
(rows as ZeroPodcastRunRow[])
|
||||
.filter((row) => row.status !== "ready")
|
||||
.map((row) => ({
|
||||
key: `podcast-run-${row.id}`,
|
||||
kind: "podcast" as const,
|
||||
entityId: row.id,
|
||||
artifactId: row.artifactId ?? undefined,
|
||||
title: row.title,
|
||||
status: runStatus(row.status),
|
||||
createdAt: new Date(row.createdAt).toISOString(),
|
||||
contentType: "markdown" as const,
|
||||
sourceThreadId: row.threadId ?? null,
|
||||
})),
|
||||
[rows]
|
||||
);
|
||||
}
|
||||
|
|
@ -8,6 +8,7 @@ import { openReportPanelAtom } from "@/atoms/chat/report-panel.atom";
|
|||
import { MobileReportPanel } from "@/components/report-panel/report-panel";
|
||||
import { Button } from "@/components/ui/button";
|
||||
import { useLibraryArtifacts } from "../hooks/use-library-artifacts";
|
||||
import { useLibraryPodcastRuns } from "../hooks/use-library-podcast-runs";
|
||||
import { useLibraryVideoRuns } from "../hooks/use-library-video-runs";
|
||||
import type { LibraryArtifact, LibraryArtifactKind } from "../model/artifact";
|
||||
import { ArtifactCard } from "./artifact-card";
|
||||
|
|
@ -64,18 +65,19 @@ function EmptyState() {
|
|||
export function ArtifactsLibrary({ workspaceId }: { workspaceId: number }) {
|
||||
const { artifacts, loading, error, refresh } = useLibraryArtifacts(workspaceId);
|
||||
const liveVideoRuns = useLibraryVideoRuns(workspaceId);
|
||||
const livePodcastRuns = useLibraryPodcastRuns(workspaceId);
|
||||
const openArtifactPanel = useSetAtom(openArtifactPanelAtom);
|
||||
const openReportPanel = useSetAtom(openReportPanelAtom);
|
||||
const [selectedMedia, setSelectedMedia] = useState<LibraryArtifact | null>(null);
|
||||
|
||||
// Delivered videos come from the Artifact API (react-query); in-flight and
|
||||
// Delivered media comes from the Artifact API (react-query); in-flight and
|
||||
// failed runs arrive by push from Zero. Merge newest-first.
|
||||
const merged = useMemo(
|
||||
() =>
|
||||
[...artifacts, ...liveVideoRuns].sort(
|
||||
[...artifacts, ...liveVideoRuns, ...livePodcastRuns].sort(
|
||||
(a, b) => new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime()
|
||||
),
|
||||
[artifacts, liveVideoRuns]
|
||||
[artifacts, liveVideoRuns, livePodcastRuns]
|
||||
);
|
||||
|
||||
const grouped = useMemo(() => {
|
||||
|
|
|
|||
|
|
@ -37,17 +37,21 @@ function MediaViewerBody({
|
|||
workspaceId: number;
|
||||
}) {
|
||||
if (artifact.kind === "podcast") {
|
||||
if (artifact.artifactId != null) {
|
||||
if (artifact.artifactId == null) {
|
||||
return (
|
||||
<PodcastPlayer
|
||||
artifactId={artifact.artifactId}
|
||||
workspaceId={workspaceId}
|
||||
podcastId={artifact.legacyEntityId}
|
||||
title={artifact.title}
|
||||
/>
|
||||
<p className="px-6 py-10 text-center text-sm text-muted-foreground">
|
||||
Podcast not available
|
||||
</p>
|
||||
);
|
||||
}
|
||||
return <PodcastPlayer podcastId={artifact.entityId} title={artifact.title} />;
|
||||
return (
|
||||
<PodcastPlayer
|
||||
artifactId={artifact.artifactId}
|
||||
workspaceId={workspaceId}
|
||||
podcastId={artifact.legacyEntityId}
|
||||
title={artifact.title}
|
||||
/>
|
||||
);
|
||||
}
|
||||
if (artifact.kind === "video") {
|
||||
if (artifact.artifactId == null) {
|
||||
|
|
|
|||
|
|
@ -6,9 +6,9 @@ import { type PodcastSpec, type PodcastStatus, podcastSpec } from "@/contracts/t
|
|||
import { queries } from "@/zero/queries";
|
||||
|
||||
/**
|
||||
* Thin live row sourced from Zero's `podcasts` publication. Drives the
|
||||
* Thin live row sourced from Zero's `podcast_runs` publication. Drives the
|
||||
* lifecycle UI by push (no polling); heavy fields (transcript, audio) stay on
|
||||
* REST and are fetched lazily when a gate or the player needs them.
|
||||
* REST/the Artifact and are fetched lazily when a gate or the player needs them.
|
||||
*/
|
||||
export interface LivePodcast {
|
||||
id: number;
|
||||
|
|
@ -18,6 +18,7 @@ export interface LivePodcast {
|
|||
specVersion: number;
|
||||
durationSeconds: number | null;
|
||||
error: string | null;
|
||||
artifactId: number | null;
|
||||
workspaceId: number;
|
||||
threadId: number | null;
|
||||
}
|
||||
|
|
@ -28,7 +29,7 @@ interface UsePodcastLiveResult {
|
|||
}
|
||||
|
||||
export function usePodcastLive(podcastId: number | undefined): UsePodcastLiveResult {
|
||||
const [row, result] = useQuery(queries.podcasts.byId({ podcastId: podcastId ?? -1 }));
|
||||
const [row, result] = useQuery(queries.podcastRuns.byId({ podcastId: podcastId ?? -1 }));
|
||||
|
||||
const podcast = useMemo<LivePodcast | undefined>(() => {
|
||||
if (!podcastId || !row) return undefined;
|
||||
|
|
@ -40,6 +41,7 @@ export function usePodcastLive(podcastId: number | undefined): UsePodcastLiveRes
|
|||
specVersion: row.specVersion,
|
||||
durationSeconds: row.durationSeconds ?? null,
|
||||
error: row.error ?? null,
|
||||
artifactId: row.artifactId ?? null,
|
||||
workspaceId: row.workspaceId,
|
||||
threadId: row.threadId ?? null,
|
||||
};
|
||||
|
|
|
|||
|
|
@ -3,7 +3,6 @@ import {
|
|||
languageOptions,
|
||||
type PodcastSpec,
|
||||
podcastDetail,
|
||||
podcastSummaryList,
|
||||
updateSpecRequest,
|
||||
voiceOption,
|
||||
} from "@/contracts/types/podcast.types";
|
||||
|
|
@ -15,14 +14,6 @@ const BASE = "/api/v1/podcasts";
|
|||
const voiceOptionList = z.array(voiceOption);
|
||||
|
||||
class PodcastsApiService {
|
||||
list = async (workspaceId: number, limit = 200) => {
|
||||
const qs = new URLSearchParams({
|
||||
workspace_id: String(workspaceId),
|
||||
limit: String(limit),
|
||||
}).toString();
|
||||
return baseApiService.get(`${BASE}?${qs}`, podcastSummaryList);
|
||||
};
|
||||
|
||||
// Full state including the deserialized brief and transcript; thin lifecycle
|
||||
// fields (status, spec, spec_version) also arrive live via Zero.
|
||||
getDetail = async (podcastId: number) => {
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ import { chatSessionQueries, commentQueries, messageQueries } from "./chat";
|
|||
import { connectorQueries, documentQueries } from "./documents";
|
||||
import { folderQueries } from "./folders";
|
||||
import { notificationQueries } from "./inbox";
|
||||
import { podcastQueries } from "./podcasts";
|
||||
import { podcastRunQueries } from "./podcast-runs";
|
||||
import { userQueries } from "./user";
|
||||
import { videoPresentationRunQueries } from "./video-presentation-runs";
|
||||
|
||||
|
|
@ -18,6 +18,6 @@ export const queries = defineQueries({
|
|||
chatSession: chatSessionQueries,
|
||||
user: userQueries,
|
||||
automationRuns: automationRunQueries,
|
||||
podcasts: podcastQueries,
|
||||
podcastRuns: podcastRunQueries,
|
||||
videoRuns: videoPresentationRunQueries,
|
||||
});
|
||||
|
|
|
|||
|
|
@ -3,13 +3,13 @@ import { z } from "zod";
|
|||
import { zql } from "../schema/index";
|
||||
import { canReadSpace, constrainToAllowedSpaces, denySpace } from "./authz";
|
||||
|
||||
export const podcastQueries = {
|
||||
export const podcastRunQueries = {
|
||||
bySpace: defineQuery(z.object({ workspaceId: z.number() }), ({ args: { workspaceId }, ctx }) => {
|
||||
const query = zql.podcasts.where("workspaceId", workspaceId);
|
||||
const query = zql.podcast_runs.where("workspaceId", workspaceId);
|
||||
if (!canReadSpace(ctx, workspaceId)) return denySpace(query).orderBy("createdAt", "desc");
|
||||
return constrainToAllowedSpaces(query, ctx).orderBy("createdAt", "desc");
|
||||
}),
|
||||
byId: defineQuery(z.object({ podcastId: z.number() }), ({ args: { podcastId }, ctx }) =>
|
||||
constrainToAllowedSpaces(zql.podcasts.where("id", podcastId), ctx).one()
|
||||
constrainToAllowedSpaces(zql.podcast_runs.where("id", podcastId), ctx).one()
|
||||
),
|
||||
};
|
||||
|
|
@ -9,7 +9,7 @@ import {
|
|||
import { documentTable, searchSourceConnectorTable } from "./documents";
|
||||
import { folderTable } from "./folders";
|
||||
import { notificationTable } from "./inbox";
|
||||
import { podcastTable } from "./podcasts";
|
||||
import { podcastRunTable } from "./podcast-runs";
|
||||
import { userTable } from "./user";
|
||||
import { videoPresentationRunTable } from "./video-presentation-runs";
|
||||
|
||||
|
|
@ -73,7 +73,7 @@ export const schema = createSchema({
|
|||
userTable,
|
||||
automationTable,
|
||||
automationRunTable,
|
||||
podcastTable,
|
||||
podcastRunTable,
|
||||
videoPresentationRunTable,
|
||||
],
|
||||
relationships: [
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@ import { json, number, string, table } from "@rocicorp/zero";
|
|||
// Mirrors PODCAST_COLS in the backend zero_publication. status drives the
|
||||
// lifecycle UI by push; spec is the reviewable brief. The bulky source_content
|
||||
// and transcript are intentionally not published and are fetched over REST.
|
||||
export const podcastTable = table("podcasts")
|
||||
export const podcastRunTable = table("podcast_runs")
|
||||
.columns({
|
||||
id: number(),
|
||||
title: string(),
|
||||
|
|
@ -12,6 +12,7 @@ export const podcastTable = table("podcasts")
|
|||
specVersion: number().from("spec_version"),
|
||||
durationSeconds: number().optional().from("duration_seconds"),
|
||||
error: string().optional(),
|
||||
artifactId: number().optional().from("artifact_id"),
|
||||
workspaceId: number().from("workspace_id"),
|
||||
threadId: number().optional().from("thread_id"),
|
||||
createdAt: number().from("created_at"),
|
||||
Loading…
Add table
Add a link
Reference in a new issue