# ============================================================ # app.py — Pipeline complet PDF → Prévisions GRU # Hugging Face Spaces (Gradio) # ============================================================ import gradio as gr import fitz # PyMuPDF import torch import torch.nn as nn import numpy as np import pickle import json import re import os import tempfile from PIL import Image from huggingface_hub import hf_hub_download from transformers import AutoImageProcessor, TableTransformerForObjectDetection import easyocr # ============================================================ # CHARGEMENT DES FICHIERS DEPUIS HUGGING FACE # ============================================================ HF_REPO_ID = "ferdaouskachouri/financial_predictionn" print("⏳ Chargement des fichiers du repo...") scaler_path = hf_hub_download(repo_id=HF_REPO_ID, filename="scaler_global.pkl") model_path = hf_hub_download(repo_id=HF_REPO_ID, filename="gru_model_global.pt") postes_path = hf_hub_download(repo_id=HF_REPO_ID, filename="poste_cols.json") targets_path = hf_hub_download(repo_id=HF_REPO_ID, filename="target_cols.json") config_path = hf_hub_download(repo_id=HF_REPO_ID, filename="model_config.json") with open(postes_path, encoding='utf-8') as f: POSTE_COLS = json.load(f) with open(targets_path, encoding='utf-8') as f: TARGET_COLS = json.load(f) with open(config_path) as f: CONFIG = json.load(f) N_FEATURES = CONFIG['n_features'] N_TARGETS = CONFIG['n_targets'] HIDDEN = CONFIG['hidden'] N_LAYERS = CONFIG['n_layers'] DROPOUT = CONFIG['dropout'] SEQ_LEN = CONFIG['seq_len'] PRED_STEPS = CONFIG['pred_steps'] with open(scaler_path, 'rb') as f: scaler_global = pickle.load(f) # ============================================================ # ARCHITECTURE GRU # ============================================================ class FinanceGRU(nn.Module): def __init__(self): super().__init__() self.gru = nn.GRU( input_size = N_FEATURES, hidden_size = HIDDEN, num_layers = N_LAYERS, dropout = DROPOUT, batch_first = True ) self.dropout = nn.Dropout(DROPOUT) self.fc = nn.Linear(HIDDEN, N_TARGETS * PRED_STEPS) def forward(self, x): out, _ = self.gru(x) out = self.dropout(out[:, -1, :]) out = self.fc(out) return out.view(-1, PRED_STEPS, N_TARGETS) DEVICE = torch.device('cpu') gru_model = FinanceGRU().to(DEVICE) gru_model.load_state_dict(torch.load(model_path, map_location=DEVICE)) gru_model.eval() print("✅ Modèle GRU chargé") # ============================================================ # CHARGEMENT TABLE TRANSFORMER + EASYOCR # ============================================================ print("⏳ Chargement Table Transformer...") tt_processor = AutoImageProcessor.from_pretrained("microsoft/table-transformer-detection") tt_model = TableTransformerForObjectDetection.from_pretrained("microsoft/table-transformer-detection") tt_model.eval() print("✅ Table Transformer chargé") print("⏳ Chargement EasyOCR...") ocr_reader = easyocr.Reader(['fr', 'en'], gpu=False) print("✅ EasyOCR chargé") # ============================================================ # KEYWORDS # ============================================================ KEYWORDS = { 'actif': [ 'immobilisations corporelles', 'immobilisations incorporelles', 'stocks nets', 'clients et comptes rattachés', 'liquidités et équivalents de liquidités', 'total des actifs', 'actifs non courants', 'actifs courants', ], 'passif': [ 'capital social', 'réserves', 'résultats reportés', 'emprunts et dettes assimilées', 'fournisseurs et comptes rattachés', 'total des capitaux propres', 'total des passifs', ], 'resultats': [ 'revenus', 'charges de personnel', 'dotations aux amortissements', "résultat d'exploitation", 'résultat net de l\'exercice', 'résultat de l\'exercice', 'impôt sur les bénéfices', ], 'flux': [ 'résultat net', 'variation de trésorerie', 'trésorerie à la clôture', 'flux de trésorerie', ], } PARASITES = [ 'notes', 'au 31', 'au 30', 'période', 'exercice', '2014','2015','2016','2017','2018','2019', '2020','2021','2022','2023','2024','2025', 'décembre', 'juin', 'page', 'suite', ] # ============================================================ # FONCTIONS PIPELINE # ============================================================ def pdf_to_images(pdf_path, zoom=2.0): doc = fitz.open(pdf_path) mat = fitz.Matrix(zoom, zoom) images = [] for page in doc: pix = page.get_pixmap(matrix=mat) img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples) images.append(img) doc.close() return images def detect_table(image, threshold=0.65): w, h = image.size if w > 1200: image = image.resize((1200, int(h * 1200 / w))) inputs = tt_processor(images=image, return_tensors="pt") with torch.no_grad(): outputs = tt_model(**inputs) target_sizes = torch.tensor([image.size[::-1]]) results = tt_processor.post_process_object_detection( outputs, threshold=threshold, target_sizes=target_sizes )[0] tables = [] for score, box in zip(results["scores"], results["boxes"]): x1, y1, x2, y2 = [int(v) for v in box.tolist()] tables.append({'bbox': [x1, y1, x2, y2], 'score': float(score)}) return tables, image def ocr_bbox(image, bbox): x1, y1, x2, y2 = bbox x1 = max(0, x1 - 5); y1 = max(0, y1 - 5) x2 = min(image.width, x2 + 5) y2 = min(image.height, y2 + 5) cropped = image.crop((x1, y1, x2, y2)) cw, ch = cropped.size if cw > 800: cropped = cropped.resize((800, int(ch * 800 / cw))) with tempfile.NamedTemporaryFile(suffix='.png', delete=False) as tmp: cropped.save(tmp.name) result = ocr_reader.readtext(tmp.name, batch_size=4) os.unlink(tmp.name) return result def clean_number(text): text = str(text).replace(' ', '').replace('\xa0', '') text = text.replace(',', '.').replace('(', '-').replace(')', '') try: val = float(re.sub(r'[^\d.\-]', '', text)) return val if abs(val) > 10 else None except: return None def is_valid_label(label): label_low = label.lower().strip() if len(label_low) < 4: return False if label_low[0].isdigit(): return False if any(p in label_low for p in PARASITES): return False if re.match(r'^[\d\s.,\-\(\)]+$', label_low): return False return True def identify_type(text): text_low = text.lower() scores = {} for t, kws in KEYWORDS.items(): score = sum(1 for kw in kws if kw in text_low) if score > 0: scores[t] = score return max(scores, key=scores.get) if scores else None def extract_pairs(ocr_result): if not ocr_result: return {} lines = {} for item in ocr_result: bbox_item, text, conf = item if conf < 0.35: continue y = int((bbox_item[0][1] + bbox_item[2][1]) / 2) y_key = round(y / 12) * 12 x = int((bbox_item[0][0] + bbox_item[2][0]) / 2) lines.setdefault(y_key, []).append({'x': x, 'text': text}) pairs = {} for y_key in sorted(lines.keys()): items = sorted(lines[y_key], key=lambda i: i['x']) if len(items) < 2: continue label = items[0]['text'].strip() if not is_valid_label(label): continue for item in items[1:]: num = clean_number(item['text']) if num is not None: pairs[label.lower()] = num break return pairs def pdf_to_financial_data(pdf_path): images = pdf_to_images(pdf_path) all_tables = {} found_types = set() for page_num, image in enumerate(images[:15]): if len(found_types) == 4: break tables = [] for threshold in [0.65, 0.50, 0.35]: tables, img_resized = detect_table(image, threshold) if tables: break if not tables: continue best = max(tables, key=lambda t: t['score']) bbox = best['bbox'] w, h = img_resized.size bbox_area = (bbox[2]-bbox[0]) * (bbox[3]-bbox[1]) / (w * h) * 100 if bbox_area < 15: continue ocr_result = ocr_bbox(img_resized, bbox) if not ocr_result: continue full_text = ' '.join([r[1] for r in ocr_result]) table_type = identify_type(full_text) if not table_type or table_type in found_types: continue pairs = extract_pairs(ocr_result) if len(pairs) < 3: continue all_tables[table_type] = pairs found_types.add(table_type) return all_tables def tables_to_feature_vector(tables): vector = np.zeros(N_FEATURES, dtype=np.float32) for ttype, pairs in tables.items(): for label, val in pairs.items(): col_name = f"{ttype}_{label}" if col_name in POSTE_COLS: idx = POSTE_COLS.index(col_name) vector[idx] = val else: for i, col in enumerate(POSTE_COLS): if label in col or col.split('_', 1)[-1] in label: vector[i] = val break return vector def predict_from_vector(feature_vector): vector_norm = scaler_global.transform(feature_vector.reshape(1, -1))[0] sequence = np.tile(vector_norm, (SEQ_LEN, 1)).astype(np.float32) X = torch.tensor(sequence[np.newaxis], dtype=torch.float32) with torch.no_grad(): Y_hat = gru_model(X).cpu().numpy()[0] previsions = {} for step, yr in enumerate([2025, 2026]): dummy = feature_vector.copy().reshape(1, -1) for j, col in enumerate(TARGET_COLS): if col in POSTE_COLS: idx = POSTE_COLS.index(col) dummy[0, idx] = float(np.clip(Y_hat[step, j], 0.0, 1.0)) denorm = scaler_global.inverse_transform(dummy)[0] previsions[str(yr)] = {} for col in TARGET_COLS: if col in POSTE_COLS: idx = POSTE_COLS.index(col) previsions[str(yr)][col] = round(float(denorm[idx]), 0) return previsions # ============================================================ # FONCTION PRINCIPALE GRADIO # ============================================================ def process_pdf(pdf_file): try: if pdf_file is None: return {"erreur": "No file received"} # Gradio type="filepath" retourne un string directement filepath = pdf_file if isinstance(pdf_file, str) else pdf_file.name # Vérifier que c'est bien un PDF en lisant les premiers bytes try: with open(filepath, 'rb') as f: header = f.read(4) if header != b'%PDF': return {"erreur": "Invalid PDF file"} except: return {"erreur": "Cannot read uploaded file"} tables = pdf_to_financial_data(filepath) if not tables: return { "erreur" : "No financial table detected in this PDF.", "conseil": "Make sure the PDF contains financial statements." } feature_vector = tables_to_feature_vector(tables) previsions = predict_from_vector(feature_vector) result = { "statut" : "succès", "tables_trouvees": list(tables.keys()), "previsions" : previsions, "targets" : TARGET_COLS, "resume" : {} } for yr in ["2025", "2026"]: result["resume"][yr] = {} for col in TARGET_COLS: val = previsions[yr].get(col, 0) label = col.replace("resultats_", "").replace("actif_", "").replace("flux_", "") result["resume"][yr][label] = f"{val:,.0f} DT" return result except Exception as e: return {"erreur": str(e)}# ============================================================ # INTERFACE GRADIO # ============================================================ with gr.Blocks(title="Financial Prediction") as demo: gr.Markdown("## 📊 Financial Predictions 2025-2026") gr.Markdown("Upload a PDF of financial statements — the pipeline extracts the data and predicts using GRU.") with gr.Row(): pdf_input = gr.File( label = "📄 Financial Statements PDF", type = "filepath" # ← plus de restriction file_types ) json_output = gr.JSON(label="📈 Results and Predictions") btn = gr.Button("🚀 Run Analysis", variant="primary") btn.click(fn=process_pdf, inputs=pdf_input, outputs=json_output) gr.Markdown("**Predicted targets:** Revenue | Operating Result | Stocks | Cash Flow") demo.launch()