Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b5c84635cf |
No files matched your search
+1
-3
@@ -1,5 +1,3 @@
|
||||
env/*
|
||||
config.json
|
||||
config.json.*
|
||||
csv/*
|
||||
logs/
|
||||
config.json.orig
|
||||
@@ -1,28 +0,0 @@
|
||||
# Consignes de maintenance
|
||||
|
||||
- Mettre à jour `docs/crealead-caldav.md` lors de toute évolution impactant
|
||||
`crealead-caldav.py`, notamment ses pré-requis, sa configuration, son
|
||||
lancement, le traitement des événements, ses sorties CSV, sa journalisation
|
||||
ou ses tests dans `tests/`.
|
||||
- Garder les exemples de configuration, les commandes d'exécution et les cas
|
||||
de test documentés cohérents avec le comportement du script.
|
||||
- Chaque nouvelle ligne écrite dans `logs/crealead-caldav.log` doit commencer
|
||||
par l'heure d'écriture en UTC au format ISO 8601 avec suffixe `Z`, entre
|
||||
crochets. Le format par défaut est :
|
||||
`[2026-10-15T12:34:56Z]: [info] - message`.
|
||||
Cela s'applique aux sorties standard et d'erreur ainsi qu'aux lignes des
|
||||
traces d'exception.
|
||||
- Tout événement non exporté doit laisser une trace au format
|
||||
`[2026-10-15T12:34:56Z]: [warning][event] - message`, sans préfixe `[info]`
|
||||
supplémentaire. Préciser le calendrier concerné, le titre, le moment de
|
||||
l'événement (début et fin disponibles) et la raison de son exclusion.
|
||||
Ajouter le fichier CSV et l'UID pour faciliter son identification.
|
||||
- Ne jamais ignorer silencieusement un événement, notamment en cas de
|
||||
`SUMMARY` absent, de dates absentes ou invalides, de document iCalendar
|
||||
illisible ou d'export CSV interrompu. Récupérer les métadonnées disponibles
|
||||
dans les données brutes si nécessaire et indiquer `indisponible` pour celles
|
||||
qui ne peuvent pas être retrouvées. Échapper les retours à la ligne des
|
||||
métadonnées et du motif pour conserver un avertissement sur une seule ligne.
|
||||
- Lors d'une évolution du traitement des événements ou de la journalisation,
|
||||
vérifier dans les tests que les événements écartés sont bien tracés avec les
|
||||
informations et les préfixes ci-dessus.
|
||||
+46
-212
@@ -17,143 +17,9 @@ import sys
|
||||
import caldav
|
||||
import json
|
||||
import csv
|
||||
from datetime import date, datetime, time, timezone
|
||||
from icalendar import Calendar
|
||||
import calendar
|
||||
from datetime import date, datetime, timezone
|
||||
import pytz
|
||||
from urllib.parse import unquote
|
||||
import re
|
||||
import traceback
|
||||
from contextlib import contextmanager, redirect_stdout, redirect_stderr
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def utc_timestamp():
|
||||
return datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
|
||||
|
||||
|
||||
class TimestampedLog:
|
||||
"""Prefix each log line, including lines written in several fragments."""
|
||||
|
||||
def __init__(self, log_file):
|
||||
self.log_file = log_file
|
||||
self.line_start = True
|
||||
|
||||
def write(self, value):
|
||||
# Keep an event warning separate from an unfinished console message.
|
||||
if value.startswith('[warning][event] - ') and not self.line_start:
|
||||
self.log_file.write('\n')
|
||||
self.line_start = True
|
||||
for fragment in value.splitlines(keepends=True):
|
||||
if self.line_start:
|
||||
self.log_file.write('[{}]: '.format(utc_timestamp()))
|
||||
if not fragment.startswith('[warning][event] - '):
|
||||
self.log_file.write('[info] - ')
|
||||
self.log_file.write(fragment)
|
||||
self.line_start = fragment.endswith(('\n', '\r'))
|
||||
return len(value)
|
||||
|
||||
def flush(self):
|
||||
self.log_file.flush()
|
||||
|
||||
|
||||
class TeeOutput:
|
||||
"""Write to the original console stream and to the shared log file."""
|
||||
|
||||
def __init__(self, stream, log_file):
|
||||
self.stream = stream
|
||||
self.log_file = log_file
|
||||
|
||||
def write(self, value):
|
||||
written = self.stream.write(value)
|
||||
self.log_file.write(value)
|
||||
self.log_file.flush()
|
||||
return written
|
||||
|
||||
def flush(self):
|
||||
self.stream.flush()
|
||||
self.log_file.flush()
|
||||
|
||||
def __getattr__(self, name):
|
||||
return getattr(self.stream, name)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def log_output():
|
||||
"""Append console output to logs/<script name>.log next to the script."""
|
||||
script_path = Path(__file__).resolve()
|
||||
log_path = script_path.parent / 'logs' / script_path.with_suffix('.log').name
|
||||
log_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with log_path.open('a', encoding='utf-8') as log_file:
|
||||
log_writer = TimestampedLog(log_file)
|
||||
with redirect_stdout(TeeOutput(sys.stdout, log_writer)), \
|
||||
redirect_stderr(TeeOutput(sys.stderr, log_writer)):
|
||||
try:
|
||||
yield
|
||||
except BaseException as error:
|
||||
# The interpreter prints the traceback after the streams are restored.
|
||||
# Save a copy here while the log is still open, without duplicating it
|
||||
# in the console. SystemExit normally has no traceback.
|
||||
if not log_writer.line_start:
|
||||
log_writer.write('\n')
|
||||
if isinstance(error, SystemExit):
|
||||
if error.code is not None and not isinstance(error.code, int):
|
||||
log_writer.write(str(error.code) + '\n')
|
||||
else:
|
||||
traceback.print_exc(file=log_writer)
|
||||
raise
|
||||
finally:
|
||||
if not log_writer.line_start:
|
||||
log_writer.write('\n')
|
||||
|
||||
|
||||
def recover_event_contexts(data):
|
||||
"""Recover identifying fields even when the iCalendar parser rejects data."""
|
||||
if isinstance(data, bytes):
|
||||
data = data.decode('utf-8', errors='replace')
|
||||
unfolded = re.sub(r'\r?\n[ \t]', '', str(data))
|
||||
blocks = re.split(r'(?im)^BEGIN:VEVENT\s*\r?$', unfolded)
|
||||
contexts = []
|
||||
for block in blocks[1:] or [unfolded]:
|
||||
context = {}
|
||||
depth = 0
|
||||
for line in block.splitlines():
|
||||
name, separator, value = line.partition(':')
|
||||
name = name.split(';', 1)[0].upper()
|
||||
if name == 'BEGIN':
|
||||
depth += 1
|
||||
elif name == 'END':
|
||||
if depth == 0:
|
||||
break
|
||||
depth -= 1
|
||||
elif depth == 0 and separator and name in ('SUMMARY', 'DTSTART', 'DTEND', 'UID'):
|
||||
context.setdefault(name, value)
|
||||
contexts.append(context)
|
||||
return contexts
|
||||
|
||||
|
||||
def log_skipped_event(calendar_find, component, reason, fallback=None):
|
||||
"""Every rejected event must have an identifiable warning in the log."""
|
||||
fallback = fallback or {}
|
||||
|
||||
def field(name):
|
||||
value = component.get(name)
|
||||
if value is None:
|
||||
return str(fallback.get(name, 'indisponible'))
|
||||
decoded = getattr(value, 'dt', None)
|
||||
if isinstance(decoded, datetime):
|
||||
return local_datetime(decoded).astimezone(timezone.utc).isoformat().replace('+00:00', 'Z')
|
||||
if isinstance(decoded, date):
|
||||
return decoded.isoformat()
|
||||
return str(value)
|
||||
|
||||
calendar_name = getattr(calendar_find.get('calendar_server'), 'name', None)
|
||||
calendar_name = calendar_name or calendar_find.get('calendar_name', 'indisponible')
|
||||
# repr keeps newlines in event metadata from breaking the warning format.
|
||||
print('[warning][event] - Calendrier={!r} ; CSV={!r} ; titre={!r} ; '
|
||||
'début={!r} ; fin={!r} ; UID={!r} ; événement non exporté : {}'.format(
|
||||
str(calendar_name), calendar_find['csv_name'], field('SUMMARY'),
|
||||
field('DTSTART'), field('DTEND'), field('UID'),
|
||||
str(reason).replace('\r', '\\r').replace('\n', '\\n')), file=sys.stderr)
|
||||
|
||||
|
||||
class EventRow():
|
||||
@@ -167,7 +33,6 @@ class EventRow():
|
||||
self.categories = ''
|
||||
self.location = ''
|
||||
self.description = ''
|
||||
self.event_context = {}
|
||||
|
||||
def set_title(self, title):
|
||||
self.title = title
|
||||
@@ -245,18 +110,21 @@ def link_calendars(config, calendars_server):
|
||||
return calendars_find
|
||||
|
||||
|
||||
def local_datetime(value):
|
||||
"""Convert an iCalendar date or datetime to Europe/Paris."""
|
||||
def split_data(data):
|
||||
"""Return string splitted by ';' separator"""
|
||||
return data.split(':')
|
||||
|
||||
|
||||
def datetime_format(date_str):
|
||||
"""Return date formated"""
|
||||
try: # Nextcloud Format
|
||||
date_formated = datetime.strptime(date_str, '%Y%m%dT%H%M%SZ')
|
||||
except: # Zimbra format
|
||||
date_formated = datetime.strptime(date_str, '%Y%m%d')
|
||||
paris_tz = pytz.timezone('Europe/Paris')
|
||||
if isinstance(value, datetime):
|
||||
result = value
|
||||
elif isinstance(value, date):
|
||||
result = datetime.combine(value, time.min)
|
||||
else:
|
||||
raise ValueError("date absente ou invalide")
|
||||
if result.tzinfo is None:
|
||||
return paris_tz.localize(result)
|
||||
return result.astimezone(paris_tz)
|
||||
date_formated = date_formated.replace(tzinfo=timezone.utc).astimezone(paris_tz)
|
||||
#date_formated = date_formated.astimezone(paris_tz)
|
||||
return date_formated
|
||||
|
||||
|
||||
def get_all_events(config, calendars_find):
|
||||
@@ -273,66 +141,44 @@ def get_all_events(config, calendars_find):
|
||||
event_rows = []
|
||||
|
||||
for event in events_fetched:
|
||||
contexts = recover_event_contexts(event.data)
|
||||
# Parse parameters (TZID, VALUE=DATE), folded lines and escaped text.
|
||||
try:
|
||||
event_calendar = Calendar.from_ical(event.data)
|
||||
except (ValueError, TypeError) as error:
|
||||
for context in contexts:
|
||||
log_skipped_event(calendar_find, {},
|
||||
'document iCalendar illisible ({})'.format(error), context)
|
||||
continue
|
||||
event_fields = list(map(split_data, event.data.splitlines()))
|
||||
|
||||
components = event_calendar.walk('VEVENT')
|
||||
if not components:
|
||||
log_skipped_event(calendar_find, {}, 'aucun composant VEVENT', contexts[0])
|
||||
|
||||
for index, component in enumerate(components):
|
||||
context = contexts[index] if index < len(contexts) else {}
|
||||
# Fix problem #2023022110004936: ignore events without a summary.
|
||||
if 'SUMMARY' not in component:
|
||||
log_skipped_event(calendar_find, component, 'propriété SUMMARY absente', context)
|
||||
continue
|
||||
|
||||
try:
|
||||
start = component.decoded('DTSTART')
|
||||
end = component.decoded('DTEND', None)
|
||||
if end is None and 'DURATION' in component:
|
||||
end = start + component.decoded('DURATION')
|
||||
start = local_datetime(start)
|
||||
end = local_datetime(end)
|
||||
except (KeyError, ValueError, TypeError) as error:
|
||||
log_skipped_event(calendar_find, component,
|
||||
'dates de début/fin absentes ou invalides ({})'.format(error), context)
|
||||
# Fix problem #2023022110004936
|
||||
has = False
|
||||
for sub in event_fields:
|
||||
if 'SUMMARY' in sub:
|
||||
has = True
|
||||
break
|
||||
if not has:
|
||||
continue
|
||||
|
||||
event_row = EventRow()
|
||||
event_row.event_context = component
|
||||
event_row.set_title(str(component['SUMMARY']))
|
||||
event_row.set_start(start)
|
||||
event_row.set_end(end)
|
||||
categories = component.get('CATEGORIES', [])
|
||||
if not isinstance(categories, list):
|
||||
categories = [categories]
|
||||
event_row.set_categories(','.join(
|
||||
str(category) for group in categories for category in group.cats))
|
||||
event_row.set_location(str(component.get('LOCATION', '')))
|
||||
event_row.set_description(clean_description(str(component.get('DESCRIPTION', ''))))
|
||||
|
||||
for field in ['SUMMARY', 'DTSTART', 'DTEND', 'CATEGORIES', 'LOCATION', 'DESCRIPTION']:
|
||||
for event_field in event_fields:
|
||||
if field in event_field[0]:
|
||||
if field == "SUMMARY":
|
||||
event_row.set_title(event_field[1])
|
||||
elif field == "DTSTART":
|
||||
date = datetime_format(event_field[1])
|
||||
event_row.set_start(date)
|
||||
elif field == "DTEND":
|
||||
date = datetime_format(event_field[1])
|
||||
event_row.set_end(date)
|
||||
elif field == "CATEGORIES":
|
||||
event_row.set_categories(event_field[1])
|
||||
elif field == "LOCATION":
|
||||
event_row.set_location(event_field[1])
|
||||
elif field == "DESCRIPTION":
|
||||
event_row.set_description(event_field[1])
|
||||
event_rows.append(event_row)
|
||||
|
||||
sorted_rows = sorted(event_rows, key=lambda row: (row.start_date, row.start_time))
|
||||
sorted_rows = sorted(event_rows, key=lambda l: datetime.strptime(str(l.start_date)+'T'+str(l.start_time), "%Y-%m-%dT%H:%M:%S"))
|
||||
calendars_to_write.append({'csv_name' : calendar_find['csv_name'],
|
||||
'calendar' : calendar_find,
|
||||
'line' : sorted_rows})
|
||||
|
||||
return calendars_to_write
|
||||
|
||||
def clean_description(value):
|
||||
if value.startswith("text/html,"):
|
||||
value = value[len("text/html,"):]
|
||||
value = unquote(value)
|
||||
value = re.sub(r'<[^>]+>', '', value)
|
||||
return value.strip().strip('"')
|
||||
|
||||
def write_csv(filename, rows):
|
||||
"""Write CSV file"""
|
||||
@@ -344,7 +190,8 @@ def write_csv(filename, rows):
|
||||
writer.writerow(row.get_csv_row())
|
||||
|
||||
|
||||
def main():
|
||||
if __name__ == "__main__":
|
||||
|
||||
# Load config
|
||||
config = load_json_config("./config.json")
|
||||
|
||||
@@ -358,18 +205,5 @@ def main():
|
||||
calendars_to_write = get_all_events(config, calendars_find)
|
||||
|
||||
# Write CSV
|
||||
for index, calendar_to_write in enumerate(calendars_to_write):
|
||||
try:
|
||||
for calendar_to_write in calendars_to_write:
|
||||
write_csv(calendar_to_write['csv_name'], calendar_to_write['line'])
|
||||
except Exception as error:
|
||||
# The current CSV may be incomplete; later calendars were not written.
|
||||
for pending in calendars_to_write[index:]:
|
||||
for row in pending['line']:
|
||||
log_skipped_event(pending['calendar'], row.event_context,
|
||||
'export CSV interrompu, écriture non garantie ({})'.format(error))
|
||||
raise
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
with log_output():
|
||||
main()
|
||||
+35
-8
@@ -139,10 +139,13 @@ def get_client_uuid_by_name(client_name, user_id):
|
||||
query = ("select uuid from oc_timemanager_client where name=\""+client_name+"\" AND user_id=\""+user_id+"\"")
|
||||
cursor.execute(query)
|
||||
|
||||
client_uuid = cursor.fetchone()[0]
|
||||
row = cursor.fetchone()
|
||||
save_db()
|
||||
|
||||
return client_uuid.decode('utf-8')
|
||||
if not row:
|
||||
return None
|
||||
|
||||
return row[0].decode('utf-8')
|
||||
|
||||
|
||||
def get_project_uuid_by_name(project_name, client_uuid, user_id):
|
||||
@@ -264,7 +267,6 @@ def create_project(name, client_uuid):
|
||||
#print(json.dumps(response.json(), indent=1))
|
||||
else:
|
||||
print("Erreur de création project")
|
||||
print(response)
|
||||
print(response.reason)
|
||||
|
||||
def create_client(name):
|
||||
@@ -315,12 +317,25 @@ def create_client(name):
|
||||
|
||||
# Vérification de la réponse
|
||||
if response.status_code == 200:
|
||||
pass
|
||||
#print("Client créé avec succès.")
|
||||
#print(json.dumps(response.json(), indent=1))
|
||||
try:
|
||||
resp = response.json()
|
||||
# Tenter d'extraire l'UUID renvoyé par l'API (structure attendue)
|
||||
client_uuid = None
|
||||
if isinstance(resp, dict):
|
||||
client_created = resp.get('data', {}).get('clients', {}).get('created', [])
|
||||
if client_created and isinstance(client_created, list):
|
||||
client_uuid = client_created[0].get('uuid')
|
||||
|
||||
if client_uuid:
|
||||
return client_uuid
|
||||
# si pas d'uuid dans la réponse, retourner None (le caller gérera)
|
||||
return None
|
||||
except Exception:
|
||||
return None
|
||||
else:
|
||||
print("Erreur de création client")
|
||||
print(response.reason)
|
||||
return None
|
||||
|
||||
|
||||
def insert_task_time(name, project_name, client_name, task_uuid, user_id, dstart, hstart, dend, hend):
|
||||
@@ -423,11 +438,23 @@ for user_info in config['calendars']:
|
||||
continue
|
||||
|
||||
# Description => Client
|
||||
client_uuid = None
|
||||
if not client_exist_by_name(client_name, user_id):
|
||||
print("Création du client {}".format(client_name))
|
||||
create_client(client_name)
|
||||
|
||||
created_uuid = create_client(client_name)
|
||||
if created_uuid:
|
||||
client_uuid = created_uuid
|
||||
else:
|
||||
# tentative de récupération depuis la base au cas où l'API a créé l'entrée
|
||||
client_uuid = get_client_uuid_by_name(client_name, user_id)
|
||||
if not client_uuid:
|
||||
print("Erreur: impossible d'obtenir l'UUID du client '{}' après création, saut.".format(client_name))
|
||||
continue
|
||||
else:
|
||||
client_uuid = get_client_uuid_by_name(client_name, user_id)
|
||||
if not client_uuid:
|
||||
print("Erreur: le client '{}' existe mais son UUID est introuvable, saut.".format(client_name))
|
||||
continue
|
||||
|
||||
# Categories => Project
|
||||
if not project_exist_by_name(project_name, client_uuid, user_id):
|
||||
|
||||
@@ -1,235 +0,0 @@
|
||||
# Export CalDAV vers CSV
|
||||
|
||||
Le script [`crealead-caldav.py`](../crealead-caldav.py) récupère des événements
|
||||
d'agendas accessibles en CalDAV, notamment Nextcloud, et produit un fichier CSV
|
||||
par calendrier configuré. Ces fichiers peuvent ensuite être utilisés par
|
||||
`csv-to-timemanager.py`.
|
||||
|
||||
## Pré-requis
|
||||
|
||||
- Python 3, avec `pip` et le module `venv`. L'environnement Docker du dépôt
|
||||
utilise Python 3.9.2.
|
||||
- Les dépendances de [`requirements.txt`](../requirements.txt), notamment
|
||||
`caldav`, `icalendar` et `pytz`.
|
||||
- Un serveur CalDAV accessible depuis la machine d'exécution et un compte
|
||||
autorisé à lire les calendriers à exporter.
|
||||
- Des droits de lecture sur `config.json` et d'écriture sur les fichiers CSV
|
||||
ainsi que sur le dossier `logs/`, situé à côté du script.
|
||||
- Bash pour utiliser le lanceur `exec-export-caldav.sh`.
|
||||
|
||||
Depuis la racine du dépôt, préparer l'environnement :
|
||||
|
||||
```bash
|
||||
python3 -m venv env
|
||||
source env/bin/activate
|
||||
python -m pip install -r requirements.txt
|
||||
```
|
||||
|
||||
L'environnement Docker fourni constitue une autre façon d'exécuter le script
|
||||
et les tests. Sa construction est décrite dans
|
||||
[`support/docker-dev/README.md`](../support/docker-dev/README.md).
|
||||
|
||||
## Usage
|
||||
|
||||
### Configuration minimale
|
||||
|
||||
Créer un fichier `config.json` à la racine du dépôt. Exemple pour exporter les
|
||||
événements d'octobre 2026 :
|
||||
|
||||
```json
|
||||
{
|
||||
"caldav_url": "https://cloud.example.com/remote.php/dav/",
|
||||
"username": "utilisateur",
|
||||
"password": "mot-de-passe-application",
|
||||
"year_start": 2026,
|
||||
"month_start": 10,
|
||||
"day_start": 1,
|
||||
"year_end": 2026,
|
||||
"month_end": 11,
|
||||
"day_end": 1,
|
||||
"calendars": [
|
||||
{
|
||||
"calendar_name": "Temps de travail",
|
||||
"csv_name": "csv/utilisateur.csv"
|
||||
}
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
| Champ | Rôle |
|
||||
| --- | --- |
|
||||
| `caldav_url` | URL du serveur CalDAV. |
|
||||
| `username`, `password` | Identifiants du compte qui peut lire les calendriers. |
|
||||
| `year_start`, `month_start`, `day_start` | Date de début de la recherche, à minuit. Les valeurs sont des nombres entiers. |
|
||||
| `year_end`, `month_end`, `day_end` | Date de fin de la recherche, à minuit. Pour couvrir un mois, utiliser le premier jour du mois suivant. |
|
||||
| `calendars` | Liste des calendriers à exporter. |
|
||||
| `calendars[].calendar_name` | Texte contenu dans le nom du calendrier distant, avec respect de la casse. Il doit correspondre à un seul calendrier. |
|
||||
| `calendars[].csv_name` | Chemin du fichier CSV à écrire. Un chemin relatif est résolu depuis le répertoire de lancement. |
|
||||
|
||||
Ajouter une entrée dans `calendars` pour chaque calendrier à exporter.
|
||||
Les champs `db_*`, `user_name` et `api_password`, utilisés par d'autres scripts
|
||||
du dépôt, ne sont pas nécessaires à cet export.
|
||||
|
||||
`config.json` est ignoré par Git. Remplacer les valeurs d'exemple par les
|
||||
informations du serveur et du compte utilisés.
|
||||
|
||||
### Lancement
|
||||
|
||||
Lancer les commandes depuis la racine du dépôt : le script lit
|
||||
`./config.json`, et le lanceur recherche `env/bin/activate` dans le répertoire
|
||||
courant. Créer au préalable les dossiers de destination des CSV ; le script
|
||||
ne les crée pas.
|
||||
|
||||
```bash
|
||||
mkdir -p csv
|
||||
bash ./exec-export-caldav.sh
|
||||
```
|
||||
|
||||
Le lanceur active l'environnement `env/`, puis exécute le script. Il est aussi
|
||||
possible de lancer directement Python :
|
||||
|
||||
```bash
|
||||
env/bin/python crealead-caldav.py
|
||||
```
|
||||
|
||||
Le script ne propose pas d'arguments en ligne de commande. La période et les
|
||||
calendriers sont définis dans `config.json`.
|
||||
|
||||
### Fichiers produits et traitement des événements
|
||||
|
||||
Chaque CSV existant est remplacé lors de l'export. Les lignes sont triées par
|
||||
date et heure de début. Le séparateur est `;` et les colonnes sont :
|
||||
|
||||
```text
|
||||
Titre;Date debut;Heure debut;Date fin;Heure fin;Categories;Emplacement;Description
|
||||
```
|
||||
|
||||
Les événements récurrents sont demandés au serveur avec `expand=True`.
|
||||
Le lecteur iCalendar prend en charge les paramètres des propriétés, comme
|
||||
`TZID` et `VALUE=DATE`, ainsi que les textes répartis sur plusieurs lignes.
|
||||
Les heures sont converties vers `Europe/Paris`. Une heure sans fuseau est
|
||||
interprétée dans ce fuseau ; une date seule est exportée à minuit.
|
||||
Si `DTEND` est absent, `DURATION` peut fournir la date de fin.
|
||||
|
||||
Les événements sans propriété `SUMMARY` sont ignorés. Ceux dont les dates
|
||||
de début ou de fin sont absentes ou invalides sont également ignorés. Chaque
|
||||
exclusion laisse un avertissement détaillé dans le journal, y compris en
|
||||
l'absence de `SUMMARY`. Un document iCalendar illisible ou une ressource sans
|
||||
composant `VEVENT` est signalé puis ignoré. Les autres événements restent
|
||||
exportés. Si un calendrier configuré est introuvable ou correspond à plusieurs
|
||||
calendriers distants, le script affiche un message et s'arrête avant l'écriture
|
||||
des CSV.
|
||||
|
||||
En cas d'erreur d'écriture d'un CSV, le script s'arrête avec une exception.
|
||||
Il laisse un avertissement pour chaque événement du calendrier en cours,
|
||||
dont le fichier peut être incomplet, et des calendriers restant à écrire.
|
||||
Les fichiers déjà écrits avec succès sont conservés.
|
||||
|
||||
### Journalisation
|
||||
|
||||
Les sorties standard et d'erreur sont affichées dans le terminal et copiées
|
||||
dans `logs/crealead-caldav.log`, à côté du script. Le dossier est créé
|
||||
automatiquement. Le journal utilise UTF-8, conserve les exécutions précédentes
|
||||
et enregistre les traces d'exception en cas de plantage pendant le traitement.
|
||||
Il n'y a pas de rotation automatique du journal. `logs/` est ignoré par Git.
|
||||
|
||||
Chaque nouvelle ligne du journal est préfixée par l'heure d'écriture en UTC,
|
||||
au format ISO 8601 `YYYY-MM-DDTHH:MM:SSZ`, puis par `[info]` :
|
||||
|
||||
```text
|
||||
[2026-10-15T12:34:56Z]: [info] - csv/utilisateur.csv ...
|
||||
```
|
||||
|
||||
Cela s'applique aussi aux messages écrits en plusieurs fragments et à chaque
|
||||
ligne d'une trace d'exception. L'horodatage est ajouté dans le fichier de
|
||||
journal, sans modifier les messages ordinaires du terminal.
|
||||
|
||||
Tout événement non exporté est signalé avec `[warning][event]`, à la place de
|
||||
`[info]`. Le message indique le calendrier distant (ou son nom configuré),
|
||||
le fichier CSV, le titre, le début, la fin, l'UID et la raison de l'exclusion :
|
||||
|
||||
```text
|
||||
[2026-10-15T12:34:56Z]: [warning][event] - Calendrier='Temps de travail' ; CSV='csv/utilisateur.csv' ; titre='Réunion' ; début='2026-10-15T08:00:00Z' ; fin='indisponible' ; UID='reunion-1' ; événement non exporté : dates de début/fin absentes ou invalides (date absente ou invalide)
|
||||
```
|
||||
|
||||
L'heure entre crochets est celle du message ; les champs `début` et `fin`
|
||||
décrivent le moment de l'événement. Les dates et heures décodées sont affichées
|
||||
en UTC, les dates de journée entière restent des dates seules. En cas de
|
||||
document illisible ou de date invalide, le script récupère les champs bruts
|
||||
disponibles. Une métadonnée absente ou irrécupérable vaut `indisponible`.
|
||||
Les retours à la ligne dans les métadonnées et le motif sont échappés pour
|
||||
garder un avertissement sur une seule ligne. Ces avertissements sont aussi
|
||||
visibles dans le terminal.
|
||||
|
||||
## Tests
|
||||
|
||||
Les tests se trouvent dans
|
||||
[`tests/test_crealead_caldav.py`](../tests/test_crealead_caldav.py) et utilisent
|
||||
`unittest`, fourni avec Python. Aucune dépendance de test supplémentaire
|
||||
n'est nécessaire.
|
||||
|
||||
### Cas couverts
|
||||
|
||||
Les 13 tests de `ExportEventsTests` vérifient :
|
||||
|
||||
- La lecture de `DTSTART` et `DTEND` avec `TZID`, leur conversion vers Paris
|
||||
et le tri chronologique des lignes.
|
||||
- La conversion des dates UTC en été et en hiver, ainsi que celle d'un autre
|
||||
fuseau horaire.
|
||||
- Les événements sur une journée entière (`VALUE=DATE`), les heures sans
|
||||
fuseau et le calcul de la fin à partir de `DURATION`.
|
||||
- L'exclusion des événements sans propriété `SUMMARY`, des dates manquantes
|
||||
ou invalides et des documents iCalendar illisibles, sans bloquer les
|
||||
événements valides, ainsi que la présence d'un avertissement contextualisé
|
||||
dans le journal pour chaque événement écarté.
|
||||
- La récupération des métadonnées de chaque événement dans un document
|
||||
illisible, la conservation d'une date brute invalide et le signalement
|
||||
d'une ressource sans composant `VEVENT`.
|
||||
- La conservation des deux-points dans les textes, des paramètres de texte,
|
||||
des catégories et des lignes repliées, ainsi que le nettoyage de la
|
||||
description HTML encodée et le contenu du CSV écrit.
|
||||
|
||||
Les 9 tests de `LogOutputTests` vérifient :
|
||||
|
||||
- La création de `logs/` à côté du script, la copie des deux sorties et
|
||||
l'écriture immédiate des messages, y compris les accents.
|
||||
- L'ajout au journal existant et le choix du nom du journal d'après celui du
|
||||
script, y compris lorsqu'une exécution se termine sans retour à la ligne.
|
||||
- L'horodatage en UTC, le préfixe `[info]` de chaque ligne, les écritures en
|
||||
plusieurs fragments, les messages sur plusieurs lignes et les lignes vides.
|
||||
- Le préfixe `[warning][event]` sans `[info]`, les métadonnées de l'événement,
|
||||
l'échappement des retours à la ligne et la séparation d'un avertissement
|
||||
après un message ordinaire inachevé.
|
||||
- La sauvegarde des traces d'exception et la restauration des sorties après
|
||||
une erreur.
|
||||
- La conservation du code de sortie d'un `SystemExit`, sans ajouter de trace
|
||||
d'exception, et la sauvegarde de son éventuel message.
|
||||
|
||||
Le test de `ExportFailureTests` provoque une erreur d'écriture CSV. Il vérifie
|
||||
qu'un avertissement est enregistré pour les événements du calendrier en
|
||||
échec et des calendriers restant à écrire, et que le premier CSV déjà écrit
|
||||
est conservé.
|
||||
|
||||
Ces tests utilisent le vrai lecteur `icalendar`, des événements synthétiques
|
||||
et des dossiers temporaires pour les CSV et les journaux. Le module CalDAV
|
||||
et les réponses du serveur sont simulés : aucun accès réseau, compte,
|
||||
`config.json` ou calendrier réel n'est nécessaire. Ils vérifient que la
|
||||
recherche demande `expand=True`, mais ne valident pas l'expansion des
|
||||
récurrences par un serveur réel ni l'import dans TimeManager.
|
||||
|
||||
### Exécution
|
||||
|
||||
Depuis la racine du dépôt, avec les dépendances installées dans `env/` :
|
||||
|
||||
```bash
|
||||
env/bin/python -B -m unittest discover -s tests -v
|
||||
```
|
||||
|
||||
L'option `-B` évite de créer des fichiers de cache Python. La suite contient
|
||||
actuellement 23 tests ; une exécution réussie se termine par `OK`.
|
||||
|
||||
Avec l'image Docker du dépôt déjà construite :
|
||||
|
||||
```bash
|
||||
docker run --rm -v "$(pwd)":/app -w /app cal-to-time-manager:dev -B -m unittest discover -s tests -v
|
||||
```
|
||||
@@ -23,8 +23,6 @@ Chaque entrée doit contenir au minimum :
|
||||
- `user_name` : login Nextcloud utilisé pour authentifier les requêtes,
|
||||
- `api_password` : mot de passe ou mot de passe d'application Nextcloud.
|
||||
|
||||
> Il est important de générer la clé API en s'étant connecté avec l'identifiant et non l'email car le script user_name utilise le champ "user_name" pour faire des vérifications en bases de données avant d'ajouter des données.
|
||||
|
||||
Colonnes CSV utilisées
|
||||
----------------------
|
||||
|
||||
|
||||
@@ -1,4 +0,0 @@
|
||||
#!/bin/bash
|
||||
|
||||
source env/bin/activate
|
||||
python3 remove-users-data-timemanager.py $1
|
||||
@@ -1,116 +0,0 @@
|
||||
import csv
|
||||
import requests
|
||||
import json
|
||||
import uuid
|
||||
import re
|
||||
import base64
|
||||
import sys
|
||||
import mysql.connector
|
||||
from unidecode import unidecode
|
||||
|
||||
##########################################
|
||||
# Config
|
||||
##########################################
|
||||
# Récupération de la config
|
||||
with open('config.json', 'r') as f:
|
||||
config = json.load(f)
|
||||
|
||||
mydb = mysql.connector.connect(
|
||||
host=config['db_host'],
|
||||
user=config['db_user'],
|
||||
password=config['db_password'],
|
||||
database=config['db_database']
|
||||
)
|
||||
cursor = mydb.cursor(buffered=True)
|
||||
|
||||
# Argument --yes pour auto-valider toutes les suppressions
|
||||
valid_args = {'-y', '--yes'}
|
||||
extra_args = [arg for arg in sys.argv[1:] if arg not in valid_args]
|
||||
|
||||
if extra_args:
|
||||
print(f"Erreur : argument(s) non reconnu(s) : {' '.join(extra_args)}")
|
||||
print(f"Usage : python {sys.argv[0]} [--yes | -y]")
|
||||
sys.exit(1)
|
||||
|
||||
auto_confirm = bool(set(sys.argv[1:]) & valid_args)
|
||||
|
||||
##########################################
|
||||
# Fonctions
|
||||
##########################################
|
||||
|
||||
def check_users_exist(user_id):
|
||||
"""Vérifie si l'utilisateur existe dans oc_users."""
|
||||
cursor.execute("SELECT uid FROM oc_users WHERE uid = %s", (user_id,))
|
||||
return cursor.fetchone() is not None
|
||||
|
||||
def remove_row_on_table_for_user(user_id, table):
|
||||
"""Supprime toutes les entrées d'une table pour un utilisateur donné."""
|
||||
query = f"DELETE FROM `{table}` WHERE user_id = %s"
|
||||
cursor.execute(query, (user_id,))
|
||||
affected = cursor.rowcount
|
||||
mydb.commit()
|
||||
print(f" [{table}] {affected} ligne(s) supprimée(s)")
|
||||
|
||||
##########################################
|
||||
# Main
|
||||
##########################################
|
||||
|
||||
tables = [
|
||||
"oc_timemanager_client",
|
||||
"oc_timemanager_commit",
|
||||
"oc_timemanager_project",
|
||||
"oc_timemanager_task",
|
||||
"oc_timemanager_time"
|
||||
]
|
||||
|
||||
# Listing de tous les utilisateurs configurés avant toute action
|
||||
print("==========================================")
|
||||
print("Utilisateurs configurés dans config.json :")
|
||||
print("==========================================")
|
||||
valid_users = []
|
||||
for user_info in config['calendars']:
|
||||
user_id = user_info.get('user_name', '')
|
||||
if not user_id:
|
||||
print(f" ⚠ [{user_id or '???'}] — infos de connexion incomplètes, ignoré")
|
||||
continue
|
||||
exists = check_users_exist(user_id)
|
||||
status = "✔ existe en BDD" if exists else "✘ introuvable en BDD"
|
||||
print(f" • {user_id} — {status}")
|
||||
if exists:
|
||||
valid_users.append(user_info)
|
||||
|
||||
print()
|
||||
|
||||
if not valid_users:
|
||||
print("Aucun utilisateur valide à traiter. Fin du script.")
|
||||
cursor.close()
|
||||
mydb.close()
|
||||
sys.exit(0)
|
||||
|
||||
if auto_confirm:
|
||||
print("Mode auto-confirmation activé (--yes) : toutes les suppressions seront effectuées sans demande.\n")
|
||||
|
||||
# Traitement par utilisateur
|
||||
for user_info in valid_users:
|
||||
user_id = user_info['user_name']
|
||||
print("==========================================")
|
||||
print(f"User : {user_id}")
|
||||
print("==========================================")
|
||||
|
||||
if auto_confirm:
|
||||
confirmed = True
|
||||
else:
|
||||
reponse = input(f"\nSupprimer toutes les données TimeManager de '{user_id}' ? [o/N] : ").strip().lower()
|
||||
confirmed = reponse in ('o', 'oui', 'y', 'yes')
|
||||
|
||||
if confirmed:
|
||||
print("Suppression en cours...")
|
||||
for table in tables:
|
||||
remove_row_on_table_for_user(user_id, table)
|
||||
print(f"✔ Données de '{user_id}' supprimées.\n")
|
||||
else:
|
||||
print(f"✘ Suppression annulée pour '{user_id}'.\n")
|
||||
|
||||
cursor.close()
|
||||
mydb.close()
|
||||
print("Terminé.")
|
||||
@@ -1,116 +0,0 @@
|
||||
import csv
|
||||
import requests
|
||||
import json
|
||||
import uuid
|
||||
import re
|
||||
import base64
|
||||
import sys
|
||||
import mysql.connector
|
||||
from unidecode import unidecode
|
||||
|
||||
##########################################
|
||||
# Config
|
||||
##########################################
|
||||
# Récupération de la config
|
||||
with open('config.json', 'r') as f:
|
||||
config = json.load(f)
|
||||
|
||||
mydb = mysql.connector.connect(
|
||||
host=config['db_host'],
|
||||
user=config['db_user'],
|
||||
password=config['db_password'],
|
||||
database=config['db_database']
|
||||
)
|
||||
cursor = mydb.cursor(buffered=True)
|
||||
|
||||
# Argument --yes pour auto-valider toutes les suppressions
|
||||
valid_args = {'-y', '--yes'}
|
||||
extra_args = [arg for arg in sys.argv[1:] if arg not in valid_args]
|
||||
|
||||
if extra_args:
|
||||
print(f"Erreur : argument(s) non reconnu(s) : {' '.join(extra_args)}")
|
||||
print(f"Usage : python {sys.argv[0]} [--yes | -y]")
|
||||
sys.exit(1)
|
||||
|
||||
auto_confirm = bool(set(sys.argv[1:]) & valid_args)
|
||||
|
||||
##########################################
|
||||
# Fonctions
|
||||
##########################################
|
||||
|
||||
def check_users_exist(user_id):
|
||||
"""Vérifie si l'utilisateur existe dans oc_users."""
|
||||
cursor.execute("SELECT uid FROM oc_users WHERE uid = %s", (user_id,))
|
||||
return cursor.fetchone() is not None
|
||||
|
||||
def remove_row_on_table_for_user(user_id, table):
|
||||
"""Supprime toutes les entrées d'une table pour un utilisateur donné."""
|
||||
query = f"DELETE FROM `{table}` WHERE user_id = %s"
|
||||
cursor.execute(query, (user_id,))
|
||||
affected = cursor.rowcount
|
||||
mydb.commit()
|
||||
print(f" [{table}] {affected} ligne(s) supprimée(s)")
|
||||
|
||||
##########################################
|
||||
# Main
|
||||
##########################################
|
||||
|
||||
tables = [
|
||||
"oc_timemanager_client",
|
||||
"oc_timemanager_commit",
|
||||
"oc_timemanager_project",
|
||||
"oc_timemanager_task",
|
||||
"oc_timemanager_time"
|
||||
]
|
||||
|
||||
# Listing de tous les utilisateurs configurés avant toute action
|
||||
print("==========================================")
|
||||
print("Utilisateurs configurés dans config.json :")
|
||||
print("==========================================")
|
||||
valid_users = []
|
||||
for user_info in config['calendars']:
|
||||
user_id = user_info.get('user_name', '')
|
||||
if not user_id:
|
||||
print(f" ⚠ [{user_id or '???'}] — infos de connexion incomplètes, ignoré")
|
||||
continue
|
||||
exists = check_users_exist(user_id)
|
||||
status = "✔ existe en BDD" if exists else "✘ introuvable en BDD"
|
||||
print(f" • {user_id} — {status}")
|
||||
if exists:
|
||||
valid_users.append(user_info)
|
||||
|
||||
print()
|
||||
|
||||
if not valid_users:
|
||||
print("Aucun utilisateur valide à traiter. Fin du script.")
|
||||
cursor.close()
|
||||
mydb.close()
|
||||
sys.exit(0)
|
||||
|
||||
if auto_confirm:
|
||||
print("Mode auto-confirmation activé (--yes) : toutes les suppressions seront effectuées sans demande.\n")
|
||||
|
||||
# Traitement par utilisateur
|
||||
for user_info in valid_users:
|
||||
user_id = user_info['user_name']
|
||||
print("==========================================")
|
||||
print(f"User : {user_id}")
|
||||
print("==========================================")
|
||||
|
||||
if auto_confirm:
|
||||
confirmed = True
|
||||
else:
|
||||
reponse = input(f"\nSupprimer toutes les données TimeManager de '{user_id}' ? [o/N] : ").strip().lower()
|
||||
confirmed = reponse in ('o', 'oui', 'y', 'yes')
|
||||
|
||||
if confirmed:
|
||||
print("Suppression en cours...")
|
||||
for table in tables:
|
||||
remove_row_on_table_for_user(user_id, table)
|
||||
print(f"✔ Données de '{user_id}' supprimées.\n")
|
||||
else:
|
||||
print(f"✘ Suppression annulée pour '{user_id}'.\n")
|
||||
|
||||
cursor.close()
|
||||
mydb.close()
|
||||
print("Terminé.")
|
||||
@@ -1,3 +0,0 @@
|
||||
#!/bin/bash
|
||||
|
||||
ssh execcsv@crl-01.ovea.com
|
||||
@@ -1,364 +0,0 @@
|
||||
import contextlib
|
||||
import csv
|
||||
import importlib.util
|
||||
import io
|
||||
import re
|
||||
import sys
|
||||
from datetime import date, datetime, time, timezone
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
from types import ModuleType, SimpleNamespace
|
||||
import unittest
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
|
||||
# The regression tests use real iCalendar parsing, with no CalDAV connection.
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
'crealead_caldav', Path(__file__).resolve().parents[1] / 'crealead-caldav.py')
|
||||
exporter = importlib.util.module_from_spec(spec)
|
||||
with patch.dict('sys.modules', {'caldav': ModuleType('caldav')}):
|
||||
spec.loader.exec_module(exporter)
|
||||
|
||||
|
||||
def event_data(properties):
|
||||
return SimpleNamespace(data=(
|
||||
'BEGIN:VCALENDAR\r\nVERSION:2.0\r\nBEGIN:VEVENT\r\n'
|
||||
+ properties.replace('\n', '\r\n')
|
||||
+ '\r\nEND:VEVENT\r\nEND:VCALENDAR\r\n'))
|
||||
|
||||
|
||||
class ExportEventsTests(unittest.TestCase):
|
||||
def export(self, *events):
|
||||
server = SimpleNamespace(name='Agenda de test', date_search=Mock(return_value=events))
|
||||
config = dict(year_start=2026, month_start=10, day_start=1,
|
||||
year_end=2026, month_end=11, day_end=1)
|
||||
errors = io.StringIO()
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
|
||||
contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(errors):
|
||||
with exporter.log_output():
|
||||
result = exporter.get_all_events(config, [
|
||||
{'csv_name': 'test.csv', 'calendar_server': server}])
|
||||
self.log = (Path(directory) / 'logs/export.log').read_text(encoding='utf-8')
|
||||
self.assertTrue(server.date_search.call_args.kwargs['expand'])
|
||||
return result[0]['line'], errors.getvalue()
|
||||
|
||||
def test_timezone_parameters_and_chronological_sort(self):
|
||||
rows, errors = self.export(
|
||||
event_data('SUMMARY:Later\nDTSTART;TZID=Europe/Paris:20261015T140000\n'
|
||||
'DTEND;TZID=Europe/Paris:20261015T150000'),
|
||||
event_data('SUMMARY:Earlier\nDTSTART:20261015T080000Z\n'
|
||||
'DTEND:20261015T090000Z'))
|
||||
self.assertEqual(errors, '')
|
||||
self.assertEqual([row.title for row in rows], ['Earlier', 'Later'])
|
||||
self.assertEqual([row.start_time for row in rows], [time(10), time(14)])
|
||||
self.assertEqual([row.end_time for row in rows], [time(11), time(15)])
|
||||
|
||||
def test_utc_conversion_in_winter(self):
|
||||
rows, errors = self.export(event_data(
|
||||
'SUMMARY:Winter\nDTSTART:20260115T080000Z\nDTEND:20260115T090000Z'))
|
||||
self.assertEqual(errors, '')
|
||||
self.assertEqual(rows[0].start_time, time(9))
|
||||
self.assertEqual(rows[0].end_time, time(10))
|
||||
|
||||
def test_other_timezone_is_converted_to_paris(self):
|
||||
rows, errors = self.export(event_data(
|
||||
'SUMMARY:Call\nDTSTART;TZID=America/New_York:20261015T090000\n'
|
||||
'DTEND;TZID=America/New_York:20261015T100000'))
|
||||
self.assertEqual(errors, '')
|
||||
self.assertEqual(rows[0].start_time, time(15))
|
||||
self.assertEqual(rows[0].end_time, time(16))
|
||||
|
||||
def test_date_only_values_keep_their_dates_at_midnight(self):
|
||||
rows, errors = self.export(event_data(
|
||||
'SUMMARY:Day\nDTSTART;VALUE=DATE:20261015\nDTEND;VALUE=DATE:20261016'))
|
||||
self.assertEqual(errors, '')
|
||||
self.assertEqual(rows[0].start_date, date(2026, 10, 15))
|
||||
self.assertEqual(rows[0].end_date, date(2026, 10, 16))
|
||||
self.assertEqual(rows[0].start_time, time(0))
|
||||
self.assertEqual(rows[0].end_time, time(0))
|
||||
|
||||
def test_floating_times_are_local(self):
|
||||
rows, errors = self.export(event_data(
|
||||
'SUMMARY:Local\nDTSTART:20261015T090000\nDTEND:20261015T100000'))
|
||||
self.assertEqual(errors, '')
|
||||
self.assertEqual(rows[0].start_time, time(9))
|
||||
self.assertEqual(rows[0].end_time, time(10))
|
||||
|
||||
def test_duration_can_supply_end_time(self):
|
||||
rows, errors = self.export(event_data(
|
||||
'SUMMARY:Duration\nDTSTART:20261015T080000Z\nDURATION:PT1H30M'))
|
||||
self.assertEqual(errors, '')
|
||||
self.assertEqual(rows[0].end_time, time(11, 30))
|
||||
|
||||
def test_missing_or_invalid_dates_are_logged_and_skipped(self):
|
||||
for dates in (
|
||||
'DTEND:20261015T090000Z',
|
||||
'DTSTART:20261015T080000Z',
|
||||
'DTSTART:invalid\nDTEND:20261015T090000Z',
|
||||
'DTSTART:20261015T080000Z\nDTEND:invalid',
|
||||
):
|
||||
with self.subTest(dates=dates):
|
||||
rows, errors = self.export(
|
||||
event_data('UID:bad-event\nSUMMARY:Invalid\n' + dates),
|
||||
event_data('SUMMARY:Valid\nDTSTART:20261015T080000Z\n'
|
||||
'DTEND:20261015T090000Z'))
|
||||
self.assertEqual([row.title for row in rows], ['Valid'])
|
||||
self.assertIn('bad-event', errors)
|
||||
self.assertIn('test.csv', errors)
|
||||
self.assertIn('événement non exporté', errors)
|
||||
self.assertIn("Calendrier='Agenda de test'", self.log)
|
||||
self.assertIn("titre='Invalid'", self.log)
|
||||
self.assertIn('début=', self.log)
|
||||
self.assertIn('fin=', self.log)
|
||||
self.assertIn('dates de début/fin absentes ou invalides', self.log)
|
||||
self.assertEqual(self.log.count('[warning][event] - '), 1)
|
||||
|
||||
def test_event_without_summary_is_still_skipped(self):
|
||||
rows, errors = self.export(event_data(
|
||||
'DTSTART:20261015T080000Z\nDTEND:20261015T090000Z'))
|
||||
self.assertEqual(rows, [])
|
||||
self.assertIn('propriété SUMMARY absente', errors)
|
||||
self.assertIn("titre='indisponible'", self.log)
|
||||
self.assertIn("début='2026-10-15T08:00:00Z'", self.log)
|
||||
self.assertIn("Calendrier='Agenda de test'", self.log)
|
||||
self.assertEqual(self.log.count('[warning][event] - '), 1)
|
||||
|
||||
def test_malformed_calendar_does_not_block_other_events(self):
|
||||
rows, errors = self.export(
|
||||
SimpleNamespace(data='not an iCalendar document'),
|
||||
event_data('SUMMARY:Valid\nDTSTART:20261015T080000Z\n'
|
||||
'DTEND:20261015T090000Z'))
|
||||
self.assertEqual([row.title for row in rows], ['Valid'])
|
||||
self.assertIn('document iCalendar illisible', errors)
|
||||
self.assertIn("titre='indisponible'", self.log)
|
||||
self.assertIn("début='indisponible'", self.log)
|
||||
|
||||
def test_malformed_document_recovers_each_event_context(self):
|
||||
rows, errors = self.export(SimpleNamespace(data=(
|
||||
'BEGIN:VEVENT\nUID:first\nSUMMARY;LANGUAGE=fr:Réunion:\n suite\n'
|
||||
'DTSTART;TZID=Europe/Paris:20261015T090000\nEND:VEVENT\n'
|
||||
'BEGIN:VEVENT\nUID:second\nSUMMARY:Autre\n'
|
||||
'DTSTART:20261016T080000Z\nEND:VEVENT\n')))
|
||||
# Multiple root components make Calendar.from_ical reject the document.
|
||||
self.assertEqual(rows, [])
|
||||
self.assertEqual(self.log.count('[warning][event] - '), 2)
|
||||
self.assertIn("titre='Réunion:suite'", self.log)
|
||||
self.assertIn("début='20261015T090000'", self.log)
|
||||
self.assertIn("titre='Autre'", self.log)
|
||||
self.assertIn("UID='second'", self.log)
|
||||
self.assertIn('document iCalendar illisible', errors)
|
||||
|
||||
def test_resource_without_event_is_logged(self):
|
||||
rows, errors = self.export(SimpleNamespace(data=(
|
||||
'BEGIN:VCALENDAR\r\nVERSION:2.0\r\nEND:VCALENDAR\r\n')))
|
||||
self.assertEqual(rows, [])
|
||||
self.assertIn('aucun composant VEVENT', errors)
|
||||
self.assertIn("Calendrier='Agenda de test'", self.log)
|
||||
self.assertEqual(self.log.count('[warning][event] - '), 1)
|
||||
|
||||
def test_invalid_date_keeps_raw_value_in_warning(self):
|
||||
rows, errors = self.export(event_data(
|
||||
'SUMMARY:Invalid\nDTSTART:invalid\nDTEND:20261015T090000Z'))
|
||||
self.assertEqual(rows, [])
|
||||
self.assertIn("début='invalid'", self.log)
|
||||
self.assertIn('dates de début/fin absentes ou invalides', errors)
|
||||
|
||||
def test_text_parameters_folding_colons_and_csv_output(self):
|
||||
rows, errors = self.export(event_data(
|
||||
'SUMMARY;LANGUAGE=fr:Projet: réunion\n'
|
||||
'DTSTART:20261015T080000Z\nDTEND:20261015T090000Z\n'
|
||||
'CATEGORIES:Projet A,Projet B\nLOCATION:Bureau: Montpellier\n'
|
||||
'DESCRIPTION:text/html,%3Cb%3EClient%3C/b%3E:\n suite'))
|
||||
self.assertEqual(errors, '')
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
filename = Path(directory) / 'export.csv'
|
||||
exporter.write_csv(filename, rows)
|
||||
with filename.open(newline='') as csv_file:
|
||||
exported = list(csv.reader(csv_file, delimiter=';'))
|
||||
self.assertEqual(len(exported), 2)
|
||||
self.assertEqual(exported[1], [
|
||||
'Projet: réunion', '2026-10-15', '10:00:00', '2026-10-15',
|
||||
'11:00:00', 'Projet A,Projet B', 'Bureau: Montpellier', 'Client:suite'])
|
||||
|
||||
|
||||
class LogOutputTests(unittest.TestCase):
|
||||
def messages(self, log):
|
||||
messages = []
|
||||
for line in log.splitlines():
|
||||
match = re.fullmatch(r'\[\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z\]: (.*)', line)
|
||||
self.assertIsNotNone(match, line)
|
||||
messages.append(match.group(1))
|
||||
return messages
|
||||
|
||||
def test_timestamp_uses_utc(self):
|
||||
with patch.object(exporter, 'datetime') as clock:
|
||||
clock.now.return_value = datetime(2026, 10, 15, 12, 34, 56, tzinfo=timezone.utc)
|
||||
self.assertEqual(exporter.utc_timestamp(), '2026-10-15T12:34:56Z')
|
||||
clock.now.assert_called_once_with(timezone.utc)
|
||||
|
||||
def test_fragmented_and_multiline_writes_have_one_prefix_per_line(self):
|
||||
console = io.StringIO()
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
|
||||
contextlib.redirect_stdout(console):
|
||||
with exporter.log_output():
|
||||
print('CSV', '...', 2)
|
||||
sys.stdout.write('Début')
|
||||
sys.stdout.write(' de ligne\n\nAutre ligne\n')
|
||||
messages = self.messages((Path(directory) / 'logs/export.log').read_text(encoding='utf-8'))
|
||||
self.assertEqual(messages, [
|
||||
'[info] - CSV ... 2', '[info] - Début de ligne', '[info] - ', '[info] - Autre ligne'])
|
||||
self.assertEqual(console.getvalue(), 'CSV ... 2\nDébut de ligne\n\nAutre ligne\n')
|
||||
|
||||
def test_event_warning_has_own_level_context_and_escaped_newlines(self):
|
||||
console, errors = io.StringIO(), io.StringIO()
|
||||
component = exporter.Calendar.from_ical(event_data(
|
||||
'UID:event-1\nSUMMARY:Réunion\\nSuite\n'
|
||||
'DTSTART;TZID=Europe/Paris:20261015T100000\n'
|
||||
'DTEND;TZID=Europe/Paris:20261015T110000').data).walk('VEVENT')[0]
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
|
||||
contextlib.redirect_stdout(console), contextlib.redirect_stderr(errors):
|
||||
with exporter.log_output():
|
||||
sys.stdout.write('En cours')
|
||||
exporter.log_skipped_event(
|
||||
{'calendar_name': 'Calendrier configuré', 'csv_name': 'test.csv'},
|
||||
component, 'motif\nsur deux lignes')
|
||||
print('Suite')
|
||||
messages = self.messages((Path(directory) / 'logs/export.log').read_text(encoding='utf-8'))
|
||||
self.assertEqual(len(messages), 3)
|
||||
self.assertEqual(messages[0], '[info] - En cours')
|
||||
self.assertTrue(messages[1].startswith('[warning][event] - '))
|
||||
self.assertNotIn('[info]', messages[1])
|
||||
self.assertIn("Calendrier='Calendrier configuré'", messages[1])
|
||||
self.assertIn("titre='Réunion\\nSuite'", messages[1])
|
||||
self.assertIn("début='2026-10-15T08:00:00Z'", messages[1])
|
||||
self.assertIn('motif\\nsur deux lignes', messages[1])
|
||||
self.assertEqual(messages[2], '[info] - Suite')
|
||||
self.assertEqual(console.getvalue(), 'En coursSuite\n')
|
||||
self.assertTrue(errors.getvalue().startswith('[warning][event] - '))
|
||||
|
||||
def test_appended_runs_without_final_newline_remain_separate(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
|
||||
contextlib.redirect_stdout(io.StringIO()):
|
||||
for message in ('Premier', 'Deuxième'):
|
||||
with exporter.log_output():
|
||||
sys.stdout.write(message)
|
||||
self.assertEqual(
|
||||
self.messages((Path(directory) / 'logs/export.log').read_text(encoding='utf-8')),
|
||||
['[info] - Premier', '[info] - Deuxième'])
|
||||
|
||||
def test_system_exit_message_is_saved(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')):
|
||||
with self.assertRaises(SystemExit):
|
||||
with exporter.log_output():
|
||||
sys.exit('Configuration invalide')
|
||||
self.assertEqual(
|
||||
self.messages((Path(directory) / 'logs/export.log').read_text(encoding='utf-8')),
|
||||
['[info] - Configuration invalide'])
|
||||
|
||||
def test_creates_logs_next_to_script_and_copies_both_streams(self):
|
||||
stdout, stderr = io.StringIO(), io.StringIO()
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
script_path = Path(directory) / 'crealead-caldav.py'
|
||||
with patch.object(exporter, '__file__', str(script_path)), \
|
||||
contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
|
||||
with exporter.log_output():
|
||||
print('Export démarré')
|
||||
print('Événement ignoré', file=sys.stderr)
|
||||
sys.stdout.write('Terminé')
|
||||
sys.stdout.flush()
|
||||
self.assertEqual(
|
||||
self.messages((Path(directory) / 'logs/crealead-caldav.log').read_text(encoding='utf-8')),
|
||||
['[info] - Export démarré', '[info] - Événement ignoré', '[info] - Terminé'])
|
||||
self.assertIs(sys.stdout, stdout)
|
||||
self.assertIs(sys.stderr, stderr)
|
||||
self.assertEqual(stdout.getvalue(), 'Export démarré\nTerminé')
|
||||
self.assertEqual(stderr.getvalue(), 'Événement ignoré\n')
|
||||
|
||||
def test_appends_on_next_execution_and_uses_script_name(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
|
||||
contextlib.redirect_stdout(io.StringIO()):
|
||||
for message in ('Premier export', 'Deuxième export'):
|
||||
with exporter.log_output():
|
||||
print(message)
|
||||
self.assertEqual(
|
||||
self.messages((Path(directory) / 'logs/export.log').read_text(encoding='utf-8')),
|
||||
['[info] - Premier export', '[info] - Deuxième export'])
|
||||
|
||||
def test_exception_traceback_is_saved_and_streams_are_restored(self):
|
||||
stdout, stderr = io.StringIO(), io.StringIO()
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
|
||||
contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
|
||||
with self.assertRaisesRegex(ValueError, 'date invalide'):
|
||||
with exporter.log_output():
|
||||
print('Avant erreur')
|
||||
raise ValueError('date invalide')
|
||||
self.assertIs(sys.stdout, stdout)
|
||||
self.assertIs(sys.stderr, stderr)
|
||||
log = (Path(directory) / 'logs/export.log').read_text(encoding='utf-8')
|
||||
messages = self.messages(log)
|
||||
self.assertEqual(messages[0], '[info] - Avant erreur')
|
||||
self.assertTrue(messages[1].startswith('[info] - Traceback'))
|
||||
self.assertTrue(all(message.startswith('[info] - ') for message in messages))
|
||||
self.assertIn('ValueError: date invalide', log)
|
||||
self.assertEqual(stdout.getvalue(), 'Avant erreur\n')
|
||||
self.assertEqual(stderr.getvalue(), '')
|
||||
|
||||
def test_system_exit_keeps_exit_code_without_logging_traceback(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
|
||||
contextlib.redirect_stderr(io.StringIO()):
|
||||
with self.assertRaises(SystemExit) as caught:
|
||||
with exporter.log_output():
|
||||
print('Connexion impossible', file=sys.stderr)
|
||||
sys.exit(2)
|
||||
self.assertEqual(caught.exception.code, 2)
|
||||
self.assertEqual(
|
||||
self.messages((Path(directory) / 'logs/export.log').read_text(encoding='utf-8')),
|
||||
['[info] - Connexion impossible'])
|
||||
|
||||
|
||||
class ExportFailureTests(unittest.TestCase):
|
||||
def test_csv_failure_logs_events_in_current_and_pending_calendars(self):
|
||||
config = dict(year_start=2026, month_start=10, day_start=1,
|
||||
year_end=2026, month_end=11, day_end=1)
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
paths = [Path(directory) / 'first.csv',
|
||||
Path(directory) / 'missing/failed.csv',
|
||||
Path(directory) / 'pending.csv']
|
||||
calendars = [
|
||||
{'csv_name': str(path), 'calendar_server': SimpleNamespace(
|
||||
name='Agenda {}'.format(index), date_search=Mock(return_value=[event_data(
|
||||
'SUMMARY:Event {}\nUID:uid-{}\nDTSTART:20261015T080000Z\n'
|
||||
'DTEND:20261015T090000Z'.format(index, index))]))}
|
||||
for index, path in enumerate(paths)
|
||||
]
|
||||
with patch.object(exporter, '__file__', str(Path(directory) / 'export.py')), \
|
||||
patch.object(exporter, 'load_json_config', return_value=config), \
|
||||
patch.object(exporter, 'init_connection', return_value=[]), \
|
||||
patch.object(exporter, 'link_calendars', return_value=calendars), \
|
||||
contextlib.redirect_stdout(io.StringIO()), \
|
||||
contextlib.redirect_stderr(io.StringIO()):
|
||||
with self.assertRaises(OSError):
|
||||
with exporter.log_output():
|
||||
exporter.main()
|
||||
log = (Path(directory) / 'logs/export.log').read_text(encoding='utf-8')
|
||||
self.assertTrue(paths[0].is_file())
|
||||
self.assertFalse(paths[2].exists())
|
||||
warnings = [line for line in log.splitlines() if '[warning][event]' in line]
|
||||
self.assertEqual(len(warnings), 2)
|
||||
for index, warning in enumerate(warnings, start=1):
|
||||
self.assertIn("Calendrier='Agenda {}'".format(index), warning)
|
||||
self.assertIn("titre='Event {}'".format(index), warning)
|
||||
self.assertIn("début='2026-10-15T08:00:00Z'", warning)
|
||||
self.assertIn('export CSV interrompu', warning)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user