cjm-transcript-workflow-management 0.0.3

A FastHTML management interface for context graph documents to list, inspect, delete, and import/export graph spines produced by transcript decomposition workflows.

cjm-transcript-workflow-management

Install

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 with virtual collection, keyboard navigation, and bulk selection
│   ├── 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

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 --> components_helpers
    components_document_detail --> utils
    components_document_detail --> html_ids
    components_document_detail --> models
    components_document_list --> components_helpers
    components_document_list --> utils
    components_document_list --> html_ids
    components_document_list --> models
    components_import_controls --> components_helpers
    components_import_controls --> html_ids
    components_import_controls --> models
    components_page_renderer --> components_helpers
    components_page_renderer --> components_import_controls
    components_page_renderer --> html_ids
    components_page_renderer --> models
    routes_core --> services_management
    routes_documents --> html_ids
    routes_documents --> routes_core
    routes_documents --> services_management
    routes_documents --> components_document_detail
    routes_documents --> models
    routes_export_ --> routes_core
    routes_export_ --> services_management
    routes_import_ --> html_ids
    routes_import_ --> models
    routes_import_ --> routes_core
    routes_import_ --> components_import_controls
    routes_import_ --> services_management
    routes_init --> html_ids
    routes_init --> components_page_renderer
    routes_init --> routes_import_
    routes_init --> routes_documents
    routes_init --> services_management
    routes_init --> models
    routes_init --> components_document_list
    routes_init --> routes_export_
    services_management --> utils
    services_management --> models

38 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

from cjm_transcript_workflow_management.routes.core import (
    DEBUG_MANAGEMENT_ROUTES
)

Variables

DEBUG_MANAGEMENT_ROUTES = False

document_detail (document_detail.ipynb)

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

Import

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

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

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."

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."

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

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

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

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."

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

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."

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."

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

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."

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

_CARD_CLS

document_list (document_list.ipynb)

Document list with virtual collection, keyboard navigation, and bulk selection

Import

from cjm_transcript_workflow_management.components.document_list import (
    build_document_columns,
    create_document_cell_renderer,
    render_toolbar,
    render_list_scripts,
    render_document_list
)

Functions

def build_document_columns() -> tuple
    "Column definitions for the document list virtual collection."

def create_document_cell_renderer(
    get_selected:Callable,  # () -> Set[str] of selected document IDs
    urls:'ManagementUrls',  # URL bundle for action buttons
    toggle_select_url:str="",  # URL for checkbox toggle route
) -> Callable:  # render_cell(item, ctx) -> FT component
    "Create a cell renderer for the document list virtual collection."

def render_toolbar(
    urls:ManagementUrls,  # URL bundle for route endpoints
    selected_count:int=0,  # Number of selected documents
    total_count:int=0,  # Total number of documents
    select_all_url:str="",  # URL for select-all toggle route
) -> Any:  # Toolbar element
    "Render the document list toolbar with Select All and bulk actions."

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

def render_document_list(
    items:list,  # Current document list (List[DocumentSummary])
    selected:set,  # Selected document IDs (Set[str])
    vc_config:'VirtualCollectionConfig',  # VC configuration
    vc_state:'VirtualCollectionState',  # VC state
    vc_ids:'VirtualCollectionHtmlIds',  # VC HTML IDs
    vc_btn_ids:'VirtualCollectionButtonIds',  # VC button IDs
    vc_urls:'VirtualCollectionUrls',  # VC route URLs
    mgmt_urls:'ManagementUrls',  # Management URLs for actions
    render_cell:Callable,  # Cell renderer callback
    select_all_url:str="",  # URL for select-all toggle route
) -> Any:  # Complete document list component
    "Render the document list with virtual collection, keyboard nav, and modals."

documents (documents.ipynb)

Document list, detail, and delete routes

Import

from cjm_transcript_workflow_management.routes.documents import (
    init_document_router
)

Functions

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)
    refresh_items:Callable,  # async () -> refresh items from service
    refresh_items_oob:Callable,  # async () -> refresh + targeted VC OOB tuple
    render_list:Callable,  # () -> rendered document list component
    render_page:Callable,  # () -> rendered full management page
    get_selected_ids:Callable,  # () -> List[str] of selected document IDs
) -> 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

from cjm_transcript_workflow_management.routes.export_ import (
    init_export_router
)

Functions

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

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."

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

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

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."

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."

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

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."

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."

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

DEBUG_MANAGEMENT_RENDER = False

html_ids (html_ids.ipynb)

HTML ID constants for the graph management interface

Import

from cjm_transcript_workflow_management.html_ids import (
    ManagementHtmlIds
)

Classes

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

from cjm_transcript_workflow_management.routes.import_ import (
    init_import_router
)

Functions

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)
    refresh_items_oob:Callable,  # async () -> refresh + targeted VC OOB tuple
) -> 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

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

Functions

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."

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

MERGE_STRATEGIES = [3 items]

init (init.ipynb)

Router assembly for management routes

Import

from cjm_transcript_workflow_management.routes.init import (
    init_management_routers
)

Functions

def init_management_routers(
    service:ManagementService,  # Service for graph queries
    prefix:str,  # Base prefix for management routes (e.g., "/manage")
) -> ManagementResult:  # Result with routers, urls, and render callables
    "Initialize all management routers with virtual collection integration."

services.management (management.ipynb)

Service layer wrapping graph plugin operations for document management

Import

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

Functions

@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."

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

@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."

@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."

@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."

@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."

@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."

@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."

@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."

@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."

@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."

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

@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."

@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

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

DEBUG_MANAGEMENT_SERVICE = False  # Enable for verbose graph query logging

Models (models.ipynb)

Data models for the graph management interface

Import

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

Classes

@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

@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

@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

@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": [...]}

@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

@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

@dataclass
class ManagementResult:
    "Result of management router initialization."
    
    routers: List[Any]  # APIRouter instances to register
    urls: ManagementUrls  # URL bundle for route endpoints
    routes: Dict[str, Callable]  # Route handler functions
    render_page: Callable  # () -> full management page component
    render_list: Callable  # () -> document list component
    refresh_items: Callable  # async () -> refresh items from service

page_renderer (page_renderer.ipynb)

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

Import

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

Functions

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."

def render_management_page(
    urls:ManagementUrls,  # URL bundle for route endpoints
    render_list_fn:Callable,  # () -> document list component
) -> 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

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

Functions

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."

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."

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."

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."

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."

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."