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
|
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from argparse import RawTextHelpFormatter
import datetime
import sys
from django.core.management.base import BaseCommand, CommandError
from django.apps import apps
from ishtar_common import models_common
APPS = (
"ishtar_common",
"archaeological_operations",
"archaeological_context_records",
"archaeological_files",
"archaeological_finds",
"archaeological_warehouse",
)
def task_main_image(quiet=False):
for model in apps.get_models():
if not issubclass(model, models_common.DocumentItem):
continue
if not hasattr(model, "main_image"):
continue
q = model.objects.filter(
main_image__isnull=True, documents__image__isnull=False
).exclude(documents__image="")
nb = q.count()
if not nb: # no image attached
continue
for item in q.all():
q_docs = item.documents.filter(image__isnull=False).exclude(image="")
item.main_image = q_docs.order_by("pk").all()[0]
item.skip_history_when_saving = True
item.save()
if not quiet:
sys.stdout.write(f"{nb} main image fixed for {model.__name__}\n")
def get_time():
return datetime.datetime.now().isoformat().split(".")[0]
TASKS = {
"main_image": {
"help": "for items with images and no main image, put the first one created as a main image.",
"action": task_main_image,
}
}
class Command(BaseCommand):
help = "Launch maintenance tasks: \n" + "\n".join(
[f"* {t}: {TASKS[t]['help']}" for t in TASKS.keys()]
)
def create_parser(self, *args, **kwargs):
parser = super(Command, self).create_parser(*args, **kwargs)
parser.formatter_class = RawTextHelpFormatter
return parser
def add_arguments(self, parser):
parser.add_argument("task")
parser.add_argument(
"--quiet", dest="quiet", action="store_true", help="Quiet output"
)
def handle(self, *args, **options):
if options["task"] not in TASKS.keys():
msg = f"{options['task']} is not a valid task. Available tasks are:\n"
msg += "\n".join(TASKS.keys())
raise CommandError(msg)
quiet = options["quiet"]
if not quiet:
sys.stdout.write(f"[{get_time()}] Processing task {options['task']}\n")
errors = TASKS[options["task"]]["action"](quiet=quiet)
if not errors:
if not quiet:
sys.stdout.write(f"[{get_time()}] Task {options['task']} finished\n")
sys.exit()
if not quiet:
sys.stdout.write("\n".join(errors))
sys.exit(1)
|
