commit b66b065da1624c3d232c6fc113758b671dc89569 Author: skylanix Date: Sun Jun 28 20:21:40 2026 +0200 first commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b697ea9 --- /dev/null +++ b/.gitignore @@ -0,0 +1,35 @@ +# Environnements virtuels +venv/ +.venv/ +env/ + +# Caches Python +__pycache__/ +*.py[cod] +*$py.class +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ + +# Distribution / build +build/ +dist/ +*.egg-info/ + +# Sorties générées +captures/ok/ +report*.html +tweets.csv + +# Modèles téléchargés (EasyOCR / Hugging Face) +.EasyOCR/ +.cache/ + +# Fichiers d'environnement / secrets +.env + +# Spécifiques aux éditeurs / OS +.vscode/ +.idea/ +.DS_Store +Thumbs.db diff --git a/.~lock.README.md# b/.~lock.README.md# new file mode 100644 index 0000000..cc28f61 --- /dev/null +++ b/.~lock.README.md# @@ -0,0 +1 @@ +,sky,skylanix,28.06.2026 18:20,/home/sky/.local/share/onlyoffice; \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..7bd9424 --- /dev/null +++ b/README.md @@ -0,0 +1,126 @@ +# Classement Automatique de Captures d'Écran de Tweets par Harcèlement + +Ce projet permet d'automatiser le classement de captures d'écran de tweets, en analysant leur contenu textuel pour détecter des types de harcèlement en français. + +Il utilise **EasyOCR** pour extraire le texte des images et un modèle d'apprentissage profond Zero-Shot multilingue de **Hugging Face** (par défaut `MoritzLaurer/mDeBERTa-v3-base-xnli-multilingual-nli-2mil7`) pour classer le texte extrait. Le système s'appuie sur une base de données **CSV** pour suivre l'état de traitement de chaque fichier. + +--- + +## Fonctionnalités + +- 🔎 **Extraction de texte (OCR)** : Reconnaissance optique de caractères robuste et multilingue grâce à EasyOCR. +- 🏷️ **Classification Zero-Shot** : Classification automatique de texte en catégories de harcèlement (*Cyberharcèlement*, *Insulte*, *Menace*, *Non-harcèlement*). +- 💾 **Base de données CSV** : Suivi persistant de chaque fichier (`pending`, `processed`, `error`). +- 📁 **Organisation automatique** : Déplacement (ou copie) des images dans des dossiers `ok/{catégorie}/`. +- 📊 **Rapport Web** : Génération d'un rapport HTML interactif (filtres, recherche, tri, lightbox) pour consulter les résultats. +- 🧪 **Mode simulation** : Option `--dry-run` pour prévisualiser le classement sans déplacer de fichiers. +- 💻 **CLI complète** : Contrôle total via la ligne de commande. + +--- + +## Catégories de classification + +Chaque image se voit attribuer l'une des catégories suivantes : + +| Catégorie | Origine | +|---|---| +| `Cyberharcèlement`, `Insulte`, `Menace`, `Non-harcèlement` | Labels Zero-Shot retenus lorsque le score dépasse le seuil. | +| `Inclassable` | Le meilleur score est en dessous du seuil de confiance (`--threshold`). | +| `Sans_Texte` | Aucun texte n'a pu être extrait de l'image. | +| `Non-classifié` | Valeur de repli en cas de résultat inattendu. | + +--- + +## Structure du Projet + +```text +classement-image/ +├── requirements.txt # Dépendances du projet +├── main.py # Point d'entrée principal +├── captures/ # Dossier d'entrée des captures à traiter +│ └── ok/ # Sortie par défaut : images classées + CSV + rapports +│ ├── tweets.csv # Base de données de suivi (générée automatiquement) +│ ├── report_*.html # Rapports visuels horodatés (--generate-report) +│ └── {catégorie}/ # Images classées par catégorie +├── tweet_classifier/ # Package principal +│ ├── ocr.py # Extraction de texte (EasyOCR) +│ ├── classifier.py # Classification Zero-Shot (Hugging Face) +│ ├── organizer.py # Orchestrateur du traitement +│ ├── database_manager.py # Gestionnaire de la base CSV +│ ├── web_generator.py # Générateur de rapport HTML (Jinja2) +│ └── cli.py # Interface en ligne de commande +└── tests/ # Suite de tests unitaires +``` + +--- + +## Installation + +Installez les dépendances : + +```bash +pip install -r requirements.txt +``` + +> ℹ️ Au premier lancement, EasyOCR et Hugging Face téléchargent automatiquement +> leurs modèles (quelques centaines de Mo). Un GPU est utilisé s'il est +> disponible, sinon le traitement s'effectue sur CPU. + +--- + +## Utilisation + +Le script `main.py` gère le traitement des images et la génération de rapports. + +### 1. Traitement des images + +Pour traiter les images du dossier `captures/` : + +```bash +python main.py -i ./captures +``` + +Les images sont classées dans des sous-dossiers de `./captures/ok/` (par défaut) +selon la catégorie détectée, et leur état est mis à jour dans +`captures/ok/tweets.csv`. + +Pour choisir un autre dossier de destination : + +```bash +python main.py -i ./captures -o ./resultats +``` + +### 2. Génération du rapport Web + +Une fois le traitement terminé, générez un rapport HTML depuis la base de données : + +```bash +python main.py --generate-report +``` + +Un fichier `report_{horodatage}.html` est créé dans le dossier de sortie +(`captures/ok/` par défaut). Ouvrez-le dans votre navigateur pour consulter la +galerie des tweets classés avec leurs images, textes et catégories détectées. + +### Options de la ligne de commande + +| Option | Raccourci | Description | Défaut | +|---|---|---|---| +| `--input-dir` | `-i` | Dossier contenant les captures à classer. | *Requis pour le traitement* | +| `--output-dir` | `-o` | Dossier de destination des images classées. | `{input-dir}/ok` | +| `--db` | *Aucun* | Chemin du fichier base de données CSV. | `captures/ok/tweets.csv` | +| `--generate-report` | *Aucun* | Génère un rapport HTML depuis le CSV (n'effectue pas de traitement). | `False` | +| `--model` | `-m` | Modèle Hugging Face Zero-Shot à utiliser. | `MoritzLaurer/mDeBERTa-v3-base-xnli-multilingual-nli-2mil7` | +| `--languages` | `-l` | Langues EasyOCR, séparées par des virgules. | `fr,en` | +| `--threshold` | `-t` | Seuil de confiance de classification (0.0 à 1.0). | `0.35` | +| `--copy` | *Aucun* | Copie les images au lieu de les déplacer. | `False` | +| `--dry-run` | *Aucun* | Simulation : ne modifie ni ne déplace les fichiers. | `False` | +| `--verbose` | `-v` | Active les logs de débogage. | `False` | + +--- + +## Exécuter les Tests + +```bash +python3 -m pytest tests/ +``` diff --git a/main.py b/main.py new file mode 100644 index 0000000..616eef2 --- /dev/null +++ b/main.py @@ -0,0 +1,8 @@ +#!/usr/bin/env python3 +""" +Point d'entrée principal du script de classement automatique de captures d'écran de tweets. +""" +from tweet_classifier.cli import main + +if __name__ == "__main__": + main() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..f29ae0d --- /dev/null +++ b/requirements.txt @@ -0,0 +1,9 @@ +# Dépendances pour le classement automatique des captures de tweets +easyocr>=1.7.1 +transformers>=4.38.0 +torch>=2.0.0 +torchvision>=0.15.0 +pillow>=10.0.0 +tqdm>=4.66.0 +jinja2>=3.1.0 +pytest>=8.0.0 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..f2514c8 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +# Package de tests unitaires diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..f75d56f --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,8 @@ +import sys +from unittest.mock import MagicMock + +# Mock global pour éviter le besoin d'installer des paquets lourds en local pour les tests unitaires +sys.modules['easyocr'] = MagicMock() +sys.modules['transformers'] = MagicMock() +sys.modules['torch'] = MagicMock() +sys.modules['torchvision'] = MagicMock() diff --git a/tests/test_organizer.py b/tests/test_organizer.py new file mode 100644 index 0000000..33cd542 --- /dev/null +++ b/tests/test_organizer.py @@ -0,0 +1,83 @@ +import pytest +from pathlib import Path +from unittest.mock import MagicMock, patch +from tweet_classifier.organizer import ImageThemeOrganizer + +@pytest.fixture +def temp_workspace(tmp_path): + """ + Crée une structure temporaire pour les tests. + """ + input_dir = tmp_path / "inputs" + input_dir.mkdir() + + # Créer quelques fichiers d'images factices + (input_dir / "tweet1.png").write_text("fake image content 1") + (input_dir / "tweet2.jpg").write_text("fake image content 2") + (input_dir / "not_an_image.txt").write_text("some text") + + return input_dir + +@patch("tweet_classifier.organizer.CSVDatabaseManager") +@patch("tweet_classifier.ocr.easyocr.Reader") +@patch("tweet_classifier.classifier.pipeline") +def test_scan_new_files(mock_pipeline, mock_reader, mock_db_class, temp_workspace): + """ + Vérifie que l'organisateur scanne les fichiers et les ajoute à la base de données. + """ + mock_db = MagicMock() + mock_db_class.return_value = mock_db + + organizer = ImageThemeOrganizer(input_dir=temp_workspace) + + organizer.scan_new_files() + + # Vérifier que add_files a été appelé avec les bonnes images + assert mock_db.add_files.called + added_files = mock_db.add_files.call_args[0][0] + assert len(added_files) == 2 + assert any(img.name == "tweet1.png" for img in added_files) + assert any(img.name == "tweet2.jpg" for img in added_files) + +@patch("tweet_classifier.organizer.CSVDatabaseManager") +@patch("tweet_classifier.ocr.easyocr.Reader") +@patch("tweet_classifier.classifier.pipeline") +def test_process_record_success(mock_pipeline, mock_reader, mock_db_class, temp_workspace): + """ + Vérifie le traitement réussi d'un enregistrement DB. + """ + mock_db = MagicMock() + mock_db_class.return_value = mock_db + + mock_reader_instance = MagicMock() + mock_reader.return_value = mock_reader_instance + mock_reader_instance.readtext.return_value = [ + (None, "Ceci est un tweet agressif", None) + ] + + mock_pipeline_instance = MagicMock() + mock_pipeline.return_value = mock_pipeline_instance + mock_pipeline_instance.return_value = { + "labels": ["Insulte"], + "scores": [0.9] + } + + organizer = ImageThemeOrganizer(input_dir=temp_workspace) + + image_path = temp_workspace / "tweet1.png" + record = { + "filename": "tweet1.png", + "filepath": str(image_path) + } + + organizer._process_record(record) + + # Vérifier le déplacement + dest_path = temp_workspace / "ok" / "Insulte" / "tweet1.png" + assert dest_path.exists() + + # Vérifier la mise à jour DB : le nouvel emplacement doit être enregistré + mock_db.update_file_status.assert_called_with( + "tweet1.png", 'processed', "Ceci est un tweet agressif", "Insulte", 0.9, + new_filepath=str(dest_path) + ) diff --git a/tests/test_web_generator.py b/tests/test_web_generator.py new file mode 100644 index 0000000..211109d --- /dev/null +++ b/tests/test_web_generator.py @@ -0,0 +1,19 @@ +import pytest +from pathlib import Path +from tweet_classifier.web_generator import WebReportGenerator + +def test_web_generator_finds_csv_in_output_dir(tmp_path): + output_dir = tmp_path / "ok" + output_dir.mkdir() + + # Create a CSV in output_dir + csv_file = output_dir / "my_tweets.csv" + csv_file.write_text("filepath,status,detected_category,confidence,ocr_text\n/path/to/img,processed,Cat,0.9,Text") + + # Instantiate with a non-existent path but pointing to the output_dir + # The WebReportGenerator will now check in output_dir + generator = WebReportGenerator(csv_path=Path("my_tweets.csv"), output_dir=output_dir) + + # This should now point to output_dir / "my_tweets.csv" + assert generator.csv_path.exists() + assert generator.csv_path == csv_file diff --git a/tweet_classifier/__init__.py b/tweet_classifier/__init__.py new file mode 100644 index 0000000..714c25f --- /dev/null +++ b/tweet_classifier/__init__.py @@ -0,0 +1,5 @@ +from .ocr import TweetOCRExtractor +from .classifier import ZeroShotClassifier +from .organizer import ImageThemeOrganizer + +__version__ = "1.0.0" diff --git a/tweet_classifier/classifier.py b/tweet_classifier/classifier.py new file mode 100644 index 0000000..1eeec43 --- /dev/null +++ b/tweet_classifier/classifier.py @@ -0,0 +1,51 @@ +import logging +from typing import List, Dict, Any, Optional +from transformers import pipeline + +logger = logging.getLogger(__name__) + +class ZeroShotClassifier: + """ + Classe responsable de la classification de texte à l'aide de modèles Hugging Face Zero-Shot. + """ + def __init__(self, model_name: str = "MoritzLaurer/mDeBERTa-v3-base-xnli-multilingual-nli-2mil7"): + """ + Initialise le classifieur zero-shot. + :param model_name: Nom du modèle Hugging Face à utiliser. + """ + self.model_name = model_name + self._pipeline = None + + @property + def classifier_pipeline(self): + """ + Initialisation tardive (lazy loading) du pipeline pour économiser de la mémoire et du temps au démarrage. + """ + if self._pipeline is None: + logger.info(f"Chargement du pipeline de classification avec le modèle {self.model_name} (ceci peut prendre quelques secondes)...") + # On laisse Hugging Face gérer le choix du device (GPU s'il est dispo, sinon CPU) + self._pipeline = pipeline("zero-shot-classification", model=self.model_name) + return self._pipeline + + def classify(self, text: str, candidate_labels: List[str] = None) -> Dict[str, Any]: + """ + Classifie un texte selon une liste de catégories candidates. + Si aucune catégorie n'est fournie, utilise les catégories de harcèlement par défaut. + :param text: Le texte à classifier. + :param candidate_labels: Liste des catégories (labels). + :return: Dictionnaire contenant les labels et leurs scores associés. + """ + if candidate_labels is None: + candidate_labels = ["Cyberharcèlement", "Insulte", "Menace", "Non-harcèlement"] + + if not text or not text.strip(): + # Si le texte est vide, on renvoie une structure vide ou par défaut + return {"labels": [], "scores": []} + + try: + # On exécute le pipeline de classification + result = self.classifier_pipeline(text, candidate_labels=candidate_labels) + return result + except Exception as e: + logger.error(f"Erreur lors de la classification du texte : {e}") + raise RuntimeError(f"Échec de la classification : {e}") from e diff --git a/tweet_classifier/cli.py b/tweet_classifier/cli.py new file mode 100644 index 0000000..a6f0094 --- /dev/null +++ b/tweet_classifier/cli.py @@ -0,0 +1,134 @@ +import argparse +import sys +import logging +from pathlib import Path +from .organizer import ImageThemeOrganizer +from .web_generator import WebReportGenerator + +def setup_logging(verbose: bool): + """ + Configure la journalisation. + """ + level = logging.DEBUG if verbose else logging.INFO + logging.basicConfig( + level=level, + format="%(asctime)s [%(levelname)s] %(name)s : %(message)s", + handlers=[ + logging.StreamHandler(sys.stdout) + ] + ) + +def main(): + parser = argparse.ArgumentParser( + description="Classement automatique de captures d'écran de tweets par thèmes à l'aide d'EasyOCR et d'Hugging Face." + ) + + parser.add_argument( + "-i", "--input-dir", + type=str, + default=None, + help="Chemin vers le dossier contenant les captures d'écran à classer." + ) + + parser.add_argument( + "-o", "--output-dir", + type=str, + default=None, + help="Dossier de destination (par défaut, utilise ./ok/)." + ) + + parser.add_argument( + "--db", + type=str, + default=str(Path(__file__).parent.parent / "captures/ok/tweets.csv"), + help="Chemin vers le fichier base de données CSV (défaut: 'captures/ok/tweets.csv')." + ) + + parser.add_argument( + "--generate-report", + action="store_true", + help="Génère un rapport HTML à partir de la base de données CSV." + ) + + parser.add_argument( + "-m", "--model", + type=str, + default="MoritzLaurer/mDeBERTa-v3-base-xnli-multilingual-nli-2mil7", + help="Modèle Hugging Face Zero-Shot à utiliser." + ) + + parser.add_argument( + "-l", "--languages", + type=str, + default="fr,en", + help="Langues pour EasyOCR, séparées par des virgules (défaut: 'fr,en')." + ) + + parser.add_argument( + "-t", "--threshold", + type=float, + default=0.35, + help="Seuil de confiance de classification (0.0 à 1.0, défaut: 0.35)." + ) + + parser.add_argument( + "--copy", + action="store_true", + help="Copie les images au lieu de les déplacer." + ) + + parser.add_argument( + "--dry-run", + action="store_true", + help="Exécute une simulation sans modifier ni déplacer les fichiers." + ) + + parser.add_argument( + "-v", "--verbose", + action="store_true", + help="Active le mode verbeux pour afficher les logs de débogage." + ) + + args = parser.parse_args() + + setup_logging(args.verbose) + + if args.generate_report: + generator = WebReportGenerator(Path(args.db)) + generator.generate() + return + + # Normal processing path + if not args.input_dir: + print("Erreur : --input-dir est requis pour le traitement.", file=sys.stderr) + sys.exit(1) + + input_path = Path(args.input_dir) + if not input_path.exists(): + print(f"Erreur : Le dossier d'entrée '{args.input_dir}' n'existe pas.", file=sys.stderr) + sys.exit(1) + + languages_list = [lang.strip() for lang in args.languages.split(",") if lang.strip()] + + output_path = Path(args.output_dir) if args.output_dir else None + + # Instanciation de l'organisateur + organizer = ImageThemeOrganizer( + input_dir=input_path, + output_dir=output_path, + ocr_languages=languages_list, + model_name=args.model, + confidence_threshold=args.threshold, + copy_only=args.copy, + dry_run=args.dry_run, + db_path=Path(args.db) + ) + + try: + organizer.run() + except Exception as e: + print(f"Une erreur critique est survenue : {e}", file=sys.stderr) + sys.exit(1) + +if __name__ == "__main__": + main() diff --git a/tweet_classifier/database_manager.py b/tweet_classifier/database_manager.py new file mode 100644 index 0000000..4062f20 --- /dev/null +++ b/tweet_classifier/database_manager.py @@ -0,0 +1,62 @@ +import csv +from pathlib import Path +from typing import List, Optional, Dict, Any + +class CSVDatabaseManager: + def __init__(self, db_path: Path): + self.db_path = db_path + self.fieldnames = ['filename', 'filepath', 'status', 'ocr_text', 'detected_category', 'confidence', 'created_at'] + self._initialize_csv() + + def _initialize_csv(self): + if not self.db_path.exists(): + self.db_path.parent.mkdir(parents=True, exist_ok=True) + with open(self.db_path, 'w', newline='', encoding='utf-8') as f: + writer = csv.DictWriter(f, fieldnames=self.fieldnames) + writer.writeheader() + + def _read_all(self) -> List[Dict[str, Any]]: + with open(self.db_path, 'r', newline='', encoding='utf-8') as f: + return list(csv.DictReader(f)) + + def _write_all(self, data: List[Dict[str, Any]]): + with open(self.db_path, 'w', newline='', encoding='utf-8') as f: + writer = csv.DictWriter(f, fieldnames=self.fieldnames) + writer.writeheader() + writer.writerows(data) + + def add_files(self, file_paths: List[Path]): + data = self._read_all() + existing_filenames = {row['filename'] for row in data} + + new_entries = [] + for path in file_paths: + if path.name not in existing_filenames: + new_entries.append({ + 'filename': path.name, + 'filepath': str(path.absolute()), + 'status': 'pending', + 'ocr_text': '', + 'detected_category': '', + 'confidence': '', + 'created_at': '' # Could add timestamp here + }) + + if new_entries: + data.extend(new_entries) + self._write_all(data) + + def get_pending_files(self) -> List[Dict[str, Any]]: + return [row for row in self._read_all() if row['status'] == 'pending'] + + def update_file_status(self, filename: str, status: str, ocr_text: Optional[str] = None, category: Optional[str] = None, confidence: Optional[float] = None, new_filepath: Optional[str] = None): + data = self._read_all() + for row in data: + if row['filename'] == filename: + row['status'] = status + if ocr_text is not None: row['ocr_text'] = ocr_text + if category is not None: row['detected_category'] = category + if confidence is not None: row['confidence'] = str(confidence) + if new_filepath is not None: row['filepath'] = new_filepath + break + self._write_all(data) diff --git a/tweet_classifier/ocr.py b/tweet_classifier/ocr.py new file mode 100644 index 0000000..ee1c904 --- /dev/null +++ b/tweet_classifier/ocr.py @@ -0,0 +1,50 @@ +import logging +from pathlib import Path +from typing import List, Optional +import easyocr + +logger = logging.getLogger(__name__) + +class TweetOCRExtractor: + """ + Classe responsable de l'extraction de texte à partir de captures d'écran en utilisant EasyOCR. + """ + def __init__(self, languages: Optional[List[str]] = None): + """ + Initialise le lecteur EasyOCR. + :param languages: Liste des langues à charger (par défaut ['fr', 'en']). + """ + if languages is None: + languages = ['fr', 'en'] + self.languages = languages + self._reader = None + + @property + def reader(self) -> easyocr.Reader: + """ + Initialisation tardive (lazy loading) d'EasyOCR pour économiser de la mémoire si non utilisé. + """ + if self._reader is None: + logger.info("Initialisation de l'OCR EasyOCR (ceci peut prendre quelques secondes)...") + self._reader = easyocr.Reader(self.languages) + return self._reader + + def extract_text(self, image_path: Path) -> str: + """ + Extrait le texte d'une image. + :param image_path: Chemin vers le fichier image. + :return: Texte brut extrait. + """ + if not image_path.exists(): + raise FileNotFoundError(f"Le fichier image n'existe pas : {image_path}") + + try: + # easyocr accepte un chemin de fichier sous forme de string + results = self.reader.readtext(str(image_path)) + # On joint les blocs de texte détectés + text_blocks = [text for (_, text, _) in results] + extracted_text = " ".join(text_blocks).strip() + return extracted_text + except Exception as e: + logger.error(f"Erreur lors de l'extraction OCR sur {image_path.name} : {e}") + raise RuntimeError(f"Échec de l'OCR : {e}") from e diff --git a/tweet_classifier/organizer.py b/tweet_classifier/organizer.py new file mode 100644 index 0000000..fdf534b --- /dev/null +++ b/tweet_classifier/organizer.py @@ -0,0 +1,152 @@ +import logging +import shutil +from pathlib import Path +from typing import List, Optional +from tqdm import tqdm + +from .ocr import TweetOCRExtractor +from .classifier import ZeroShotClassifier +from .database_manager import CSVDatabaseManager + +logger = logging.getLogger(__name__) + +# Extensions d'images supportées +SUPPORTED_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.webp', '.bmp', '.tiff'} + +class ImageThemeOrganizer: + """ + Orchestrateur principal du traitement : parcourt les images, extrait le texte, + classifie, et organise les fichiers dans des sous-dossiers via une base de données. + """ + def __init__( + self, + input_dir: Path, + output_dir: Optional[Path] = None, + ocr_languages: Optional[List[str]] = None, + model_name: str = "MoritzLaurer/mDeBERTa-v3-base-xnli-multilingual-nli-2mil7", + confidence_threshold: float = 0.35, + copy_only: bool = False, + dry_run: bool = False, + db_path: Path = Path("captures/ok/tweets.csv") + ): + self.input_dir = Path(input_dir) + self.output_dir = Path(output_dir) if output_dir else self.input_dir / "ok" + self.confidence_threshold = confidence_threshold + self.copy_only = copy_only + self.dry_run = dry_run + + self.db = CSVDatabaseManager(db_path) + + # Initialisation des modules (lazy) + self.ocr_extractor = TweetOCRExtractor(languages=ocr_languages) + self.classifier = ZeroShotClassifier(model_name=model_name) + + def scan_new_files(self): + """ + Trouve tous les fichiers images dans le dossier d'entrée et les ajoute à la base de données. + """ + images = [] + for file in self.input_dir.iterdir(): + if file.is_file() and file.suffix.lower() in SUPPORTED_EXTENSIONS: + images.append(file) + + if images: + self.db.add_files(images) + logger.info(f"Ajout de {len(images)} fichiers à la base de données.") + + def process_pending(self): + """ + Traite tous les fichiers en attente dans la base de données. + """ + pending_files = self.db.get_pending_files() + if not pending_files: + logger.info("Aucune image en attente.") + return + + for record in tqdm(pending_files, desc="Traitement des tweets"): + self._process_record(record) + + def _process_record(self, record: dict): + image_path = Path(record['filepath']) + logger.info(f"Traitement de l'image : {image_path.name}") + + try: + # 1. Extraction du texte + text = self.ocr_extractor.extract_text(image_path) + + dest_category = "Non-classifié" + + if not text: + logger.info(f" Aucun texte extrait pour {image_path.name}.") + dest_category = "Sans_Texte" + confidence = 1.0 + else: + # 2. Classification du texte (Harassment categories) + classification_result = self.classifier.classify(text) + + # Récupérer la meilleure catégorie + if classification_result and classification_result.get("labels"): + best_label = classification_result["labels"][0] + best_score = classification_result["scores"][0] + + logger.info(f" Classification : {best_label} (score: {best_score:.2f})") + + if best_score >= self.confidence_threshold: + dest_category = best_label + confidence = best_score + else: + dest_category = "Inclassable" + confidence = best_score + else: + dest_category = "Inclassable" + confidence = 0.0 + + # 3. Organisation physique du fichier + dest_path = self._organize_file(image_path, dest_category) + + # 4. Mise à jour DB (on enregistre le nouvel emplacement du fichier + # afin que le rapport HTML pointe vers l'image déplacée). + self.db.update_file_status( + record['filename'], 'processed', text, dest_category, confidence, + new_filepath=str(dest_path) + ) + + except Exception as e: + logger.error(f" Échec du traitement de {image_path.name} : {e}") + self.db.update_file_status(record['filename'], 'error') + + def _organize_file(self, image_path: Path, category: str) -> Path: + """ + Crée le dossier de destination et y déplace ou copie le fichier image. + + Retourne le chemin de destination du fichier (utilisé pour la base de + données et le rapport HTML). + """ + dest_dir = self.output_dir / category + dest_path = dest_dir / image_path.name + + if self.dry_run: + logger.info(f"[DRY-RUN] Déplacer {image_path.name} -> {dest_dir.name}/") + return dest_path + + # Créer le dossier de catégorie si nécessaire + dest_dir.mkdir(parents=True, exist_ok=True) + + try: + if self.copy_only: + shutil.copy2(image_path, dest_path) + else: + shutil.move(image_path, dest_path) + logger.info(f" Organisé : {image_path.name} -> {dest_dir.name}/") + except Exception as e: + logger.error(f"Impossible d'organiser le fichier {image_path.name} vers {dest_dir} : {e}") + + return dest_path + + def run(self) -> dict: + """ + Exécute le processus complet. + """ + self.scan_new_files() + self.process_pending() + return {} diff --git a/tweet_classifier/web_generator.py b/tweet_classifier/web_generator.py new file mode 100644 index 0000000..043ea0f --- /dev/null +++ b/tweet_classifier/web_generator.py @@ -0,0 +1,353 @@ +import csv +from pathlib import Path +from datetime import datetime +from urllib.parse import quote +from collections import Counter +from jinja2 import Template + +# Couleurs par catégorie (avec repli sur une couleur neutre). +CATEGORY_COLORS = { + "Cyberharcèlement": "#8e44ad", + "Menace": "#c0392b", + "Insulte": "#d35400", + "Harcèlement": "#e74c3c", + "Non-harcèlement": "#27ae60", + "Sans_Texte": "#7f8c8d", + "Inclassable": "#95a5a6", + "Non-classifié": "#bdc3c7", +} +DEFAULT_COLOR = "#34495e" + +TEMPLATE = """ + + + + + + Rapport de classification + + + +
+

Rapport de classification des tweets

+

Généré le {{ generated_at }} — {{ items|length }} élément(s)

+ +
+
{{ items|length }}
Total
+ {% for cat, count in category_counts %} +
+
{{ count }}
+
{{ cat }}
+
+ {% endfor %} +
+ +
+
+ + {% for cat, count in category_counts %} + + {% endfor %} +
+ + +
+
+ + + + + + + + +""" + + +class WebReportGenerator: + def __init__(self, csv_path: Path, output_dir: Path = Path("captures/ok")): + self.csv_path = Path(csv_path).resolve() + self.output_dir = Path(output_dir).resolve() + + # Si le CSV n'existe pas, on le cherche dans output_dir. + if not self.csv_path.exists(): + potential_path = self.output_dir / self.csv_path.name + if potential_path.exists(): + self.csv_path = potential_path + + def _resolve_relative_path(self, row: dict) -> str: + """ + Détermine le chemin de l'image relatif au rapport HTML (placé dans + output_dir), encodé pour une URL. + + Robustesse : si le chemin enregistré n'existe pas (CSV obsolète d'avant + le déplacement), on reconstruit le chemin attendu + ``output_dir/catégorie/fichier``. + """ + image_path = Path(row['filepath']) + category = row.get('detected_category') or '' + + candidates = [image_path] + if category: + candidates.append(self.output_dir / category / image_path.name) + candidates.append(self.output_dir / image_path.name) + + chosen = next((c for c in candidates if c.exists()), image_path) + + try: + relative = chosen.relative_to(self.output_dir) + except ValueError: + # Repli : catégorie/fichier, sinon juste le nom du fichier. + relative = Path(category) / image_path.name if category else Path(image_path.name) + + # Encodage URL (espaces, apostrophes typographiques, accents…) en + # préservant les séparateurs de dossiers. + return quote(relative.as_posix()) + + @staticmethod + def _confidence_fields(raw_value: str) -> dict: + try: + value = float(raw_value) + except (TypeError, ValueError): + value = 0.0 + pct = round(value * 100) + if pct >= 60: + conf_class = "conf-high" + elif pct >= 35: + conf_class = "conf-mid" + else: + conf_class = "conf-low" + return {"confidence_value": value, "confidence_pct": pct, "conf_class": conf_class} + + def generate(self): + items = [] + if self.csv_path.exists(): + with open(self.csv_path, 'r', encoding='utf-8') as f: + reader = csv.DictReader(f) + for row in reader: + if row.get('status') == 'processed': + row['relative_filepath'] = self._resolve_relative_path(row) + row.update(self._confidence_fields(row.get('confidence'))) + items.append(row) + + # Tri par défaut : confiance décroissante. + items.sort(key=lambda r: r['confidence_value'], reverse=True) + + category_counts = Counter(item['detected_category'] for item in items) + # Catégories triées par effectif décroissant. + sorted_counts = sorted(category_counts.items(), key=lambda kv: (-kv[1], kv[0])) + + self.output_dir.mkdir(parents=True, exist_ok=True) + + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + output_path = self.output_dir / f"report_{timestamp}.html" + + template = Template(TEMPLATE) + html = template.render( + items=items, + category_counts=sorted_counts, + category_colors=CATEGORY_COLORS, + default_color=DEFAULT_COLOR, + generated_at=datetime.now().strftime("%d/%m/%Y à %H:%M"), + ) + + with open(output_path, 'w', encoding='utf-8') as f: + f.write(html) + print(f"Rapport généré : {output_path}") + return output_path