#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Module pour extraire les images intégrées dans les messages HTML d'Odoo.
Complémentaire à l'extracteur d'images actuel, il détecte spécifiquement les images
référencées dans le HTML avec des balises
et les associe aux pièces jointes.
"""
import os
import re
import logging
import json
from typing import Dict, List, Any, Optional, Set, Tuple
from bs4 import BeautifulSoup
class HtmlImageExtractor:
"""
Extracteur d'images intégrées dans les messages HTML d'Odoo.
"""
def __init__(self, ticket_dir: str):
"""
Initialise l'extracteur d'images HTML.
Args:
ticket_dir: Répertoire du ticket contenant les messages et pièces jointes
"""
self.ticket_dir = ticket_dir
self.messages_file = os.path.join(ticket_dir, "messages_raw.json")
self.attachments_file = os.path.join(ticket_dir, "attachments_info.json")
self.output_file = os.path.join(ticket_dir, "embedded_images.json")
# Cache pour les données
self._messages = None
self._attachments = None
def _load_messages(self) -> List[Dict[str, Any]]:
"""
Charge les messages du ticket.
Returns:
Liste des messages
"""
if self._messages is not None:
return self._messages
# Essayer différents fichiers potentiels contenant les messages
possible_message_files = [
self.messages_file, # messages_raw.json
os.path.join(self.ticket_dir, "messages.json"),
os.path.join(self.ticket_dir, "all_messages.json")
]
for msg_file in possible_message_files:
if os.path.exists(msg_file):
try:
with open(msg_file, 'r', encoding='utf-8') as f:
data = json.load(f)
# Gérer différents formats de fichiers de messages
if isinstance(data, dict):
if "messages" in data:
self._messages = data["messages"]
return self._messages
# Si pas de clé messages mais d'autres clés, essayer de trouver les messages
for key, value in data.items():
if isinstance(value, list) and len(value) > 0:
# Vérifier si ça ressemble à des messages (ont une clé body ou content)
if isinstance(value[0], dict) and any(k in value[0] for k in ["body", "content"]):
self._messages = value
return self._messages
# Si c'est directement une liste, supposer que ce sont des messages
elif isinstance(data, list):
self._messages = data
return self._messages
except Exception as e:
logging.error(f"Erreur lors du chargement du fichier {msg_file}: {e}")
# Si on arrive ici, aucun fichier valide n'a été trouvé
logging.error(f"Aucun fichier de messages valide trouvé dans: {self.ticket_dir}")
return []
def _load_attachments(self) -> Dict[int, Dict[str, Any]]:
"""
Charge les pièces jointes du ticket.
Returns:
Dictionnaire des pièces jointes indexées par ID
"""
if self._attachments is not None:
return self._attachments
# Essayer différents fichiers potentiels de pièces jointes
possible_attachment_files = [
self.attachments_file, # attachments_info.json
os.path.join(self.ticket_dir, "attachments.json")
]
for att_file in possible_attachment_files:
if os.path.exists(att_file):
try:
with open(att_file, 'r', encoding='utf-8') as f:
attachments = json.load(f)
if not isinstance(attachments, list):
# Si ce n'est pas une liste mais un dict, essayer de trouver une liste
for key, value in attachments.items():
if isinstance(value, list) and len(value) > 0:
attachments = value
break
if not isinstance(attachments, list):
continue
# Indexer par ID pour un accès rapide
self._attachments = {}
for att in attachments:
if "id" in att:
self._attachments[att["id"]] = att
elif "attachment_id" in att:
self._attachments[att["attachment_id"]] = att
if self._attachments:
return self._attachments
except Exception as e:
logging.error(f"Erreur lors du chargement des pièces jointes depuis {att_file}: {e}")
# Si on arrive ici, aucun fichier valide n'a été trouvé
logging.error(f"Aucun fichier de pièces jointes valide trouvé dans: {self.ticket_dir}")
return {}
def _get_tag_attribute(self, tag: Any, attr_name: str, default: str = "") -> str:
"""
Récupère un attribut d'une balise HTML de manière sécurisée.
Args:
tag: Balise HTML
attr_name: Nom de l'attribut à récupérer
default: Valeur par défaut si l'attribut n'existe pas
Returns:
Valeur de l'attribut
"""
# Vérifier que c'est bien une balise et qu'elle a la méthode get
if not hasattr(tag, 'get'):
return default
# Récupérer l'attribut si présent
try:
value = tag.get(attr_name)
return str(value) if value is not None else default
except (AttributeError, KeyError, TypeError):
return default
def extract_image_references(self) -> Dict[str, Any]:
"""
Extrait les références aux images dans le HTML des messages.
Returns:
Dictionnaire contenant les références d'images trouvées
"""
messages = self._load_messages()
attachments_by_id = self._load_attachments()
if not messages:
logging.error("Aucun message trouvé pour l'extraction d'images")
return {"status": "error", "message": "Aucun message trouvé", "references": []}
# Stocker les références trouvées
image_references = []
# Ensemble pour dédupliquer
processed_ids = set()
for message in messages:
# Déterminer quel champ contient le contenu HTML
message_body = None
message_id = None
# Différentes possibilités de noms pour le contenu HTML et l'ID
for body_key in ["body", "content", "html_content", "message"]:
if body_key in message and message[body_key]:
message_body = message[body_key]
break
for id_key in ["id", "message_id", "uid"]:
if id_key in message:
message_id = message[id_key]
break
if not message_body or not message_id:
continue
# Méthode 1: Extraction depuis les balises
dans le HTML
try:
# Analyser le HTML du message
soup = BeautifulSoup(message_body, "html.parser")
# Trouver toutes les balises
img_tags = soup.find_all("img")
for img in img_tags:
# Utiliser la méthode sécurisée pour récupérer les attributs
src = self._get_tag_attribute(img, "src")
# Ignorer les images vides ou data URLs
if not src or src.startswith("data:"):
continue
# Différents patterns de référence d'images
image_id = None
# Pattern 1: /web/image/ID?access_token=...
match = re.search(r"/web/image/(\d+)", src)
if match:
image_id = int(match.group(1))
# Pattern 2: /web/content/ID?...
if not image_id:
match = re.search(r"/web/content/(\d+)", src)
if match:
image_id = int(match.group(1))
# Pattern 3: /web/static/ID?...
if not image_id:
match = re.search(r"/web/static/(\d+)", src)
if match:
image_id = int(match.group(1))
# Pattern 4: /web/binary/image?id=ID&...
if not image_id:
match = re.search(r"[?&]id=(\d+)", src)
if match:
image_id = int(match.group(1))
if image_id:
self._ajouter_reference_image(image_id, message_id, attachments_by_id, img, processed_ids, image_references)
except Exception as e:
logging.error(f"Erreur lors de l'analyse des balises
dans le message {message_id}: {e}")
# Méthode 2: Extraction depuis les références textuelles à la fin du message
# Format: "- Nom_image.ext (image/type) [ID: 12345]"
try:
# Chercher des références d'images dans le texte du message
# Pattern des références: "- Nom_fichier.ext (type/mime) [ID: 12345]"
img_ref_pattern = r"- ([^\(\)]+) \(([^\(\)]*)\) \[ID: (\d+)\]"
for match in re.finditer(img_ref_pattern, message_body):
try:
nom_fichier = match.group(1).strip()
mimetype = match.group(2).strip()
image_id = int(match.group(3))
if image_id in processed_ids:
continue
processed_ids.add(image_id)
# Vérifier si cette image est dans nos pièces jointes
if image_id in attachments_by_id:
attachment = attachments_by_id[image_id]
# Récupérer le chemin local de l'image
local_path = attachment.get("local_path")
# Vérifier que le fichier existe réellement
if local_path and not os.path.exists(local_path):
# Essayer de chercher dans d'autres endroits potentiels du répertoire
for root, dirs, files in os.walk(self.ticket_dir):
for file in files:
if os.path.basename(local_path) == file:
local_path = os.path.join(root, file)
break
if os.path.exists(local_path):
break
image_ref = {
"image_id": image_id,
"message_id": message_id,
"attachment_id": attachment.get("id", attachment.get("attachment_id", 0)),
"name": nom_fichier, # Utiliser le nom trouvé dans le texte
"mimetype": mimetype, # Utiliser le type MIME trouvé dans le texte
"file_size": attachment.get("file_size", 0),
"local_path": local_path,
"source": "text_reference"
}
image_references.append(image_ref)
logging.info(f"Image référencée trouvée dans le texte: ID {image_id} dans message {message_id}")
except Exception as e:
logging.error(f"Erreur lors du traitement de la référence d'image textuelle: {e}")
except Exception as e:
logging.error(f"Erreur lors de l'analyse des références textuelles dans le message {message_id}: {e}")
# Sauvegarder les références trouvées
result = {
"status": "success" if image_references else "warning",
"message": f"Trouvé {len(image_references)} références d'images intégrées",
"references": image_references
}
try:
with open(self.output_file, 'w', encoding='utf-8') as f:
json.dump(result, f, indent=2, ensure_ascii=False)
logging.info(f"Références d'images sauvegardées dans: {self.output_file}")
except Exception as e:
logging.error(f"Erreur lors de la sauvegarde des références d'images: {e}")
return result
def _ajouter_reference_image(self, image_id: int, message_id: str, attachments_by_id: Dict[int, Dict[str, Any]],
img: Any, processed_ids: Set[int], image_references: List[Dict[str, Any]]) -> None:
"""
Ajoute une référence d'image à la liste des références
Args:
image_id: ID de l'image
message_id: ID du message
attachments_by_id: Dictionnaire des pièces jointes indexées par ID
img: Balise HTML de l'image
processed_ids: Ensemble des IDs déjà traités
image_references: Liste des références d'images à enrichir
"""
# Éviter les duplications
if image_id in processed_ids:
return
processed_ids.add(image_id)
# Vérifier si cette image est dans nos pièces jointes
if image_id in attachments_by_id:
attachment = attachments_by_id[image_id]
# Récupérer les attributs de manière sécurisée
width = self._get_tag_attribute(img, "width")
height = self._get_tag_attribute(img, "height")
alt = self._get_tag_attribute(img, "alt")
# Récupérer le chemin local de l'image
local_path = attachment.get("local_path")
# Vérifier que le fichier existe réellement
if local_path and not os.path.exists(local_path):
# Essayer de chercher dans d'autres endroits potentiels du répertoire
for root, dirs, files in os.walk(self.ticket_dir):
for file in files:
if os.path.basename(local_path) == file:
local_path = os.path.join(root, file)
break
if os.path.exists(local_path):
break
image_ref = {
"image_id": image_id,
"message_id": message_id,
"attachment_id": attachment.get("id", attachment.get("attachment_id", 0)),
"name": attachment.get("name", ""),
"mimetype": attachment.get("mimetype", ""),
"file_size": attachment.get("file_size", 0),
"local_path": local_path,
"img_width": width,
"img_height": height,
"img_alt": alt,
"source": "html_img_tag"
}
image_references.append(image_ref)
logging.info(f"Image intégrée trouvée: ID {image_id} dans message {message_id}")
def get_image_paths(self) -> List[str]:
"""
Récupère les chemins des images référencées.
Returns:
Liste des chemins locaux des images référencées
"""
try:
# Tenter d'extraire les images
data = self.extract_image_references()
# Récupérer les chemins locaux des images
paths = []
for ref in data.get("references", []):
path = ref.get("local_path")
if path and os.path.exists(path):
paths.append(path)
if not paths:
logging.warning("Aucune image intégrée trouvée ou les chemins sont invalides")
return paths
except Exception as e:
logging.error(f"Erreur lors de la récupération des chemins d'images: {e}")
return []
def extract_images_from_ticket(ticket_dir: str) -> List[str]:
"""
Fonction utilitaire pour extraire les images intégrées dans les messages HTML d'un ticket.
Args:
ticket_dir: Répertoire du ticket contenant les messages et pièces jointes
Returns:
Liste des chemins locaux des images référencées
"""
extractor = HtmlImageExtractor(ticket_dir)
return extractor.get_image_paths()
if __name__ == "__main__":
# Test avec un répertoire de ticket spécifique
import sys
if len(sys.argv) > 1:
ticket_dir = sys.argv[1]
else:
# Utiliser un répertoire de test par défaut
ticket_dir = "./output/ticket_T0241/T0241_20250409_141018"
if not os.path.exists(ticket_dir):
print(f"Répertoire introuvable: {ticket_dir}")
sys.exit(1)
print(f"Extraction des images intégrées dans le HTML pour le ticket: {os.path.basename(ticket_dir)}")
extractor = HtmlImageExtractor(ticket_dir)
result = extractor.extract_image_references()
print(f"Statut: {result['status']}")
print(f"Message: {result['message']}")
print(f"Nombre de références trouvées: {len(result['references'])}")
for i, ref in enumerate(result["references"]):
print(f"\nImage {i+1}:")
print(f" ID: {ref['image_id']}")
print(f" Nom: {ref['name']}")
print(f" Type: {ref['mimetype']}")
print(f" Taille: {ref['file_size']} octets")
print(f" Chemin local: {ref['local_path']}")
# Récupérer les chemins des images
paths = extractor.get_image_paths()
print(f"\nChemins des images ({len(paths)}):")
for path in paths:
print(f" {path}")