Import my .zsh_eternal_history to the atuin db

zsh_eternal_to_atuin.py

#!/usr/bin/env python3

# Load history from ~/.zsh_eternal_history and save it to the atuin history database
# Modified from Mark Hepburn's code
# https://blog.markhepburn.com/posts/atuin-and-per-directory-history-import/

# My eternal history command was copied from someone elses, *years* ago.

# ZSH_ETERNAL_HISTORY=~/.zsh_eternal_history
# [[ ! -f $ZSH_ETERNAL_HISTORY ]] && touch $ZSH_ETERNAL_HISTORY
# precmd () { echo -e $$\\t$USER\\t$HOSTNAME\\tscreen $WINDOW\\t`command date +%D%t%T%t%Y%t%s`\\t$PWD"$(history -1)" >> $ZSH_ETERNAL_HISTORY }

#9560	huntert	timcent	screen 0	07/20/13 04:08:08 2013 1374289688	/home/huntert 2032  touch ~/.zsh_eternal_history
#9560	huntert	timcent	screen 0	07/20/13 04:08:10 2013 1374289690	/home/huntert 2033  ls
#9560	huntert	timcent	screen 0	07/20/13 04:09:57 2013 1374289797	/home/huntert 2034  man virt-install
#9560	huntert	timcent	screen 0	07/20/13 04:10:06 2013 1374289806	/home/huntert 2035  sudo yum list updates
#tab between pid, user, hostname, screen 0, date, and path historyline command

import sqlite3
import sys
import uuid
import re
import argparse

debug=0
limit=0

def load_history(fpath):
    """
    Load history from fpath, return a list of dicts

    We will process each line in the file
    - if it matches the pattern, we will place it in 'newentry'
    - if it does not match the pattern, we will assume it is a continuation of the last entry's cmd
    """
    global debug, limit

    history = []
    # last command added to history
    lastcmd = ''
    lastline = ''
    # next entry to be added to history
    newentry = {}
    # regex to match the line
    p = re.compile(r'^(?P<pid>\d+)\t(?P<user>\w+)\t(?P<hostname>[^\t]+)\t(?P<screen>[^\t]+)\t' +
                   r'(?P<date>[^\t]+)\t(?P<path>.+?) +(?P<line>\d+) +(?P<cmd>.*)')

    with open(fpath, 'rb') as f:
        for lc, line in enumerate(iter(f)):
            try:
                line = line.decode('utf-8')
                if limit>0 and lc>limit:
                    print(f"Stopping after {limit} lines")
                    break

                line = line.strip()
                if debug>0:
                    print(f"line {lc}: {line}")

                match = p.match(line)
                if not match:
                    if not newentry:
                        print(f'Unexpected line {lc} -- did not match pattern, but no previous entry: {line}')
                        print("Stopping without writing")
                        sys.exit(1)
                    # this line is a continuation of the last entry
                    newentry['command'] += '\n' + line
                    continue

                # we have a new entry
                if (len(match.groups()) < 8):
                    print(f'Unexpected line {lc} -- found {len(match.groups())} groups, expected 8: {line}')
                    print("Stopping without writing")
                    sys.exit(1)

                # write the last entry
                if newentry:
                    if (len(newentry) < 8):
                        print(f'Unexpected line {lc-1}: {lastline}')
                        print("Stopping without writing")
                        sys.exit(1)
                    if newentry['command'] != lastcmd: # Ignore dups, since the timestamp doesn't always have enough resolution
                        history.append(newentry)
                        lastcmd = newentry['command']
                        if (debug>1):
                            print(f"writing new entry: {newentry}")
                        newentry = {}
                    else:
                        print(f"Skipping duplicate entry ({newentry['command']})")

                # prepare entry for this line
                lastline = line
                newentry['id'] = uuid.uuid4().hex
                if match.group('date'):
                    newentry['timestamp'] = int(match.group('date').split(' ')[-1]) * 1e9
                else:
                    print(f'Unexpected line {lc} (no date): {lastline}')
                    print("Stopping without writing")
                    sys.exit(1)
                newentry['duration'] = 0
                newentry['exit'] = 0
                newentry['command'] = match.group('cmd')
                newentry['cwd'] = match.group('path')
                newentry['session'] = match.group('pid')
                newentry['hostname'] = f"{match.group('hostname')}:{match.group('user')}"

            except Exception as e:
                raise Exception(f'Error: {e} at line {lc}: {line}')
    return history


INSERT_SQL = """
INSERT INTO history (id, timestamp, duration, exit, command, cwd, session, hostname)
  VALUES (:id, :timestamp, :duration, :exit, :command, :cwd, :session, :hostname)
  ON CONFLICT DO NOTHING
"""

def main():
    global debug, limit

    parser = argparse.ArgumentParser(description='Convert zsh eternal history to atuin database')
    parser.add_argument('input_file', help='Input file containing zsh eternal history')
    parser.add_argument('database_file', help='Output sqlite database file')
    parser.add_argument('-d', '--debug', type=int, default=0, help='Debug level')
    parser.add_argument('--limit', type=int, default=0, help='Limit the number of lines to process')
    args = parser.parse_args()

    debug = args.debug
    limit = args.limit

    db = sqlite3.connect(args.database_file)
    cur = db.cursor()

    print(f'Loading history from {args.input_file}, writing to {args.database_file}')

    history = load_history(args.input_file)

    # my zsh_eternal_history has 11777 lines, we'll split at 1000
    for i in range(0, len(history), 1000):
        count = cur.execute("SELECT COUNT(*) FROM history").fetchone()[0]
        print(f'Adding {len(history[i:i+1000])} commands to existing total {count}', end='')
        try:
            cur.executemany(INSERT_SQL, history[i:i+1000])
        except Exception as e:
            print(f'Error: {e} at range {i}:{i+1000}')
            break
        updatecount = cur.execute("SELECT COUNT(*) FROM history").fetchone()[0]
        print(f'...added {updatecount-count} entries')

    db.commit()
    db.close()

if __name__ == '__main__':
    main()