search filestore for orphaned files (deleted from ir.attachments)

filestore_search_orphans.py

#!/usr/bin/env python3
import sys
import os
import time
import psycopg2

def walk_filestore(filestore_path):
    for root, dirs, files in os.walk(filestore_path):
        if "checklist" not in root.split("/")[-2:] and files:
            for file in files:
                assert root.split("/")[-1] == file[:2], f"files in abnormal location in filestore: {root}  -  {file}"
                yield root, file

def is_file_orphan(sql_cr, file_name):
    # true if file does not exist on the database anymore
    sql_cr.execute("SELECT count(id) from ir_attachment where checksum = %s", (file_name, ))
    nbr = sql_cr.fetchone()[0]
    return nbr == 0

def delete_file(file_path):
    print(f"deleting --> {file_path}")
    assert os.path.isfile(file_path), f"not a file, something is weird with {file_path}"
    os.remove(file_path)
    pass

def delete_orphans(filestore_path, dbname, dry_run=True):
    if not dry_run:
        print("this will delete stuff, starting in 5 seconds")
        time.sleep(6)
        print("ok, let's go")

    orphans, ok = 0, 0
    with psycopg2.connect(database=dbname) as cnx, cnx.cursor() as cr:
        for dir_path, file_name in walk_filestore(filestore_path):
            # print(dir_path, file_name)
            if is_file_orphan(cr, file_name):
                orphans += 1
                # print(f"nuke file {file_name}")
                if not dry_run:
                    delete_file(f"{dir_path}/{file_name}")
            else:
                ok += 1
    print(f"orphans count: {orphans}")
    print(f"regular files count: {ok}")

if __name__ == "__main__":
    argv = sys.argv
    if "--help" in argv or len(argv) < 3:
        print("filestore_search_ophanes.py dbname filestore_location [--for-real]")
    else:
        delete_orphans(argv[2], argv[1], "--for-real" not in argv)