cjm-transcript-source-select 0.0.1
FastHTML source selection component for transcript decomposition workflows, with federated database browsing, drag-drop ordering, and keyboard navigation.
License: Apache-2.0
Requires Python: >=3.12
Keywords: nbdev jupyter notebook python

# cjm-transcript-source-select


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

## Install

``` bash
pip install cjm_transcript_source_select
```

## Project Structure

    nbs/
    ├── components/ (6)
    │   ├── helpers.ipynb          # Shared helper functions for the selection module
    │   ├── local_files.ipynb      # Local files browser for importing external .db files
    │   ├── preview_panel.ipynb    # Collapsible preview panel for displaying selected content
    │   ├── selection_queue.ipynb  # Selection queue component with drag-drop reordering
    │   ├── source_browser.ipynb   # Source browser components for displaying and filtering transcription sources
    │   └── step_renderer.ipynb    # Phase 1 step renderer: Source Selection & Ordering with two-column layout and collapsible preview
    ├── routes/ (6)
    │   ├── core.ipynb         # Selection step state management helpers
    │   ├── filtering.ipynb    # Filtering, grouping, and keyboard navigation route handlers
    │   ├── init.ipynb         # Router assembly for Phase 1 selection routes
    │   ├── local_files.ipynb  # Local files browser route handlers
    │   ├── queue.ipynb        # Selection queue route handlers for Phase 1
    │   └── tabs.ipynb         # Tab switching route handlers
    ├── services/ (2)
    │   ├── source.ipynb        # Source service for federated transcription queries via DuckDB
    │   └── source_utils.ipynb  # Source record operations for metadata extraction, grouping, and validation
    ├── html_ids.ipynb  # HTML ID constants for Phase 1: Source Selection & Ordering
    ├── models.ipynb    # Data models and URL bundles for Phase 1: Source Selection & Ordering
    └── utils.ipynb     # Display formatting and word counting utilities for the selection step

Total: 17 notebooks across 3 directories

## Module Dependencies

``` mermaid
graph LR
    components_helpers[components.helpers<br/>helpers]
    components_local_files[components.local_files<br/>local_files]
    components_preview_panel[components.preview_panel<br/>preview_panel]
    components_selection_queue[components.selection_queue<br/>selection_queue]
    components_source_browser[components.source_browser<br/>source_browser]
    components_step_renderer[components.step_renderer<br/>step_renderer]
    html_ids[html_ids<br/>html_ids]
    models[models<br/>models]
    routes_core[routes.core<br/>core]
    routes_filtering[routes.filtering<br/>filtering]
    routes_init[routes.init<br/>init]
    routes_local_files[routes.local_files<br/>local_files]
    routes_queue[routes.queue<br/>queue]
    routes_tabs[routes.tabs<br/>tabs]
    services_source[services.source<br/>source]
    services_source_utils[services.source_utils<br/>source_utils]
    utils[utils<br/>utils]

    components_helpers --> models
    components_local_files --> components_helpers
    components_local_files --> html_ids
    components_preview_panel --> html_ids
    components_selection_queue --> html_ids
    components_source_browser --> services_source_utils
    components_source_browser --> utils
    components_source_browser --> html_ids
    components_step_renderer --> components_helpers
    components_step_renderer --> components_source_browser
    components_step_renderer --> utils
    components_step_renderer --> components_preview_panel
    components_step_renderer --> html_ids
    components_step_renderer --> components_local_files
    components_step_renderer --> components_selection_queue
    components_step_renderer --> models
    routes_core --> components_step_renderer
    routes_core --> models
    routes_core --> components_selection_queue
    routes_core --> components_source_browser
    routes_core --> services_source
    routes_filtering --> routes_core
    routes_filtering --> services_source_utils
    routes_filtering --> models
    routes_filtering --> components_source_browser
    routes_filtering --> services_source
    routes_init --> routes_queue
    routes_init --> routes_local_files
    routes_init --> models
    routes_init --> routes_filtering
    routes_init --> routes_tabs
    routes_init --> services_source
    routes_init --> routes_core
    routes_local_files --> components_local_files
    routes_local_files --> routes_core
    routes_local_files --> services_source
    routes_local_files --> models
    routes_queue --> routes_core
    routes_queue --> services_source_utils
    routes_queue --> models
    routes_queue --> components_preview_panel
    routes_queue --> services_source
    routes_tabs --> routes_local_files
    routes_tabs --> components_local_files
    routes_tabs --> services_source_utils
    routes_tabs --> routes_core
    routes_tabs --> components_source_browser
    routes_tabs --> models
    routes_tabs --> components_step_renderer
    routes_tabs --> services_source
```

*50 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`)

> Selection step state management helpers

#### Import

``` python
from cjm_transcript_source_select.routes.core import (
    DEBUG_SELECTION_STATE,
    WorkflowStateStore
)
```

#### Functions

``` python
def _get_step_state(
    state_store: WorkflowStateStore,  # The workflow state store
    workflow_id: str,  # The workflow identifier
    session_id: str  # Session identifier string
) -> Dict[str, Any]:  # Step state dictionary
    "Get the selection step state from the workflow state store."
```

``` python
def _get_active_source_tab(
    state_store: WorkflowStateStore,  # The workflow state store
    workflow_id: str,  # The workflow identifier
    session_id: str  # Session identifier string
) -> str:  # Active tab: "db" or "files"
    "Get the currently active source tab from workflow state."
```

``` python
def _build_queue_response(
    state_store: WorkflowStateStore,  # The workflow state store
    workflow_id: str,  # The workflow identifier
    source_service: SourceService,  # The source service for querying transcriptions
    session_id: str,  # Session identifier string
    selected_sources: List[Dict[str, str]],  # Current selected sources after mutation
    urls: SelectionUrls,  # URL bundle for rendering
    include_stats: bool = True,  # Include OOB stats swap
    include_source_list: bool = True,  # Include conditional OOB source list swap
    grouping_mode: str = None,  # Override grouping mode for source list rendering
) -> Union[Any, Tuple]:  # Single component or tuple of components with OOB swaps
    "Build the standard response for queue-mutating handlers."
```

``` python
def _update_step_state(
    "Update the selection step state in the workflow state store."
```

#### Variables

``` python
DEBUG_SELECTION_STATE = False
```

### filtering (`filtering.ipynb`)

> Filtering, grouping, and keyboard navigation route handlers

#### Import

``` python
from cjm_transcript_source_select.routes.filtering import (
    init_filtering_router
)
```

#### Functions

``` python
def _handle_source_filter(
    state_store: WorkflowStateStore,  # The workflow state store
    workflow_id: str,  # The workflow identifier
    source_service: SourceService,  # The source service for queries
    request,  # FastHTML request object
    sess,  # FastHTML session object
    search: str,  # Search term from input
    urls: SelectionUrls,  # URL bundle for rendering
):  # Filtered source list component
    "Filter transcription sources by search term."
```

``` python
def _handle_grouping_change(
    state_store: WorkflowStateStore,  # The workflow state store
    workflow_id: str,  # The workflow identifier
    source_service: SourceService,  # The source service for queries
    request,  # FastHTML request object
    sess,  # FastHTML session object
    grouping_mode: str,  # New grouping mode: "media_path" or "batch_id"
    urls: SelectionUrls,  # URL bundle for rendering
):  # Updated source list component
    "Change the grouping mode and re-render the source list."
```

``` python
def _handle_selection_toggle_focused(
    state_store: WorkflowStateStore,  # The workflow state store
    workflow_id: str,  # The workflow identifier
    source_service: SourceService,  # The source service for queries
    request,  # FastHTML request object
    sess,  # FastHTML session object
    record_id: str,  # Job ID from focused row (via hx-include)
    provider_id: str,  # Plugin name from focused row (via hx-include)
    urls: SelectionUrls,  # URL bundle for rendering
):  # Queue component with OOB stats, optionally with OOB source list
    "Toggle selection of the focused row (keyboard shortcut handler)."
```

``` python
def _handle_keyboard_reorder(
    state_store: WorkflowStateStore,  # The workflow state store
    workflow_id: str,  # The workflow identifier
    source_service: SourceService,  # The source service for queries
    request,  # FastHTML request object
    sess,  # FastHTML session object
    record_id: str,  # Record ID of item to move
    provider_id: str,  # Provider ID of item to move
    direction: str,  # Direction to move: "up" or "down"
    urls: SelectionUrls,  # URL bundle for rendering
):  # Queue component, optionally with OOB source list
    "Move an item up or down in the selection queue via keyboard."
```

``` python
def init_filtering_router(
    state_store: WorkflowStateStore,  # The workflow state store
    workflow_id: str,  # The workflow identifier
    source_service: SourceService,  # The source service for queries
    prefix: str,  # Route prefix (e.g., "/workflow/selection/filtering")
    urls: SelectionUrls,  # URL bundle for rendering
) -> Tuple[APIRouter, Dict[str, Callable]]:  # (router, route_dict)
    "Initialize filtering and keyboard navigation routes."
```

### helpers (`helpers.ipynb`)

> Shared helper functions for the selection module

#### Import

``` python
from cjm_transcript_source_select.components.helpers import *
```

#### Functions

``` python
def _get_selection_state(
    ctx: InteractionContext  # Interaction context with state
) -> SelectionStepState:  # Typed selection step state
    "Get the full selection step state from context."
```

``` python
def _get_selected_sources(
    ctx: InteractionContext  # Interaction context with state
) -> List[SelectedSource]:  # List of selected source dicts
    "Get the list of selected sources from step state."
```

``` python
def _get_grouping_mode(
    ctx: InteractionContext  # Interaction context with state
) -> str:  # Grouping mode: "media_path" or "batch_id"
    "Get the current grouping mode from step state."
```

``` python
def _generate_sortable_init_script(
    container_selector: str = ".sortable",  # CSS selector for sortable containers
    handle_selector: str = ".drag-handle",  # CSS selector for drag handles
    animation_ms: int = 150,  # Animation duration in milliseconds
) -> str:  # JavaScript initialization script
    "Generate Sortable.js initialization script for htmx integration."
```

### html_ids (`html_ids.ipynb`)

> HTML ID constants for Phase 1: Source Selection & Ordering

#### Import

``` python
from cjm_transcript_source_select.html_ids import (
    SelectionHtmlIds
)
```

#### Classes

``` python
class SelectionHtmlIds:
    "HTML ID constants for Phase 1: Source Selection & Ordering."
    
    def as_selector(
            id_str:str  # The HTML ID to convert
        ) -> str:  # CSS selector with # prefix
        "Convert an ID to a CSS selector format."
    
    def source_checkbox(
            record_id:str,  # Record identifier
            provider_id:str  # Provider identifier
        ) -> str:  # HTML ID for the source checkbox
        "Generate HTML ID for a source selection checkbox."
    
    def source_row(
            record_id:str,  # Record identifier
            provider_id:str  # Provider identifier
        ) -> str:  # HTML ID for the source row
        "Generate HTML ID for a source browser row."
    
    def queue_item(
            record_id:str,  # Record identifier
            provider_id:str  # Provider identifier
        ) -> str:  # HTML ID for the queue item
        "Generate HTML ID for a queue item."
```

### init (`init.ipynb`)

> Router assembly for Phase 1 selection routes

#### Import

``` python
from cjm_transcript_source_select.routes.init import (
    init_selection_routers
)
```

#### Functions

``` python
def init_selection_routers(
    state_store: WorkflowStateStore,  # The workflow state store
    source_service: SourceService,  # The source service for queries
    workflow_id: str,  # The workflow identifier
    prefix: str,  # Base prefix for selection routes (e.g., "/workflow/selection")
) -> Tuple[List[APIRouter], SelectionUrls, Dict[str, Callable]]:  # (routers, urls, merged_routes)
    "Initialize and return all selection routers with URL bundle."
```

### local_files (`local_files.ipynb`)

> Local files browser for importing external .db files

#### Import

``` python
from cjm_transcript_source_select.components.local_files import *
```

#### Functions

``` python
def _get_external_db_paths(
    ctx: InteractionContext  # Interaction context with state
) -> List[str]:  # List of external database paths
    "Get the list of external database paths from step state."
```

``` python
def _get_current_browse_path(
    ctx: InteractionContext  # Interaction context with state
) -> str:  # Current browse path
    "Get the current browse path from step state."
```

``` python
def _get_file_browser_state(
    step_state: Dict[str, Any],  # Selection step state dictionary
    default_path: Optional[str] = None  # Default path if no state exists
) -> BrowserState:  # BrowserState for file browser
    "Get or create BrowserState from step state."
```

``` python
def _create_db_browser_config() -> FileBrowserConfig:  # Configured FileBrowserConfig for .db file selection
    "Create file browser config for .db file selection."
```

``` python
def _render_external_sources_list(
    external_paths: List[str],  # List of added external database paths
    remove_url: str,  # URL for removing external source
) -> Any:  # External sources list component
    "Render the list of added external database sources with scrollable paths."
```

``` python
def _render_local_files_browser(
    browser_state: Optional[BrowserState] = None,  # Current browser state
    external_paths: Optional[List[str]] = None,  # List of added external database paths
    provider: Optional[LocalFileSystemProvider] = None,  # File system provider
    config: Optional[FileBrowserConfig] = None,  # Browser configuration
    navigate_url: str = "",  # URL for browsing directories
    select_url: str = "",  # URL for adding external source (maps path to db_path)
    remove_url: str = "",  # URL for removing external source
    refresh_url: str = "",  # URL for refreshing browser
    path_input_url: str = "",  # URL for direct path input
    home_path: Optional[str] = None,  # Home directory path
    error_message: Optional[str] = None,  # Error message to display
) -> Any:  # Local files browser component
    "Render the local files browser for adding external .db files."
```

### local_files (`local_files.ipynb`)

> Local files browser route handlers

#### Import

``` python
from cjm_transcript_source_select.routes.local_files import (
    init_local_files_router
)
```

#### Functions

``` python
def _get_local_files_provider() -> LocalFileSystemProvider:
    """Get or create the local files provider singleton."""
    global _local_files_provider
    if _local_files_provider is None
    "Get or create the local files provider singleton."
```

``` python
def _handle_browse_directory(
    state_store: WorkflowStateStore,  # The workflow state store
    workflow_id: str,  # The workflow identifier
    source_service: SourceService,  # The source service for external db ops
    request,  # FastHTML request object
    sess,  # FastHTML session object
    path: str,  # Directory path to browse
    urls: SelectionUrls,  # URL bundle for rendering
):  # Local files browser component
    "Browse a directory and return the local files browser component."
```

``` python
def _get_local_files_config() -> FileBrowserConfig:
    """Get or create the local files config singleton."""
    global _local_files_config
    if _local_files_config is None
    "Get or create the local files config singleton."
```

``` python
def _handle_add_external_source(
    state_store: WorkflowStateStore,  # The workflow state store
    workflow_id: str,  # The workflow identifier
    source_service: SourceService,  # The source service for external db ops
    request,  # FastHTML request object
    sess,  # FastHTML session object
    path: str,  # Path to the .db file (from file-browser select_url)
    urls: SelectionUrls,  # URL bundle for rendering
):  # Local files browser component
    "Toggle an external database source (add if not present, remove if present)."
```

``` python
def _handle_remove_external_source(
    state_store: WorkflowStateStore,  # The workflow state store
    workflow_id: str,  # The workflow identifier
    source_service: SourceService,  # The source service for external db ops
    request,  # FastHTML request object
    sess,  # FastHTML session object
    db_path: str,  # Path to the .db file to remove
    urls: SelectionUrls,  # URL bundle for rendering
):  # Local files browser component
    "Remove an external database source from the Added Sources list."
```

``` python
def init_local_files_router(
    state_store: WorkflowStateStore,  # The workflow state store
    workflow_id: str,  # The workflow identifier
    source_service: SourceService,  # The source service for external db ops
    prefix: str,  # Route prefix (e.g., "/workflow/selection/local_files")
    urls: SelectionUrls,  # URL bundle for rendering
) -> Tuple[APIRouter, Dict[str, Callable]]:  # (router, route_dict)
    "Initialize local files browser routes."
```

#### Variables

``` python
_local_files_provider: Optional[LocalFileSystemProvider] = None
_local_files_config: Optional[FileBrowserConfig] = None
```

### models (`models.ipynb`)

> Data models and URL bundles for Phase 1: Source Selection & Ordering

#### Import

``` python
from cjm_transcript_source_select.models import (
    SelectionStepState,
    SelectionUrls
)
```

#### Classes

``` python
class SelectionStepState(TypedDict):
    "State for Phase 1: Source Selection & Ordering."
```

``` python
@dataclass
class SelectionUrls:
    "URL bundle for Phase 1 selection route handlers and renderers."
    
    add: str = ''  # Add source to queue
    remove: str = ''  # Remove source from queue
    reorder: str = ''  # Reorder queue items
    clear: str = ''  # Clear all from queue
    select_all: str = ''  # Select all in a group
    preview: str = ''  # Preview source content
    toggle_focused: str = ''  # Toggle focused row selection
    keyboard_reorder: str = ''  # Keyboard reorder (Shift+Up/Down)
    filter: str = ''  # Filter source list
    grouping_change: str = ''  # Change grouping mode
    browse_directory: str = ''  # Browse directory
    add_external: str = ''  # Add external .db source
    remove_external: str = ''  # Remove external .db source
    tab_switch: str = ''  # Switch source tabs
```

### preview_panel (`preview_panel.ipynb`)

> Collapsible preview panel for displaying selected content

#### Import

``` python
from cjm_transcript_source_select.components.preview_panel import *
```

#### Functions

``` python
def _render_preview_panel(
    preview_record_id: Optional[str] = None,  # Job ID being previewed
    preview_text: Optional[str] = None,  # Text content to preview
    is_open: bool = False,  # Whether the collapse should be open
) -> Any:  # Preview panel component (collapsible, full-width)
    "Render the collapsible preview panel for displaying selected content."
```

### queue (`queue.ipynb`)

> Selection queue route handlers for Phase 1

#### Import

``` python
from cjm_transcript_source_select.routes.queue import (
    init_queue_router
)
```

#### Functions

``` python
def _handle_selection_add(
    state_store: WorkflowStateStore,  # The workflow state store
    workflow_id: str,  # The workflow identifier
    source_service: SourceService,  # The source service for queries
    request,  # FastHTML request object
    sess,  # FastHTML session object
    record_id: str,  # Job ID to add
    provider_id: str,  # Plugin name for the source
    urls: SelectionUrls,  # URL bundle for rendering
):  # Queue component with OOB stats, optionally with OOB source list
    "Add a source to the selection queue."
```

``` python
def _handle_selection_remove(
    state_store: WorkflowStateStore,  # The workflow state store
    workflow_id: str,  # The workflow identifier
    source_service: SourceService,  # The source service for queries
    request,  # FastHTML request object
    sess,  # FastHTML session object
    record_id: str,  # Job ID to remove
    urls: SelectionUrls,  # URL bundle for rendering
):  # Queue component with OOB stats, optionally with OOB source list
    "Remove a source from the selection queue."
```

``` python
async def _handle_selection_reorder(
    state_store: WorkflowStateStore,  # The workflow state store
    workflow_id: str,  # The workflow identifier
    source_service: SourceService,  # The source service for queries
    request,  # FastHTML request object
    sess,  # FastHTML session object
    urls: SelectionUrls,  # URL bundle for rendering
):  # Updated queue component
    "Reorder items in the selection queue based on SortableJS result."
```

``` python
def _handle_selection_clear(
    state_store: WorkflowStateStore,  # The workflow state store
    workflow_id: str,  # The workflow identifier
    source_service: SourceService,  # The source service for queries
    request,  # FastHTML request object
    sess,  # FastHTML session object
    urls: SelectionUrls,  # URL bundle for rendering
):  # Queue component with OOB stats, optionally with OOB source list
    "Clear all items from the selection queue."
```

``` python
def _handle_selection_select_all(
    state_store: WorkflowStateStore,  # The workflow state store
    workflow_id: str,  # The workflow identifier
    source_service: SourceService,  # The source service for queries
    request,  # FastHTML request object
    sess,  # FastHTML session object
    group_key: str,  # Group key to select all transcriptions for
    grouping_mode: str,  # Current grouping mode: "media_path" or "batch_id"
    urls: SelectionUrls,  # URL bundle for rendering
):  # Queue component with OOB stats, optionally with OOB source list
    "Select all transcriptions for a given group."
```

``` python
def _handle_selection_preview(
    source_service: SourceService,  # The source service for queries
    request,  # FastHTML request object
    record_id: str,  # Job ID to preview
    provider_id: str,  # Plugin name for the source
):  # Full preview panel component (collapsible, open with content)
    "Get preview panel for a selected source."
```

``` python
def init_queue_router(
    state_store: WorkflowStateStore,  # The workflow state store
    workflow_id: str,  # The workflow identifier
    source_service: SourceService,  # The source service for queries
    prefix: str,  # Route prefix (e.g., "/workflow/selection/queue")
    urls: SelectionUrls,  # URL bundle for rendering (populated after all routers created)
) -> Tuple[APIRouter, Dict[str, Callable]]:  # (router, route_dict)
    "Initialize queue management routes."
```

### selection_queue (`selection_queue.ipynb`)

> Selection queue component with drag-drop reordering

#### Import

``` python
from cjm_transcript_source_select.components.selection_queue import *
```

#### Functions

``` python
def _render_queue_item(
    source: Dict[str, str],  # Source dict with record_id and provider_id
    index: int,  # Position in queue (1-based)
    remove_url: str,  # URL for removing from queue
) -> Any:  # Queue item element
    "Render a single item in the selection queue."
```

``` python
def _render_selection_queue(
    selected_sources: List[Dict[str, str]],  # List of selected sources in order
    remove_url: str,  # URL for removing from queue
    reorder_url: str,  # URL for reordering queue
    clear_url: str,  # URL for clearing all
) -> Any:  # Queue panel component
    "Render the selection queue panel with drag-drop reordering."
```

### source (`source.ipynb`)

> Source service for federated transcription queries via DuckDB

#### Import

``` python
from cjm_transcript_source_select.services.source import (
    VALID_DB_EXTENSIONS,
    TranscriptionDBProvider,
    SourceService,
    validate_and_toggle_external_db
)
```

#### Functions

``` python
def validate_and_toggle_external_db(
    source_service: SourceService,  # Source service for duplicate detection
    path: str,  # Path to the .db file
    external_paths: List[str],  # Current external database paths
    valid_extensions: List[str] = None,  # Valid file extensions (default: VALID_DB_EXTENSIONS)
) -> Tuple[List[str], Optional[str]]:  # (updated_paths, error_message or None)
    "Validate and toggle an external database path in the external paths list."
```

#### Classes

``` python
class TranscriptionDBProvider:
    def __init__(
        self,
        db_path: str,  # Path to SQLite database file
        name: str,  # Display name for this provider
        provider_id: Optional[str] = None  # Unique ID (defaults to db_path)
    )
    "SourceProvider for transcription SQLite databases."
    
    def __init__(
            self,
            db_path: str,  # Path to SQLite database file
            name: str,  # Display name for this provider
            provider_id: Optional[str] = None  # Unique ID (defaults to db_path)
        )
        "Initialize provider for a transcription database."
    
    def provider_id(self) -> str:  # Unique identifier
            """Unique identifier for this provider instance."""
            return self._id
        
        @property
        def provider_name(self) -> str:  # Display name
        "Unique identifier for this provider instance."
    
    def provider_name(self) -> str:  # Display name
            """Human-readable name for display."""
            return self._name
        
        @property
        def provider_type(self) -> str:  # Provider category
        "Human-readable name for display."
    
    def provider_type(self) -> str:  # Provider category
            """Provider type category."""
            return "transcription_db"
        
        @property
        def db_path(self) -> Path:  # Database file path
        "Provider type category."
    
    def db_path(self) -> Path:  # Database file path
            """Path to the underlying database file."""
            return self._db_path
        
        def is_available(self) -> bool:  # Whether database exists and is accessible
        "Path to the underlying database file."
    
    def is_available(self) -> bool:  # Whether database exists and is accessible
            """Check if the database file exists and is accessible."""
            return self._db_path.exists() and self._db_path.suffix == '.db'
        
        def validate_schema(self) -> Tuple[bool, str]:  # (is_valid, error_message)
        "Check if the database file exists and is accessible."
    
    def validate_schema(self) -> Tuple[bool, str]:  # (is_valid, error_message)
            """Check if database has valid transcription schema."""
            if not self.is_available()
        "Check if database has valid transcription schema."
    
    def query_records(
            self,
            limit: int = 100  # Maximum records to return
        ) -> List[SourceRecord]:  # List of source records
        "Query transcription records from the database."
    
    def get_source_block(
            self,
            record_id: str  # Job ID to fetch
        ) -> Optional[SourceBlock]:  # SourceBlock or None if not found
        "Fetch a specific transcription as a SourceBlock."
    
    def from_plugin(
            cls,
            meta: PluginMeta  # Plugin metadata with manifest containing db_path
        ) -> Optional["TranscriptionDBProvider"]:  # Provider or None if no valid db_path
        "Create provider from plugin metadata."
    
    def from_external_path(
            cls,
            path: str  # Path to external database file
        ) -> Optional["TranscriptionDBProvider"]:  # Provider or None if path invalid
        "Create provider from an external database path."
```

``` python
class SourceService:
    def __init__(
        self,
        plugin_manager: PluginManager,  # Plugin manager for discovering plugin sources
        source_categories: List[str] = None,  # Plugin categories to query (default: ['transcription'])
        external_paths: List[str] = None  # External database paths
    )
    "Service for federated access to content sources via providers."
    
    def __init__(
            self,
            plugin_manager: PluginManager,  # Plugin manager for discovering plugin sources
            source_categories: List[str] = None,  # Plugin categories to query (default: ['transcription'])
            external_paths: List[str] = None  # External database paths
        )
        "Initialize the source service."
    
    def add_provider(
            self,
            provider: SourceProvider  # Provider instance to add
        ) -> bool:  # True if added, False if ID already exists
        "Add a source provider."
    
    def remove_provider(
            self,
            provider_id: str  # ID of provider to remove
        ) -> bool:  # True if removed, False if not found
        "Remove a source provider by ID."
    
    def get_provider(
            self,
            provider_id: str  # ID of provider to get
        ) -> Optional[SourceProvider]:  # Provider or None if not found
        "Get a provider by ID."
    
    def get_providers(self) -> List[SourceProvider]:  # List of all providers
            """Get all registered providers."""
            return list(self._providers.values())
        
        def get_provider_by_name(
            self,
            name: str  # Provider name to search for
        ) -> Optional[SourceProvider]:  # Provider or None if not found
        "Get all registered providers."
    
    def get_provider_by_name(
            self,
            name: str  # Provider name to search for
        ) -> Optional[SourceProvider]:  # Provider or None if not found
        "Find a provider by its display name."
    
    def has_provider_for_path(
            self,
            path: str  # Path to check
        ) -> Tuple[bool, Optional[str]]:  # (has_duplicate, existing_provider_name)
        "Check if any provider uses the same resolved database path."
    
    def add_plugin_providers(self) -> int:  # Number of providers added
            """Discover and add providers from loaded plugins."""
            added = 0
            for category in self._categories
        "Discover and add providers from loaded plugins."
    
    def set_external_paths(
            self,
            paths: List[str]  # List of external database paths to set
        ) -> None
        "Set external database paths (replaces existing external providers)."
    
    def add_external_path(
            self,
            path: str  # External database path to add
        ) -> bool:  # True if added, False if already exists or invalid
        "Add an external database as a provider."
    
    def remove_external_path(
            self,
            path: str  # External database path to remove
        ) -> bool:  # True if removed, False if not found
        "Remove an external database provider."
    
    def get_external_paths(self) -> List[str]:  # List of external database paths
            """Get list of external database paths."""
            paths = []
            for pid, provider in self._providers.items()
        "Get list of external database paths."
    
    def get_available_sources(self) -> List[Dict[str, Any]]:  # List of source info dicts
            """Get list of available sources (for UI display)."""
            # First ensure plugin providers are loaded
            self.add_plugin_providers()
            
            sources = []
            for provider in self._providers.values()
        "Get list of available sources (for UI display)."
    
    def query_transcriptions(
            self,
            provider_name: Optional[str] = None,  # Filter by provider name (None for all)
            limit: int = 100  # Maximum number of results per provider
        ) -> List[Dict[str, Any]]:  # List of transcription records
        "Query records from all providers (or a specific one)."
    
    def get_transcription_by_id(
            self,
            record_id: str,  # Record ID to fetch
            provider_id: str  # Provider ID that owns this record
        ) -> Optional[SourceBlock]:  # SourceBlock or None if not found
        "Get a specific transcription as a SourceBlock."
    
    def get_source_blocks(
            self,
            selections: List[Dict[str, str]]  # List of {record_id, provider_id} dicts
        ) -> List[SourceBlock]:  # Ordered list of SourceBlocks
        "Fetch multiple records as SourceBlocks in order."
```

#### Variables

``` python
VALID_DB_EXTENSIONS = [3 items]
```

### source_browser (`source_browser.ipynb`)

> Source browser components for displaying and filtering transcription
> sources

#### Import

``` python
from cjm_transcript_source_select.components.source_browser import *
```

#### Functions

``` python
def _render_grouping_selector(
    grouping_mode: str,  # Current grouping mode: "media_path" or "batch_id"
    grouping_change_url: str,  # URL for changing grouping mode
) -> Any:  # Grouping selector component
    "Render the dropdown for selecting grouping mode."
```

``` python
def _render_source_row(
    record: Dict[str, Any],  # Transcription record
    is_selected: bool,  # Whether this source is selected
    add_url: str,  # URL for adding to queue
    remove_url: str,  # URL for removing from queue
    preview_url: str,  # URL for previewing content
    is_first: bool = False,  # Whether this is the first row (gets initial focus)
) -> Any:  # Table row element
    "Render a single source row in the browser table."
```

``` python
def _render_group_header(
    group_key: str,  # The group key (media_path or batch_id value)
    record_count: int,  # Number of records in this group
    select_all_url: str,  # URL for selecting all in group
    grouping_mode: str = "media_path",  # Current grouping mode
) -> Any:  # Table row for group header
    "Render a group header row."
```

``` python
def _render_audio_group_header(
    media_path: str,  # Path to audio file
    record_count: int,  # Number of records in this group
    select_all_url: str,  # URL for selecting all in group
) -> Any:  # Table row for group header
    "Render a group header row for an audio file (legacy wrapper)."
```

``` python
def _render_source_list(
    transcriptions: List[Dict[str, Any]],  # Available transcription records
    selected_sources: List[Dict[str, str]],  # Currently selected sources
    add_url: str,  # URL for adding to queue
    remove_url: str,  # URL for removing from queue
    preview_url: str,  # URL for previewing content
    select_all_url: str,  # URL for selecting all in a group
    grouping_mode: str = "media_path",  # Grouping mode: "media_path" or "batch_id"
    oob: bool = False,  # Whether to include hx-swap-oob for out-of-band swap
) -> Any:  # Source list container with table
    "Render the source list table with grouped rows."
```

``` python
def _render_source_browser(
    transcriptions: List[Dict[str, Any]],  # Available transcription records
    sources: List[Dict[str, Any]],  # Available source plugins (unused, kept for API compat)
    selected_sources: List[Dict[str, str]],  # Currently selected sources
    add_url: str,  # URL for adding to queue
    remove_url: str,  # URL for removing from queue
    preview_url: str,  # URL for previewing content
    select_all_url: str,  # URL for selecting all in a group
    filter_url: str,  # URL for filtering sources
    grouping_mode: str = "media_path",  # Current grouping mode
    grouping_change_url: str = "",  # URL for changing grouping mode
) -> Any:  # Source browser component
    "Render the source browser panel with search filtering and grouped table."
```

### source_utils (`source_utils.ipynb`)

> Source record operations for metadata extraction, grouping, and
> validation

#### Import

``` python
from cjm_transcript_source_select.services.source_utils import (
    extract_batch_id,
    extract_model_name,
    group_transcriptions,
    group_transcriptions_by_audio,
    is_source_selected,
    filter_transcriptions,
    select_all_in_group,
    toggle_source_selection,
    reorder_item,
    reorder_sources,
    calculate_next_tab,
    check_audio_exists,
    validate_browse_path
)
```

#### Functions

``` python
def extract_batch_id(
    metadata: Any  # Metadata dict or JSON string
) -> str:  # Batch ID or "No Batch ID"
    "Extract batch_id from transcription metadata."
```

``` python
def extract_model_name(
    metadata: Any  # Metadata dict or JSON string
) -> str:  # Formatted model name for display
    "Extract and format model name from transcription metadata."
```

``` python
def group_transcriptions(
    transcriptions: List[Dict[str, Any]],  # List of transcription records
    group_by: str = "media_path"  # Grouping mode: "media_path" or "batch_id"
) -> Dict[str, List[Dict[str, Any]]]:  # Grouped transcriptions
    "Group transcription records by the specified field."
```

``` python
def group_transcriptions_by_audio(
    transcriptions: List[Dict[str, Any]]  # List of transcription records
) -> Dict[str, List[Dict[str, Any]]]:  # Grouped by media_path
    "Group transcription records by their source audio file."
```

``` python
def is_source_selected(
    record_id: str,  # Job ID to check
    selected_sources: List[Dict[str, str]]  # List of selected sources
) -> bool:  # True if source is selected
    "Check if a source is in the selected list."
```

``` python
def filter_transcriptions(
    transcriptions: List[Dict[str, Any]],  # List of transcription records to filter
    search_text: str,  # Search term for case-insensitive substring matching
) -> List[Dict[str, Any]]:  # Filtered transcription records
    "Filter transcriptions by substring match across record_id, media_path, and text fields."
```

``` python
def select_all_in_group(
    transcriptions: List[Dict[str, Any]],  # All transcription records
    group_key: str,  # Group key to match against
    grouping_mode: str,  # Grouping mode: "media_path" or "batch_id"
    selected_sources: List[Dict[str, str]],  # Current selections
) -> List[Dict[str, str]]:  # Updated selections with new items appended
    "Add all transcriptions matching a group key to the selection list, skipping duplicates."
```

``` python
def toggle_source_selection(
    record_id: str,  # Job ID to toggle
    provider_id: str,  # Plugin name for the source
    selected_sources: List[Dict[str, str]],  # Current selections
) -> List[Dict[str, str]]:  # Updated selections
    "Toggle a source in or out of the selection list."
```

``` python
def reorder_item(
    selected_sources: List[Dict[str, str]],  # Current selections
    record_id: str,  # Record ID of item to move
    provider_id: str,  # Provider ID of item to move
    direction: str,  # Direction: "up" or "down"
) -> List[Dict[str, str]]: