import os
import io
import json
import re
import pandas as pd
import pdfplumber
import unicodedata
import traceback
from flask import Flask, request, render_template_string, Response, send_from_directory, redirect
app = Flask(__name__)
# Dossiers et fichiers
TEMP_DIR = "/app/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 = "/app/corresp_IGPDE.csv"
ODS_FILE = "/app/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''
@app.route("/")
def home():
ensure_database()
return render_template_string(f'{COMMON_STYLE}{get_nav("home")}')
@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")}')
@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")}')
@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}'
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)
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'