feat: intégration du parsing backend (PDF, Excel, VBA) et gestion sécurisée des champs prompteur/concaténeur vides dans les templates
Build and Push Docker Image / build-and-push (push) Successful in 41s

This commit is contained in:
xavier committed 2026-10-08 10:25:02 +02:00
1 parent c779c51a03
commit 44bf0b868e
4 files changed
+93 -26

No files matched your search

+54 -2
View File
@@ -3,7 +3,7 @@ from fastapi.responses import RedirectResponse, FileResponse
from fastapi.staticfiles import StaticFiles
from sqlalchemy.orm import Session
from typing import List
import os, json, asyncio, shutil
import os, json, asyncio, shutil, base64, io
from datetime import datetime
import pytz
@@ -147,7 +147,59 @@ async def create_message(project_id: int, message: MessageCreate, db: Session =
if message.files:
files_text = ""
for f in message.files:
files_text += f"\n\n[Fichier attaché : {f.name}]\n```\n{f.content}\n```"
content = f.content
if content.startswith("data:"):
try:
header, b64data = content.split(",", 1)
file_bytes = base64.b64decode(b64data)
ext = f.name.split('.')[-1].lower()
extracted_text = ""
if ext == 'pdf':
try:
import PyPDF2
reader = PyPDF2.PdfReader(io.BytesIO(file_bytes))
extracted_text = "\n".join([page.extract_text() for page in reader.pages if page.extract_text()])
except ImportError:
extracted_text = "[Erreur: L'administrateur doit exécuter 'pip install PyPDF2' sur le serveur pour lire les PDF.]"
elif ext in ['xls', 'xlsx', 'xlsm', 'xlsb']:
try:
import openpyxl
wb = openpyxl.load_workbook(io.BytesIO(file_bytes), data_only=False)
for sheet_name in wb.sheetnames:
sheet = wb[sheet_name]
extracted_text += f"\n--- Feuille : {sheet_name} ---\n"
for row in sheet.iter_rows(values_only=True):
row_vals = [str(cell) if cell is not None else "" for cell in row]
if any(row_vals):
extracted_text += "\t".join(row_vals) + "\n"
# Extraction des Macros VBA
if ext in ['xlsm', 'xlsb', 'xls']:
try:
from oletools.olevba import VBA_Parser
vbaparser = VBA_Parser("filename", data=file_bytes)
if vbaparser.detect_vba_macros():
extracted_text += "\n\n--- MACROS VBA DETECTEES ---\n"
for (filename, stream_path, vba_filename, vba_code) in vbaparser.extract_macros():
extracted_text += f"\n// Module: {vba_filename}\n{vba_code}\n"
except ImportError:
extracted_text += "\n[Extraction VBA impossible: L'administrateur doit exécuter 'pip install oletools' sur le serveur.]\n"
except Exception as e:
extracted_text += f"\n[Erreur de lecture VBA interne: {str(e)}]\n"
except ImportError:
extracted_text = "[Erreur: L'administrateur doit exécuter 'pip install openpyxl' sur le serveur pour lire Excel.]"
else:
extracted_text = f"[Fichier binaire non supporté textuellement : {f.name}]"
files_text += f"\n\n[Fichier attaché : {f.name}]\n```text\n{extracted_text}\n```"
except Exception as e:
files_text += f"\n\n[Fichier attaché : {f.name} - ERREUR DE DECODAGE: {str(e)}]"
else:
files_text += f"\n\n[Fichier attaché : {f.name}]\n```\n{content}\n```"
final_content = message.content + files_text
db.add(Message(role=message.role, content=final_content, project_id=project_id))
+9 -5
View File
@@ -16,10 +16,9 @@ def determine_domain(model_id: str) -> str:
return "Texte Polyvalent"
async def translate_en_to_fr(client: httpx.AsyncClient, text: str) -> str:
"""Traduit automatiquement l'anglais vers le français via l'API publique Google Translate."""
if not text: return "Aucune description fournie."
try:
short_text = text[:300].strip() # Limite stricte pour l'API gratuite
short_text = text[:300].strip()
url = f"https://translate.googleapis.com/translate_a/single?client=gtx&sl=en&tl=fr&dt=t&q={urllib.parse.quote(short_text)}"
resp = await client.get(url, timeout=4.0)
if resp.status_code == 200:
@@ -205,7 +204,10 @@ async def run_orchestrator(db: Session, history: list, settings: SystemSettings,
final_response = ""
try:
p_mod = config.get("prompter", "gemini-3.5-flash-lite")
# Fallback sécurisé en cas de champ vide
p_mod = config.get("prompter")
if not p_mod: p_mod = "gemini-3.5-flash-lite"
p_client, p_prov = get_client_for_model(db, p_mod, settings)
prompt_system = "You are an expert prompt engineer. Translate and optimize the user request into clear, precise English tailored for AI execution."
optimized = await ask_agent(p_client, p_mod, [{"role": "system", "content": prompt_system}, {"role": "user", "content": user_prompt}], p_prov)
@@ -222,7 +224,9 @@ async def run_orchestrator(db: Session, history: list, settings: SystemSettings,
w_tasks.append(ask_agent(w_client, w, formatted_history + [{"role": "user", "content": optimized}], w_prov))
responses = await asyncio.gather(*w_tasks, return_exceptions=True)
c_mod = config.get("concatenator", "gemini-3.5-flash-lite")
c_mod = config.get("concatenator")
if not c_mod: c_mod = "gemini-3.5-flash-lite"
c_client, c_prov = get_client_for_model(db, c_mod, settings)
concat_system = (
@@ -233,7 +237,7 @@ async def run_orchestrator(db: Session, history: list, settings: SystemSettings,
"You may translate code comments into French if appropriate, but leave code syntax strictly intact."
)
synth = f"User Request: {user_prompt}\n\n" + "\n".join([f"--- EXPERT {i+1} ---\n{r}" for i, r in enumerate(responses)])
synth = f"User Request: {user_prompt}\n\n" + "\n".join([f"--- EXPERT {i+1} ---\n{str(r)}" for i, r in enumerate(responses)])
final_response = await ask_agent(c_client, c_mod, [{"role": "system", "content": concat_system}, {"role": "user", "content": synth}], c_prov)
except Exception as e: