# -*- coding: utf-8 -*- """ PLANNING TEAM - base de donnees et calculs. Tout tient dans un fichier SQLite (travail/planning.db) : sauvegarde = copie du fichier. Aucune dependance externe : sqlite3, zipfile et xml.etree sont dans la bibliotheque standard. Le cycle de reference fait 2 semaines (S1 = semaines ISO impaires, S2 = paires). Il se reconduit indefiniment ; le quotidien consiste a poser des exceptions par-dessus (lot 2). """ import os import sqlite3 import zipfile import xml.etree.ElementTree as ET BASE = os.path.dirname(os.path.abspath(__file__)) PROJET = os.path.dirname(BASE) CHEMIN_DB = os.path.join(BASE, 'planning.db') CLASSEUR = os.path.join( PROJET, 'docs', 'Emploi du temps BASE SEPTEMBRE 2026 planning sur 2 semaines.xlsx') NS = '{http://schemas.openxmlformats.org/spreadsheetml/2006/main}' JOURS = ['lundi', 'mardi', 'mercredi', 'jeudi', 'vendredi', 'samedi'] # Horaires d'ouverture, en minutes depuis minuit. # Etabli par le calcul : personne n'est planifie entre 12h30 et 14h dans le cycle. OUVERTURE = { 1: [(510, 750), (840, 1170)], # lundi 8h30-12h30 / 14h-19h30 2: [(510, 750), (840, 1170)], # mardi 3: [(510, 750), (840, 1170)], # mercredi 4: [(510, 750), (840, 1170)], # jeudi 5: [(510, 750), (840, 1170)], # vendredi 6: [(510, 750), (840, 1110)], # samedi 8h30-12h30 / 14h-18h30 } # La grille affichee couvre toute l'amplitude du classeur : 8h30 -> 19h30, soit 22 # demi-heures, y compris la coupure du midi et la fin de journee du samedi. Les creneaux # fermes restent saisissables : c'est la qu'on note le temps administratif (Nicolas, # lundi 12h30-13h). Ils ne comptent simplement pas dans la tenue du comptoir. GRILLE_DEBUT = 510 # 8h30 GRILLE_FIN = 1170 # 19h30 (fin du dernier creneau, qui commence a 19h) PHARMACIENS = 'pharmacien' # La rayonniste (Patricia) range les rayons, elle ne tient pas le comptoir. On compte # donc deux choses a chaque demi-heure : les PRESENTS dans l'officine, et ceux qui sont # reellement AU COMPTOIR. C'est ce second nombre que la regle des 3 personnes surveille. HORS_COMPTOIR = {'rayonniste'} MINI_COMPTOIR = 3 # jamais moins de 3 personnes au comptoir... DERNIERE_DEMI_HEURE = True # ...sauf sur la derniere demi-heure de la journee # Effectif de reference, feuilles « Gwen a 80% » du classeur de septembre 2026. # # L'initiale est celle de la grille : elle distingue Sa/St et MC/M, contrairement au # tableau recapitulatif du classeur qui les confond. # # La couleur est celle de la POLICE dans le classeur Excel : c'est la que se trouve le # code couleur par personne, pas dans le remplissage des cellules. Les quatre couleurs # de theme ont ete resolues avec leur teinte (Sariaka, Marie-Cecile, Stephanie, # Gwendoline). Manon et Lilou n'apparaissent pas dans la grille : leurs couleurs sont # choisies hors des teintes deja prises, a valider. # Les couleurs sont celles du classeur Excel, MODERNISEES : chacun garde sa teinte — # Aurore reste bleue, Thomas reste rouge — mais toutes sont ramenees a une luminosite # et une saturation voisines. Les couleurs pures du classeur (#FF0000, #FF00FF, # #00CCFF) vibraient les unes contre les autres et supportaient mal le fond sombre. # La couleur d'origine est rappelee en commentaire, pour garder le lien avec le classeur. EFFECTIF = [ # initiale, prenom, nom, fonction, contrat, entree, groupe, couleur ('A', 'Aurore', 'Labroy', 'pharmacien', 32.0, '2014-04-09', 'Pharmaciens', '#5271E0'), # bleu <- 3366FF ('N', 'Nicolas', 'Coue', 'pharmacien', 35.0, '2018-01-02', 'Pharmaciens', '#E5A33C'), # ambre <- FFC000 ('Sa', 'Sariaka', 'Rakotoanadahy', 'pharmacien', 35.0, '2024-01-01', 'Pharmaciens', '#9C8F5B'), # kaki <- 948A54 ('MC', 'Marie-Cecile', 'Barthelet', 'pharmacien', 27.0, '2022-01-24', 'Pharmaciens', '#6F9450'), # olive <- 77933C ('V', 'Valerie', 'Chantemesse', 'preparateur', 35.0, '2005-03-01', 'Preparateurs', '#C75CAE'), # magenta <- FF00FF ('E', 'Elodie', 'Lescovec', 'preparateur', 35.0, '2017-01-11', 'Preparateurs', '#3CB3D4'), # cyan <- 00CCFF ('G', 'Gwendoline', 'Venet', 'preparateur', 28.0, '2023-09-18', 'Preparateurs', '#E8814A'), # orange <- F79646 ('St', 'Stephanie', 'Passot', 'preparateur', 35.0, '2025-01-06', 'Preparateurs', '#9DBB5E'), # vert <- 9BBB59 ('T', 'Thomas', 'Jamet', 'preparateur', 35.0, '2021-09-13', 'Preparateurs', '#D9564F'), # rouge <- FF0000 ('C', 'Chloe', 'Constant', 'apprenti', 18.0, '2025-09-01', 'Apprentie', '#8D6E4F'), # brun <- 996600 ('P', 'Patricia', 'Pognant-Gros', 'rayonniste', 28.0, '2016-06-08', 'Rayonniste', '#8659BE'), # violet <- 7030A0 # Les deux etudiantes partagent la meme couleur : elles ne sont pas dans le cycle # de reference et n'apparaissent qu'en exception. Teinte choisie hors des onze # du classeur. ('M', 'Manon', 'Chevalier', 'etudiant', None, '2023-09-09', 'Etudiantes', '#3EA895'), # sarcelle ('L', 'Lilou', '', 'etudiant', None, '2026-09-01', 'Etudiantes', '#3EA895'), ] GROUPES = ['Pharmaciens', 'Preparateurs', 'Apprentie', 'Rayonniste', 'Etudiantes'] # Le titulaire n'a ni contrat ni creneaux : il n'apparait ni dans la grille ni dans les # compteurs d'heures. Il existe malgre tout dans la base, parce qu'il lui faut un compte # avec un acces complet. TITULAIRE = ('J', 'Jerome', 'Cuvillier', 'titulaire', None, None, 'Direction', '#1F6763') SCHEMA = """ CREATE TABLE IF NOT EXISTS salarie ( id INTEGER PRIMARY KEY, initiale TEXT NOT NULL UNIQUE, prenom TEXT NOT NULL, nom TEXT NOT NULL DEFAULT '', fonction TEXT NOT NULL, groupe TEXT NOT NULL, groupe_rang INTEGER NOT NULL DEFAULT 0, couleur TEXT NOT NULL DEFAULT '#5A5B5D', contrat_h_moyen REAL, date_entree TEXT, paye_a_l_heure INTEGER NOT NULL DEFAULT 0, ordre INTEGER NOT NULL DEFAULT 0, actif INTEGER NOT NULL DEFAULT 1, hors_planning INTEGER NOT NULL DEFAULT 0 ); CREATE TABLE IF NOT EXISTS creneau_type ( id INTEGER PRIMARY KEY, salarie_id INTEGER NOT NULL REFERENCES salarie(id) ON DELETE CASCADE, semaine_cycle INTEGER NOT NULL CHECK (semaine_cycle IN (1, 2)), jour INTEGER NOT NULL CHECK (jour BETWEEN 1 AND 6), debut_min INTEGER NOT NULL, fin_min INTEGER NOT NULL, hors_comptoir INTEGER NOT NULL DEFAULT 0, ajpp INTEGER NOT NULL DEFAULT 0, CHECK (fin_min > debut_min) ); -- Copie figee du cycle tel qu'il sort du classeur. Elle ne bouge qu'a l'import et sert -- au bouton « revenir a l'emploi du temps de base », ligne par ligne. CREATE TABLE IF NOT EXISTS creneau_base ( id INTEGER PRIMARY KEY, salarie_id INTEGER NOT NULL, semaine_cycle INTEGER NOT NULL, jour INTEGER NOT NULL, debut_min INTEGER NOT NULL, fin_min INTEGER NOT NULL, hors_comptoir INTEGER NOT NULL DEFAULT 0, ajpp INTEGER NOT NULL DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_creneau_cycle ON creneau_type (semaine_cycle, jour, salarie_id); CREATE INDEX IF NOT EXISTS idx_base_cycle ON creneau_base (semaine_cycle, jour, salarie_id); """ # ---------------------------------------------------------------- connexion def connexion(): cx = sqlite3.connect(CHEMIN_DB) cx.row_factory = sqlite3.Row cx.execute('PRAGMA foreign_keys = ON') return cx def creer_schema(cx): cx.executescript(SCHEMA) _migrer(cx) cx.commit() # Colonnes ajoutees apres coup. On les ajoute au lieu de recreer les tables : recreer # effacerait les conges et les demandes, qui pointent sur salarie. MIGRATIONS = { 'salarie': [ ('groupe', "TEXT NOT NULL DEFAULT ''"), ('groupe_rang', 'INTEGER NOT NULL DEFAULT 0'), ('couleur', "TEXT NOT NULL DEFAULT '#5A5B5D'"), ('hors_planning', 'INTEGER NOT NULL DEFAULT 0'), ], 'creneau_type': [('ajpp', 'INTEGER NOT NULL DEFAULT 0')], 'creneau_base': [('ajpp', 'INTEGER NOT NULL DEFAULT 0')], } def _colonnes(cx, table): return {l[1] for l in cx.execute('PRAGMA table_info(%s)' % table)} def _migrer(cx): for table, colonnes in MIGRATIONS.items(): existantes = _colonnes(cx, table) if not existantes: continue for nom, definition in colonnes: if nom not in existantes: cx.execute('ALTER TABLE %s ADD COLUMN %s %s' % (table, nom, definition)) # ------------------------------------------------------- lecture du classeur def _chaines_partagees(zf): chaines = [] with zf.open('xl/sharedStrings.xml') as f: for si in ET.parse(f).getroot(): chaines.append(''.join(t.text or '' for t in si.iter(NS + 't'))) return chaines def _feuille(zf, nom_cherche): with zf.open('xl/workbook.xml') as f: noms = [s.get('name') for s in ET.parse(f).getroot().iter(NS + 'sheet')] if nom_cherche not in noms: raise ValueError(u'Feuille introuvable : %s (feuilles : %s)' % (nom_cherche, ', '.join(noms))) return 'xl/worksheets/sheet%d.xml' % (noms.index(nom_cherche) + 1) def _styles_remplis(zf): """L'ensemble des index de style dont la cellule porte un remplissage visible.""" with zf.open('xl/styles.xml') as f: racine = ET.parse(f).getroot() remplis = [] for fill in racine.find(NS + 'fills'): motif = fill.find(NS + 'patternFill') remplis.append(bool(motif is not None and motif.get('patternType') not in (None, 'none', 'gray125'))) resultat = set() for i, xf in enumerate(racine.find(NS + 'cellXfs')): if remplis[int(xf.get('fillId', 0))]: resultat.add(i) return resultat def _cellules(zf, chemin, chaines, styles_remplis): """{ 'C4': (valeur, remplie) } pour toutes les cellules non vides.""" valeurs = {} with zf.open(chemin) as f: for c in ET.parse(f).getroot().iter(NS + 'c'): v = c.find(NS + 'v') if v is None or v.text is None: continue texte = chaines[int(v.text)] if c.get('t') == 's' else v.text texte = (texte or '').strip() if texte: style = c.get('s') valeurs[c.get('r')] = (texte, style is not None and int(style) in styles_remplis) return valeurs def _num_colonne(lettres): n = 0 for ch in lettres: n = n * 26 + (ord(ch) - 64) return n def _lettres_colonne(n): s = '' while n > 0: n, r = divmod(n - 1, 26) s = chr(65 + r) + s return s def lire_grille(chemin_xlsx, nom_feuille): """ Lit une feuille du planning de base. La grille couvre les colonnes C a BP : six groupes de onze colonnes, un par jour, la ligne 2 donnant l'initiale de chaque colonne. Les lignes 4 a 25 sont les demi-heures de 8h30 a 19h30. Le remplissage de fond marque les heures d'AJPP (verifie sur Elodie : mardi et jeudi en S1, jeudi en S2, soit 17 h et 8,5 h — exactement l'ecart annonce entre « 35 h » et « 18 h / 26,5 h sans AJPP »). Renvoie { (initiale, jour) : { minute : ajpp } } """ with zipfile.ZipFile(chemin_xlsx) as zf: chaines = _chaines_partagees(zf) cell = _cellules(zf, _feuille(zf, nom_feuille), chaines, _styles_remplis(zf)) premiere, derniere = _num_colonne('C'), _num_colonne('BP') grille = {} for i in range(premiere, derniere + 1): col = _lettres_colonne(i) entete = cell.get(col + '2') if not entete: continue initiale, jour = entete[0], (i - premiere) // 11 + 1 for ligne in range(4, 26): case = cell.get('%s%d' % (col, ligne)) if case: grille.setdefault((entete[0], jour), {})[510 + (ligne - 4) * 30] = case[1] return grille def _plages(minutes_ajpp, jour): """ Transforme { minute : ajpp } en plages continues (debut, fin, hors_comptoir, ajpp). Deux demi-heures ne sont recollees que si elles partagent le meme statut. """ morceaux = [] for minute in sorted(minutes_ajpp): ajpp = 1 if minutes_ajpp[minute] else 0 hors = 0 if _dans_ouverture(jour, minute) else 1 if (morceaux and morceaux[-1][1] == minute and morceaux[-1][2] == hors and morceaux[-1][3] == ajpp): morceaux[-1][1] = minute + 30 else: morceaux.append([minute, minute + 30, hors, ajpp]) return [tuple(m) for m in morceaux] def _dans_ouverture(jour, minute): return any(debut <= minute < fin for debut, fin in OUVERTURE[jour]) # ------------------------------------------------------------------- import def importer(cx, chemin_xlsx=None, bavard=True): """(Re)construit la base a partir du classeur de reference.""" chemin_xlsx = chemin_xlsx or CLASSEUR creer_schema(cx) # L'import ne reconstruit QUE ce qui vient du classeur : le cycle de reference et # la fiche de chaque salarie. Il ne doit jamais toucher aux conges, aux absences # ni aux demandes. # # D'ou deux precautions : on ne supprime pas la table salarie (avec les cles # etrangeres actives, un DROP TABLE supprime d'abord toutes les lignes, ce qui # propage les ON DELETE CASCADE et efface soldes, absences et demandes), et on # conserve l'identifiant de chaque personne en la reconnaissant a son initiale. cx.execute('DELETE FROM creneau_type') cx.execute('DELETE FROM creneau_base') par_initiale = {l['initiale']: l['id'] for l in cx.execute('SELECT id, initiale FROM salarie')} for ordre, ligne in enumerate(EFFECTIF): initiale, prenom, nom, fonction, contrat, entree, groupe, couleur = ligne champs = (prenom, nom, fonction, groupe, GROUPES.index(groupe), couleur, contrat, entree, 1 if contrat is None else 0, ordre) if initiale in par_initiale: cx.execute( """UPDATE salarie SET prenom = ?, nom = ?, fonction = ?, groupe = ?, groupe_rang = ?, couleur = ?, contrat_h_moyen = ?, date_entree = ?, paye_a_l_heure = ?, ordre = ?, actif = 1 WHERE initiale = ?""", champs + (initiale,)) else: cur = cx.execute( """INSERT INTO salarie (prenom, nom, fonction, groupe, groupe_rang, couleur, contrat_h_moyen, date_entree, paye_a_l_heure, ordre, initiale) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", champs + (initiale,)) par_initiale[initiale] = cur.lastrowid # Le titulaire : present dans la base pour son compte, absent du planning. initiale, prenom, nom, fonction, _, _, groupe, couleur = TITULAIRE if initiale in par_initiale: cx.execute("""UPDATE salarie SET prenom = ?, nom = ?, fonction = ?, groupe = ?, couleur = ?, hors_planning = 1, actif = 1 WHERE initiale = ?""", (prenom, nom, fonction, groupe, couleur, initiale)) else: cur = cx.execute( """INSERT INTO salarie (initiale, prenom, nom, fonction, groupe, groupe_rang, couleur, ordre, hors_planning) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1)""", (initiale, prenom, nom, fonction, groupe, len(GROUPES), couleur, 99)) par_initiale[initiale] = cur.lastrowid # Quelqu'un qui a quitte l'equipe est desactive, jamais supprime : ses absences # et son historique restent consultables. connus = [l[0] for l in EFFECTIF] + [TITULAIRE[0]] cx.execute('UPDATE salarie SET actif = 0 WHERE initiale NOT IN (%s)' % ','.join('?' * len(connus)), connus) total, ajpp_total, ignores = 0, 0, set() for semaine, feuille in ((1, u'S1 (Gwen \xe0 80%)'), (2, u'S2 (Gwen \xe0 80%)')): for (initiale, jour), minutes in lire_grille(chemin_xlsx, feuille).items(): sid = par_initiale.get(initiale) if sid is None: ignores.add(initiale) continue for debut, fin, hors, ajpp in _plages(minutes, jour): cx.execute( """INSERT INTO creneau_type (salarie_id, semaine_cycle, jour, debut_min, fin_min, hors_comptoir, ajpp) VALUES (?, ?, ?, ?, ?, ?, ?)""", (sid, semaine, jour, debut, fin, hors, ajpp)) total += 1 if ajpp: ajpp_total += (fin - debut) / 60.0 cx.execute( """INSERT INTO creneau_base (salarie_id, semaine_cycle, jour, debut_min, fin_min, hors_comptoir, ajpp) SELECT salarie_id, semaine_cycle, jour, debut_min, fin_min, hors_comptoir, ajpp FROM creneau_type""") cx.commit() if bavard: print(u'Import termine : %d salaries, %d creneaux, %.1f h d\'AJPP.' % (len(EFFECTIF), total, ajpp_total)) if ignores: print(u'Initiales du classeur sans salarie connu : %s' % ', '.join(sorted(ignores))) return total # ------------------------------------------------------------------ lecture def salaries(cx, tous=False): """ Les gens du planning. Le titulaire en est exclu : il a un compte, pas de creneaux. `tous=True` le ramene, pour la gestion des acces. """ condition = 'actif = 1' if tous else 'actif = 1 AND hors_planning = 0' lignes = cx.execute( """SELECT id, initiale, prenom, nom, fonction, groupe, groupe_rang, couleur, contrat_h_moyen, date_entree, paye_a_l_heure, hors_planning FROM salarie WHERE %s ORDER BY groupe_rang, ordre""" % condition).fetchall() return [dict(l) for l in lignes] def creneaux(cx, semaine_cycle): lignes = cx.execute( """SELECT id, salarie_id, jour, debut_min, fin_min, hors_comptoir, ajpp FROM creneau_type WHERE semaine_cycle = ? ORDER BY jour, debut_min""", (semaine_cycle,)).fetchall() return [dict(l) for l in lignes] def heures_par_semaine(cx): """{ salarie_id: {1: (total, ajpp), 2: (total, ajpp)} }""" resultat = {} for l in cx.execute( """SELECT salarie_id, semaine_cycle, SUM(fin_min - debut_min) AS minutes, SUM(CASE WHEN ajpp THEN fin_min - debut_min ELSE 0 END) AS minutes_ajpp FROM creneau_type GROUP BY salarie_id, semaine_cycle"""): resultat.setdefault(l['salarie_id'], {})[l['semaine_cycle']] = ( l['minutes'] / 60.0, l['minutes_ajpp'] / 60.0) return resultat def creneaux_grille(): """Les 22 demi-heures affichees, 8h30 -> 19h30, communes a tous les jours.""" return list(range(GRILLE_DEBUT, GRILLE_FIN, 30)) def creneaux_ouverture(jour): """Les demi-heures ou l'officine est ouverte ce jour-la.""" pas = [] for debut, fin in OUVERTURE[jour]: pas.extend(range(debut, fin, 30)) return pas def derniere_demi_heure(jour): """La derniere demi-heure d'ouverture, toleree sous le seuil de 3 personnes.""" pas = creneaux_ouverture(jour) return pas[-1] if pas else None def presence(cx, semaine_cycle): """ Pour chaque demi-heure d'ouverture : qui tient le comptoir, combien, combien de pharmaciens, et si une regle est enfreinte. """ gens = salaries(cx) fonctions = {s['id']: s['fonction'] for s in gens} initiales = {s['id']: s['initiale'] for s in gens} occupe = {} for c in creneaux(cx, semaine_cycle): if c['hors_comptoir']: continue # le travail administratif ne tient pas le comptoir for minute in range(c['debut_min'], c['fin_min'], 30): occupe.setdefault((c['jour'], minute), []).append(c['salarie_id']) resultat = {} for jour in range(1, 7): ouverts = set(creneaux_ouverture(jour)) derniere = derniere_demi_heure(jour) jour_resultat = [] for minute in creneaux_grille(): ouvert = minute in ouverts ids = occupe.get((jour, minute), []) nb_pharmaciens = sum(1 for i in ids if fonctions.get(i) == PHARMACIENS) au_comptoir = [i for i in ids if fonctions.get(i) not in HORS_COMPTOIR] jour_resultat.append({ 'minute': minute, 'ouvert': ouvert, 'initiales': [initiales[i] for i in ids], 'nb': len(ids), 'comptoir': len(au_comptoir), 'pharmaciens': nb_pharmaciens, # Les regles ne s'appliquent que pendant l'ouverture, et le seuil de 3 # porte sur le comptoir, pas sur les presents. 'manque_pharmacien': ouvert and nb_pharmaciens == 0, 'sous_effectif': (ouvert and len(au_comptoir) < MINI_COMPTOIR and not (DERNIERE_DEMI_HEURE and minute == derniere)), }) resultat[jour] = jour_resultat return resultat def compteurs(cx): """Par salarie : heures S1, S2, moyenne du cycle, ecart au contrat, AJPP, samedis.""" heures = heures_par_semaine(cx) samedis = {} for l in cx.execute( """SELECT salarie_id, semaine_cycle FROM creneau_type WHERE jour = 6 GROUP BY salarie_id, semaine_cycle"""): samedis.setdefault(l['salarie_id'], set()).add(l['semaine_cycle']) resultat = [] for s in salaries(cx): h = heures.get(s['id'], {}) h1, a1 = h.get(1, (0.0, 0.0)) h2, a2 = h.get(2, (0.0, 0.0)) moyenne = (h1 + h2) / 2.0 contrat = s['contrat_h_moyen'] resultat.append({ 'salarie_id': s['id'], 'initiale': s['initiale'], 'prenom': s['prenom'], 'fonction': s['fonction'], 'groupe': s['groupe'], 'couleur': s['couleur'], 'contrat': contrat, 's1': round(h1, 2), 's2': round(h2, 2), 'moyenne': round(moyenne, 2), 'ecart': None if contrat is None else round(moyenne - contrat, 2), 'ajpp_s1': round(a1, 2), 'ajpp_s2': round(a2, 2), 'samedis': sorted(samedis.get(s['id'], [])), }) return resultat # ------------------------------------------------------------------ ecriture def basculer_creneau(cx, salarie_id, semaine_cycle, jour, minute): """ Ajoute ou retire une demi-heure, puis refusionne les plages contigues de meme statut. Renvoie True si la demi-heure est desormais travaillee. """ ligne = cx.execute( """SELECT id, debut_min, fin_min, ajpp FROM creneau_type WHERE salarie_id = ? AND semaine_cycle = ? AND jour = ? AND debut_min <= ? AND fin_min > ?""", (salarie_id, semaine_cycle, jour, minute, minute)).fetchone() if ligne: # elle existe -> on la retire cx.execute('DELETE FROM creneau_type WHERE id = ?', (ligne['id'],)) if ligne['debut_min'] < minute: _inserer(cx, salarie_id, semaine_cycle, jour, ligne['debut_min'], minute, ligne['ajpp']) if minute + 30 < ligne['fin_min']: _inserer(cx, salarie_id, semaine_cycle, jour, minute + 30, ligne['fin_min'], ligne['ajpp']) cx.commit() return False # L'AJPP est une propriete de la JOURNEE : une demi-heure ajoutee a une journee # deja entierement en AJPP en herite. Sans cela la journee deviendrait mixte, et # le bouton AJPP ne serait plus reversible. jour_ajpp = cx.execute( """SELECT MIN(ajpp) AS tous FROM creneau_type WHERE salarie_id = ? AND semaine_cycle = ? AND jour = ?""", (salarie_id, semaine_cycle, jour)).fetchone()['tous'] _inserer(cx, salarie_id, semaine_cycle, jour, minute, minute + 30, 1 if jour_ajpp else 0) _fusionner(cx, salarie_id, semaine_cycle, jour) cx.commit() return True def basculer_ajpp(cx, salarie_id, semaine_cycle, jour): """ Marque ou demarque une JOURNEE entiere en AJPP. L'AJPP se compte en jours, pas en creneaux : le cabinet attend « Lu 24/08 + Je 27/08, 2 jours, 17,5 h ». On bascule donc tout ce que la personne travaille ce jour-la, d'un coup. """ lignes = cx.execute( """SELECT id, ajpp FROM creneau_type WHERE salarie_id = ? AND semaine_cycle = ? AND jour = ?""", (salarie_id, semaine_cycle, jour)).fetchall() if not lignes: return None nouveau = 0 if all(l['ajpp'] for l in lignes) else 1 cx.execute( """UPDATE creneau_type SET ajpp = ? WHERE salarie_id = ? AND semaine_cycle = ? AND jour = ?""", (nouveau, salarie_id, semaine_cycle, jour)) _fusionner(cx, salarie_id, semaine_cycle, jour) cx.commit() return nouveau def reinitialiser_ligne(cx, salarie_id, semaine_cycle, jour): """ Remet une ligne — une personne, un jour — telle qu'elle sort du classeur. Renvoie True si quelque chose a change. """ avant = cx.execute( """SELECT debut_min, fin_min, hors_comptoir, ajpp FROM creneau_type WHERE salarie_id = ? AND semaine_cycle = ? AND jour = ? ORDER BY debut_min""", (salarie_id, semaine_cycle, jour)).fetchall() apres = cx.execute( """SELECT debut_min, fin_min, hors_comptoir, ajpp FROM creneau_base WHERE salarie_id = ? AND semaine_cycle = ? AND jour = ? ORDER BY debut_min""", (salarie_id, semaine_cycle, jour)).fetchall() cx.execute( """DELETE FROM creneau_type WHERE salarie_id = ? AND semaine_cycle = ? AND jour = ?""", (salarie_id, semaine_cycle, jour)) for l in apres: cx.execute( """INSERT INTO creneau_type (salarie_id, semaine_cycle, jour, debut_min, fin_min, hors_comptoir, ajpp) VALUES (?, ?, ?, ?, ?, ?, ?)""", (salarie_id, semaine_cycle, jour, l['debut_min'], l['fin_min'], l['hors_comptoir'], l['ajpp'])) cx.commit() return [tuple(l) for l in avant] != [tuple(l) for l in apres] def lignes_modifiees(cx, semaine_cycle): """ Les couples (salarie, jour) qui different du classeur — pour signaler dans la grille ce qui a ete retouche a la main. """ requete = """ SELECT salarie_id, jour, GROUP_CONCAT(debut_min || '-' || fin_min || ':' || ajpp) AS empreinte FROM %s WHERE semaine_cycle = ? GROUP BY salarie_id, jour""" actuel = {(l['salarie_id'], l['jour']): l['empreinte'] for l in cx.execute(requete % 'creneau_type', (semaine_cycle,))} origine = {(l['salarie_id'], l['jour']): l['empreinte'] for l in cx.execute(requete % 'creneau_base', (semaine_cycle,))} cles = set(actuel) | set(origine) return ['%d|%d' % cle for cle in sorted(cles) if actuel.get(cle) != origine.get(cle)] def jours_ajpp(cx, semaine_cycle): """{ salarie_id : [jours entierement en AJPP] } pour une semaine du cycle.""" resultat = {} for l in cx.execute( """SELECT salarie_id, jour, MIN(ajpp) AS tous, SUM(fin_min - debut_min) AS minutes FROM creneau_type WHERE semaine_cycle = ? GROUP BY salarie_id, jour""", (semaine_cycle,)): if l['tous']: resultat.setdefault(l['salarie_id'], []).append( {'jour': l['jour'], 'heures': round(l['minutes'] / 60.0, 2)}) return resultat def _inserer(cx, salarie_id, semaine_cycle, jour, debut, fin, ajpp): minutes = {m: bool(ajpp) for m in range(debut, fin, 30)} for d, f, hors, marque in _plages(minutes, jour): cx.execute( """INSERT INTO creneau_type (salarie_id, semaine_cycle, jour, debut_min, fin_min, hors_comptoir, ajpp) VALUES (?, ?, ?, ?, ?, ?, ?)""", (salarie_id, semaine_cycle, jour, d, f, hors, marque)) def _fusionner(cx, salarie_id, semaine_cycle, jour): """Recolle les plages qui se touchent et partagent le meme statut.""" lignes = cx.execute( """SELECT id, debut_min, fin_min, hors_comptoir, ajpp FROM creneau_type WHERE salarie_id = ? AND semaine_cycle = ? AND jour = ? ORDER BY debut_min""", (salarie_id, semaine_cycle, jour)).fetchall() precedente = None for l in lignes: if (precedente and precedente['fin_min'] == l['debut_min'] and precedente['hors_comptoir'] == l['hors_comptoir'] and precedente['ajpp'] == l['ajpp']): cx.execute('UPDATE creneau_type SET fin_min = ? WHERE id = ?', (l['fin_min'], precedente['id'])) cx.execute('DELETE FROM creneau_type WHERE id = ?', (l['id'],)) precedente = {'id': precedente['id'], 'debut_min': precedente['debut_min'], 'fin_min': l['fin_min'], 'hors_comptoir': l['hors_comptoir'], 'ajpp': l['ajpp']} else: precedente = l # ---------------------------------------------------------------- diagnostic def etat(cx): """Un resume lisible, pour verifier l'import depuis la ligne de commande.""" lignes = [u'Base : %s' % CHEMIN_DB, u''] lignes.append(u'%-4s %-13s %-13s %7s %7s %9s %8s' % (u'Ini', u'Prenom', u'Groupe', u'S1', u'S2', u'Moyenne', u'Contrat')) groupe_courant = None for c in compteurs(cx): if c['groupe'] != groupe_courant: groupe_courant = c['groupe'] lignes.append(u'-- %s' % groupe_courant) contrat = u'-' if c['contrat'] is None else u'%.2f' % c['contrat'] notes = u'' if c['ecart'] is not None and abs(c['ecart']) >= 0.01: notes += u' ecart %+0.2f h' % c['ecart'] if c['ajpp_s1'] or c['ajpp_s2']: notes += u' AJPP %.1f h / %.1f h' % (c['ajpp_s1'], c['ajpp_s2']) lignes.append(u'%-4s %-13s %-13s %7.2f %7.2f %9.2f %8s%s' % (c['initiale'], c['prenom'], c['groupe'], c['s1'], c['s2'], c['moyenne'], contrat, notes)) for semaine in (1, 2): alertes = [] for jour, pas in presence(cx, semaine).items(): for p in pas: if p['manque_pharmacien'] or p['sous_effectif']: alertes.append(u' %-9s %02dh%02d : %d pers, %d pharm' % (JOURS[jour - 1], p['minute'] // 60, p['minute'] % 60, p['nb'], p['pharmaciens'])) lignes.append(u'') lignes.append(u'Semaine %d : %s' % (semaine, u'aucune alerte' if not alertes else u'%d alertes' % len(alertes))) lignes.extend(alertes) return u'\n'.join(lignes) if __name__ == '__main__': import sys cx = connexion() if len(sys.argv) > 1 and sys.argv[1] == 'import': importer(cx) print(etat(cx)) cx.close()