cjm-transcript-workflow-management 0.0.1
A FastHTML management interface for context graph documents to list, inspect, delete, and import/export graph spines produced by transcript decomposition workflows.
License: Apache-2.0
Requires Python: >=3.12
Keywords: nbdev jupyter notebook python

# cjm-transcript-workflow-management


<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->

## Install

``` bash
pip install cjm_transcript_workflow_management
```

## Project Structure

    nbs/
    ├── components/ (5)
    │   ├── document_detail.ipynb  # Document detail dashboard with info, stats, integrity checks, and samples
    │   ├── document_list.ipynb    # Document list table with toolbar and row actions
    │   ├── helpers.ipynb          # Shared rendering helpers for the management interface
    │   ├── import_controls.ipynb  # Import UI with file input, merge strategy selector, and result display
    │   └── page_renderer.ipynb    # Main management page renderer composing header, toolbar buttons, and document list
    ├── routes/ (5)
    │   ├── core.ipynb       # Request helpers for management routes
    │   ├── documents.ipynb  # Document list, detail, and delete routes
    │   ├── export_.ipynb    # Export routes for single document and full database JSON file downloads
    │   ├── import_.ipynb    # Import route for file upload with JSON validation and merge strategy
    │   └── init.ipynb       # Router assembly for management routes
    ├── services/ (1)
    │   └── management.ipynb  # Service layer wrapping graph plugin operations for document management
    ├── html_ids.ipynb  # HTML ID constants for the graph management interface
    ├── models.ipynb    # Data models for the graph management interface
    └── utils.ipynb     # Formatting utilities for the management interface

Total: 14 notebooks across 3 directories

## Module Dependencies

``` mermaid
graph LR
    components_document_detail[components.document_detail<br/>document_detail]
    components_document_list[components.document_list<br/>document_list]
    components_helpers[components.helpers<br/>helpers]
    components_import_controls[components.import_controls<br/>import_controls]
    components_page_renderer[components.page_renderer<br/>page_renderer]
    html_ids[html_ids<br/>html_ids]
    models[models<br/>Models]
    routes_core[routes.core<br/>core]
    routes_documents[routes.documents<br/>documents]
    routes_export_[routes.export_<br/>export_]
    routes_import_[routes.import_<br/>import_]
    routes_init[routes.init<br/>init]
    services_management[services.management<br/>services.management]
    utils[utils<br/>utils]

    components_document_detail --> html_ids
    components_document_detail --> models
    components_document_detail --> utils
    components_document_detail --> components_helpers
    components_document_list --> html_ids
    components_document_list --> utils
    components_document_list --> components_helpers
    components_document_list --> models
    components_import_controls --> html_ids
    components_import_controls --> components_helpers
    components_import_controls --> models
    components_page_renderer --> html_ids
    components_page_renderer --> components_document_list
    components_page_renderer --> models
    components_page_renderer --> components_helpers
    components_page_renderer --> components_import_controls
    routes_core --> services_management
    routes_documents --> html_ids
    routes_documents --> components_helpers
    routes_documents --> services_management
    routes_documents --> components_page_renderer
    routes_documents --> routes_core
    routes_documents --> components_document_list
    routes_documents --> components_document_detail
    routes_documents --> models
    routes_export_ --> services_management
    routes_export_ --> routes_core
    routes_import_ --> html_ids
    routes_import_ --> services_management
    routes_import_ --> models
    routes_import_ --> routes_core
    routes_import_ --> components_document_list
    routes_import_ --> components_import_controls
    routes_init --> routes_documents
    routes_init --> services_management
    routes_init --> routes_export_
    routes_init --> routes_import_
    routes_init --> models
    services_management --> models
    services_management --> utils
```

*40 cross-module dependencies detected*

## CLI Reference

No CLI commands found in this project.

## Module Overview

Detailed documentation for each module in the project:

### core (`core.ipynb`)

> Request helpers for management routes

#### Import

``` python
from cjm_transcript_workflow_management.routes.core import (
    DEBUG_MANAGEMENT_ROUTES
)
```

#### Variables

``` python
DEBUG_MANAGEMENT_ROUTES = False
```

### document_detail (`document_detail.ipynb`)

> Document detail dashboard with info, stats, integrity checks, and
> samples

#### Import

``` python
from cjm_transcript_workflow_management.components.document_detail import (
    render_detail_header,
    render_document_info,
    render_segment_stats,
    render_sources_info,
    render_integrity_checks,
    render_sample_segments,
    render_detail_scripts,
    render_document_detail,
    render_detail_error
)
```

#### Functions

``` python
def _render_stat_row(
    label:str,  # Label text
    value:str,  # Value text
) -> Any:  # Flexbox row element
    "Render a label-value row for stat display."
```

``` python
def _render_check_row(
    passed:bool,  # Whether the check passed
    label:str,  # Check description
    detail:str="",  # Optional detail text (e.g., counts)
) -> Any:  # Flexbox row with icon
    "Render a pass/fail check row with icon."
```

``` python
def render_detail_header(
    detail:DocumentDetail,  # Document detail data
    urls:ManagementUrls,  # URL bundle for route endpoints
) -> Any:  # Header element with navigation and actions
    "Render the detail view header with Back, Export, and Delete buttons."
```

``` python
def render_document_info(
    detail:DocumentDetail,  # Document detail data
) -> Any:  # Card element with document info
    "Render the document info card."
```

``` python
def render_segment_stats(
    detail:DocumentDetail,  # Document detail data
) -> Any:  # Card element with segment stats
    "Render the segment statistics card."
```

``` python
def render_sources_info(
    detail:DocumentDetail,  # Document detail data
) -> Any:  # Card element with source plugin info
    "Render the source traceability card."
```

``` python
def render_integrity_checks(
    detail:DocumentDetail,  # Document detail data
) -> Any:  # Card element with integrity check rows
    "Render the integrity checks card with pass/fail indicators."
```

``` python
def _render_sample_row(
    sample:SegmentSample,  # Segment sample data
) -> Any:  # Flexbox row with index, text, and timing
    "Render a single sample segment row."
```

``` python
def _render_sample_list(
    samples:List[SegmentSample],  # List of segment samples
    label:str,  # Section label (e.g., "First", "Last")
) -> Any:  # Flexbox column with label and rows
    "Render a labeled list of sample segments."
```

``` python
def render_sample_segments(
    detail:DocumentDetail,  # Document detail data
) -> Any:  # Card element with sample segment lists
    "Render the sample segments card with first and last segments."
```

``` python
def render_detail_scripts(
    urls:ManagementUrls,  # URL bundle for route endpoints
) -> Any:  # Script element
    "Render client-side JavaScript for delete from detail view."
```

``` python
def render_document_detail(
    detail:DocumentDetail,  # Document detail data
    urls:ManagementUrls,  # URL bundle for route endpoints
) -> Any:  # Complete detail dashboard
    "Render the complete document detail dashboard."
```

``` python
def render_detail_error(
    message:str="Document not found.",  # Error message
    urls:ManagementUrls=None,  # URL bundle for Back to List
) -> Any:  # Error state element
    "Render an error state for the detail view."
```

#### Variables

``` python
_CARD_CLS
```

### document_list (`document_list.ipynb`)

> Document list table with toolbar and row actions

#### Import

``` python
from cjm_transcript_workflow_management.components.document_list import (
    render_toolbar,
    render_document_row,
    render_document_table,
    render_list_scripts,
    render_document_list
)
```

#### Functions

``` python
def render_toolbar(
    urls:ManagementUrls,  # URL bundle for route endpoints
    doc_count:int=0,  # Number of documents in the list
) -> Any:  # Toolbar element
    "Render the document list toolbar with Select All and bulk actions."
```

``` python
def render_document_row(
    doc:DocumentSummary,  # Document summary data
    urls:ManagementUrls,  # URL bundle for route endpoints
) -> Any:  # Table row element
    "Render a single document row in the list table."
```

``` python
def render_document_table(
    documents:List[DocumentSummary],  # List of document summaries
    urls:ManagementUrls,  # URL bundle for route endpoints
) -> Any:  # Table element wrapped in scrollable container
    "Render the document list table."
```

``` python
def render_list_scripts(
    urls:ManagementUrls,  # URL bundle for route endpoints
) -> Any:  # Script element
    "Render client-side JavaScript for checkbox and modal management."
```

``` python
def render_document_list(
    documents:List[DocumentSummary],  # List of document summaries
    urls:ManagementUrls,  # URL bundle for route endpoints
) -> Any:  # Complete document list component
    "Render the complete document list with toolbar, table, and modals."
```

### documents (`documents.ipynb`)

> Document list, detail, and delete routes

#### Import

``` python
from cjm_transcript_workflow_management.routes.documents import (
    init_document_router
)
```

#### Functions

``` python
def init_document_router(
    service:ManagementService,  # Service for graph queries
    prefix:str,  # Route prefix (e.g., "/manage/documents")
    urls:ManagementUrls,  # URL bundle (populated after init)
) -> Tuple[APIRouter, Dict[str, Callable]]:  # (router, routes dict)
    "Initialize document list, detail, and delete routes."
```

### export\_ (`export_.ipynb`)

> Export routes for single document and full database JSON file
> downloads

#### Import

``` python
from cjm_transcript_workflow_management.routes.export_ import (
    init_export_router
)
```

#### Functions

``` python
def _sanitize_filename(
    name:str,  # Raw filename string
) -> str:  # Filesystem-safe filename
    "Remove characters unsafe for filenames."
```

``` python
def _bundle_to_json_response(
    bundle_dict:dict,  # Serialized ExportBundle
    filename:str,  # Download filename (e.g., "document.json")
) -> Response:  # Starlette Response with JSON content and download headers
    "Create a file download response from an export bundle dict."
```

``` python
def init_export_router(
    service:ManagementService,  # Service for graph queries
    prefix:str,  # Route prefix (e.g., "/manage/export")
) -> Tuple[APIRouter, Dict[str, Callable]]:  # (router, routes dict)
    "Initialize export routes for single document and full database downloads."
```

### helpers (`helpers.ipynb`)

> Shared rendering helpers for the management interface

#### Import

``` python
from cjm_transcript_workflow_management.components.helpers import (
    DEBUG_MANAGEMENT_RENDER,
    render_section_header,
    render_icon_button,
    render_media_type_badge,
    render_alert,
    render_delete_modal,
    render_empty_state
)
```

#### Functions

``` python
def render_section_header(
    title:str,  # Section title text
    icon_name:str,  # Lucide icon name (kebab-case)
) -> Any:  # Header element with icon and title
    "Render a section header with icon."
```

``` python
def render_icon_button(
    icon_name:str,  # Lucide icon name (kebab-case)
    label:str,  # Accessible label text
    color:str=None,  # DaisyUI button color class (e.g., btn_colors.error)
    size:str=None,  # DaisyUI button size class (e.g., btn_sizes.sm)
    **kwargs  # Additional HTML attributes (onclick, hx_post, etc.)
) -> Any:  # Button element with icon
    "Render a button with an icon and accessible label."
```

``` python
def render_media_type_badge(
    media_type:str,  # Media type string (e.g., "audio")
) -> Any:  # Badge element
    "Render a badge for media type display."
```

``` python
def render_alert(
    message:str,  # Alert message text
    color:str=None,  # DaisyUI alert color class (e.g., alert_colors.success)
    alert_id:str="",  # Optional HTML ID for the alert
) -> Any:  # Alert element
    "Render a DaisyUI alert message."
```

``` python
def render_delete_modal(
    modal_id:str,  # HTML ID for the dialog element
    body_id:str,  # HTML ID for the modal body (for HTMX swaps)
    title:str="Delete Document?",  # Modal title text
    confirm_attrs:dict=None,  # Attributes for the confirm button (hx_delete, etc.)
) -> Any:  # Dialog element
    "Render a delete confirmation modal using HTML5 dialog."
```

``` python
def render_empty_state(
    message:str="No documents found.",  # Primary message
    detail:str="Complete a workflow to create a document, or import one.",  # Secondary detail
) -> Any:  # Empty state element
    "Render an empty state placeholder."
```

#### Variables

``` python
DEBUG_MANAGEMENT_RENDER = False
```

### html_ids (`html_ids.ipynb`)

> HTML ID constants for the graph management interface

#### Import

``` python
from cjm_transcript_workflow_management.html_ids import (
    ManagementHtmlIds
)
```

#### Classes

``` python
class ManagementHtmlIds:
    "HTML ID constants for the graph management interface."
    
    def as_selector(
            id_str: str  # The HTML ID to convert
        ) -> str:  # CSS selector with # prefix
        "Convert an ID to a CSS selector format."
```

### import\_ (`import_.ipynb`)

> Import route for file upload with JSON validation and merge strategy

#### Import

``` python
from cjm_transcript_workflow_management.routes.import_ import (
    init_import_router
)
```

#### Functions

``` python
def init_import_router(
    service:ManagementService,  # Service for graph queries
    prefix:str,  # Route prefix (e.g., "/manage/import")
    urls:ManagementUrls,  # URL bundle (for list refresh)
) -> Tuple[APIRouter, Dict[str, Callable]]:  # (router, routes dict)
    "Initialize import route for file upload with merge strategy."
```

### import_controls (`import_controls.ipynb`)

> Import UI with file input, merge strategy selector, and result display

#### Import

``` python
from cjm_transcript_workflow_management.components.import_controls import (
    MERGE_STRATEGIES,
    render_import_result,
    render_import_controls
)
```

#### Functions

``` python
def render_import_result(
    result:ImportResult,  # Import operation result
) -> Any:  # Alert element showing import outcome
    "Render the import result as a success or error alert."
```

``` python
def render_import_controls(
    urls:ManagementUrls,  # URL bundle for route endpoints
) -> Any:  # Import form with file input, strategy selector, and result area
    "Render the import section with file input, merge strategy, and submit button."
```

#### Variables

``` python
MERGE_STRATEGIES = [3 items]
```

### init (`init.ipynb`)

> Router assembly for management routes

#### Import

``` python
from cjm_transcript_workflow_management.routes.init import (
    init_management_routers
)
```

#### Functions

``` python
def init_management_routers(
    service:ManagementService,  # Service for graph queries
    prefix:str,  # Base prefix for management routes (e.g., "/manage")
) -> Tuple[List[APIRouter], ManagementUrls, Dict[str, Callable]]:  # (routers, urls, routes)
    "Initialize and return all management routers with URL bundle."
```

### services.management (`management.ipynb`)

> Service layer wrapping graph plugin operations for document management

#### Import

``` python
from cjm_transcript_workflow_management.services.management import (
    DEBUG_MANAGEMENT_SERVICE,
    ManagementService
)
```

#### Functions

``` python
@patch
async def list_documents_async(self:ManagementService) -> List[DocumentSummary]:  # All documents sorted newest first
    """List all documents with summary info."""
    if DEBUG_MANAGEMENT_SERVICE
    "List all documents with summary info."
```

``` python
@patch
def list_documents(self:ManagementService) -> List[DocumentSummary]:  # All documents sorted newest first
    "List all documents with summary info synchronously."
```

``` python
@patch
async def get_document_detail_async(
    self:ManagementService,
    document_id: str,  # UUID of the Document node
) -> Optional[DocumentDetail]:  # Full detail or None if not found
    "Get full document detail with integrity checks and samples."
```

``` python
@patch
def get_document_detail(
    self:ManagementService,
    document_id: str,  # UUID of the Document node
) -> Optional[DocumentDetail]:  # Full detail or None if not found
    "Get full document detail with integrity checks and samples synchronously."
```

``` python
@patch
async def delete_document_async(
    self:ManagementService,
    document_id: str,  # UUID of the Document node to delete
) -> bool:  # True if deletion succeeded
    "Delete a single document and all its segments via cascade."
```

``` python
@patch
def delete_document(
    self:ManagementService,
    document_id: str,  # UUID of the Document node to delete
) -> bool:  # True if deletion succeeded
    "Delete a single document and all its segments synchronously."
```

``` python
@patch
async def delete_documents_async(
    self:ManagementService,
    document_ids: List[str],  # UUIDs of Document nodes to delete
) -> int:  # Number of documents successfully deleted
    "Delete multiple documents and all their segments via cascade."
```

``` python
@patch
def delete_documents(
    self:ManagementService,
    document_ids: List[str],  # UUIDs of Document nodes to delete
) -> int:  # Number of documents successfully deleted
    "Delete multiple documents and all their segments synchronously."
```

``` python
@patch
async def export_document_async(
    self:ManagementService,
    document_id: str,  # UUID of the Document node to export
) -> Optional[ExportBundle]:  # Export bundle or None if not found
    "Export a single document's subgraph as an ExportBundle."
```

``` python
@patch
def export_document(
    self:ManagementService,
    document_id: str,  # UUID of the Document node to export
) -> Optional[ExportBundle]:  # Export bundle or None if not found
    "Export a single document's subgraph synchronously."
```

``` python
@patch
async def export_all_async(self:ManagementService) -> Optional[ExportBundle]:  # Export bundle or None if error
    """Export the entire graph database as an ExportBundle."""
    if DEBUG_MANAGEMENT_SERVICE
    "Export the entire graph database as an ExportBundle."
```

``` python
@patch
def export_all(self:ManagementService) -> Optional[ExportBundle]:  # Export bundle or None if error
    "Export the entire graph database synchronously."
```

``` python
@patch
async def import_graph_async(
    self:ManagementService,
    bundle_data: Dict[str, Any],  # Parsed JSON from export file
    merge_strategy: str = "skip",  # skip, overwrite, or merge
) -> ImportResult:  # Result with counts and any errors
    "Validate and import graph data from an export bundle."
```

``` python
@patch
def import_graph(
    self:ManagementService,
    bundle_data: Dict[str, Any],  # Parsed JSON from export file
    merge_strategy: str = "skip",  # skip, overwrite, or merge
) -> ImportResult:  # Result with counts and any errors
    "Validate and import graph data synchronously."
```

#### Classes

``` python
class ManagementService:
    def __init__(
        self,
        plugin_manager: PluginManager,  # Plugin manager for accessing graph plugin
        plugin_name: str = "cjm-graph-plugin-sqlite",  # Name of the graph plugin
    )
    "Service wrapping graph plugin operations for document management."
    
    def __init__(
            self,
            plugin_manager: PluginManager,  # Plugin manager for accessing graph plugin
            plugin_name: str = "cjm-graph-plugin-sqlite",  # Name of the graph plugin
        )
        "Initialize with plugin manager."
    
    def is_available(self) -> bool:  # True if plugin is loaded and ready
            """Check if the graph plugin is available."""
            return self._manager.get_plugin(self._plugin_name) is not None
        
        # --- Plugin action wrappers ---
        
        async def _get_context_async(
            self,
            node_id: str,  # UUID of the node to query
            depth: int = 1,  # Traversal depth
        ) -> Optional[GraphContext]:  # GraphContext or None if error
        "Check if the graph plugin is available."
```

#### Variables

``` python
DEBUG_MANAGEMENT_SERVICE = False  # Enable for verbose graph query logging
```

### Models (`models.ipynb`)

> Data models for the graph management interface

#### Import

``` python
from cjm_transcript_workflow_management.models import (
    SegmentSample,
    DocumentSummary,
    DocumentDetail,
    ExportBundle,
    ImportResult,
    ManagementUrls
)
```

#### Classes

``` python
@dataclass
class SegmentSample:
    "Lightweight segment snapshot for detail view display."
    
    index: int  # Segment position in the document
    text: str  # Segment text content
    start_time: float  # Start time in seconds
    end_time: float  # End time in seconds
```

``` python
@dataclass
class DocumentSummary:
    "Summary of a single document for list display."
    
    document_id: str  # Document node UUID
    title: str  # Document title from properties
    media_type: str  # e.g. "audio"
    segment_count: int  # Number of Segment nodes
    total_duration: float  # Sum of segment durations in seconds
    created_at: float  # Unix timestamp when created
```

``` python
@dataclass
class DocumentDetail:
    "Full document information for the detail dashboard."
    
    document_id: str  # Document node UUID
    title: str  # Document title
    media_type: str  # e.g. "audio"
    created_at: float  # Unix timestamp
    updated_at: float  # Unix timestamp
    segment_count: int  # Total number of segments
    total_duration: float  # Sum of segment durations in seconds
    avg_segment_duration: float  # Average segment duration in seconds
    has_starts_with: bool  # Document has a STARTS_WITH edge
    next_chain_complete: bool  # All NEXT edges form a complete chain
    next_count: int  # Number of NEXT edges found
    part_of_complete: bool  # All segments have PART_OF edges
    part_of_count: int  # Number of PART_OF edges found
    all_have_timing: bool  # All segments have start_time/end_time
    segments_missing_timing: int  # Count of segments without timing
    all_have_sources: bool  # All segments have source references
    segments_missing_sources: int  # Count of segments without sources
    all_checks_passed: bool  # True if all integrity checks pass
    source_plugins: List[str] = field(...)  # Unique plugin names from sources
    first_segments: List[SegmentSample] = field(...)  # First N segments
    last_segments: List[SegmentSample] = field(...)  # Last N segments
```

``` python
@dataclass
class ExportBundle:
    "Metadata wrapper for exported graph data."
    
    format: str  # Always "cjm-context-graph"
    version: str  # Semantic version, e.g. "1.0.0"
    exported_at: str  # ISO 8601 datetime string
    source_plugin: str  # Plugin that produced the data
    document_count: int  # Number of Document nodes in the export
    graph: Dict[str, Any]  # {"nodes": [...], "edges": [...]}
```

``` python
@dataclass
class ImportResult:
    "Result of a graph import operation."
    
    success: bool  # Whether the import succeeded
    nodes_created: int  # Number of nodes created
    edges_created: int  # Number of edges created
    nodes_skipped: int  # Number of nodes skipped (already exist)
    edges_skipped: int  # Number of edges skipped (already exist)
    errors: List[str] = field(...)  # Error messages if any
```

``` python
@dataclass
class ManagementUrls:
    "URL bundle for management route endpoints."
    
    management_page: str  # GET: full page (header + import + list)
    list_documents: str  # GET: document list only
    document_detail: str  # GET: + ?doc_id=...
    delete_document: str  # POST: + doc_id in form data
    delete_selected: str  # POST: bulk delete
    export_document: str  # GET: + ?doc_id=...
    export_all: str  # GET: full database export
    import_graph: str  # POST: file upload import
```

### page_renderer (`page_renderer.ipynb`)

> Main management page renderer composing header, toolbar buttons, and
> document list

#### Import

``` python
from cjm_transcript_workflow_management.components.page_renderer import (
    render_page_header,
    render_management_page
)
```

#### Functions

``` python
def render_page_header(
    urls:ManagementUrls,  # URL bundle for route endpoints
) -> Any:  # Header element with title and action buttons
    "Render the management page header with title and top-level actions."
```

``` python
def render_management_page(
    documents:List[DocumentSummary],  # List of document summaries
    urls:ManagementUrls,  # URL bundle for route endpoints
) -> Any:  # Complete management page component
    "Render the complete management page with header, import section, and document list."
```

### utils (`utils.ipynb`)

> Formatting utilities for the management interface

#### Import

``` python
from cjm_transcript_workflow_management.utils import (
    format_duration,
    format_duration_short,
    format_date,
    format_datetime,
    truncate_text,
    format_time_range
)
```

#### Functions

``` python
def format_duration(
    seconds: Optional[float]  # Duration in seconds
) -> str:  # Formatted string (MM:SS or H:MM:SS)
    "Format duration for display in document list and detail views."
```

``` python
def format_duration_short(
    seconds: Optional[float]  # Duration in seconds
) -> str:  # Formatted string (e.g., "10.3s")
    "Format duration as compact seconds for average display."
```

``` python
def format_date(
    timestamp: Optional[float]  # Unix timestamp
) -> str:  # Formatted date string (e.g., "Feb 19, 2026")
    "Format unix timestamp as a human-readable date."
```

``` python
def format_datetime(
    timestamp: Optional[float]  # Unix timestamp
) -> str:  # Formatted datetime string (e.g., "Feb 19, 2026 12:00")
    "Format unix timestamp as a human-readable date and time."
```

``` python
def truncate_text(
    text: Optional[str],  # Full text to truncate
    max_length: int = 60  # Maximum length before truncation
) -> str:  # Truncated text with ellipsis if needed
    "Truncate text for table and sample display."
```

``` python
def format_time_range(
    start: Optional[float],  # Start time in seconds
    end: Optional[float]     # End time in seconds
) -> str:  # Formatted range (e.g., "0.0s - 2.1s")
    "Format a time range for sample segment display."
```