path: root/src/hircine/cli.py

import argparse
import asyncio
import configparser
import importlib.metadata
import os
import sys
from datetime import datetime, timezone

import alembic.config

import hircine.db as db
from hircine import codename
from hircine.config import init_dir_structure
from hircine.scanner import Scanner


class SubcommandHelpFormatter(argparse.RawDescriptionHelpFormatter):
    def _format_action(self, action):
        parts = super(argparse.RawDescriptionHelpFormatter, self)._format_action(action)
        if action.nargs == argparse.PARSER:
            parts = "\n".join(parts.split("\n")[1:])
        return parts


def init(config, dirs, engine, args):
    if os.path.exists(dirs.database):
        sys.exit("Database already initialized.")

    dirs.mkdirs()

    print("Initializing database...")
    asyncio.run(db.initialize(engine))
    print("Done.")


def scan(config, dirs, engine, args):
    db.ensuredb(dirs)

    asyncio.run(db.ensure_current_revision(engine))

    scanner = Scanner(config, dirs, reprocess=args.reprocess)
    asyncio.run(scanner.scan())
    scanner.report()


def backup(config, dirs, engine, args, tag="manual"):
    db.ensuredb(dirs)

    os.makedirs(dirs.backups, exist_ok=True)

    date = datetime.now(tz=timezone.utc).strftime("%Y-%m-%d_%H%M%S")
    filename = f"{os.path.basename(dirs.database)}.{tag}.{date}"
    path = os.path.join(dirs.backups, filename)

    asyncio.run(db.backup(engine, path))


def migrate(config, dirs, engine, args):
    db.ensuredb(dirs)

    backup(config, dirs, engine, args, tag="pre-migrate")
    alembic.config.main(argv=["--config", db.alembic_ini, "upgrade", "head"])


def vacuum(config, dirs, engine, args):
    db.ensuredb(dirs)

    asyncio.run(db.vacuum(engine))


def version(config, dirs, engine, args):
    version = importlib.metadata.metadata("hircine")["Version"]
    print(f'hircine {version} "{codename}"')


def main():
    parser = argparse.ArgumentParser(
        prog="hircine", formatter_class=SubcommandHelpFormatter
    )
    parser.add_argument("-C", dest="dir", help="run as if launched in DIR")

    subparsers = parser.add_subparsers(title="commands", required=True)

    parser_init = subparsers.add_parser("init", help="initialize a database")
    parser_init.set_defaults(func=init)

    parser_import = subparsers.add_parser("import", help="import archives")
    parser_import.set_defaults(func=scan)
    parser_import.add_argument(
        "-r", "--reprocess", action="store_true", help="reprocess all image files"
    )

    parser_migrate = subparsers.add_parser("migrate", help="run database migrations")
    parser_migrate.set_defaults(func=migrate)

    parser_backup = subparsers.add_parser(
        "backup", help="create a backup of the database"
    )
    parser_backup.set_defaults(func=backup)

    parser_vacuum = subparsers.add_parser(
        "vacuum", help="repack and optimize the database"
    )
    parser_vacuum.set_defaults(func=vacuum)

    parser_version = subparsers.add_parser("version", help="show version and exit")
    parser_version.set_defaults(func=version)

    args = parser.parse_args()

    if args.dir:
        try:
            os.chdir(args.dir)
        except OSError as e:
            sys.exit(e)

    dirs = init_dir_structure()

    config = configparser.ConfigParser()
    config.read(dirs.config)

    engine = db.configure(dirs)

    args.func(config, dirs, engine, args)


if __name__ == "__main__":
    main()