PartsInventorySystem/app/admin.py

"""Recovery CLI.

Everyday password changes happen in the UI. This exists for the one case the UI
cannot help with — a forgotten password — and for the initial handover.

    docker exec -it parts python -m app.admin set-password
    docker exec parts python -m app.admin show-status
    docker exec parts python -m app.admin check-images
    docker exec parts python -m app.admin prune-images
"""

import getpass
import sys

from . import auth, db


def set_password(argv):
    password = argv[0] if argv else None
    if password is None:
        password = getpass.getpass("New password: ")
        if password != getpass.getpass("Repeat: "):
            print("Passwords did not match.", file=sys.stderr)
            return 1
    problem = auth.password_problem(password)
    if problem:
        print(problem, file=sys.stderr)
        return 1
    with db.session() as conn:
        auth.set_password(conn, password)
    print("Password set. All existing sessions have been signed out.")
    return 0


def clear_password(_argv):
    """Fall back to the PARTS_PASSWORD environment variable again."""
    with db.session() as conn:
        conn.execute("DELETE FROM settings WHERE key = ?", (auth.PASSWORD_KEY,))
        auth.bump_epoch(conn)
    print("Stored password cleared; PARTS_PASSWORD from the environment is live again.")
    print("All existing sessions have been signed out.")
    return 0


# An upload writes its file before inserting the row, so for a moment a live
# file legitimately has no row. Pruning anything younger than this would delete
# a photo out from under a request that is still in flight.
PRUNE_MIN_AGE_SECONDS = 3600


def prune_images(argv):
    """Delete image files with no row pointing at them.

    Only files older than an hour, unless --all is given — and --all is only
    safe with the app stopped, because a file younger than its row is exactly
    what an in-flight upload looks like.
    """
    import os
    import time

    ignore_age = "--all" in argv
    with db.session() as conn:
        known = {
            os.path.basename(db.image_path(r["token"], r["mime"]))
            for r in conn.execute("SELECT token, mime FROM part_images").fetchall()
        }
    removed = skipped = 0
    now = time.time()
    for name in os.listdir(db.IMAGE_DIR):
        if name in known:
            continue
        path = os.path.join(db.IMAGE_DIR, name)
        if not ignore_age and now - os.path.getmtime(path) < PRUNE_MIN_AGE_SECONDS:
            skipped += 1
            continue
        os.remove(path)
        removed += 1
    print(f"{removed} orphaned image file(s) removed; {len(known)} referenced file(s) kept.")
    if skipped:
        print(f"{skipped} skipped as too recent to be certain they are orphans "
              f"(stop the app and re-run with --all to include them).")
    return 0


def check_images(_argv):
    """Report drift in both directions between the database and the files."""
    import os

    with db.session() as conn:
        rows = conn.execute("SELECT token, mime, part_id FROM part_images").fetchall()
    expected = {os.path.basename(db.image_path(r["token"], r["mime"])): r for r in rows}
    present = set(os.listdir(db.IMAGE_DIR))
    missing = sorted(set(expected) - present)
    orphans = sorted(present - set(expected))
    print(f"{len(expected)} referenced photo(s), {len(present)} file(s) on disk")
    for name in missing:
        print(f"  MISSING FILE  part {expected[name]['part_id']}  {name}")
    for name in orphans:
        print(f"  orphan file   {name}")
    # A referenced photo with no file is data loss; a stray file is only clutter.
    return 1 if missing else 0


def list_image_files(_argv):
    """Every filename the database expects to exist. Used by the backup script."""
    import os

    with db.session() as conn:
        for r in conn.execute("SELECT token, mime FROM part_images ORDER BY id").fetchall():
            print(os.path.basename(db.image_path(r["token"], r["mime"])))
    return 0


def show_status(_argv):
    import os

    with db.session() as conn:
        stored = auth.stored_hash(conn)
        images = conn.execute("SELECT COUNT(*) AS n, COALESCE(SUM(bytes), 0) AS b "
                              "FROM part_images").fetchone()
        print("password source :", "database (set in the app)" if stored else "PARTS_PASSWORD env var")
        print("session epoch   :", auth.current_epoch(conn))
        print("auth enabled    :", auth.auth_enabled())
        print("images          :", f"{images['n']} rows, {images['b'] / 1024 / 1024:.1f} MB",
              f"({len(os.listdir(db.IMAGE_DIR))} files in {db.IMAGE_DIR})")
    return 0


COMMANDS = {"set-password": set_password, "clear-password": clear_password,
            "show-status": show_status, "prune-images": prune_images,
            "check-images": check_images, "list-image-files": list_image_files}


def main(argv=None):
    argv = list(sys.argv[1:] if argv is None else argv)
    if not argv or argv[0] not in COMMANDS:
        print("usage: python -m app.admin {" + "|".join(COMMANDS) + "}", file=sys.stderr)
        return 2
    db.init()
    return COMMANDS[argv[0]](argv[1:])


if __name__ == "__main__":
    raise SystemExit(main())