transcript_cli.py

#!/usr/bin/env python
"""
transcript_cli.py - command line interface for retrieving audio transcripts from rewinddb.

call flow:
1. parse command line arguments using argparse
2. connect to rewinddb database
3. determine time range based on arguments:
   - if --relative is provided, calculate relative time range from now
   - if --from and --to are provided, use specific time range
4. query audio transcripts for the specified time range
5. format and display results
6. close database connection

the cli supports two main query modes:
- relative time queries (e.g., "1 hour", "5h", "30m", "2d", "1w")
- specific time range queries with --from and --to timestamps

examples:
  python transcript_cli.py --relative "1 hour"
  python transcript_cli.py --relative "5 hours"
  python transcript_cli.py --relative "5h"
  python transcript_cli.py --relative "30m"
  python transcript_cli.py --relative "2d"
  python transcript_cli.py --relative "1w"
  python transcript_cli.py --from "2023-05-11 13:00:00" --to "2023-05-11 17:00:00"
"""

import argparse
import datetime
from datetime import timezone
import re
import sys
import os
import shutil
from pathlib import Path

import rewinddb
import rewinddb.utils


def convert_to_local_time(dt):
    """convert a utc datetime to local time.

    args:
        dt: datetime object in utc

    returns:
        datetime object in local time
    """

    if dt is None:
        return None

    # if datetime has no timezone info, assume it's utc
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)

    # convert to local time
    return dt.astimezone()


def parse_relative_time(time_str):
    """parse a relative time string into timedelta components.

    args:
        time_str: string like "1 hour", "5 hours", "30 minutes" or short form "5h", "3m", "10d", "2w"

    returns:
        dict with keys for days, hours, minutes, seconds

    raises:
        ValueError: if the time string format is invalid
    """

    time_str = time_str.lower().strip()
    time_components = {"days": 0, "hours": 0, "minutes": 0, "seconds": 0}

    # short form pattern (e.g., "5h", "3m", "10d", "2w")
    short_patterns = {
        r"^(\d+)w$": lambda x: {"days": int(x) * 7},
        r"^(\d+)d$": lambda x: {"days": int(x)},
        r"^(\d+)h$": lambda x: {"hours": int(x)},
        r"^(\d+)m$": lambda x: {"minutes": int(x)},
        r"^(\d+)s$": lambda x: {"seconds": int(x)}
    }

    # check for short form patterns first
    for pattern, handler in short_patterns.items():
        match = re.search(pattern, time_str)
        if match:
            component_values = handler(match.group(1))
            for component, value in component_values.items():
                time_components[component] = value
            return time_components

    # long form patterns
    patterns = {
        r"(\d+)\s*(?:day|days)": "days",
        r"(\d+)\s*(?:hour|hours|hr|hrs)": "hours",
        r"(\d+)\s*(?:minute|minutes|min|mins)": "minutes",
        r"(\d+)\s*(?:second|seconds|sec|secs)": "seconds",
        r"(\d+)\s*(?:week|weeks)": "weeks"
    }

    # try to match each pattern
    found_match = False
    for pattern, component in patterns.items():
        match = re.search(pattern, time_str)
        if match:
            if component == "weeks":
                time_components["days"] += int(match.group(1)) * 7
            else:
                time_components[component] = int(match.group(1))
            found_match = True

    if not found_match:
        raise ValueError(f"invalid time format: {time_str}. use format like '1 hour', '5h', '30m', '2d', '1w'.")

    return time_components


def get_transcripts_relative(db, time_str, speech_source=None):
    """get audio transcripts from a relative time period.

    args:
        db: rewinddb instance
        time_str: relative time string (e.g., "1 hour", "5 hours")
        speech_source: optional filter for speech source ('me' for user voice, 'others' for other speakers)

    returns:
        list of transcript dictionaries
    """

    try:
        time_components = parse_relative_time(time_str)
        return db.get_audio_transcripts_relative(speech_source=speech_source, **time_components)
    except ValueError as e:
        print(f"error: {e}", file=sys.stderr)
        sys.exit(1)


def get_transcripts_absolute(db, from_time_str, to_time_str, speech_source=None):
    """get audio transcripts from a specific time range.

    args:
        db: rewinddb instance
        from_time_str: start time string in format "YYYY-MM-DD HH:MM:SS", "YYYY-MM-DD HH:MM", "YYYY-MM-DD", "HH:MM:SS", or "HH:MM"
        to_time_str: end time string in format "YYYY-MM-DD HH:MM:SS", "YYYY-MM-DD HH:MM", "YYYY-MM-DD", "HH:MM:SS", or "HH:MM"
        speech_source: optional filter for speech source ('me' for user voice, 'others' for other speakers)

    returns:
        list of transcript dictionaries
    """

    def normalize_time_string(time_str, is_end_time=False):
        """normalize time string to handle various formats."""
        # check if time_str is time-only format (HH:MM or HH:MM:SS)
        if len(time_str) <= 8 and ':' in time_str:
            today = datetime.datetime.now().strftime("%Y-%m-%d")
            # if it's HH:MM format, add :00 for seconds
            if time_str.count(':') == 1:
                time_str = f"{time_str}:00"
            time_str = f"{today} {time_str}"
        # check if it's date-only format (YYYY-MM-DD)
        elif len(time_str) == 10 and time_str.count('-') == 2:
            if is_end_time:
                time_str = f"{time_str} 23:59:59"
            else:
                time_str = f"{time_str} 00:00:00"
        # check if it's date with HH:MM format
        elif ' ' in time_str and time_str.split(' ')[1].count(':') == 1:
            time_str = f"{time_str}:00"

        return time_str

    try:
        # get local timezone for proper conversion
        local_tz = datetime.datetime.now().astimezone().tzinfo

        # normalize time strings to handle various formats
        from_time_str = normalize_time_string(from_time_str, is_end_time=False)
        to_time_str = normalize_time_string(to_time_str, is_end_time=True)

        # parse as naive datetime first
        from_time_naive = datetime.datetime.strptime(from_time_str, "%Y-%m-%d %H:%M:%S")
        to_time_naive = datetime.datetime.strptime(to_time_str, "%Y-%m-%d %H:%M:%S")

        # add local timezone info and convert to UTC for database query
        from_time = from_time_naive.replace(tzinfo=local_tz).astimezone(timezone.utc)
        to_time = to_time_naive.replace(tzinfo=local_tz).astimezone(timezone.utc)

        return db.get_audio_transcripts_absolute(from_time, to_time, speech_source)
    except ValueError as e:
        print(f"error: invalid time format. use format 'YYYY-MM-DD HH:MM:SS', 'YYYY-MM-DD HH:MM', 'YYYY-MM-DD', 'HH:MM:SS', or 'HH:MM'.", file=sys.stderr)
        sys.exit(1)


def export_own_voice_by_day(db, from_time_str, to_time_str, format_type='text'):
    """export user's own voice transcripts organized by day.

    args:
        db: rewinddb instance
        from_time_str: start time string
        to_time_str: end time string
        format_type: 'text' for text output, 'audio' for audio file export (future)

    returns:
        dictionary with dates as keys and formatted output as values
    """

    def normalize_time_string(time_str, is_end_time=False):
        """normalize time string to handle various formats."""
        # check if time_str is time-only format (HH:MM or HH:MM:SS)
        if len(time_str) <= 8 and ':' in time_str:
            today = datetime.datetime.now().strftime("%Y-%m-%d")
            # if it's HH:MM format, add :00 for seconds
            if time_str.count(':') == 1:
                time_str = f"{time_str}:00"
            time_str = f"{today} {time_str}"
        # check if it's date-only format (YYYY-MM-DD)
        elif len(time_str) == 10 and time_str.count('-') == 2:
            if is_end_time:
                time_str = f"{time_str} 23:59:59"
            else:
                time_str = f"{time_str} 00:00:00"
        # check if it's date with HH:MM format
        elif ' ' in time_str and time_str.split(' ')[1].count(':') == 1:
            time_str = f"{time_str}:00"

        return time_str

    try:
        # get local timezone for proper conversion
        local_tz = datetime.datetime.now().astimezone().tzinfo

        # normalize time strings to handle various formats
        from_time_str = normalize_time_string(from_time_str, is_end_time=False)
        to_time_str = normalize_time_string(to_time_str, is_end_time=True)

        # parse as naive datetime first
        from_time_naive = datetime.datetime.strptime(from_time_str, "%Y-%m-%d %H:%M:%S")
        to_time_naive = datetime.datetime.strptime(to_time_str, "%Y-%m-%d %H:%M:%S")

        # add local timezone info and convert to UTC for database query
        from_time = from_time_naive.replace(tzinfo=local_tz).astimezone(timezone.utc)
        to_time = to_time_naive.replace(tzinfo=local_tz).astimezone(timezone.utc)

        # get own voice transcripts organized by day
        transcripts_by_day = db.get_own_voice_transcripts_by_day(from_time, to_time)
        
        return transcripts_by_day
    except ValueError as e:
        print(f"error: invalid time format. use format 'YYYY-MM-DD HH:MM:SS', 'YYYY-MM-DD HH:MM', 'YYYY-MM-DD', 'HH:MM:SS', or 'HH:MM'.", file=sys.stderr)
        sys.exit(1)


def format_own_voice_export(transcripts_by_day, format_type='text'):
    """format own voice transcripts for export.

    args:
        transcripts_by_day: dictionary with dates as keys and transcript lists as values
        format_type: 'text' for text output, 'json' for JSON output

    returns:
        formatted string for export
    """
    if format_type == 'json':
        import json
        export_data = {}
        for date, transcripts in sorted(transcripts_by_day.items()):
            words = [t['word'] for t in transcripts]
            export_data[date] = {
                'word_count': len(words),
                'text': ' '.join(words),
                'words': words
            }
        return json.dumps(export_data, indent=2, default=str)
    
    else:  # text format
        output_lines = []
        total_words = 0
        
        for date in sorted(transcripts_by_day.keys()):
            transcripts = transcripts_by_day[date]
            words = [t['word'] for t in transcripts]
            word_count = len(words)
            total_words += word_count
            
            output_lines.append(f"\n=== {date} ===")
            output_lines.append(f"Word count: {word_count}")
            output_lines.append("")
            
            # join words into readable text
            text = ' '.join(words)
            output_lines.append(text)
            output_lines.append("")
        
        output_lines.insert(0, f"Own Voice Export - Total words: {total_words}")
        output_lines.insert(1, f"Date range: {min(transcripts_by_day.keys())} to {max(transcripts_by_day.keys())}")
        output_lines.insert(2, "=" * 50)
        
        return '\n'.join(output_lines)


def export_own_voice_audio(transcripts_by_day, output_dir):
    """export audio files for own voice transcripts.

    args:
        transcripts_by_day: dictionary with dates as keys and transcript lists as values
        output_dir: directory to save audio files

    returns:
        summary of exported files
    """
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)
    
    exported_files = []
    skipped_files = []
    total_copied = 0
    
    # group transcripts by audio file to avoid duplicates
    audio_files_by_day = {}
    
    for date, transcripts in transcripts_by_day.items():
        audio_files_by_day[date] = {}
        for transcript in transcripts:
            audio_path = transcript.get('audio_path')
            if audio_path and os.path.exists(audio_path):
                audio_id = transcript['audio_id']
                if audio_id not in audio_files_by_day[date]:
                    audio_files_by_day[date][audio_id] = {
                        'path': audio_path,
                        'audio_start_time': transcript['audio_start_time'],
                        'duration': transcript['audio_duration'],
                        'words': []
                    }
                audio_files_by_day[date][audio_id]['words'].append(transcript['word'])
    
    # export files organized by day
    for date, audio_files in audio_files_by_day.items():
        day_dir = output_path / date
        day_dir.mkdir(exist_ok=True)
        
        # create summary file for the day
        day_summary = []
        day_summary.append(f"Voice Export Summary for {date}")
        day_summary.append("=" * 40)
        day_summary.append("")
        
        for audio_id, info in audio_files.items():
            src_path = info['path']
            if not os.path.exists(src_path):
                skipped_files.append(f"{date}/{audio_id} - file not found")
                continue
            
            # create descriptive filename
            time_str = info['audio_start_time'].strftime("%H%M%S")
            word_count = len(info['words'])
            filename = f"{time_str}_audio{audio_id}_{word_count}words.m4a"
            dest_path = day_dir / filename
            
            try:
                shutil.copy2(src_path, dest_path)
                file_size = os.path.getsize(dest_path) / 1024 / 1024  # MB
                
                exported_files.append(str(dest_path))
                total_copied += 1
                
                # add to summary
                transcript_text = ' '.join(info['words'][:20])  # first 20 words
                if len(info['words']) > 20:
                    transcript_text += "..."
                
                day_summary.append(f"File: {filename}")
                day_summary.append(f"  Size: {file_size:.1f} MB")
                day_summary.append(f"  Duration: {info['duration']/1000:.1f} seconds")
                day_summary.append(f"  Words: {word_count}")
                day_summary.append(f"  Preview: {transcript_text}")
                day_summary.append("")
                
            except Exception as e:
                skipped_files.append(f"{date}/{audio_id} - copy error: {e}")
        
        # save day summary
        summary_file = day_dir / "transcript_summary.txt"
        with open(summary_file, 'w', encoding='utf-8') as f:
            f.write('\n'.join(day_summary))
    
    # create overall summary
    overall_summary = []
    overall_summary.append("Own Voice Audio Export Summary")
    overall_summary.append("=" * 50)
    overall_summary.append(f"Total files exported: {total_copied}")
    overall_summary.append(f"Total files skipped: {len(skipped_files)}")
    overall_summary.append(f"Export directory: {output_path.absolute()}")
    overall_summary.append("")
    
    if skipped_files:
        overall_summary.append("Skipped files:")
        for skipped in skipped_files:
            overall_summary.append(f"  - {skipped}")
        overall_summary.append("")
    
    overall_summary.append("Exported files by day:")
    for date in sorted(audio_files_by_day.keys()):
        count = len([f for f in exported_files if f"/{date}/" in f])
        overall_summary.append(f"  {date}: {count} files")
    
    return {
        'exported_files': exported_files,
        'skipped_files': skipped_files,
        'total_copied': total_copied,
        'summary': '\n'.join(overall_summary)
    }


def parse_arguments():
    """parse command line arguments.

    returns:
        parsed argument namespace
    """

    parser = argparse.ArgumentParser(
        description="retrieve audio transcripts from rewinddb",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
examples:
  %(prog)s --relative "1 hour"
  %(prog)s --relative "5 hours"
  %(prog)s --relative "5h"
  %(prog)s --relative "30m"
  %(prog)s --relative "2d"
  %(prog)s --relative "1w"
  %(prog)s --from "2023-05-11 13:00:00" --to "2023-05-11 17:00:00"
  %(prog)s --from "2023-05-11" --to "2023-05-12"  # uses 00:00:00 and 23:59:59
  %(prog)s --from "13:00:00" --to "17:00:00"  # uses today's date
  %(prog)s --from "13:00" --to "17:00"  # uses today's date, HH:MM format
  %(prog)s --relative "7 days" --debug
  %(prog)s --relative "1 hour" --env-file /path/to/.env
  %(prog)s --relative "1 day" --utc  # display times in UTC instead of local time
  %(prog)s --relative "1 hour" --speech-source me  # filter for own voice only
  %(prog)s --relative "1 day" --speech-source others  # filter for other speakers
  %(prog)s --export-own-voice "2025-01-01 to 2025-01-31"  # export own voice by day
  %(prog)s --export-own-voice "2025-01-01 to 2025-01-31" --export-format json --save-to my_voice.json
  %(prog)s --export-own-voice "2025-01-01 to 2025-01-31" --export-format audio --audio-export-dir ./my_voice_audio
"""
    )

    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("-r", "--relative", metavar="TIME", help="relative time period (e.g., '1 hour', '5h', '3m', '10d', '2w')")
    group.add_argument("--from", dest="from_time", metavar="DATETIME",
                       help="start time in format 'YYYY-MM-DD HH:MM:SS', 'YYYY-MM-DD HH:MM', 'YYYY-MM-DD' (uses 00:00:00), 'HH:MM:SS', or 'HH:MM' (uses today's date)")
    group.add_argument("--export-own-voice", dest="export_own_voice", metavar="DATERANGE",
                       help="export own voice transcripts by day in format 'YYYY-MM-DD to YYYY-MM-DD' (e.g., '2025-01-01 to 2025-01-31')")

    parser.add_argument("--to", dest="to_time", metavar="DATETIME",
                       help="end time in format 'YYYY-MM-DD HH:MM:SS', 'YYYY-MM-DD HH:MM', 'YYYY-MM-DD' (uses 23:59:59), 'HH:MM:SS', or 'HH:MM' (uses today's date)")
    parser.add_argument("--debug", action="store_true", help="enable debug output")
    parser.add_argument("--env-file", metavar="FILE", help="path to .env file with database configuration")
    parser.add_argument("--utc", action="store_true", help="display times in UTC instead of local time")
    parser.add_argument("--speech-source", choices=['me', 'others'], help="filter by speech source ('me' for own voice, 'others' for other speakers)")
    parser.add_argument("--export-format", choices=['text', 'json', 'audio'], default='text', help="export format for own voice (default: text)")
    parser.add_argument("--save-to", metavar="FILE", help="save export output to file instead of displaying")
    parser.add_argument("--audio-export-dir", metavar="DIR", help="directory to export audio files (required when --export-format audio)")

    args = parser.parse_args()

    # validate that if --from is provided, --to is also provided
    if args.from_time and not args.to_time:
        parser.error("--to is required when --from is provided")
    
    # validate export-own-voice format
    if args.export_own_voice:
        if " to " not in args.export_own_voice:
            parser.error("--export-own-voice must be in format 'YYYY-MM-DD to YYYY-MM-DD'")
        try:
            parts = args.export_own_voice.split(" to ")
            if len(parts) != 2:
                raise ValueError()
            # validate date format
            for part in parts:
                datetime.datetime.strptime(part.strip(), "%Y-%m-%d")
        except ValueError:
            parser.error("--export-own-voice dates must be in format 'YYYY-MM-DD to YYYY-MM-DD'")
    
    # validate audio export requirements
    if args.export_format == 'audio':
        if not args.export_own_voice:
            parser.error("--export-format audio can only be used with --export-own-voice")
        if not args.audio_export_dir:
            parser.error("--audio-export-dir is required when --export-format is audio")

    return args


# No replacement needed - removing the DirectSqliteAccess class


def main():
    """main function for the transcript cli tool."""

    args = parse_arguments()

    try:
        # connect to the database using rewinddb library
        print("connecting to rewind database...")
        with rewinddb.RewindDB(args.env_file) as db:
            # handle export own voice mode
            if args.export_own_voice:
                parts = args.export_own_voice.split(" to ")
                from_date = parts[0].strip()
                to_date = parts[1].strip()
                
                print(f"exporting own voice transcripts from {from_date} to {to_date}...")
                transcripts_by_day = export_own_voice_by_day(db, from_date, to_date)
                
                if not transcripts_by_day:
                    print("no own voice transcripts found for the specified date range.")
                    return
                
                # handle different export formats
                if args.export_format == 'audio':
                    # export audio files
                    print("exporting audio files...")
                    export_result = export_own_voice_audio(transcripts_by_day, args.audio_export_dir)
                    
                    print(export_result['summary'])
                    
                    # also save text summary if requested
                    if args.save_to:
                        formatted_output = format_own_voice_export(transcripts_by_day, 'text')
                        with open(args.save_to, 'w', encoding='utf-8') as f:
                            f.write(formatted_output)
                        print(f"\ntext summary also saved to {args.save_to}")
                else:
                    # format text/json output
                    formatted_output = format_own_voice_export(transcripts_by_day, args.export_format)
                    
                    # save to file or display
                    if args.save_to:
                        with open(args.save_to, 'w', encoding='utf-8') as f:
                            f.write(formatted_output)
                        print(f"own voice transcripts exported to {args.save_to}")
                        
                        # show summary
                        total_days = len(transcripts_by_day)
                        total_words = sum(len(transcripts) for transcripts in transcripts_by_day.values())
                        print(f"exported {total_words} words across {total_days} days")
                    else:
                        print(formatted_output)
                
                return
            
            # handle regular transcript retrieval
            speech_source = args.speech_source
            
            # get transcripts based on the specified time range
            if args.relative:
                source_msg = f" (speech source: {speech_source})" if speech_source else ""
                print(f"retrieving transcripts from the last {args.relative}{source_msg}...")
                transcripts = get_transcripts_relative(db, args.relative, speech_source)
            else:
                source_msg = f" (speech source: {speech_source})" if speech_source else ""
                print(f"retrieving transcripts from {args.from_time} to {args.to_time}{source_msg}...")
                transcripts = get_transcripts_absolute(db, args.from_time, args.to_time, speech_source)

            # format and display results
            if not transcripts:
                print("no transcripts found for the specified time range.")
                return

            print(f"found {len(transcripts)} transcript words.")

            # convert timestamps to local time if not using UTC
            if not args.utc:
                for transcript in transcripts:
                    if 'absolute_time' in transcript:
                        transcript['absolute_time'] = convert_to_local_time(transcript['absolute_time'])
                    if 'audio_start_time' in transcript:
                        transcript['audio_start_time'] = convert_to_local_time(transcript['audio_start_time'])

            formatted = rewinddb.utils.format_transcript(transcripts)
            print("\ntranscripts:")
            print(formatted)

    except FileNotFoundError as e:
        print(f"error: {e}", file=sys.stderr)
        print("check your DB_PATH setting in .env file", file=sys.stderr)
        sys.exit(1)
    except ConnectionError as e:
        print(f"error: {e}", file=sys.stderr)
        print("check your DB_PASSWORD setting in .env file", file=sys.stderr)
        sys.exit(1)
    except Exception as e:
        print(f"unexpected error: {e}", file=sys.stderr)
        print(f"error type: {type(e).__name__}")
        import traceback
        traceback.print_exc()
        sys.exit(1)


if __name__ == "__main__":
    main()