ranger plugin for directory history

dir_history.py

import os.path
import sqlite3
import subprocess
import time
import traceback

import ranger.api
import ranger.api.commands


db_path = os.path.expanduser('~/.config/ranger/dir-history.s3db')
db_is_new = not os.path.exists(db_path)
db = sqlite3.connect(db_path)
#db.execute('PRAGMA case_sensitive_like = ON')

if db_is_new:
    db.execute('CREATE TABLE [history] ([directory] TEXT UNIQUE, [time] INTEGER)')
    db.execute('CREATE INDEX [history-index] ON [history] ([time] DESC)')


def handle_cd(signal):
    db.execute('INSERT OR REPLACE INTO [history] ([directory], [time]) VALUES (?, ?)',
               (signal.new.path, time.time_ns()))
    db.commit()


HOOK_INIT_OLD = ranger.api.hook_init


def hook_init(fm):
    fm.signal_bind('cd', handle_cd)
    HOOK_INIT_OLD(fm)


ranger.api.hook_init = hook_init


class dir_history_search(ranger.api.commands.Command):
    def execute(self):
        self.fm.ui.suspend()
        try:
            p = subprocess.Popen(
                ["fzf", "--read0", "--tiebreak=index", "--scheme=history"],
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
            )
            try:
                c = db.cursor()
                for row in c.execute('SELECT [directory] FROM [history] ORDER BY [time] DESC'):
                    p.stdin.write(row[0].encode('utf-8'))
                    p.stdin.write(bytes([0]))
                p.stdin.close()
                directory = p.stdout.read().decode('utf-8')
            finally:
                r = p.wait()
            if r != 0:
                raise Exception('fzf exited with status ' + str(r))
        finally:
            self.fm.ui.initialize()
        self.fm.execute_console("cd " + directory.rstrip('\n'))