cjm-transcript-verify

Install

pip install cjm_transcript_verify

Project Structure

nbs/
├── components/ (5)
│   ├── helpers.ipynb               # State getters for the verify step from InteractionContext
│   ├── integrity_checks.ipynb      # Integrity check display with pass/fail indicators
│   ├── sample_segments.ipynb       # Sample segment display with first/last segments and jump-to-index
│   ├── step_renderer.ipynb         # Main verify step renderer combining all dashboard components
│   └── verification_summary.ipynb  # Summary cards for document info, segments stats, and source info
├── routes/ (4)
│   ├── core.ipynb    # Verify step state management helpers for routes
│   ├── init.ipynb    # Router assembly for Phase 4 verify routes
│   ├── sample.ipynb  # Jump-to-index route for fetching a single segment by index
│   └── verify.ipynb  # Verification route that queries graph and computes results
├── services/ (1)
│   └── verify.ipynb  # Service layer for querying graph database and computing verification results
├── html_ids.ipynb  # HTML ID constants for Phase 4: Verify
├── models.ipynb    # Verify step state and verification result models for Phase 4: Verify
└── utils.ipynb     # Formatting utilities for verification display

Total: 13 notebooks across 3 directories

Module Dependencies

graph LR
    components_helpers[components.helpers<br/>helpers]
    components_integrity_checks[components.integrity_checks<br/>integrity_checks]
    components_sample_segments[components.sample_segments<br/>sample_segments]
    components_step_renderer[components.step_renderer<br/>step_renderer]
    components_verification_summary[components.verification_summary<br/>verification_summary]
    html_ids[html_ids<br/>html_ids]
    models[models<br/>models]
    routes_core[routes.core<br/>core]
    routes_init[routes.init<br/>init]
    routes_sample[routes.sample<br/>sample]
    routes_verify[routes.verify<br/>verify]
    services_verify[services.verify<br/>services.verify]
    utils[utils<br/>utils]

    components_helpers --> models
    components_integrity_checks --> html_ids
    components_integrity_checks --> models
    components_sample_segments --> models
    components_sample_segments --> utils
    components_sample_segments --> html_ids
    components_step_renderer --> components_sample_segments
    components_step_renderer --> models
    components_step_renderer --> html_ids
    components_step_renderer --> components_integrity_checks
    components_step_renderer --> components_verification_summary
    components_verification_summary --> html_ids
    components_verification_summary --> utils
    components_verification_summary --> models
    routes_core --> models
    routes_init --> models
    routes_init --> services_verify
    routes_init --> routes_core
    routes_init --> routes_sample
    routes_init --> routes_verify
    routes_sample --> models
    routes_sample --> components_sample_segments
    routes_sample --> services_verify
    routes_sample --> routes_core
    routes_verify --> models
    routes_verify --> services_verify
    routes_verify --> routes_core
    routes_verify --> components_step_renderer
    services_verify --> utils
    services_verify --> models

30 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)

Verify step state management helpers for routes

Import

from cjm_transcript_verify.routes.core import (
    WorkflowStateStore,
    DEBUG_VERIFY_STATE,
    VerifyContext
)

Functions

def _get_verify_state(
    state_store:WorkflowStateStore,  # The workflow state store
    workflow_id:str,  # The workflow identifier
    session_id:str  # Session identifier string
) -> VerifyStepState:  # Verify step state dictionary
    "Get the verify step state from the workflow state store."

def _get_review_state(
    state_store:WorkflowStateStore,  # The workflow state store
    workflow_id:str,  # The workflow identifier
    session_id:str  # Session identifier string
) -> Dict[str, Any]:  # Review step state dictionary
    "Get the review step state (for document_id fallback)."

def _load_verify_context(
    state_store:WorkflowStateStore,  # The workflow state store
    workflow_id:str,  # The workflow identifier
    session_id:str  # Session identifier string
) -> VerifyContext:  # Common verify state values
    "Load commonly-needed verify state values with review fallback."

def _update_verify_state(
    state_store:WorkflowStateStore,  # The workflow state store
    workflow_id:str,  # The workflow identifier
    session_id:str,  # Session identifier string
    document_id:str=None,  # Document ID (None = don't change)
    media_path:str=None,  # Media path (None = don't change)
) -> None
    "Update the verify step state in the workflow state store."

Classes

class VerifyContext(NamedTuple):
    "Common verify state values loaded by handlers."

Variables

DEBUG_VERIFY_STATE = False

helpers (helpers.ipynb)

State getters for the verify step from InteractionContext

Import

from cjm_transcript_verify.components.helpers import *

Functions

def _get_verify_state(
    ctx:InteractionContext  # Interaction context with state
) -> VerifyStepState:  # Typed verify step state
    "Get the full verify step state from context."

def _get_document_id(
    ctx:InteractionContext  # Interaction context with state
) -> Optional[str]:  # Document UUID or None if not set
    "Get the document ID to verify from context."

def _get_media_path(
    ctx:InteractionContext  # Interaction context with state
) -> Optional[str]:  # Media path or None if not set
    "Get the media path for display context."

def _get_document_id_from_review(
    ctx:InteractionContext  # Interaction context with state
) -> Optional[str]:  # Document UUID or None if not found
    "Get the document ID from review step state (fallback)."

def _get_document_id_any(
    ctx:InteractionContext  # Interaction context with state
) -> Optional[str]:  # Document UUID or None if not found anywhere
    "Get document ID from verify state, falling back to review state."

html_ids (html_ids.ipynb)

HTML ID constants for Phase 4: Verify

Import

from cjm_transcript_verify.html_ids import (
    VerifyHtmlIds
)

Classes

class VerifyHtmlIds:
    "HTML ID constants for Phase 4: Verify."
    
    def as_selector(
            id_str: str  # The HTML ID to convert
        ) -> str:  # CSS selector with # prefix
        "Convert an ID to a CSS selector format."

init (init.ipynb)

Router assembly for Phase 4 verify routes

Import

from cjm_transcript_verify.routes.init import (
    init_verify_routers
)

Functions

def init_verify_routers(
    state_store:WorkflowStateStore,  # The workflow state store
    workflow_id:str,  # The workflow identifier
    prefix:str,  # Base prefix for verify routes (e.g., "/workflow/verify")
    verify_service:VerifyService,  # Service for graph queries
) -> Tuple[List[APIRouter], VerifyUrls, Dict[str, Callable]]:  # (routers, urls, routes)
    "Initialize and return all verify routers with URL bundle."

integrity_checks (integrity_checks.ipynb)

Integrity check display with pass/fail indicators

Import

from cjm_transcript_verify.components.integrity_checks import (
    render_integrity_checks
)

Functions

def _render_check_row(
    passed:bool,  # Whether the check passed
    label:str,  # Check description
    detail:str="",  # Optional detail (e.g., "246/246")
) -> Any:  # Check row element
    "Render an integrity check row with pass/fail icon."

def render_integrity_checks(
    result:VerificationResult,  # Verification result with integrity data
) -> Any:  # Integrity checks card
    "Render the structure integrity checks section."

models (models.ipynb)

Verify step state and verification result models for Phase 4: Verify

Import

from cjm_transcript_verify.models import (
    VerifyStepState,
    SegmentSample,
    VerificationResult,
    VerifyUrls
)

Classes

class VerifyStepState(TypedDict):
    "State for Phase 4: Verify."

@dataclass
class SegmentSample:
    "Lightweight segment data for sample display."
    
    index: int  # Segment index in document
    text: str  # Truncated text (~60 chars)
    start_time: Optional[float]  # Start time in seconds
    end_time: Optional[float]  # End time in seconds
    
    def duration(self) -> Optional[float]:  # Computed duration in seconds
            """Compute duration from start and end times."""
            if self.start_time is not None and self.end_time is not None
        "Compute duration from start and end times."
    
    def to_dict(self) -> dict:  # Serializable dictionary
            """Convert to dictionary for serialization."""
            return {
                "index": self.index,
        "Convert to dictionary for serialization."
    
    def from_dict(cls, data: dict) -> "SegmentSample":  # Reconstructed instance
        "Create from dictionary."

@dataclass
class VerificationResult:
    "Complete verification data from graph database queries."
    
    document_id: str  # UUID of Document node
    document_title: str  # Document title
    document_media_type: str  # Media type ('audio', 'video', 'text')
    segment_count: int  # Total number of segments
    total_duration: float  # Total duration in seconds
    avg_segment_duration: float  # Average segment duration in seconds
    has_starts_with: bool  # Whether STARTS_WITH edge exists
    starts_with_count: int  # Number of STARTS_WITH edges (should be 1)
    next_chain_complete: bool  # Whether NEXT chain is complete
    next_count: int  # Number of NEXT edges (should be segment_count - 1)
    part_of_complete: bool  # Whether all PART_OF edges exist
    part_of_count: int  # Number of PART_OF edges (should be segment_count)
    all_have_timing: bool  # Whether all segments have timing
    segments_missing_timing: int  # Count of segments without timing
    all_have_sources: bool  # Whether all segments have sources
    segments_missing_sources: int  # Count of segments without sources
    source_plugins: List[str] = field(...)  # Unique plugin names
    first_segments: List[SegmentSample] = field(...)  # First 3 segments
    last_segments: List[SegmentSample] = field(...)  # Last 3 segments
    
    def all_checks_passed(self) -> bool:  # Whether all integrity checks passed
            """Check if all integrity checks passed."""
            return (
                self.has_starts_with and
                self.starts_with_count == 1 and
                self.next_chain_complete and
                self.part_of_complete and
                self.all_have_timing and
                self.all_have_sources
            )
        
        def to_dict(self) -> dict:  # Serializable dictionary
        "Check if all integrity checks passed."
    
    def to_dict(self) -> dict:  # Serializable dictionary
            """Convert to dictionary for serialization."""
            return {
                "document_id": self.document_id,
        "Convert to dictionary for serialization."
    
    def from_dict(cls, data: dict) -> "VerificationResult":  # Reconstructed instance
        "Create from dictionary."

@dataclass
class VerifyUrls:
    "URL bundle for Phase 4 verify route handlers."
    
    verify: str = ''  # Main verification computation route
    sample: str = ''  # Jump-to-index sample fetch route

sample (sample.ipynb)

Jump-to-index route for fetching a single segment by index

Import

from cjm_transcript_verify.routes.sample import (
    DEBUG_SAMPLE_ROUTES,
    init_sample_router
)

Functions

def init_sample_router(
    state_store:WorkflowStateStore,  # The workflow state store
    workflow_id:str,  # The workflow identifier
    prefix:str,  # Route prefix (e.g., "/workflow/verify")
    verify_service:VerifyService,  # Service for graph queries
    urls:VerifyUrls,  # URL bundle (will be populated)
) -> Tuple[APIRouter, Dict[str, Callable]]:  # (router, routes dict)
    "Initialize sample route for jump-to-index segment fetching."

Variables

DEBUG_SAMPLE_ROUTES = False

sample_segments (sample_segments.ipynb)

Sample segment display with first/last segments and jump-to-index

Import

from cjm_transcript_verify.components.sample_segments import (
    render_sample_row,
    render_sample_list,
    render_jump_to_index,
    render_jump_result,
    render_sample_segments
)

Functions

def render_sample_row(
    sample:SegmentSample,  # Segment sample to display
) -> Any:  # Sample row element
    "Render a single segment sample row."

def render_sample_list(
    samples:List[SegmentSample],  # Samples to display
    label:str,  # Label for this list (e.g., "First", "Last")
    container_id:str="",  # Optional container ID
) -> Any:  # Sample list element
    "Render a list of segment samples with label."

def render_jump_to_index(
    urls:VerifyUrls=None,  # URL bundle for routes
    max_index:int=0,  # Maximum valid index for placeholder
) -> Any:  # Jump-to-index form with loading indicator
    "Render the jump-to-index input form with loading state."

def render_jump_result(
    sample:Optional[SegmentSample]=None,  # Fetched sample or None
    error:str="",  # Error message if any
) -> Any:  # Jump result display
    "Render the jump-to-index result (single segment or error)."

def render_sample_segments(
    result:VerificationResult,  # Verification result with samples
    urls:VerifyUrls=None,  # URL bundle for routes
) -> Any:  # Sample segments card
    "Render the full sample segments section."

step_renderer (step_renderer.ipynb)

Main verify step renderer combining all dashboard components

Import

from cjm_transcript_verify.components.step_renderer import (
    DEBUG_VERIFY_RENDER,
    render_verify_header,
    render_verify_error,
    render_verify_loading,
    render_verify_step
)

Functions

def render_verify_header(
    all_passed:bool,  # Whether all integrity checks passed
) -> Any:  # Header element with title and status badge
    "Render the verify step header with status badge."

def render_verify_error(
    message:str="Unable to load verification data",  # Error message
) -> Any:  # Error display element
    "Render an error state for the verify step."

def render_verify_loading() -> Any:  # Loading indicator
    "Render a loading state for the verify step."

def render_verify_step(
    result:Optional[VerificationResult]=None,  # Verification result or None for error
    urls:VerifyUrls=None,  # URL bundle for routes
    error:str="",  # Error message if result is None
) -> Any:  # Complete verify step component
    "Render the complete verify step with all dashboard components."

Variables

DEBUG_VERIFY_RENDER = False

utils (utils.ipynb)

Formatting utilities for verification display

Import

from cjm_transcript_verify.utils import (
    format_duration_mmss,
    format_duration_seconds,
    format_time_range,
    truncate_text
)

Functions

def format_duration_mmss(
    seconds: Optional[float]  # Duration in seconds
) -> str:  # Formatted string (mm:ss)
    "Format duration in seconds as mm:ss for summary display."

def format_duration_seconds(
    seconds: Optional[float]  # Duration in seconds
) -> str:  # Formatted string (e.g., "10.3s")
    "Format duration in seconds as Xs or X.Xs for average 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 time range for sample segment display."

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 sample segment display."

verification_summary (verification_summary.ipynb)

Summary cards for document info, segments stats, and source info

Import

from cjm_transcript_verify.components.verification_summary import (
    render_document_section,
    render_segments_section,
    render_sources_section,
    render_verification_summary
)

Functions

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

def _render_section_header(
    icon:Any,  # Lucide icon component
    title:str,  # Section title
) -> Any:  # Section header element
    "Render a section header with icon."

def render_document_section(
    result:VerificationResult,  # Verification result with document info
) -> Any:  # Document info card
    "Render the document info section."

def render_segments_section(
    result:VerificationResult,  # Verification result with segment stats
) -> Any:  # Segments stats card
    "Render the segments statistics section."

def render_sources_section(
    result:VerificationResult,  # Verification result with source info
) -> Any:  # Sources info card
    "Render the source traceability section."

def render_verification_summary(
    result:VerificationResult,  # Verification result to display
) -> Any:  # Summary section with all info cards
    "Render the full verification summary with document, segments, and sources."

verify (verify.ipynb)

Verification route that queries graph and computes results

Import

from cjm_transcript_verify.routes.verify import (
    DEBUG_VERIFY_ROUTES,
    init_verify_router
)

Functions

def init_verify_router(
    state_store:WorkflowStateStore,  # The workflow state store
    workflow_id:str,  # The workflow identifier
    prefix:str,  # Route prefix (e.g., "/workflow/verify")
    verify_service:VerifyService,  # Service for graph queries
    urls:VerifyUrls,  # URL bundle (will be populated)
) -> Tuple[APIRouter, Dict[str, Callable]]:  # (router, routes dict)
    "Initialize verify route that computes verification results."

Variables

DEBUG_VERIFY_ROUTES = False

services.verify (verify.ipynb)

Service layer for querying graph database and computing verification results

Import

from cjm_transcript_verify.services.verify import (
    DEBUG_VERIFY_SERVICE,
    VerifyService
)

Classes

class VerifyService:
    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 for querying graph database and computing verification results."
    
    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
        
        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."
    
    async def verify_document_async(
            self,
            document_id: str,  # UUID of the Document node to verify
        ) -> Optional[VerificationResult]:  # Verification results or None if error/not found
        "Query graph and compute verification results for a document."
    
    def verify_document(
            self,
            document_id: str,  # UUID of the Document node to verify
        ) -> Optional[VerificationResult]:  # Verification results or None if error/not found
        "Query graph and compute verification results for a document synchronously."
    
    async def get_segment_by_index_async(
            self,
            document_id: str,  # UUID of the Document node
            index: int,  # Segment index to fetch
        ) -> Optional[SegmentSample]:  # Segment sample or None if error/not found
        "Fetch a single segment by index for jump-to-index feature."
    
    def get_segment_by_index(
            self,
            document_id: str,  # UUID of the Document node
            index: int,  # Segment index to fetch
        ) -> Optional[SegmentSample]:  # Segment sample or None if error/not found
        "Fetch a single segment by index for jump-to-index feature synchronously."
    
    async def get_segment_count_async(
            self,
            document_id: str,  # UUID of the Document node
        ) -> int:  # Number of segments or 0 if error
        "Get total segment count for index validation."
    
    def get_segment_count(
            self,
            document_id: str,  # UUID of the Document node
        ) -> int:  # Number of segments or 0 if error
        "Get total segment count for index validation synchronously."

Variables

DEBUG_VERIFY_SERVICE = False  # Enable for verbose graph query logging