ocr_cli.py

#!/usr/bin/env python
"""
ocr_cli.py - command line interface for retrieving screen OCR data 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 screen OCR data for the specified time range
5. optionally filter by application if --app is specified
6. deduplicate OCR data to remove entries with very similar text content
7. format and display results (with conditional app name display)
8. 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

additional features:
- list all applications that have OCR data with --list-apps
- filter OCR data by specific application with --app
- automatic deduplication of similar text content (keeps most recent entries)
- improved display format when filtering by app (hides redundant app names)

examples:
  python ocr_cli.py --relative "1 hour"
  python ocr_cli.py --relative "5 hours"
  python ocr_cli.py --relative "5h"
  python ocr_cli.py --relative "30m"
  python ocr_cli.py --relative "2d"
  python ocr_cli.py --relative "1w"
  python ocr_cli.py --from "2023-05-11 13:00:00" --to "2023-05-11 17:00:00"
  python ocr_cli.py --list-apps
  python ocr_cli.py --relative "1 day" --app "com.apple.Safari"
"""

import argparse
import datetime
from datetime import timezone
import re
import sys
import hashlib
import time
# Removed SequenceMatcher import - now using fast fingerprint approach
# Removed defaultdict import - no longer needed with fingerprint approach

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_ocr_data_relative(db, time_str):
    """get screen OCR data from a relative time period.

    args:
        db: rewinddb instance
        time_str: relative time string (e.g., "1 hour", "5 hours")

    returns:
        list of OCR data dictionaries
    """

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


def get_ocr_data_absolute(db, from_time_str, to_time_str):
    """get screen OCR data 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"

    returns:
        list of OCR data 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_screen_ocr_text_absolute(from_time, to_time)
    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 get_applications_with_ocr_data(db, time_str=None, from_time_str=None, to_time_str=None):
    """get list of applications that have OCR data.

    args:
        db: rewinddb instance
        time_str: optional relative time string (e.g., "1 week")
        from_time_str: optional start time string
        to_time_str: optional end time string

    returns:
        sorted list of unique application names
    """

    # get OCR data for the specified time range
    if time_str:
        ocr_data = get_ocr_data_relative(db, time_str)
    elif from_time_str and to_time_str:
        ocr_data = get_ocr_data_absolute(db, from_time_str, to_time_str)
    else:
        # default to last week if no time range specified
        ocr_data = get_ocr_data_relative(db, "1 week")

    # extract unique application names
    applications = set()
    for item in ocr_data:
        app = item.get('application')
        if app:
            applications.add(app)

    return sorted(list(applications))


def filter_ocr_data_by_app(ocr_data, app_name):
    """filter OCR data to only include entries from a specific application.

    args:
        ocr_data: list of OCR data dictionaries
        app_name: application name to filter by

    returns:
        filtered list of OCR data dictionaries
    """

    return [item for item in ocr_data if item.get('application') == app_name]


def normalize_text_for_similarity(text):
    """normalize text for similarity comparison.

    args:
        text: input text string

    returns:
        normalized text string suitable for similarity matching
    """
    if not text:
        return ""

    # convert to lowercase, remove extra whitespace, strip
    normalized = re.sub(r'\s+', ' ', text.lower().strip())
    # remove common punctuation that doesn't affect meaning but keep structure
    normalized = re.sub(r'[.,;:!?"\'\-_(){}[\]]+', '', normalized)
    return normalized


# Removed calculate_text_similarity - now using fast fingerprint approach


# Removed old fuzzy matching functions - now using fast fingerprint approach


def deduplicate_ocr_data_fuzzy(ocr_data, similarity_threshold=None, debug=False):
    """fast, lossy deduplication using improved text fingerprinting for maximum speed.

    uses multi-part text fingerprinting instead of similarity calculations:
    - creates fingerprints using first 30 chars + words 3-5 + last 20 chars + length
    - uses dictionary lookups for O(1) duplicate detection
    - catches variations in middle of text (like "CONANT H V S" vs "CONANT H V")
    - prioritizes speed over perfect accuracy
    - target: process 1000+ entries in under 1 second

    keeps the most recent entry when duplicates are found.

    args:
        ocr_data: list of OCR data dictionaries
        similarity_threshold: ignored (kept for compatibility)
        debug: whether to show timing and debug information

    returns:
        tuple of (deduplicated_ocr_data, num_duplicates_removed)
    """
    if not ocr_data:
        return ocr_data, 0

    start_time = time.time()

    if debug:
        print(f"starting fast fingerprint deduplication with {len(ocr_data)} entries...")

    # sort by frame_time to process chronologically (keep most recent)
    sorted_data = sorted(ocr_data, key=lambda x: x.get('frame_time', datetime.datetime.min))

    # dictionary to track fingerprints: fingerprint -> (index, item)
    fingerprint_to_item = {}
    deduplicated = []
    duplicates_removed = 0

    for current_item in sorted_data:
        current_text = current_item.get('text', '')
        current_app = current_item.get('application', '')

        # skip items with no meaningful text
        if not current_text.strip():
            deduplicated.append(current_item)
            continue

        # create improved fingerprint using multiple text parts
        normalized_text = normalize_text_for_similarity(current_text)
        text_len = len(normalized_text)

        # create fingerprint components
        first_part = normalized_text[:30] if len(normalized_text) > 30 else normalized_text
        last_part = normalized_text[-20:] if len(normalized_text) > 20 else ""

        # get middle words (words 3-5) to catch variations like "V S" vs "V" vs "U2"
        words = normalized_text.split()
        middle_words = ""
        if len(words) >= 5:
            middle_words = " ".join(words[2:5])  # words 3-5 (0-indexed)
        elif len(words) >= 3:
            middle_words = " ".join(words[2:])   # remaining words after first 2

        # combine into fingerprint: app:length:first30:middle_words:last20
        fingerprint = f"{current_app}:{text_len}:{first_part}:{middle_words}:{last_part}"

        if fingerprint in fingerprint_to_item:
            # this is a duplicate - replace the existing item with the more recent one
            existing_index, existing_item = fingerprint_to_item[fingerprint]
            deduplicated[existing_index] = current_item
            fingerprint_to_item[fingerprint] = (existing_index, current_item)
            duplicates_removed += 1
        else:
            # this is a new unique item
            new_index = len(deduplicated)
            deduplicated.append(current_item)
            fingerprint_to_item[fingerprint] = (new_index, current_item)

    elapsed = time.time() - start_time

    if debug:
        print(f"fast fingerprint deduplication completed in {elapsed:.3f} seconds")
        print(f"processed {len(ocr_data)} entries, removed {duplicates_removed} duplicates")
        print(f"processing rate: {len(ocr_data)/elapsed:.0f} entries/second")

    return deduplicated, duplicates_removed


def deduplicate_ocr_data_fast(ocr_data, debug=False):
    """legacy fast hash-based deduplication (kept for compatibility).

    this is the old exact-match approach. use deduplicate_ocr_data_fuzzy for better results.

    args:
        ocr_data: list of OCR data dictionaries
        debug: whether to show timing information

    returns:
        tuple of (deduplicated_ocr_data, num_duplicates_removed)
    """
    if not ocr_data:
        return ocr_data, 0

    start_time = time.time() if debug else None

    # sort by frame_time to process chronologically
    sorted_data = sorted(ocr_data, key=lambda x: x.get('frame_time', datetime.datetime.min))

    # use dictionary to track unique text hashes
    hash_to_item = {}
    deduplicated = []
    duplicates_removed = 0

    for current_item in sorted_data:
        current_text = current_item.get('text', '')
        current_app = current_item.get('application', '')

        # skip items with no meaningful text
        if not current_text.strip():
            deduplicated.append(current_item)
            continue

        # create hash for this text + app combination
        normalized_text = normalize_text_for_similarity(current_text)
        hash_input = f"{current_app}:{normalized_text}"
        text_hash = hashlib.md5(hash_input.encode('utf-8')).hexdigest()

        if text_hash in hash_to_item:
            # this is a duplicate - replace the existing item with the more recent one
            existing_index, existing_item = hash_to_item[text_hash]
            deduplicated[existing_index] = current_item
            hash_to_item[text_hash] = (existing_index, current_item)
            duplicates_removed += 1
        else:
            # this is a new unique item
            new_index = len(deduplicated)
            deduplicated.append(current_item)
            hash_to_item[text_hash] = (new_index, current_item)

    if debug and start_time:
        elapsed = time.time() - start_time
        print(f"hash-based deduplication completed in {elapsed:.3f} seconds")

    return deduplicated, duplicates_removed


def deduplicate_ocr_data(ocr_data, similarity_threshold=None, debug=False):
    """deduplicate OCR data using fast fingerprint-based matching.

    uses fast text fingerprinting instead of similarity calculations for maximum speed.
    can catch similar entries like:
    - "CONANT H V S + home DMs Activity Later..."
    - "CONANT H V + home DMs Activity Later..."
    - "CONANT H U2 + home DMs Activity Later..."

    keeps the most recent entry when duplicates are found.

    args:
        ocr_data: list of OCR data dictionaries
        similarity_threshold: ignored (kept for compatibility)
        debug: whether to show timing and debug information

    returns:
        tuple of (deduplicated_ocr_data, num_duplicates_removed)
    """
    return deduplicate_ocr_data_fuzzy(ocr_data, similarity_threshold, debug=debug)


# Compatibility functions for existing test files
def normalize_text_for_deduplication(text):
    """legacy function name - redirects to normalize_text_for_similarity for compatibility."""
    return normalize_text_for_similarity(text)


def create_text_hash(text, app_name=""):
    """legacy function for creating text hash - kept for compatibility with test files."""
    normalized_text = normalize_text_for_similarity(text)
    hash_input = f"{app_name}:{normalized_text}"
    return hashlib.md5(hash_input.encode('utf-8')).hexdigest()


def format_ocr_data_with_text(ocr_data, show_app_name=True):
    """format OCR data into readable text with actual text content.

    converts a list of OCR data dictionaries into a formatted string
    with timestamps, applications, and extracted text content.

    args:
        ocr_data: list of OCR data dictionaries
        show_app_name: whether to show application name in the output

    returns:
        formatted string representation of the OCR data
    """

    if not ocr_data:
        return "no OCR data available."

    # group by frame time and application for better readability
    frames = {}
    for item in ocr_data:
        frame_time = item['frame_time']
        application = item.get('application', 'Unknown')
        window = item.get('window', 'Unknown')
        text = item.get('text', '')

        # create a key for grouping
        time_str = frame_time.strftime('%Y-%m-%d %H:%M:%S')
        key = f"{time_str}_{application}_{window}"

        if key not in frames:
            frames[key] = {
                'time': frame_time,
                'application': application,
                'window': window,
                'texts': []
            }

        if text.strip():  # only add non-empty text
            frames[key]['texts'].append(text.strip())

    # sort frames by time
    sorted_frames = sorted(frames.items(), key=lambda x: x[1]['time'])

    # format each frame
    result = []
    for key, frame in sorted_frames:
        time_str = frame['time'].strftime('%Y-%m-%d %H:%M:%S')

        if show_app_name:
            app_str = f"{frame['application']}"
            if frame['window'] and frame['window'] != 'Unknown':
                app_str += f" - {frame['window']}"
            result.append(f"[{time_str}] {app_str}")
        else:
            # when not showing app name, still show window if available
            if frame['window'] and frame['window'] != 'Unknown':
                result.append(f"[{time_str}] {frame['window']}")
            else:
                result.append(f"[{time_str}]")

        # show text content
        if frame['texts']:
            for text in frame['texts']:
                # show full text content without truncation
                result.append(f"  {text}")
        else:
            result.append("  (no text content)")

        result.append("")  # empty line between frames

    return "\n".join(result)


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

    returns:
        parsed argument namespace
    """

    parser = argparse.ArgumentParser(
        description="retrieve screen OCR data 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 --list-apps  # list all applications with OCR data
  %(prog)s --relative "1 day" --app "com.apple.Safari"  # filter by application
"""
    )

    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("--list-apps", action="store_true", help="list all applications that have OCR data")

    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("--app", metavar="APPLICATION", help="filter OCR data by specific application name")
    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")

    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 that --app can only be used with time range queries
    if args.app and args.list_apps:
        parser.error("--app cannot be used with --list-apps")

    return args


def main():
    """main function for the OCR 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:

            if args.list_apps:
                # list all applications with OCR data
                print("retrieving applications with OCR data...")
                applications = get_applications_with_ocr_data(db)

                if not applications:
                    print("no applications found with OCR data.")
                    return

                print(f"found {len(applications)} applications with OCR data:")
                for app in applications:
                    print(f"  {app}")
                return

            # get OCR data based on the specified time range
            if args.relative:
                print(f"retrieving OCR data from the last {args.relative}...")
                ocr_data = get_ocr_data_relative(db, args.relative)
            else:
                print(f"retrieving OCR data from {args.from_time} to {args.to_time}...")
                ocr_data = get_ocr_data_absolute(db, args.from_time, args.to_time)

            # filter by application if specified
            if args.app:
                print(f"filtering OCR data for application: {args.app}")
                original_count = len(ocr_data)
                ocr_data = filter_ocr_data_by_app(ocr_data, args.app)
                filtered_count = len(ocr_data)
                print(f"filtered from {original_count} to {filtered_count} OCR entries.")

            # deduplicate OCR data
            if ocr_data:
                if args.debug:
                    print("deduplicating OCR data using fast fingerprint matching...")
                else:
                    print("deduplicating OCR data...")
                ocr_data, duplicates_removed = deduplicate_ocr_data(ocr_data, debug=args.debug)
                if duplicates_removed > 0:
                    print(f"removed {duplicates_removed} duplicate entries.")
                else:
                    print("no duplicates found.")

            # format and display results
            if not ocr_data:
                if args.app:
                    print(f"no OCR data found for application '{args.app}' in the specified time range.")
                else:
                    print("no OCR data found for the specified time range.")
                return

            print(f"found {len(ocr_data)} OCR entries after deduplication.")

            # convert timestamps to local time if not using UTC
            if not args.utc:
                for item in ocr_data:
                    if 'frame_time' in item:
                        item['frame_time'] = convert_to_local_time(item['frame_time'])

            # format with conditional app name display
            show_app_name = not bool(args.app)  # hide app name when filtering by specific app
            formatted = format_ocr_data_with_text(ocr_data, show_app_name=show_app_name)
            print("\nOCR data:")
            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()