Spaces:
Runtime error
Runtime error
File size: 5,439 Bytes
34b0226 0649f1b 34b0226 4a3d0a5 fa8df4c 34b0226 b74c6ae 0649f1b fa8df4c bdd1640 34b0226 0649f1b 34b0226 0649f1b fa8df4c 4a3d0a5 0649f1b 34b0226 070c931 fa6e4a3 070c931 34b0226 fa6e4a3 34b0226 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 | """Artifacts are every generated thing (summaries, audio, video, launch kits, stills). Creating one always
means: artifact row + queued job, returned together so the UI can stream progress."""
import uuid
from sqlalchemy import select
from sqlalchemy.orm import Session, object_session
from app.models import Artifact, Job
from app.services.jobs import create_job, serialize_job
from app.services.storage import storage
from app.slides.build import slots as slide_slots
from app.slides.build import view as slide_pages
TYPE_LABELS = {
"summary": "Summary",
"audio_overview": "Audio overview",
"video": "Video",
"quote_card": "Quote card",
"carousel": "Carousel",
"launch_kit": "Launch Kit",
"mind_map": "Mind Constellation",
"quiz": "Quiz",
"flashcards": "Flashcards",
"report": "Report",
"infographic": "Infographic",
"slide_deck": "Slide deck",
"upload": "Upload",
}
def start_artifact(
db: Session,
workspace_id: uuid.UUID,
type_: str,
*,
title: str,
notebook_id: uuid.UUID | None = None,
document_id: uuid.UUID | None = None,
content: dict | None = None,
job_kind: str | None = None,
params: dict | None = None,
max_attempts: int = 2,
) -> tuple[Artifact, Job]:
artifact = Artifact(workspace_id=workspace_id, type=type_, notebook_id=notebook_id, document_id=document_id,
status="pending", content_json={"title": title, **(content or {})})
db.add(artifact)
db.commit()
job = create_job(db, workspace_id, job_kind or type_, {"artifact_id": str(artifact.id), **(params or {})},
artifact_id=artifact.id, max_attempts=max_attempts)
return artifact, job
def latest_jobs(db: Session, artifact_ids: list[uuid.UUID]) -> dict[uuid.UUID, Job]:
if not artifact_ids:
return {}
jobs = db.scalars(select(Job).where(Job.artifact_id.in_(artifact_ids)).order_by(Job.created_at)).all()
return {j.artifact_id: j for j in jobs} # later rows win, so this is the newest job per artifact
def report_blocks(a: Artifact) -> list[dict]:
"""A report's blocks, each infographic block with its page: the child artifact's html and status (or "missing"
once it was deleted). Read when the report is read, so one still being written shows up when it is ready."""
from app.pipeline.infographic import infographic_html # not at import: pipeline imports this module
blocks = list((a.content_json or {}).get("blocks") or [])
ids = [uuid.UUID(b["artifact_id"]) for b in blocks if b.get("type") == "infographic" and b.get("artifact_id")]
db = object_session(a)
kids = {k.id: k for k in db.scalars(select(Artifact).where(Artifact.id.in_(ids), Artifact.workspace_id == a.workspace_id))} \
if ids and db else {}
out = []
for b in blocks:
if b.get("type") != "infographic":
out.append(b)
continue
kid = kids.get(uuid.UUID(b["artifact_id"])) if b.get("artifact_id") else None
ready = bool(kid and kid.status == "ready" and (kid.content_json or {}).get("content"))
out.append({**b, "status": kid.status if kid else "missing",
"html": infographic_html(kid.content_json) if ready else None,
"html_landscape": infographic_html(kid.content_json, "landscape") if ready else None})
return out
def serialize_artifact(a: Artifact, job: Job | None = None) -> dict:
content = a.content_json or {}
if a.type == "infographic" and a.status == "ready":
from app.pipeline.infographic import infographic_html # not at import: pipeline imports this module
content = {**content, "html": infographic_html(content), "html_landscape": infographic_html(content, "landscape")}
if a.type == "slide_deck" and a.status == "ready":
content = {**content, "slides_html": slide_pages(content), "slide_slots": slide_slots(content)}
if a.type == "report" and any(b.get("type") == "infographic" for b in content.get("blocks") or []):
content = {**content, "blocks": report_blocks(a)}
slides = content.get("slide_keys") or []
return {
"id": str(a.id),
"type": a.type,
"type_label": TYPE_LABELS.get(a.type, a.type),
"title": content.get("title") or TYPE_LABELS.get(a.type, a.type),
"status": a.status,
"notebook_id": str(a.notebook_id) if a.notebook_id else None,
"document_id": str(a.document_id) if a.document_id else None,
"content": content,
"storage_key": a.storage_key,
# blog2video videos live on blog2video's storage (video_url once rendered)
"url": storage.presign_get(a.storage_key) if a.storage_key else content.get("video_url"),
"download_url": (storage.presign_get(a.storage_key, download_name=download_name(a)) if a.storage_key
else content.get("video_url")),
"provider": content.get("provider"),
"slide_urls": [storage.presign_get(k) for k in slides],
"created_at": a.created_at.isoformat() if a.created_at else None,
"job": serialize_job(job) if job else None,
}
def download_name(a: Artifact) -> str:
"""The file name a download is saved under: the artifact's title and its file's extension."""
title = (a.content_json or {}).get("title") or a.type
ext = (a.storage_key or "").rsplit(".", 1)[-1]
return f"{title[:60]}.{ext}"
|