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
129
130
131
132
133
134
135
|
# Copyright (C) 2017 Alban Gruin
#
# celcatsanitizer is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# celcatsanitizer is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with celcatsanitizer. If not, see <http://www.gnu.org/licenses/>.
import datetime
from django.core.management.base import BaseCommand
from django.db import transaction
from django.db.models import Min
from edt.models import Course, Timetable
from edt.utils import get_week, tz_now
from ._private import delete_courses_in_week, get_events, get_update_date, get_weeks, get_xml
@transaction.atomic
def process_timetable_week(timetable, soup, weeks_in_soup, force, year=None, week=None):
begin, end = get_week(year, week)
# Si on force la mise à jour, on définit de moment
# de la mise à jour au début de la semaine
if force:
today = begin
else:
today = tz_now()
# On récupère la mise à jour la plus ancienne dans les cours de l’emploi du temps
# commençant à partir de maintenant
last_update_date = Course.objects.filter(timetable=timetable, begin__gte=today)
if year is not None and week is not None:
# Si jamais on traite une semaine spécifique, on limite les cours sélectionnés
# à ceux qui commencent entre le début du traitement et la fin de la semaine
last_update_date = last_update_date.filter(begin__lt=end)
last_update_date = last_update_date.aggregate(Min("last_update")) \
["last_update__min"]
# Date de mise à jour de Celcat, utilisée à des fins de statistiques
new_update_date = get_update_date(soup)
# On ne fait pas la mise à jour si jamais la dernière date de MàJ est plus récente
# que celle indiquée par Celcat.
# Attention, le champ last_update de la classe Course représente l’heure à laquelle
# le cours a été inséré dans la base de données, et non pas la date indiquée par
# Celcat.
if not force and last_update_date is not None and new_update_date is not None and \
last_update_date >= new_update_date:
return
if year is not None and week is not None:
# On efface la semaine à partir de maintenant si jamais
# on demande le traitement d’une seule semaine
delete_courses_in_week(timetable, year, week, today)
else:
# Sinon, on efface tous les cours à partir de maintenant.
# Précisément, on prend la plus grande valeur entre la première semaine
# présente dans Celcat et maintenant.
delete_from = min(weeks_in_soup.values())
if not force:
# Si jamais on force la MàJ, on efface tout à partir de la première semaine
delete_from = max(delete_from, today)
Course.objects.filter(timetable=timetable, begin__gte=delete_from).delete()
# Tous les cours commençant sur la période traitée
# sont parsés, puis enregistrés dans la base de données.
for course in get_events(timetable, soup, weeks_in_soup, today, year, week):
course.save()
# On renseigne la date de mise à jour de Celcat, à des fins de statistiques
timetable.last_update_date = new_update_date
timetable.save()
def process_timetable(timetable, force, year=None, weeks=None):
soup = get_xml(timetable.url)
weeks_in_soup = get_weeks(soup)
if year is not None and weeks is not None:
for week in weeks:
process_timetable_week(timetable, soup, weeks_in_soup, force, year, week)
else:
process_timetable_week(timetable, soup, weeks_in_soup, force)
class Command(BaseCommand):
help = "Fetches registered celcat timetables"
def add_arguments(self, parser):
parser.add_argument("--all", const=True, default=False, action="store_const")
parser.add_argument("--force", const=True, default=False, action="store_const")
parser.add_argument("--week", type=int, choices=range(1, 54), nargs="+")
parser.add_argument("--year", type=int, nargs=1)
def handle(self, *args, **options):
year = None
errcount = 0
if options["all"]:
weeks = None
elif options["week"] is None:
_, week, day = tz_now().isocalendar()
if day >= 6:
year, week, _ = (tz_now() + datetime.timedelta(weeks=1)).isocalendar()
weeks = [week]
else:
weeks = options["week"]
if not options["all"]:
if options["year"] is None and year is None:
year = tz_now().year
elif year is None:
year = options["year"][0]
for timetable in Timetable.objects.all():
self.stdout.write("Processing {0}".format(timetable))
try:
process_timetable(timetable, options["force"], year, weeks)
except Exception as exc:
self.stderr.write(
self.style.ERROR("Failed to process {0}: {1}".format(timetable, exc)))
errcount += 1
if errcount == 0:
self.stdout.write(self.style.SUCCESS("Done."))
else:
self.stdout.write(self.style.ERROR("Done with {0} errors.".format(errcount)))
|
