import os import io import json import re import pandas as pd import pdfplumber import unicodedata import traceback import webbrowser from threading import Timer from flask import Flask, request, render_template_string, Response, send_from_directory, redirect app = Flask(__name__) # --- ADAPTATION DES CHEMINS (AUTO-DETECTION) --- # Si on est dans Docker, on utilise /app, sinon le dossier local BASE_DIR = "/app" if os.path.exists("/app") else os.getcwd() TEMP_DIR = os.path.join(BASE_DIR, "downloads") if not os.path.exists(TEMP_DIR): os.makedirs(TEMP_DIR) TEMP_DATA_FILE = os.path.join(TEMP_DIR, "temp_data.json") TEMP_MISSING_FILE = os.path.join(TEMP_DIR, "temp_missing.json") CSV_FILE = os.path.join(BASE_DIR, "corresp_IGPDE.csv") ODS_FILE = os.path.join(BASE_DIR, "corresp_IGPDE.ods") RESULT_FILE = os.path.join(TEMP_DIR, "resultat.xlsx") # --- GESTION DE LA BASE DE DONNÉES --- def ensure_database(): if os.path.exists(CSV_FILE): return if os.path.exists(ODS_FILE): try: df = pd.read_excel(ODS_FILE, engine="odf") df.to_csv(CSV_FILE, index=False, encoding='utf-8-sig') return except: pass df_empty = pd.DataFrame(columns=["Nouvelle nomenclature", "Nomenclature IGPDE"]) df_empty.to_csv(CSV_FILE, index=False, encoding='utf-8-sig') # --- FONCTION DE NORMALISATION --- def super_norm(texte): if not texte or not isinstance(texte, str): return "" t = "".join(c for c in unicodedata.normalize('NFD', texte) if unicodedata.category(c) != 'Mn') t = re.sub(r'[^a-zA-Z0-9]', '', t) return t.lower() # --- DESIGN CSS --- COMMON_STYLE = """ """ def get_nav(active_page): h = "active" if active_page=="home" else "" c = "active" if active_page=="corresp" else "" r = "active" if active_page=="result" else "" return f'
🏠 Accueil📋 Table de Correspondance📊 Dernier RĂ©sultat
' @app.route("/") def home(): ensure_database() return render_template_string(f'{COMMON_STYLE}{get_nav("home")}
📄

Extraction PDF

') @app.route("/view_correspondence") def view_correspondence(): ensure_database() df = pd.read_csv(CSV_FILE, encoding='utf-8-sig').fillna('') table_html = df.to_html(index=False) return render_template_string(f'{COMMON_STYLE}{get_nav("corresp")}

Table de Correspondance (CSV)

đŸ“„ TĂ©lĂ©charger la base (CSV)
{table_html}
') @app.route("/view_result") def view_result(): if not os.path.exists(RESULT_FILE): return "Aucun résultat." df = pd.read_excel(RESULT_FILE).fillna('') table_html = df.to_html(index=False) return render_template_string(f'{COMMON_STYLE}{get_nav("result")}

Dernier RĂ©sultat Extrait

đŸ“„ TĂ©lĂ©charger le rĂ©sultat (Excel)
{table_html}
') @app.route("/process", methods=["POST"]) def process(): ensure_database() file = request.files['pdf_file'] if not file: return "Fichier manquant" file_content = file.read() def generate(): yield f'{COMMON_STYLE}

Analyse en cours...

' all_formations = [] try: with pdfplumber.open(io.BytesIO(file_content)) as pdf: total_pages = len(pdf.pages) for i, page in enumerate(pdf.pages): page_num = i + 1 percent = int((page_num / total_pages) * 85) try: check_zone = page.crop((page.width * 0.5, 0, page.width, 400)) if "Référence" in (check_zone.extract_text() or ""): bandeaux = sorted([r for r in page.rects if (r['x1'] - r['x0']) > page.width * 0.7 and r['top'] < 300 and 15 < r['height'] < 200], key=lambda r: r['top']) if len(bandeaux) >= 2: titre_brut = " ".join((page.crop((0, max(0, bandeaux[0]['top'] - 2), page.width, min(page.height, bandeaux[0]['bottom'] + 2))).extract_text() or "").replace('\n', ' ').split()) categorie_str = " ".join((page.crop((0, max(0, bandeaux[1]['top'] - 2), page.width, min(page.height, bandeaux[1]['bottom'] + 2))).extract_text() or "").replace('\n', ' ').split()) else: header = page.crop((0, 60, page.width, 250)) lines = [l.strip() for l in (header.extract_text() or "").split('\n') if l.strip() and not (len(l) > 90 or l.endswith('.'))] titre_brut, categorie_str = (" ".join(lines[:-1]), lines[-1]) if len(lines) >= 2 else (lines[0] if lines else "", "") chapitre, sous_chap = (categorie_str.split(">")[0].strip(), categorie_str.split(">")[1].strip()) if ">" in categorie_str else (categorie_str.strip(), "") data = {'Nom_IGPDE_Key': f"{chapitre} > {sous_chap}" if sous_chap else chapitre, 'Chapitre': chapitre, 'Sous-chapitre': sous_chap, 'Titre': titre_brut, 'Page_Source': page_num} table = page.crop((page.width * 0.6, 180, page.width, 600)).extract_table() if table: for row in table: if len(row) >= 2 and row[0]: cle = row[0].replace('\n', ' ').strip() for col in ["Référence", "Durée", "Type", "Publics éligibles", "Niveau", "Sessions", "Tarifs"]: if col.lower() in cle.lower(): data[col] = row[1].replace('\n', ' ').strip() break all_formations.append(data) # Nettoyage sécurisé pour les logs JS t_log = titre_brut[:40].replace("'", " ").replace('"', ' ') yield f"" except Exception: pass # --- MAPPING --- mapping_dict = {} df_map = pd.read_csv(CSV_FILE, encoding='utf-8-sig') for _, row in df_map.dropna(subset=['Nomenclature IGPDE']).iterrows(): mapping_dict[super_norm(str(row['Nomenclature IGPDE']))] = str(row['Nouvelle nomenclature']).strip() missing = [] seen_missing_norm = set() for f in all_formations: full_norm, chap_norm = super_norm(f['Nom_IGPDE_Key']), super_norm(f['Chapitre']) if full_norm not in mapping_dict and chap_norm not in mapping_dict: if full_norm not in seen_missing_norm: missing.append(f['Nom_IGPDE_Key']); seen_missing_norm.add(full_norm) with open(TEMP_DATA_FILE, "w", encoding="utf-8") as f: json.dump(all_formations, f) if missing: with open(TEMP_MISSING_FILE, "w", encoding="utf-8") as f: json.dump(missing, f) yield "" else: yield "" except Exception as e: traceback.print_exc() yield f"" yield "" return Response(generate(), mimetype='text/html') @app.route("/ask_mapping") def ask_mapping(): with open(TEMP_MISSING_FILE, "r", encoding="utf-8") as f: missing = json.load(f) df_db = pd.read_csv(CSV_FILE, encoding='utf-8-sig') known = sorted(df_db['Nouvelle nomenclature'].dropna().unique().tolist()) options_html = "".join([f'