def reconstruct_url(schema: str, host_parts: list, path: str) -> str | None:
"""Reconstruit une URL à partir de schema, host_parts, et path."""
if not schema or not host_parts or not path: return None
host = ".".join(part for part in host_parts if part)
# S'assurer que le path commence par / s'il n'est pas vide
path = path if path.startswith('/') or not path else '/' + path
return f"{schema}//{host}{path}"
# --- Définitions des autres fonctions utilitaires ---
# (get_cache_filepath, load_from_cache, save_to_cache, encrypt_data, decrypt_data,
# load_extract_definitions (modifiée pour utiliser la nouvelle structure interne),
# save_extract_definitions, fetch_with_jina, fetch_with_tika)
# ... (Copiez ici le code complet de ces fonctions depuis la CELLULE 3 / Bloc Python 3 de ma réponse précédente) ...
# ... Assurez-vous que load_extract_definitions utilise la nouvelle structure
# et que fetch_with_jina/tika appellent la nouvelle reconstruct_url si nécessaire ...
# Exemple de load_extract_definitions (ajusté légèrement pour fallback)
def load_extract_definitions(config_file: Path, key: bytes) -> list:
"""Charge, déchiffre et décompresse les définitions depuis le fichier."""
global EXTRACT_SOURCES, DEFAULT_EXTRACT_SOURCES # Utiliser les structures définies globalement
fallback_definitions = EXTRACT_SOURCES if EXTRACT_SOURCES else DEFAULT_EXTRACT_SOURCES
if not config_file.exists():
print(f"Fichier config '{config_file}' non trouvé. Utilisation définitions en mémoire.")
return fallback_definitions[:]
if not key:
print("Clé chiffrement absente. Chargement config impossible. Utilisation définitions en mémoire.")
return fallback_definitions[:]
print(f"Chargement et déchiffrement de '{config_file}'...")
try:
with open(config_file, 'rb') as f: encrypted_data = f.read()
decrypted_compressed_data = decrypt_data(encrypted_data, key)
if not decrypted_compressed_data: raise ValueError("Échec déchiffrement.")
decompressed_data = gzip.decompress(decrypted_compressed_data)
definitions = json.loads(decompressed_data.decode('utf-8'))
print("✅ Définitions chargées et déchiffrées.")
# Validation plus robuste
if not isinstance(definitions, list) or not all(
isinstance(item, dict) and
"source_name" in item and
"source_type" in item and
"schema" in item and
"host_parts" in item and
"path" in item and
isinstance(item.get("extracts"), list)
for item in definitions
):
print("⚠️ Format définitions invalide après chargement. Utilisation définitions en mémoire.")
return fallback_definitions[:]
# Mettre à jour EXTRACT_SOURCES global si chargement OK
EXTRACT_SOURCES = definitions
print(f"-> {len(EXTRACT_SOURCES)} définitions chargées depuis fichier.")
return definitions # Retourner les définitions chargées
except Exception as e:
print(f"❌ Erreur chargement/traitement '{config_file}': {e}. Utilisation définitions en mémoire.")
return fallback_definitions[:]
# --- Assurez-vous que toutes les autres fonctions utilitaires sont collées ici ---
def get_cache_filepath(url: str) -> Path:
url_hash = hashlib.sha256(url.encode()).hexdigest()
if 'CACHE_DIR' not in globals() or not isinstance(CACHE_DIR, Path): raise NameError("CACHE_DIR non définie.")
return CACHE_DIR / f"{url_hash}.txt"
def load_from_cache(url: str) -> str | None:
filepath = get_cache_filepath(url)
if filepath.exists():
try:
print(f" -> Lecture depuis cache : {filepath.name}")
return filepath.read_text(encoding='utf-8')
except Exception as e: print(f" -> Erreur lecture cache {filepath.name}: {e}"); return None
return None
def save_to_cache(url: str, text: str):
if not text: print(" -> Texte vide, non sauvegardé."); return
filepath = get_cache_filepath(url)
try:
filepath.write_text(text, encoding='utf-8'); print(f" -> Texte sauvegardé : {filepath.name}")
except Exception as e: print(f" -> Erreur sauvegarde cache {filepath.name}: {e}")
def encrypt_data(data: bytes, key: bytes) -> bytes | None:
if not key: print("Erreur: Clé chiffrement manquante."); return None
try: f = Fernet(key); return f.encrypt(data)
except Exception as e: print(f"Erreur chiffrement: {e}"); return None
def decrypt_data(encrypted_data: bytes, key: bytes) -> bytes | None:
if not key: print("Erreur: Clé chiffrement manquante."); return None
try: f = Fernet(key); return f.decrypt(encrypted_data)
except (InvalidToken, InvalidSignature, Exception) as e: print(f"Erreur déchiffrement: {e}"); return None
def save_extract_definitions(definitions: list, config_file: Path, key: bytes):
if not key: print("Clé chiffrement absente. Sauvegarde annulée."); return False
if not isinstance(definitions, list): print("Erreur: définitions non valides."); return False
print(f"Préparation sauvegarde vers '{config_file}'...")
try:
json_data = json.dumps(definitions, indent=2, ensure_ascii=False).encode('utf-8'); compressed_data = gzip.compress(json_data); encrypted_data = encrypt_data(compressed_data, key)
if not encrypted_data: raise ValueError("Échec chiffrement.")
with open(config_file, 'wb') as f: f.write(encrypted_data)
print(f"✅ Définitions sauvegardées dans '{config_file}'.")
return True
except Exception as e: print(f"❌ Erreur sauvegarde chiffrée: {e}"); return False
PLAINTEXT_EXTENSIONS = ['.txt', '.md', '.json', '.csv', '.xml', '.py', '.js', '.html', '.htm'] # Extensions à traiter comme texte simple
def fetch_direct_text(source_url, timeout=60):
"""Récupère contenu texte brut d'URL, utilise cache fichier."""
# ... (Code INCHANGÉ) ...
cached_text = load_from_cache(source_url)
if cached_text is not None: return cached_text
print(f"-> Téléchargement direct depuis : {source_url}...")
headers = {'User-Agent': 'Agent-Analysis-Notebook/1.0'}
try:
response = requests.get(source_url, headers=headers, timeout=timeout)
response.raise_for_status()
# Décoder en UTF-8, ignorer les erreurs potentielles
texte_brut = response.content.decode('utf-8', errors='ignore')
print(f" -> Contenu direct récupéré (longueur {len(texte_brut)}).")
save_to_cache(source_url, texte_brut)
return texte_brut
except requests.exceptions.RequestException as e:
raise ConnectionError(f"Erreur téléchargement direct ({source_url}): {e}") from e
def fetch_with_jina(source_url, timeout=90):
"""Récupère et extrait via Jina, utilise cache fichier."""
# ... (Code INCHANGÉ) ...
cached_text = load_from_cache(source_url);
if cached_text is not None: return cached_text
jina_url = f"{JINA_READER_PREFIX}{source_url}"; print(f"-> Récupération via Jina : {jina_url}...")
headers = {'Accept': 'text/markdown', 'User-Agent': 'Agent-Analysis-Notebook/1.0'}
try: response = requests.get(jina_url, headers=headers, timeout=timeout); response.raise_for_status()
except requests.exceptions.RequestException as e: raise ConnectionError(f"Erreur Jina ({jina_url}): {e}") from e
content = response.text; md_start_marker = "Markdown Content:"; md_start_index = content.find(md_start_marker)
texte_brut = content[md_start_index + len(md_start_marker):].strip() if md_start_index != -1 else content
print(f" -> Contenu Jina récupéré (longueur {len(texte_brut)})."); save_to_cache(source_url, texte_brut); return texte_brut
# --- fetch_with_tika MODIFIÉE ---
def fetch_with_tika(source_url=None, file_content=None, file_name="fichier",
raw_file_cache_path: Path | str | None = None, # Chemin cache brut
timeout_dl=60, timeout_tika=600):
"""
Traite une source via Tika. Tente d'abord de lire le cache texte.
Si URL fournie, vérifie le cache brut puis télécharge si besoin.
Vérifie l'extension: si texte simple, utilise fetch_direct_text au lieu de Tika.
Sinon, envoie le contenu brut (téléchargé ou fourni) à Tika.
"""
cache_key = source_url if source_url else f"file://{file_name}"
# 1. Vérifier cache TEXTE FINAL
cached_text = load_from_cache(cache_key)
if cached_text is not None: return cached_text
global TIKA_SERVER_URL
content_to_send = None
original_filename_or_path = file_name # Par défaut (pour upload)
# 2. Obtenir contenu BRUT (si URL) et vérifier extension
if source_url:
original_filename_or_path = Path(source_url).name # Pour check extension et cache brut
# Vérifier si c'est un type texte simple connu basé sur l'URL
if any(source_url.lower().endswith(ext) for ext in PLAINTEXT_EXTENSIONS):
print(f" -> URL détectée comme texte simple ({source_url}). Utilisation fetch direct au lieu de Tika.")
# On ne passe pas par Tika, on télécharge directement le texte
# La fonction fetch_direct_text gère son propre cache .txt
return fetch_direct_text(source_url)
# Si pas texte simple, gérer cache brut et téléchargement
if raw_file_cache_path:
raw_path = Path(raw_file_cache_path)
if raw_path.exists() and raw_path.stat().st_size > 0:
try:
print(f" -> Lecture fichier brut depuis cache local : {raw_path.name}")
content_to_send = raw_path.read_bytes()
except Exception as e_read_raw:
print(f" -> ⚠️ Erreur lecture cache brut {raw_path.name}: {e_read_raw}. Re-téléchargement...")
content_to_send = None
if content_to_send is None: # Si cache brut absent ou erreur lecture
print(f"-> Téléchargement (pour Tika) depuis : {source_url}...")
try:
response_dl = requests.get(source_url, stream=True, timeout=timeout_dl); response_dl.raise_for_status()
content_to_send = response_dl.content; print(f" -> Doc téléchargé ({len(content_to_send)} bytes).")
if raw_file_cache_path: # Sauvegarder si chemin fourni
try: save_path = Path(raw_file_cache_path); save_path.parent.mkdir(parents=True, exist_ok=True); save_path.write_bytes(content_to_send); print(f" -> Doc brut sauvegardé: {save_path.resolve()}")
except Exception as e_save: print(f" -> ⚠️ Erreur sauvegarde brut: {e_save}")
except requests.exceptions.RequestException as e: raise ConnectionError(f"Erreur téléchargement {source_url}: {e}") from e
elif file_content:
# Cas: Fichier uploadé (pas d'URL source)
print(f"-> Utilisation contenu fichier '{file_name}' ({len(file_content)} bytes)...")
content_to_send = file_content
# Vérifier si upload est texte simple
if any(file_name.lower().endswith(ext) for ext in PLAINTEXT_EXTENSIONS):
print(" -> Fichier uploadé détecté comme texte simple. Lecture directe.")
try:
texte_brut = file_content.decode('utf-8', errors='ignore')
save_to_cache(cache_key, texte_brut) # Sauver dans cache texte
return texte_brut
except Exception as e_decode:
print(f" -> ⚠️ Erreur décodage fichier texte '{file_name}': {e_decode}. Tentative avec Tika...")
# Si erreur décodage, on laisse Tika essayer
else:
raise ValueError("fetch_with_tika: Il faut soit source_url soit file_content.")
# 3. Envoyer à Tika (seulement si nécessaire et contenu valide)
if not content_to_send:
print(" -> ⚠️ Contenu brut vide ou non récupéré. Impossible d'envoyer à Tika.")
save_to_cache(cache_key, "") # Sauver cache vide pour éviter retry
return ""
print(f"-> Envoi contenu à Tika ({TIKA_SERVER_URL})... (Timeout={timeout_tika}s)")
headers = { 'Accept': 'text/plain', 'Content-Type': 'application/octet-stream', 'X-Tika-OCRLanguage': 'fra+eng' }
try:
response_tika = requests.put(TIKA_SERVER_URL, data=content_to_send, headers=headers, timeout=timeout_tika)
response_tika.raise_for_status()
texte_brut = response_tika.text
if not texte_brut: print(f" -> Warning: Tika status {response_tika.status_code} sans texte.")
else: print(f" -> Texte Tika extrait (longueur {len(texte_brut)}).")
# 4. Sauvegarder le TEXTE EXTRAIT dans le cache .txt
save_to_cache(cache_key, texte_brut)
return texte_brut
except requests.exceptions.Timeout: print(f" -> ❌ Timeout Tika ({timeout_tika}s)."); raise ConnectionError(f"Timeout Tika")
except requests.exceptions.RequestException as e: raise ConnectionError(f"Erreur Tika: {e}") from e
def verify_extract_definitions(definitions_list: list):
"""
Vérifie la présence des marqueurs start/end pour chaque extrait défini.
Récupère le texte complet (via cache ou fetch) pour chaque source.
Retourne un résumé des vérifications.
"""
print("\n🔬 Lancement de la vérification des marqueurs d'extraits...")
results = []
total_checks = 0
total_errors = 0
if not definitions_list or definitions_list == DEFAULT_EXTRACT_SOURCES:
return "Aucune définition valide à vérifier."
for source_idx, source_info in enumerate(definitions_list):
source_name = source_info.get("source_name", f"Source Inconnue #{source_idx+1}")
print(f"\n--- Vérification Source: '{source_name}' ---")
source_errors = 0
source_checks = 0
texte_brut_source = None
reconstructed_url = None
try:
# Reconstruire l'URL
reconstructed_url = reconstruct_url(
source_info.get("schema"), source_info.get("host_parts", []), source_info.get("path")
)
if not reconstructed_url:
print(" -> ❌ Erreur: URL Invalide.")
results.append(f"<li>{source_name}: URL invalide</li>")
total_errors += len(source_info.get("extracts", [])) # Compter tous les extraits comme erreurs
total_checks += len(source_info.get("extracts", []))
continue
# Récupérer le texte complet (via cache ou fetch)
source_type = source_info.get("source_type")
cache_key = reconstructed_url # Clé cache pour le texte complet
texte_brut_source = load_from_cache(cache_key)
if texte_brut_source is None:
print(f" -> Cache texte absent. Récupération (type: {source_type})...")
if source_type == "jina": texte_brut_source = fetch_with_jina(reconstructed_url)
elif source_type == "direct_download": texte_brut_source = fetch_direct_text(reconstructed_url)
elif source_type == "tika":
is_plaintext = any(source_info.get("path", "").lower().endswith(ext) for ext in PLAINTEXT_EXTENSIONS)
if is_plaintext: texte_brut_source = fetch_direct_text(reconstructed_url)
else:
# On ne peut pas raisonnablement vérifier les marqueurs si Tika échoue ici
print(" -> ⚠️ Impossible de vérifier les marqueurs car nécessite Tika (potentiellement long/échoué).")
texte_brut_source = None # Marquer comme non vérifiable
else:
print(f" -> ⚠️ Type source inconnu '{source_type}'. Vérification impossible.")
texte_brut_source = None
# Vérifier les marqueurs si le texte a été obtenu
if texte_brut_source is not None:
print(f" -> Texte complet récupéré (longueur: {len(texte_brut_source)}). Vérification des extraits...")
extracts = source_info.get("extracts", [])
if not extracts: print(" -> Aucun extrait défini pour cette source.")
for extract_idx, extract_info in enumerate(extracts):
extract_name = extract_info.get("extract_name", f"Extrait #{extract_idx+1}")
start_marker = extract_info.get("start_marker")
end_marker = extract_info.get("end_marker")
total_checks += 1
source_checks += 1
marker_errors = []
if not start_marker or not end_marker:
marker_errors.append("Marqueur(s) Manquant(s)")
else:
start_found = start_marker in texte_brut_source
end_found_after_start = False
if start_found:
try:
start_pos = texte_brut_source.index(start_marker)
end_found_after_start = end_marker in texte_brut_source[start_pos + len(start_marker):]
except ValueError: # Ne devrait pas arriver si start_found est True
start_found = False
if not start_found: marker_errors.append("Début NON TROUVÉ")
if not end_found_after_start: marker_errors.append("Fin NON TROUVÉE (après début)")
if marker_errors:
print(f" -> ❌ Problème Extrait '{extract_name}': {', '.join(marker_errors)}")
results.append(f"<li>{source_name} -> {extract_name}: <strong style='color:red;'>{', '.join(marker_errors)}</strong></li>")
source_errors += 1
total_errors += 1
else:
print(f" -> ✅ OK: Extrait '{extract_name}'")
# results.append(f"<li>{source_name} -> {extract_name}: OK</li>") # Optionnel: lister aussi les OK
else: # Cas où le texte brut n'a pas pu être récupéré (ex: erreur Tika pendant la vérif)
num_extracts = len(source_info.get("extracts",[]))
results.append(f"<li>{source_name}: Vérification impossible (texte source non obtenu)</li>")
total_errors += num_extracts
total_checks += num_extracts
except Exception as e_verify:
print(f" -> ❌ Erreur inattendue vérification source '{source_name}': {e_verify}")
num_extracts = len(source_info.get("extracts",[]))
results.append(f"<li>{source_name}: Erreur Vérification Générale ({type(e_verify).__name__})</li>")
total_errors += num_extracts
total_checks += num_extracts # Compter comme échec si erreur globale
summary = f"--- Résultat Vérification ---<br/>{total_checks} extraits vérifiés. <strong style='color: {'red' if total_errors > 0 else 'green'};'>{total_errors} erreur(s) trouvée(s).</strong>"
if results:
summary += "<br/>Détails erreurs :<ul>" + "".join(results) + "</ul>"
else:
summary += "<br/>Tous les marqueurs semblent corrects."
# Fix f-string backslash error - extract HTML cleanup outside f-string
console_summary = summary.replace('<br/>', '\n').replace('<li>', '- ').replace('</li>', '').replace('<ul>', '').replace('</ul>', '').replace('<strong>', '').replace('</strong>', '')
print(f"\n{console_summary}") # Affichage console simple
return summary
print("Fonctions utilitaires UI_Configuration définies.")