1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
|
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()
|
