utility script to transfer phashes from one stash db to another

stash_phash_transfer.py

#!/usr/bin/env python
# coding: utf-8

# utility script for https://github.com/stashapp/stash
# 2024, poontology

# finds fingerprints from destination db that are without phashes then finds
# phashes from source db that matches the oshashes and copies the phashes over
# (benefit being to avoid rehashing thousands of large video files)
# NOTE: target db files need to be writable (sudo or chown them)

import sqlite3
import sys
import os


def find_oshashes_wo_phash(cur):
    cur.execute("""select file_id, fingerprint
                   from files_fingerprints
                   where type='oshash' and file_id in (
                     select file_id from files_fingerprints
                     group by file_id having count()=1
                   )""")
    return cur.fetchall()


def find_phashes_for_oshashes(cur, oshashes):
    phashes = {}
    for file_id, oshash in oshashes:
        cur.execute("""select fingerprint
                       from files_fingerprints
                       where type='phash' and file_id=(
                         select file_id from files_fingerprints
                         where fingerprint=?
                       )""", (oshash,))
        res = cur.fetchone()
        if res:
            phashes[file_id] = res[0]
    return phashes


def insert_phashes(cur, phashes):
    for file_id, hash in phashes.items():
        cur.execute("""insert into files_fingerprints (file_id,type,fingerprint)
                       values(?,'phash',?)""", (file_id, hash))


if __name__ == '__main__':
    if len(sys.argv) != 3:
        print("Syntax:", sys.argv[0], "[src_stash_with_phashes.sqlite]",
              "[dst_stash_with_oshashes.sqlite]")
        sys.exit(1)

    srcfile, dstfile = sys.argv[1:3]
    for fn in [srcfile, dstfile]:
        if not os.access(fn, os.R_OK):
            print("Error: file not readable: ", srcfile)
            sys.exit(2)
    for suffix in ['', '-shm', '-wal']:
        fn = dstfile + suffix
        if os.path.exists(fn) and not os.access(fn, os.W_OK):
            print("Error: file not writable:", dstfile + suffix)
            sys.exit(2)

    srccon = sqlite3.connect(srcfile)
    dstcon = sqlite3.connect(dstfile)

    try:
        cur = dstcon.cursor()
        oshashes = find_oshashes_wo_phash(cur)
        cur = srccon.cursor()
        phashes = find_phashes_for_oshashes(cur, oshashes)

        print("found", len(phashes), "phashes for", len(oshashes),
              "lonely oshashes")
        input("Press enter to continue (ctrl-c to cancel)")

        cur = dstcon.cursor()
        insert_phashes(cur, phashes)
        dstcon.commit()
        print("done")
    finally:
        dstcon.close()
        srccon.close()