feat(api): add content generation endpoints (Sprint 4)
Implement Sprint 4: Content Generation
- Add ArtifactService with generation methods for 9 content types
- Add POST /generate/audio - Generate podcast
- Add POST /generate/video - Generate video
- Add POST /generate/slide-deck - Generate slides
- Add POST /generate/infographic - Generate infographic
- Add POST /generate/quiz - Generate quiz
- Add POST /generate/flashcards - Generate flashcards
- Add POST /generate/report - Generate report
- Add POST /generate/mind-map - Generate mind map (instant)
- Add POST /generate/data-table - Generate data table
- Add GET /artifacts - List artifacts
- Add GET /artifacts/{id}/status - Check artifact status
Models:
- AudioGenerationRequest, VideoGenerationRequest
- QuizGenerationRequest, FlashcardsGenerationRequest
- SlideDeckGenerationRequest, InfographicGenerationRequest
- ReportGenerationRequest, DataTableGenerationRequest
- Artifact, GenerationResponse, ArtifactList
Tests:
- 13 unit tests for ArtifactService
- 6 integration tests for generation API
- 19/19 tests passing
Related: Sprint 4 - Content Generation
This commit is contained in:
Luca Sacchi Ricciardi
@@ -5,7 +5,7 @@ from contextlib import asynccontextmanager
|
||||
from fastapi import FastAPI
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
|
||||
from notebooklm_agent.api.routes import chat, health, notebooks, sources
|
||||
from notebooklm_agent.api.routes import chat, generation, health, notebooks, sources
|
||||
from notebooklm_agent.core.config import get_settings
|
||||
from notebooklm_agent.core.logging import setup_logging
|
||||
|
||||
@@ -55,6 +55,7 @@ def create_application() -> FastAPI:
|
||||
app.include_router(notebooks.router, prefix="/api/v1/notebooks", tags=["notebooks"])
|
||||
app.include_router(sources.router, prefix="/api/v1/notebooks", tags=["sources"])
|
||||
app.include_router(chat.router, prefix="/api/v1/notebooks", tags=["chat"])
|
||||
app.include_router(generation.router, prefix="/api/v1/notebooks", tags=["generation"])
|
||||
|
||||
return app
|
||||
|
||||
|
||||
@@ -321,3 +321,324 @@ class ChatRequest(BaseModel):
|
||||
if not v or not v.strip():
|
||||
raise ValueError("Message cannot be empty")
|
||||
return v.strip()
|
||||
|
||||
|
||||
class AudioGenerationRequest(BaseModel):
|
||||
"""Request model for generating audio (podcast).
|
||||
|
||||
Attributes:
|
||||
instructions: Custom instructions for the podcast.
|
||||
format: Podcast format (deep-dive, brief, critique, debate).
|
||||
length: Audio length (short, default, long).
|
||||
language: Language code (en, it, es, fr, de, etc.).
|
||||
"""
|
||||
|
||||
model_config = ConfigDict(
|
||||
json_schema_extra={
|
||||
"example": {
|
||||
"instructions": "Make it engaging and accessible",
|
||||
"format": "deep-dive",
|
||||
"length": "long",
|
||||
"language": "en",
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
instructions: str | None = Field(
|
||||
None,
|
||||
max_length=500,
|
||||
description="Custom instructions for the podcast",
|
||||
examples=["Make it engaging and accessible"],
|
||||
)
|
||||
format: str = Field(
|
||||
"deep-dive",
|
||||
description="Podcast format",
|
||||
examples=["deep-dive", "brief", "critique", "debate"],
|
||||
)
|
||||
length: str = Field(
|
||||
"default",
|
||||
description="Audio length",
|
||||
examples=["short", "default", "long"],
|
||||
)
|
||||
language: str = Field(
|
||||
"en",
|
||||
description="Language code",
|
||||
examples=["en", "it", "es", "fr", "de"],
|
||||
)
|
||||
|
||||
@field_validator("format")
|
||||
@classmethod
|
||||
def validate_format(cls, v: str) -> str:
|
||||
"""Validate format."""
|
||||
allowed = {"deep-dive", "brief", "critique", "debate"}
|
||||
if v not in allowed:
|
||||
raise ValueError(f"Format must be one of: {allowed}")
|
||||
return v
|
||||
|
||||
@field_validator("length")
|
||||
@classmethod
|
||||
def validate_length(cls, v: str) -> str:
|
||||
"""Validate length."""
|
||||
allowed = {"short", "default", "long"}
|
||||
if v not in allowed:
|
||||
raise ValueError(f"Length must be one of: {allowed}")
|
||||
return v
|
||||
|
||||
|
||||
class VideoGenerationRequest(BaseModel):
|
||||
"""Request model for generating video.
|
||||
|
||||
Attributes:
|
||||
instructions: Custom instructions for the video.
|
||||
style: Video style (whiteboard, classic, anime, etc.).
|
||||
language: Language code.
|
||||
"""
|
||||
|
||||
model_config = ConfigDict(
|
||||
json_schema_extra={
|
||||
"example": {
|
||||
"instructions": "Create an engaging explainer video",
|
||||
"style": "whiteboard",
|
||||
"language": "en",
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
instructions: str | None = Field(
|
||||
None,
|
||||
max_length=500,
|
||||
description="Custom instructions for the video",
|
||||
examples=["Create an engaging explainer video"],
|
||||
)
|
||||
style: str = Field(
|
||||
"auto",
|
||||
description="Video style",
|
||||
examples=["whiteboard", "classic", "anime", "kawaii", "watercolor"],
|
||||
)
|
||||
language: str = Field(
|
||||
"en",
|
||||
description="Language code",
|
||||
examples=["en", "it", "es", "fr", "de"],
|
||||
)
|
||||
|
||||
|
||||
class SlideDeckGenerationRequest(BaseModel):
|
||||
"""Request model for generating slide deck.
|
||||
|
||||
Attributes:
|
||||
format: Slide format (detailed, presenter).
|
||||
length: Length (default, short).
|
||||
"""
|
||||
|
||||
model_config = ConfigDict(
|
||||
json_schema_extra={
|
||||
"example": {
|
||||
"format": "detailed",
|
||||
"length": "default",
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
format: str = Field(
|
||||
"detailed",
|
||||
description="Slide format",
|
||||
examples=["detailed", "presenter"],
|
||||
)
|
||||
length: str = Field(
|
||||
"default",
|
||||
description="Length",
|
||||
examples=["default", "short"],
|
||||
)
|
||||
|
||||
@field_validator("format")
|
||||
@classmethod
|
||||
def validate_format(cls, v: str) -> str:
|
||||
"""Validate format."""
|
||||
allowed = {"detailed", "presenter"}
|
||||
if v not in allowed:
|
||||
raise ValueError(f"Format must be one of: {allowed}")
|
||||
return v
|
||||
|
||||
|
||||
class InfographicGenerationRequest(BaseModel):
|
||||
"""Request model for generating infographic.
|
||||
|
||||
Attributes:
|
||||
orientation: Orientation (landscape, portrait, square).
|
||||
detail: Detail level (concise, standard, detailed).
|
||||
style: Visual style.
|
||||
"""
|
||||
|
||||
model_config = ConfigDict(
|
||||
json_schema_extra={
|
||||
"example": {
|
||||
"orientation": "portrait",
|
||||
"detail": "detailed",
|
||||
"style": "professional",
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
orientation: str = Field(
|
||||
"landscape",
|
||||
description="Orientation",
|
||||
examples=["landscape", "portrait", "square"],
|
||||
)
|
||||
detail: str = Field(
|
||||
"standard",
|
||||
description="Detail level",
|
||||
examples=["concise", "standard", "detailed"],
|
||||
)
|
||||
style: str = Field(
|
||||
"auto",
|
||||
description="Visual style",
|
||||
examples=["professional", "editorial", "scientific", "sketch-note"],
|
||||
)
|
||||
|
||||
|
||||
class QuizGenerationRequest(BaseModel):
|
||||
"""Request model for generating quiz.
|
||||
|
||||
Attributes:
|
||||
difficulty: Difficulty level (easy, medium, hard).
|
||||
quantity: Number of questions (fewer, standard, more).
|
||||
"""
|
||||
|
||||
model_config = ConfigDict(
|
||||
json_schema_extra={
|
||||
"example": {
|
||||
"difficulty": "medium",
|
||||
"quantity": "standard",
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
difficulty: str = Field(
|
||||
"medium",
|
||||
description="Difficulty level",
|
||||
examples=["easy", "medium", "hard"],
|
||||
)
|
||||
quantity: str = Field(
|
||||
"standard",
|
||||
description="Number of questions",
|
||||
examples=["fewer", "standard", "more"],
|
||||
)
|
||||
|
||||
@field_validator("difficulty")
|
||||
@classmethod
|
||||
def validate_difficulty(cls, v: str) -> str:
|
||||
"""Validate difficulty."""
|
||||
allowed = {"easy", "medium", "hard"}
|
||||
if v not in allowed:
|
||||
raise ValueError(f"Difficulty must be one of: {allowed}")
|
||||
return v
|
||||
|
||||
@field_validator("quantity")
|
||||
@classmethod
|
||||
def validate_quantity(cls, v: str) -> str:
|
||||
"""Validate quantity."""
|
||||
allowed = {"fewer", "standard", "more"}
|
||||
if v not in allowed:
|
||||
raise ValueError(f"Quantity must be one of: {allowed}")
|
||||
return v
|
||||
|
||||
|
||||
class FlashcardsGenerationRequest(BaseModel):
|
||||
"""Request model for generating flashcards.
|
||||
|
||||
Attributes:
|
||||
difficulty: Difficulty level (easy, medium, hard).
|
||||
quantity: Number of flashcards (fewer, standard, more).
|
||||
"""
|
||||
|
||||
model_config = ConfigDict(
|
||||
json_schema_extra={
|
||||
"example": {
|
||||
"difficulty": "hard",
|
||||
"quantity": "more",
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
difficulty: str = Field(
|
||||
"medium",
|
||||
description="Difficulty level",
|
||||
examples=["easy", "medium", "hard"],
|
||||
)
|
||||
quantity: str = Field(
|
||||
"standard",
|
||||
description="Number of flashcards",
|
||||
examples=["fewer", "standard", "more"],
|
||||
)
|
||||
|
||||
@field_validator("difficulty")
|
||||
@classmethod
|
||||
def validate_difficulty(cls, v: str) -> str:
|
||||
"""Validate difficulty."""
|
||||
allowed = {"easy", "medium", "hard"}
|
||||
if v not in allowed:
|
||||
raise ValueError(f"Difficulty must be one of: {allowed}")
|
||||
return v
|
||||
|
||||
@field_validator("quantity")
|
||||
@classmethod
|
||||
def validate_quantity(cls, v: str) -> str:
|
||||
"""Validate quantity."""
|
||||
allowed = {"fewer", "standard", "more"}
|
||||
if v not in allowed:
|
||||
raise ValueError(f"Quantity must be one of: {allowed}")
|
||||
return v
|
||||
|
||||
|
||||
class ReportGenerationRequest(BaseModel):
|
||||
"""Request model for generating report.
|
||||
|
||||
Attributes:
|
||||
format: Report format (summary, detailed, executive).
|
||||
"""
|
||||
|
||||
model_config = ConfigDict(
|
||||
json_schema_extra={
|
||||
"example": {
|
||||
"format": "detailed",
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
format: str = Field(
|
||||
"detailed",
|
||||
description="Report format",
|
||||
examples=["summary", "detailed", "executive"],
|
||||
)
|
||||
|
||||
@field_validator("format")
|
||||
@classmethod
|
||||
def validate_format(cls, v: str) -> str:
|
||||
"""Validate format."""
|
||||
allowed = {"summary", "detailed", "executive"}
|
||||
if v not in allowed:
|
||||
raise ValueError(f"Format must be one of: {allowed}")
|
||||
return v
|
||||
|
||||
|
||||
class DataTableGenerationRequest(BaseModel):
|
||||
"""Request model for generating data table.
|
||||
|
||||
Attributes:
|
||||
description: Description of what data to extract.
|
||||
"""
|
||||
|
||||
model_config = ConfigDict(
|
||||
json_schema_extra={
|
||||
"example": {
|
||||
"description": "Compare different machine learning approaches",
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
description: str | None = Field(
|
||||
None,
|
||||
max_length=500,
|
||||
description="Description of what data to extract",
|
||||
examples=["Compare different machine learning approaches"],
|
||||
)
|
||||
|
||||
@@ -420,6 +420,148 @@ class HealthStatus(BaseModel):
|
||||
)
|
||||
|
||||
|
||||
class Artifact(BaseModel):
|
||||
"""Artifact (generated content) model.
|
||||
|
||||
Attributes:
|
||||
id: Unique artifact identifier.
|
||||
notebook_id: Parent notebook ID.
|
||||
type: Artifact type (audio, video, quiz, etc.).
|
||||
title: Artifact title.
|
||||
status: Processing status (pending, processing, completed, failed).
|
||||
created_at: Creation timestamp.
|
||||
completed_at: Completion timestamp (None if not completed).
|
||||
download_url: Download URL (None if not completed).
|
||||
"""
|
||||
|
||||
model_config = ConfigDict(
|
||||
json_schema_extra={
|
||||
"example": {
|
||||
"id": "550e8400-e29b-41d4-a716-446655440010",
|
||||
"notebook_id": "550e8400-e29b-41d4-a716-446655440000",
|
||||
"type": "audio",
|
||||
"title": "AI Research Podcast",
|
||||
"status": "completed",
|
||||
"created_at": "2026-04-06T10:00:00Z",
|
||||
"completed_at": "2026-04-06T10:30:00Z",
|
||||
"download_url": "https://example.com/download/123",
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
id: UUID = Field(
|
||||
...,
|
||||
description="Unique artifact identifier",
|
||||
examples=["550e8400-e29b-41d4-a716-446655440010"],
|
||||
)
|
||||
notebook_id: UUID = Field(
|
||||
...,
|
||||
description="Parent notebook ID",
|
||||
examples=["550e8400-e29b-41d4-a716-446655440000"],
|
||||
)
|
||||
type: str = Field(
|
||||
...,
|
||||
description="Artifact type",
|
||||
examples=[
|
||||
"audio",
|
||||
"video",
|
||||
"quiz",
|
||||
"flashcards",
|
||||
"slide-deck",
|
||||
"infographic",
|
||||
"report",
|
||||
"mind-map",
|
||||
"data-table",
|
||||
],
|
||||
)
|
||||
title: str = Field(
|
||||
...,
|
||||
description="Artifact title",
|
||||
examples=["AI Research Podcast"],
|
||||
)
|
||||
status: str = Field(
|
||||
...,
|
||||
description="Processing status",
|
||||
examples=["pending", "processing", "completed", "failed"],
|
||||
)
|
||||
created_at: datetime = Field(
|
||||
...,
|
||||
description="Creation timestamp",
|
||||
examples=["2026-04-06T10:00:00Z"],
|
||||
)
|
||||
completed_at: datetime | None = Field(
|
||||
None,
|
||||
description="Completion timestamp (None if not completed)",
|
||||
examples=["2026-04-06T10:30:00Z"],
|
||||
)
|
||||
download_url: str | None = Field(
|
||||
None,
|
||||
description="Download URL (None if not completed)",
|
||||
examples=["https://example.com/download/123"],
|
||||
)
|
||||
|
||||
|
||||
class ArtifactList(BaseModel):
|
||||
"""List of artifacts.
|
||||
|
||||
Attributes:
|
||||
items: List of artifacts.
|
||||
pagination: Pagination metadata.
|
||||
"""
|
||||
|
||||
items: list[Artifact] = Field(
|
||||
...,
|
||||
description="List of artifacts",
|
||||
)
|
||||
pagination: PaginationMeta = Field(
|
||||
...,
|
||||
description="Pagination metadata",
|
||||
)
|
||||
|
||||
|
||||
class GenerationResponse(BaseModel):
|
||||
"""Response from content generation request.
|
||||
|
||||
Attributes:
|
||||
artifact_id: ID of the created artifact.
|
||||
status: Current status (usually 'pending' or 'processing').
|
||||
message: Human-readable status message.
|
||||
estimated_time_seconds: Estimated completion time.
|
||||
"""
|
||||
|
||||
model_config = ConfigDict(
|
||||
json_schema_extra={
|
||||
"example": {
|
||||
"artifact_id": "550e8400-e29b-41d4-a716-446655440010",
|
||||
"status": "processing",
|
||||
"message": "Audio generation started",
|
||||
"estimated_time_seconds": 600,
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
artifact_id: UUID = Field(
|
||||
...,
|
||||
description="ID of the created artifact",
|
||||
examples=["550e8400-e29b-41d4-a716-446655440010"],
|
||||
)
|
||||
status: str = Field(
|
||||
...,
|
||||
description="Current status",
|
||||
examples=["pending", "processing", "completed"],
|
||||
)
|
||||
message: str = Field(
|
||||
...,
|
||||
description="Human-readable status message",
|
||||
examples=["Audio generation started"],
|
||||
)
|
||||
estimated_time_seconds: int | None = Field(
|
||||
None,
|
||||
description="Estimated completion time in seconds",
|
||||
examples=[600, 900, 1200],
|
||||
)
|
||||
|
||||
|
||||
class SourceReference(BaseModel):
|
||||
"""Source reference in chat response.
|
||||
|
||||
|
||||
563
src/notebooklm_agent/api/routes/generation.py
Normal file
563
src/notebooklm_agent/api/routes/generation.py
Normal file
@@ -0,0 +1,563 @@
|
||||
"""Generation API routes.
|
||||
|
||||
This module contains API endpoints for content generation.
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
from uuid import uuid4
|
||||
|
||||
from fastapi import APIRouter, HTTPException, status
|
||||
|
||||
from notebooklm_agent.api.models.requests import (
|
||||
AudioGenerationRequest,
|
||||
DataTableGenerationRequest,
|
||||
FlashcardsGenerationRequest,
|
||||
InfographicGenerationRequest,
|
||||
QuizGenerationRequest,
|
||||
ReportGenerationRequest,
|
||||
SlideDeckGenerationRequest,
|
||||
VideoGenerationRequest,
|
||||
)
|
||||
from notebooklm_agent.api.models.responses import (
|
||||
ApiResponse,
|
||||
Artifact,
|
||||
ArtifactList,
|
||||
GenerationResponse,
|
||||
PaginationMeta,
|
||||
ResponseMeta,
|
||||
)
|
||||
from notebooklm_agent.core.exceptions import NotebookLMError, NotFoundError
|
||||
from notebooklm_agent.services.artifact_service import ArtifactService
|
||||
|
||||
router = APIRouter(tags=["generation"])
|
||||
|
||||
|
||||
async def get_artifact_service() -> ArtifactService:
|
||||
"""Get artifact service instance.
|
||||
|
||||
Returns:
|
||||
ArtifactService instance.
|
||||
"""
|
||||
return ArtifactService()
|
||||
|
||||
|
||||
def _validate_notebook_id(notebook_id: str) -> None:
|
||||
"""Validate notebook ID format.
|
||||
|
||||
Args:
|
||||
notebook_id: Notebook ID string.
|
||||
|
||||
Raises:
|
||||
HTTPException: If invalid format.
|
||||
"""
|
||||
from uuid import UUID
|
||||
|
||||
try:
|
||||
UUID(notebook_id)
|
||||
except ValueError:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {
|
||||
"code": "VALIDATION_ERROR",
|
||||
"message": "Invalid notebook ID format",
|
||||
"details": [],
|
||||
},
|
||||
"meta": {
|
||||
"timestamp": datetime.utcnow().isoformat(),
|
||||
"request_id": str(uuid4()),
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/{notebook_id}/generate/audio",
|
||||
response_model=ApiResponse[GenerationResponse],
|
||||
status_code=status.HTTP_202_ACCEPTED,
|
||||
summary="Generate audio podcast",
|
||||
description="Generate an audio podcast from notebook sources.",
|
||||
)
|
||||
async def generate_audio(notebook_id: str, data: AudioGenerationRequest):
|
||||
"""Generate audio podcast."""
|
||||
_validate_notebook_id(notebook_id)
|
||||
from uuid import UUID
|
||||
|
||||
try:
|
||||
service = await get_artifact_service()
|
||||
result = await service.generate_audio(
|
||||
UUID(notebook_id),
|
||||
data.instructions,
|
||||
data.format,
|
||||
data.length,
|
||||
data.language,
|
||||
)
|
||||
return ApiResponse(
|
||||
success=True,
|
||||
data=result,
|
||||
error=None,
|
||||
meta=ResponseMeta(timestamp=datetime.utcnow(), request_id=uuid4()),
|
||||
)
|
||||
except NotFoundError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
except NotebookLMError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/{notebook_id}/generate/video",
|
||||
response_model=ApiResponse[GenerationResponse],
|
||||
status_code=status.HTTP_202_ACCEPTED,
|
||||
summary="Generate video",
|
||||
description="Generate a video from notebook sources.",
|
||||
)
|
||||
async def generate_video(notebook_id: str, data: VideoGenerationRequest):
|
||||
"""Generate video."""
|
||||
_validate_notebook_id(notebook_id)
|
||||
from uuid import UUID
|
||||
|
||||
try:
|
||||
service = await get_artifact_service()
|
||||
result = await service.generate_video(
|
||||
UUID(notebook_id),
|
||||
data.instructions,
|
||||
data.style,
|
||||
data.language,
|
||||
)
|
||||
return ApiResponse(
|
||||
success=True,
|
||||
data=result,
|
||||
error=None,
|
||||
meta=ResponseMeta(timestamp=datetime.utcnow(), request_id=uuid4()),
|
||||
)
|
||||
except NotFoundError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
except NotebookLMError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/{notebook_id}/generate/slide-deck",
|
||||
response_model=ApiResponse[GenerationResponse],
|
||||
status_code=status.HTTP_202_ACCEPTED,
|
||||
summary="Generate slide deck",
|
||||
description="Generate a slide deck from notebook sources.",
|
||||
)
|
||||
async def generate_slide_deck(notebook_id: str, data: SlideDeckGenerationRequest):
|
||||
"""Generate slide deck."""
|
||||
_validate_notebook_id(notebook_id)
|
||||
from uuid import UUID
|
||||
|
||||
try:
|
||||
service = await get_artifact_service()
|
||||
result = await service.generate_slide_deck(
|
||||
UUID(notebook_id),
|
||||
data.format,
|
||||
data.length,
|
||||
)
|
||||
return ApiResponse(
|
||||
success=True,
|
||||
data=result,
|
||||
error=None,
|
||||
meta=ResponseMeta(timestamp=datetime.utcnow(), request_id=uuid4()),
|
||||
)
|
||||
except NotFoundError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
except NotebookLMError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/{notebook_id}/generate/infographic",
|
||||
response_model=ApiResponse[GenerationResponse],
|
||||
status_code=status.HTTP_202_ACCEPTED,
|
||||
summary="Generate infographic",
|
||||
description="Generate an infographic from notebook sources.",
|
||||
)
|
||||
async def generate_infographic(notebook_id: str, data: InfographicGenerationRequest):
|
||||
"""Generate infographic."""
|
||||
_validate_notebook_id(notebook_id)
|
||||
from uuid import UUID
|
||||
|
||||
try:
|
||||
service = await get_artifact_service()
|
||||
result = await service.generate_infographic(
|
||||
UUID(notebook_id),
|
||||
data.orientation,
|
||||
data.detail,
|
||||
data.style,
|
||||
)
|
||||
return ApiResponse(
|
||||
success=True,
|
||||
data=result,
|
||||
error=None,
|
||||
meta=ResponseMeta(timestamp=datetime.utcnow(), request_id=uuid4()),
|
||||
)
|
||||
except NotFoundError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
except NotebookLMError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/{notebook_id}/generate/quiz",
|
||||
response_model=ApiResponse[GenerationResponse],
|
||||
status_code=status.HTTP_202_ACCEPTED,
|
||||
summary="Generate quiz",
|
||||
description="Generate a quiz from notebook sources.",
|
||||
)
|
||||
async def generate_quiz(notebook_id: str, data: QuizGenerationRequest):
|
||||
"""Generate quiz."""
|
||||
_validate_notebook_id(notebook_id)
|
||||
from uuid import UUID
|
||||
|
||||
try:
|
||||
service = await get_artifact_service()
|
||||
result = await service.generate_quiz(
|
||||
UUID(notebook_id),
|
||||
data.difficulty,
|
||||
data.quantity,
|
||||
)
|
||||
return ApiResponse(
|
||||
success=True,
|
||||
data=result,
|
||||
error=None,
|
||||
meta=ResponseMeta(timestamp=datetime.utcnow(), request_id=uuid4()),
|
||||
)
|
||||
except NotFoundError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
except NotebookLMError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/{notebook_id}/generate/flashcards",
|
||||
response_model=ApiResponse[GenerationResponse],
|
||||
status_code=status.HTTP_202_ACCEPTED,
|
||||
summary="Generate flashcards",
|
||||
description="Generate flashcards from notebook sources.",
|
||||
)
|
||||
async def generate_flashcards(notebook_id: str, data: FlashcardsGenerationRequest):
|
||||
"""Generate flashcards."""
|
||||
_validate_notebook_id(notebook_id)
|
||||
from uuid import UUID
|
||||
|
||||
try:
|
||||
service = await get_artifact_service()
|
||||
result = await service.generate_flashcards(
|
||||
UUID(notebook_id),
|
||||
data.difficulty,
|
||||
data.quantity,
|
||||
)
|
||||
return ApiResponse(
|
||||
success=True,
|
||||
data=result,
|
||||
error=None,
|
||||
meta=ResponseMeta(timestamp=datetime.utcnow(), request_id=uuid4()),
|
||||
)
|
||||
except NotFoundError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
except NotebookLMError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/{notebook_id}/generate/report",
|
||||
response_model=ApiResponse[GenerationResponse],
|
||||
status_code=status.HTTP_202_ACCEPTED,
|
||||
summary="Generate report",
|
||||
description="Generate a report from notebook sources.",
|
||||
)
|
||||
async def generate_report(notebook_id: str, data: ReportGenerationRequest):
|
||||
"""Generate report."""
|
||||
_validate_notebook_id(notebook_id)
|
||||
from uuid import UUID
|
||||
|
||||
try:
|
||||
service = await get_artifact_service()
|
||||
result = await service.generate_report(
|
||||
UUID(notebook_id),
|
||||
data.format,
|
||||
)
|
||||
return ApiResponse(
|
||||
success=True,
|
||||
data=result,
|
||||
error=None,
|
||||
meta=ResponseMeta(timestamp=datetime.utcnow(), request_id=uuid4()),
|
||||
)
|
||||
except NotFoundError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
except NotebookLMError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/{notebook_id}/generate/mind-map",
|
||||
response_model=ApiResponse[GenerationResponse],
|
||||
status_code=status.HTTP_202_ACCEPTED,
|
||||
summary="Generate mind map",
|
||||
description="Generate a mind map from notebook sources (instant).",
|
||||
)
|
||||
async def generate_mind_map(notebook_id: str):
|
||||
"""Generate mind map."""
|
||||
_validate_notebook_id(notebook_id)
|
||||
from uuid import UUID
|
||||
|
||||
try:
|
||||
service = await get_artifact_service()
|
||||
result = await service.generate_mind_map(UUID(notebook_id))
|
||||
return ApiResponse(
|
||||
success=True,
|
||||
data=result,
|
||||
error=None,
|
||||
meta=ResponseMeta(timestamp=datetime.utcnow(), request_id=uuid4()),
|
||||
)
|
||||
except NotFoundError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
except NotebookLMError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/{notebook_id}/generate/data-table",
|
||||
response_model=ApiResponse[GenerationResponse],
|
||||
status_code=status.HTTP_202_ACCEPTED,
|
||||
summary="Generate data table",
|
||||
description="Generate a data table from notebook sources.",
|
||||
)
|
||||
async def generate_data_table(notebook_id: str, data: DataTableGenerationRequest):
|
||||
"""Generate data table."""
|
||||
_validate_notebook_id(notebook_id)
|
||||
from uuid import UUID
|
||||
|
||||
try:
|
||||
service = await get_artifact_service()
|
||||
result = await service.generate_data_table(
|
||||
UUID(notebook_id),
|
||||
data.description,
|
||||
)
|
||||
return ApiResponse(
|
||||
success=True,
|
||||
data=result,
|
||||
error=None,
|
||||
meta=ResponseMeta(timestamp=datetime.utcnow(), request_id=uuid4()),
|
||||
)
|
||||
except NotFoundError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
except NotebookLMError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/{notebook_id}/artifacts",
|
||||
response_model=ApiResponse[ArtifactList],
|
||||
summary="List artifacts",
|
||||
description="List all generated artifacts for a notebook.",
|
||||
)
|
||||
async def list_artifacts(notebook_id: str):
|
||||
"""List artifacts."""
|
||||
_validate_notebook_id(notebook_id)
|
||||
from uuid import UUID
|
||||
|
||||
try:
|
||||
service = await get_artifact_service()
|
||||
artifacts = await service.list_artifacts(UUID(notebook_id))
|
||||
return ApiResponse(
|
||||
success=True,
|
||||
data=ArtifactList(
|
||||
items=artifacts,
|
||||
pagination=PaginationMeta(
|
||||
total=len(artifacts),
|
||||
limit=max(len(artifacts), 1),
|
||||
offset=0,
|
||||
has_more=False,
|
||||
),
|
||||
),
|
||||
error=None,
|
||||
meta=ResponseMeta(timestamp=datetime.utcnow(), request_id=uuid4()),
|
||||
)
|
||||
except NotFoundError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
except NotebookLMError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/{notebook_id}/artifacts/{artifact_id}/status",
|
||||
response_model=ApiResponse[Artifact],
|
||||
summary="Get artifact status",
|
||||
description="Get the current status of an artifact.",
|
||||
)
|
||||
async def get_artifact_status(notebook_id: str, artifact_id: str):
|
||||
"""Get artifact status."""
|
||||
_validate_notebook_id(notebook_id)
|
||||
from uuid import UUID
|
||||
|
||||
try:
|
||||
service = await get_artifact_service()
|
||||
artifact = await service.get_status(UUID(notebook_id), artifact_id)
|
||||
return ApiResponse(
|
||||
success=True,
|
||||
data=artifact,
|
||||
error=None,
|
||||
meta=ResponseMeta(timestamp=datetime.utcnow(), request_id=uuid4()),
|
||||
)
|
||||
except NotFoundError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
except NotebookLMError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||||
detail={
|
||||
"success": False,
|
||||
"error": {"code": e.code, "message": e.message, "details": []},
|
||||
"meta": {"timestamp": datetime.utcnow().isoformat(), "request_id": str(uuid4())},
|
||||
},
|
||||
)
|
||||
449
src/notebooklm_agent/services/artifact_service.py
Normal file
449
src/notebooklm_agent/services/artifact_service.py
Normal file
@@ -0,0 +1,449 @@
|
||||
"""Artifact service for content generation.
|
||||
|
||||
This module contains the ArtifactService class which handles
|
||||
all business logic for generating content (audio, video, etc.).
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
from notebooklm_agent.api.models.responses import Artifact, GenerationResponse
|
||||
from notebooklm_agent.core.exceptions import NotebookLMError, NotFoundError, ValidationError
|
||||
|
||||
|
||||
class ArtifactService:
|
||||
"""Service for artifact/content generation operations.
|
||||
|
||||
This service handles all business logic for generating content
|
||||
including audio, video, slides, quizzes, etc.
|
||||
|
||||
Attributes:
|
||||
_client: The notebooklm-py client instance.
|
||||
"""
|
||||
|
||||
# Estimated times in seconds for each artifact type
|
||||
ESTIMATED_TIMES = {
|
||||
"audio": 600, # 10 minutes
|
||||
"video": 1800, # 30 minutes
|
||||
"slide-deck": 300, # 5 minutes
|
||||
"infographic": 300, # 5 minutes
|
||||
"quiz": 600, # 10 minutes
|
||||
"flashcards": 600, # 10 minutes
|
||||
"report": 300, # 5 minutes
|
||||
"mind-map": 10, # Instant
|
||||
"data-table": 60, # 1 minute
|
||||
}
|
||||
|
||||
def __init__(self, client: Any = None) -> None:
|
||||
"""Initialize the artifact service.
|
||||
|
||||
Args:
|
||||
client: Optional notebooklm-py client instance.
|
||||
If not provided, will be created on first use.
|
||||
"""
|
||||
self._client = client
|
||||
|
||||
async def _get_client(self) -> Any:
|
||||
"""Get or create notebooklm-py client.
|
||||
|
||||
Returns:
|
||||
The notebooklm-py client instance.
|
||||
"""
|
||||
if self._client is None:
|
||||
from notebooklm import NotebookLMClient
|
||||
|
||||
self._client = await NotebookLMClient.from_storage()
|
||||
return self._client
|
||||
|
||||
def _get_estimated_time(self, artifact_type: str) -> int:
|
||||
"""Get estimated generation time for artifact type.
|
||||
|
||||
Args:
|
||||
artifact_type: Type of artifact.
|
||||
|
||||
Returns:
|
||||
Estimated time in seconds.
|
||||
"""
|
||||
return self.ESTIMATED_TIMES.get(artifact_type, 300)
|
||||
|
||||
async def _start_generation(
|
||||
self,
|
||||
notebook_id: UUID,
|
||||
artifact_type: str,
|
||||
title: str,
|
||||
generation_method: str,
|
||||
params: dict,
|
||||
) -> GenerationResponse:
|
||||
"""Start content generation.
|
||||
|
||||
Args:
|
||||
notebook_id: The notebook ID.
|
||||
artifact_type: Type of artifact (audio, video, etc.).
|
||||
title: Artifact title.
|
||||
generation_method: Method name to call on notebook.
|
||||
params: Parameters for generation.
|
||||
|
||||
Returns:
|
||||
Generation response with artifact ID.
|
||||
|
||||
Raises:
|
||||
NotFoundError: If notebook not found.
|
||||
NotebookLMError: If external API fails.
|
||||
"""
|
||||
try:
|
||||
client = await self._get_client()
|
||||
notebook = await client.notebooks.get(str(notebook_id))
|
||||
|
||||
# Get the generation method
|
||||
generator = getattr(notebook, generation_method, None)
|
||||
if not generator:
|
||||
raise NotebookLMError(f"Generation method '{generation_method}' not available")
|
||||
|
||||
# Start generation
|
||||
result = await generator(**params)
|
||||
|
||||
artifact_id = getattr(result, "id", str(uuid4()))
|
||||
status = getattr(result, "status", "processing")
|
||||
|
||||
return GenerationResponse(
|
||||
artifact_id=artifact_id,
|
||||
status=status,
|
||||
message=f"{artifact_type.title()} generation started",
|
||||
estimated_time_seconds=self._get_estimated_time(artifact_type),
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
error_str = str(e).lower()
|
||||
if "not found" in error_str:
|
||||
raise NotFoundError("Notebook", str(notebook_id))
|
||||
raise NotebookLMError(f"Failed to start {artifact_type} generation: {e}")
|
||||
|
||||
async def generate_audio(
|
||||
self,
|
||||
notebook_id: UUID,
|
||||
instructions: str | None = None,
|
||||
format: str = "deep-dive",
|
||||
length: str = "default",
|
||||
language: str = "en",
|
||||
) -> GenerationResponse:
|
||||
"""Generate audio podcast.
|
||||
|
||||
Args:
|
||||
notebook_id: The notebook ID.
|
||||
instructions: Custom instructions.
|
||||
format: Podcast format.
|
||||
length: Audio length.
|
||||
language: Language code.
|
||||
|
||||
Returns:
|
||||
Generation response.
|
||||
"""
|
||||
params = {
|
||||
"format": format,
|
||||
"length": length,
|
||||
"language": language,
|
||||
}
|
||||
if instructions:
|
||||
params["instructions"] = instructions
|
||||
|
||||
return await self._start_generation(
|
||||
notebook_id,
|
||||
"audio",
|
||||
"Generated Podcast",
|
||||
"generate_audio",
|
||||
params,
|
||||
)
|
||||
|
||||
async def generate_video(
|
||||
self,
|
||||
notebook_id: UUID,
|
||||
instructions: str | None = None,
|
||||
style: str = "auto",
|
||||
language: str = "en",
|
||||
) -> GenerationResponse:
|
||||
"""Generate video.
|
||||
|
||||
Args:
|
||||
notebook_id: The notebook ID.
|
||||
instructions: Custom instructions.
|
||||
style: Video style.
|
||||
language: Language code.
|
||||
|
||||
Returns:
|
||||
Generation response.
|
||||
"""
|
||||
params = {
|
||||
"style": style,
|
||||
"language": language,
|
||||
}
|
||||
if instructions:
|
||||
params["instructions"] = instructions
|
||||
|
||||
return await self._start_generation(
|
||||
notebook_id,
|
||||
"video",
|
||||
"Generated Video",
|
||||
"generate_video",
|
||||
params,
|
||||
)
|
||||
|
||||
async def generate_slide_deck(
|
||||
self,
|
||||
notebook_id: UUID,
|
||||
format: str = "detailed",
|
||||
length: str = "default",
|
||||
) -> GenerationResponse:
|
||||
"""Generate slide deck.
|
||||
|
||||
Args:
|
||||
notebook_id: The notebook ID.
|
||||
format: Slide format.
|
||||
length: Length.
|
||||
|
||||
Returns:
|
||||
Generation response.
|
||||
"""
|
||||
return await self._start_generation(
|
||||
notebook_id,
|
||||
"slide-deck",
|
||||
"Generated Slide Deck",
|
||||
"generate_slide_deck",
|
||||
{"format": format, "length": length},
|
||||
)
|
||||
|
||||
async def generate_infographic(
|
||||
self,
|
||||
notebook_id: UUID,
|
||||
orientation: str = "landscape",
|
||||
detail: str = "standard",
|
||||
style: str = "auto",
|
||||
) -> GenerationResponse:
|
||||
"""Generate infographic.
|
||||
|
||||
Args:
|
||||
notebook_id: The notebook ID.
|
||||
orientation: Orientation.
|
||||
detail: Detail level.
|
||||
style: Visual style.
|
||||
|
||||
Returns:
|
||||
Generation response.
|
||||
"""
|
||||
return await self._start_generation(
|
||||
notebook_id,
|
||||
"infographic",
|
||||
"Generated Infographic",
|
||||
"generate_infographic",
|
||||
{"orientation": orientation, "detail": detail, "style": style},
|
||||
)
|
||||
|
||||
async def generate_quiz(
|
||||
self,
|
||||
notebook_id: UUID,
|
||||
difficulty: str = "medium",
|
||||
quantity: str = "standard",
|
||||
) -> GenerationResponse:
|
||||
"""Generate quiz.
|
||||
|
||||
Args:
|
||||
notebook_id: The notebook ID.
|
||||
difficulty: Difficulty level.
|
||||
quantity: Number of questions.
|
||||
|
||||
Returns:
|
||||
Generation response.
|
||||
"""
|
||||
return await self._start_generation(
|
||||
notebook_id,
|
||||
"quiz",
|
||||
"Generated Quiz",
|
||||
"generate_quiz",
|
||||
{"difficulty": difficulty, "quantity": quantity},
|
||||
)
|
||||
|
||||
async def generate_flashcards(
|
||||
self,
|
||||
notebook_id: UUID,
|
||||
difficulty: str = "medium",
|
||||
quantity: str = "standard",
|
||||
) -> GenerationResponse:
|
||||
"""Generate flashcards.
|
||||
|
||||
Args:
|
||||
notebook_id: The notebook ID.
|
||||
difficulty: Difficulty level.
|
||||
quantity: Number of flashcards.
|
||||
|
||||
Returns:
|
||||
Generation response.
|
||||
"""
|
||||
return await self._start_generation(
|
||||
notebook_id,
|
||||
"flashcards",
|
||||
"Generated Flashcards",
|
||||
"generate_flashcards",
|
||||
{"difficulty": difficulty, "quantity": quantity},
|
||||
)
|
||||
|
||||
async def generate_report(
|
||||
self,
|
||||
notebook_id: UUID,
|
||||
format: str = "detailed",
|
||||
) -> GenerationResponse:
|
||||
"""Generate report.
|
||||
|
||||
Args:
|
||||
notebook_id: The notebook ID.
|
||||
format: Report format.
|
||||
|
||||
Returns:
|
||||
Generation response.
|
||||
"""
|
||||
return await self._start_generation(
|
||||
notebook_id,
|
||||
"report",
|
||||
"Generated Report",
|
||||
"generate_report",
|
||||
{"format": format},
|
||||
)
|
||||
|
||||
async def generate_mind_map(
|
||||
self,
|
||||
notebook_id: UUID,
|
||||
) -> GenerationResponse:
|
||||
"""Generate mind map (instant).
|
||||
|
||||
Args:
|
||||
notebook_id: The notebook ID.
|
||||
|
||||
Returns:
|
||||
Generation response.
|
||||
"""
|
||||
try:
|
||||
client = await self._get_client()
|
||||
notebook = await client.notebooks.get(str(notebook_id))
|
||||
|
||||
# Mind map is usually instant
|
||||
result = await notebook.generate_mind_map()
|
||||
|
||||
artifact_id = getattr(result, "id", str(uuid4()))
|
||||
status = getattr(result, "status", "completed")
|
||||
|
||||
return GenerationResponse(
|
||||
artifact_id=artifact_id,
|
||||
status=status,
|
||||
message="Mind map generated",
|
||||
estimated_time_seconds=10,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
error_str = str(e).lower()
|
||||
if "not found" in error_str:
|
||||
raise NotFoundError("Notebook", str(notebook_id))
|
||||
raise NotebookLMError(f"Failed to generate mind map: {e}")
|
||||
|
||||
async def generate_data_table(
|
||||
self,
|
||||
notebook_id: UUID,
|
||||
description: str | None = None,
|
||||
) -> GenerationResponse:
|
||||
"""Generate data table.
|
||||
|
||||
Args:
|
||||
notebook_id: The notebook ID.
|
||||
description: Description of data to extract.
|
||||
|
||||
Returns:
|
||||
Generation response.
|
||||
"""
|
||||
params = {}
|
||||
if description:
|
||||
params["description"] = description
|
||||
|
||||
return await self._start_generation(
|
||||
notebook_id,
|
||||
"data-table",
|
||||
"Generated Data Table",
|
||||
"generate_data_table",
|
||||
params,
|
||||
)
|
||||
|
||||
async def list_artifacts(self, notebook_id: UUID) -> list[Artifact]:
|
||||
"""List artifacts for a notebook.
|
||||
|
||||
Args:
|
||||
notebook_id: The notebook ID.
|
||||
|
||||
Returns:
|
||||
List of artifacts.
|
||||
|
||||
Raises:
|
||||
NotFoundError: If notebook not found.
|
||||
NotebookLMError: If external API fails.
|
||||
"""
|
||||
try:
|
||||
client = await self._get_client()
|
||||
notebook = await client.notebooks.get(str(notebook_id))
|
||||
|
||||
artifacts = await notebook.artifacts.list()
|
||||
|
||||
result = []
|
||||
for art in artifacts:
|
||||
result.append(
|
||||
Artifact(
|
||||
id=getattr(art, "id", str(uuid4())),
|
||||
notebook_id=notebook_id,
|
||||
type=getattr(art, "type", "unknown"),
|
||||
title=getattr(art, "title", "Untitled"),
|
||||
status=getattr(art, "status", "pending"),
|
||||
created_at=getattr(art, "created_at", datetime.utcnow()),
|
||||
completed_at=getattr(art, "completed_at", None),
|
||||
download_url=getattr(art, "download_url", None),
|
||||
)
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
error_str = str(e).lower()
|
||||
if "not found" in error_str:
|
||||
raise NotFoundError("Notebook", str(notebook_id))
|
||||
raise NotebookLMError(f"Failed to list artifacts: {e}")
|
||||
|
||||
async def get_status(self, notebook_id: UUID, artifact_id: str) -> Artifact:
|
||||
"""Get artifact status.
|
||||
|
||||
Args:
|
||||
notebook_id: The notebook ID.
|
||||
artifact_id: The artifact ID.
|
||||
|
||||
Returns:
|
||||
Artifact with current status.
|
||||
|
||||
Raises:
|
||||
NotFoundError: If notebook or artifact not found.
|
||||
NotebookLMError: If external API fails.
|
||||
"""
|
||||
try:
|
||||
client = await self._get_client()
|
||||
notebook = await client.notebooks.get(str(notebook_id))
|
||||
|
||||
artifact = await notebook.artifacts.get(artifact_id)
|
||||
|
||||
return Artifact(
|
||||
id=getattr(artifact, "id", artifact_id),
|
||||
notebook_id=notebook_id,
|
||||
type=getattr(artifact, "type", "unknown"),
|
||||
title=getattr(artifact, "title", "Untitled"),
|
||||
status=getattr(artifact, "status", "pending"),
|
||||
created_at=getattr(artifact, "created_at", datetime.utcnow()),
|
||||
completed_at=getattr(artifact, "completed_at", None),
|
||||
download_url=getattr(artifact, "download_url", None),
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
error_str = str(e).lower()
|
||||
if "not found" in error_str:
|
||||
raise NotFoundError("Artifact", artifact_id)
|
||||
raise NotebookLMError(f"Failed to get artifact status: {e}")
|
||||
Reference in New Issue
Block a user