🛡️ Foodwatch Arnaques Database
Base de données complète avec images des arnaques alimentaires
Version démo avec données réalistes
#!/usr/bin/env python3 """ Foodwatch Arnaques Analyzer - Version finale avec base persistante et images Système de scraping incrémental avec sauvegarde des photos """ import streamlit as st import pandas as pd import sqlite3 import requests from bs4 import BeautifulSoup import plotly.express as px import plotly.graph_objects as go import json import re import time from datetime import datetime, timedelta import logging from typing import Dict, List, Optional, Tuple from dataclasses import dataclass, asdict import io import os import sys from pathlib import Path from urllib.parse import urljoin, urlparse import numpy as np import random import tempfile import hashlib import base64 import uuid # Configuration Streamlit st.set_page_config( page_title="🛡️ Foodwatch Arnaques DB", page_icon="🛡️", layout="wide", initial_sidebar_state="expanded" ) # Variables d'environnement pour HF Spaces os.environ["STREAMLIT_SERVER_HEADLESS"] = "true" os.environ["STREAMLIT_BROWSER_GATHER_USAGE_STATS"] = "false" os.environ["STREAMLIT_GLOBAL_GATHER_USAGE_STATS"] = "false" # CSS personnalisé st.markdown(""" """, unsafe_allow_html=True) @dataclass class ArnaqueProduit: """Structure complète pour une arnaque avec métadonnées images""" id: Optional[int] = None nom_produit: str = "" marque: str = "" supermarche: str = "" ville: str = "" date_signalement: str = "" type_arnaque: str = "" description: str = "" prix: str = "" ingredients_problematiques: str = "" origine_reelle: str = "" origine_affichee: str = "" additifs_controverses: List[str] = None url_image_original: str = "" image_filename: str = "" image_hash: str = "" image_size: str = "" image_downloaded: bool = False url_source: str = "" date_scraping: str = "" date_last_update: str = "" content_hash: str = "" def __post_init__(self): if self.additifs_controverses is None: self.additifs_controverses = [] if not self.date_scraping: self.date_scraping = datetime.now().isoformat() if not self.date_last_update: self.date_last_update = self.date_scraping if not self.content_hash: content_str = f"{self.nom_produit}_{self.marque}_{self.description}" self.content_hash = hashlib.md5(content_str.encode()).hexdigest()[:12] class FoodwatchAdvancedDB: """Système avancé de base de données avec gestion des images""" def __init__(self): # Configuration des chemins if 'SPACE_ID' in os.environ: self.base_dir = tempfile.mkdtemp() self.db_path = os.path.join(self.base_dir, "foodwatch_arnaques.db") self.images_dir = os.path.join(self.base_dir, "images") else: self.base_dir = Path("foodwatch_data") self.base_dir.mkdir(exist_ok=True) self.db_path = self.base_dir / "foodwatch_arnaques.db" self.images_dir = self.base_dir / "images" self.images_dir.mkdir(exist_ok=True) os.makedirs(self.images_dir, exist_ok=True) self.base_url = "https://www.foodwatch.org" self.mur_arnaques_url = "https://www.foodwatch.org/fr/agir/mur-des-arnaques-etiquettes" # Configuration session HTTP self.session = requests.Session() self.session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8', 'Accept-Language': 'fr-FR,fr;q=0.9,en;q=0.8', 'Accept-Encoding': 'gzip, deflate', 'DNT': '1', 'Connection': 'keep-alive' }) # Patterns pour extraction self.additif_patterns = [ r'E\d{3,4}[a-z]?', r'nitrite[s]?\s+ajouté[s]?', r'nitrate[s]?\s+ajouté[s]?', r'glutamate', r'diphosphate', r'huile\s+de\s+palme' ] self.types_arnaques = [ "Arnaque au prix", "Arnaque à l'origine", "Plein de vide", "Ingrédients masqués", "Arnaque au visuel", "Intox détox", "Made in France trompeur", "Shrinkflation", "Cheapflation" ] self.init_database() def init_database(self): """Initialise la base de données avec tables complètes""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # Table principale des arnaques cursor.execute(""" CREATE TABLE IF NOT EXISTS arnaques ( id INTEGER PRIMARY KEY AUTOINCREMENT, nom_produit TEXT NOT NULL, marque TEXT, supermarche TEXT, ville TEXT, date_signalement DATE, type_arnaque TEXT, description TEXT, prix TEXT, ingredients_problematiques TEXT, origine_reelle TEXT, origine_affichee TEXT, additifs_controverses TEXT, url_image_original TEXT, image_filename TEXT, image_hash TEXT, image_size TEXT, image_downloaded BOOLEAN DEFAULT FALSE, url_source TEXT, date_scraping DATETIME DEFAULT CURRENT_TIMESTAMP, date_last_update DATETIME DEFAULT CURRENT_TIMESTAMP, content_hash TEXT UNIQUE, is_active BOOLEAN DEFAULT TRUE, scraping_session_id TEXT ) """) # Table des sessions de scraping cursor.execute(""" CREATE TABLE IF NOT EXISTS scraping_sessions ( id TEXT PRIMARY KEY, start_time DATETIME, end_time DATETIME, pages_scraped INTEGER, products_found INTEGER, products_new INTEGER, products_updated INTEGER, images_downloaded INTEGER, status TEXT, error_log TEXT ) """) # Table des additifs de référence cursor.execute(""" CREATE TABLE IF NOT EXISTS additifs_references ( code_additif TEXT PRIMARY KEY, nom_additif TEXT, categorie TEXT, risques_sante TEXT, reglementation_ue TEXT, alternatives TEXT, niveau_risque INTEGER DEFAULT 1 ) """) # Index pour les performances cursor.execute("CREATE INDEX IF NOT EXISTS idx_content_hash ON arnaques(content_hash)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_date_scraping ON arnaques(date_scraping)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_type_arnaque ON arnaques(type_arnaque)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_marque ON arnaques(marque)") # Insertion des additifs de référence self.populate_additifs_reference(cursor) # Insertion de données d'exemple si base vide cursor.execute("SELECT COUNT(*) FROM arnaques") if cursor.fetchone()[0] == 0: self.insert_sample_data(cursor) conn.commit() conn.close() except Exception as e: st.error(f"Erreur initialisation base: {e}") def populate_additifs_reference(self, cursor): """Remplit la base de référence des additifs""" additifs_ref = [ ("E250", "Nitrite de sodium", "Conservateur", "Cancérigène possible (CIRC 2A)", "Autorisé avec limites", "Sel de céleri", 4), ("E252", "Nitrate de potassium", "Conservateur", "Cancérigène possible", "Autorisé avec limites", "Conservation naturelle", 4), ("E621", "Glutamate monosodique", "Exhausteur de goût", "Maux de tête, allergies", "Autorisé", "Levure nutritionnelle", 3), ("E450", "Diphosphates", "Stabilisant", "Hyperactivité, troubles digestifs", "Autorisé", "Phosphates naturels", 3), ("E951", "Aspartame", "Édulcorant", "Débat scientifique en cours", "Autorisé", "Stévia, sucre de coco", 2), ("E407", "Carraghénanes", "Épaississant", "Inflammation intestinale possible", "Autorisé", "Agar-agar", 3), ] cursor.executemany(""" INSERT OR IGNORE INTO additifs_references (code_additif, nom_additif, categorie, risques_sante, reglementation_ue, alternatives, niveau_risque) VALUES (?, ?, ?, ?, ?, ?, ?) """, additifs_ref) def insert_sample_data(self, cursor): """Insère des données d'exemple avec images""" session_id = str(uuid.uuid4())[:8] sample_data = [ ("Suprêmes au goût frais de Homard", "Coraya", "Carrefour", "Paris", "2024-01-15", "Ingrédients masqués", "Affiche 'homard' en grandes lettres mais n'en contient aucune trace, contient du glutamate", "4.99€", "Glutamate (E621)", "", "", "[]", "https://example.com/coraya_homard.jpg", "", "", "", False, "https://www.foodwatch.org/fr/agir/mur-des-arnaques-etiquettes", session_id), ("Pain de mie 100% français", "Jacquet", "E.Leclerc", "Lyon", "2024-01-10", "Arnaque à l'origine", "Blé importé d'Ukraine malgré l'affichage tricolore français", "2.50€", "", "Ukraine", "France", "[]", "https://example.com/jacquet_pain.jpg", "", "", "", False, "https://www.foodwatch.org/fr/agir/mur-des-arnaques-etiquettes", session_id), ] for data in sample_data: content_str = f"{data[0]}_{data[1]}_{data[6]}" content_hash = hashlib.md5(content_str.encode()).hexdigest()[:12] cursor.execute(""" INSERT OR IGNORE INTO arnaques (nom_produit, marque, supermarche, ville, date_signalement, type_arnaque, description, prix, ingredients_problematiques, origine_reelle, origine_affichee, additifs_controverses, url_image_original, image_filename, image_hash, image_size, image_downloaded, url_source, scraping_session_id, content_hash) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (*data, content_hash)) def scrape_with_demo_data(self, max_pages: int = 5) -> Dict: """Version de démonstration avec données réalistes""" session_id = str(uuid.uuid4())[:8] start_time = datetime.now() st.info(f"🚀 **Session de scraping {session_id} démarrée (mode démo)**") # Données de démonstration réalistes demo_products = [ { "nom_produit": "Jambon de Parme italien", "marque": "Aoste", "supermarche": "Carrefour", "ville": "Paris", "description": "Étiquette indique 'Jambon de Parme' avec drapeau italien mais fabriqué en France", "type_arnaque": "Arnaque à l'origine", "prix": "6.99€", "ingredients_problematiques": "", "url_image_original": "https://via.placeholder.com/300x200/FF6B35/FFFFFF?text=Jambon+Parme" }, { "nom_produit": "Céréales Kids Multivitamines", "marque": "Kellogg's", "supermarche": "E.Leclerc", "ville": "Lyon", "description": "Marketing santé avec vitamines ajoutées mais 35% de sucre", "type_arnaque": "Intox détox", "prix": "4.49€", "ingredients_problematiques": "Sucre, E102 (Tartrazine)", "url_image_original": "https://via.placeholder.com/300x200/4CAF50/FFFFFF?text=Cereales+Kids" }, { "nom_produit": "Pizza Margherita Artisanale", "marque": "Buitoni", "supermarche": "Monoprix", "ville": "Marseille", "description": "Emballage 30% plus grand que nécessaire, donne l'impression d'une grande pizza", "type_arnaque": "Plein de vide", "prix": "3.79€", "ingredients_problematiques": "", "url_image_original": "https://via.placeholder.com/300x200/9C27B0/FFFFFF?text=Pizza+XXL" }, { "nom_produit": "Charcuterie Sans Nitrites", "marque": "Fleury Michon", "supermarche": "Auchan", "ville": "Toulouse", "description": "Affiche 'sans nitrites' mais contient des nitrites naturels de céleri non mentionnés", "type_arnaque": "Ingrédients masqués", "prix": "5.99€", "ingredients_problematiques": "Nitrites cachés (céleri), E250", "url_image_original": "https://via.placeholder.com/300x200/F44336/FFFFFF?text=Sans+Nitrites" }, { "nom_produit": "Format Familial Chocolat", "marque": "Milka", "supermarche": "Casino", "ville": "Nice", "description": "Prix au kilo 25% plus élevé que format standard pour même produit", "type_arnaque": "Arnaque au prix", "prix": "4.89€", "ingredients_problematiques": "Huile de palme", "url_image_original": "https://via.placeholder.com/300x200/FF9800/FFFFFF?text=Format+XL" } ] progress_bar = st.progress(0) status_text = st.empty() produits_nouveaux = 0 produits_maj = 0 images_telechargees = 0 try: for page in range(1, max_pages + 1): status_text.text(f"🔍 Traitement page {page}/{max_pages}") progress_bar.progress(page / max_pages) # Simulation délai respectueux time.sleep(random.uniform(1, 3)) # Sélection d'un produit de démo pour cette page if page <= len(demo_products): demo = demo_products[page - 1] produit = ArnaqueProduit( nom_produit=demo["nom_produit"], marque=demo["marque"], supermarche=demo["supermarche"], ville=demo["ville"], description=demo["description"], type_arnaque=demo["type_arnaque"], prix=demo["prix"], ingredients_problematiques=demo["ingredients_problematiques"], url_image_original=demo["url_image_original"], url_source=self.mur_arnaques_url, date_signalement=(datetime.now() - timedelta(days=random.randint(1, 30))).strftime("%Y-%m-%d") ) produit.additifs_controverses = demo["ingredients_problematiques"].split(", ") if demo["ingredients_problematiques"] else [] # Vérification si le produit existe déjà existing_id = self.check_existing_product(produit.content_hash) if existing_id: self.update_existing_product(existing_id, produit) produits_maj += 1 st.info(f"📝 Produit mis à jour: {produit.nom_produit}") else: product_id = self.save_new_product(produit) produits_nouveaux += 1 st.success(f"✅ Nouveau produit: {produit.nom_produit}") images_telechargees += 1 progress_bar.progress(1.0) status_text.text("✅ Scraping terminé") # Enregistrement de la session end_time = datetime.now() conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(""" INSERT INTO scraping_sessions (id, start_time, end_time, pages_scraped, products_found, products_new, products_updated, images_downloaded, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'TERMINE') """, (session_id, start_time, end_time, max_pages, produits_nouveaux + produits_maj, produits_nouveaux, produits_maj, images_telechargees)) conn.commit() conn.close() except Exception as e: st.error(f"❌ Erreur: {e}") results = { 'session_id': session_id, 'produits_extraits': produits_nouveaux + produits_maj, 'produits_nouveaux': produits_nouveaux, 'produits_maj': produits_maj, 'images_telechargees': images_telechargees, 'errors': [] } return results def check_existing_product(self, content_hash: str) -> Optional[int]: """Vérifie si un produit existe déjà""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute("SELECT id FROM arnaques WHERE content_hash = ?", (content_hash,)) result = cursor.fetchone() conn.close() return result[0] if result else None except: return None def save_new_product(self, produit: ArnaqueProduit) -> int: """Sauvegarde un nouveau produit""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(""" INSERT INTO arnaques (nom_produit, marque, supermarche, ville, date_signalement, type_arnaque, description, prix, ingredients_problematiques, origine_reelle, origine_affichee, additifs_controverses, url_image_original, url_source, content_hash, scraping_session_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( produit.nom_produit, produit.marque, produit.supermarche, produit.ville, produit.date_signalement, produit.type_arnaque, produit.description, produit.prix, produit.ingredients_problematiques, produit.origine_reelle, produit.origine_affichee, json.dumps(produit.additifs_controverses), produit.url_image_original, produit.url_source, produit.content_hash, produit.date_scraping )) product_id = cursor.lastrowid conn.commit() conn.close() return product_id def update_existing_product(self, product_id: int, produit: ArnaqueProduit): """Met à jour un produit existant""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(""" UPDATE arnaques SET date_last_update = ?, description = ?, prix = ?, ingredients_problematiques = ?, url_image_original = ? WHERE id = ? """, ( datetime.now().isoformat(), produit.description, produit.prix, produit.ingredients_problematiques, produit.url_image_original, product_id )) conn.commit() conn.close() def get_database_stats(self) -> Dict: """Statistiques complètes de la base de données""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() stats = {} # Statistiques générales cursor.execute("SELECT COUNT(*) FROM arnaques WHERE is_active = TRUE") stats['total_produits'] = cursor.fetchone()[0] cursor.execute("SELECT COUNT(*) FROM arnaques WHERE image_downloaded = TRUE") stats['produits_avec_images'] = cursor.fetchone()[0] cursor.execute("SELECT COUNT(DISTINCT marque) FROM arnaques WHERE marque IS NOT NULL") stats['total_marques'] = cursor.fetchone()[0] cursor.execute("SELECT COUNT(DISTINCT supermarche) FROM arnaques WHERE supermarche IS NOT NULL") stats['total_supermarches'] = cursor.fetchone()[0] # Sessions de scraping cursor.execute("SELECT COUNT(*) FROM scraping_sessions") stats['sessions_scraping'] = cursor.fetchone()[0] cursor.execute("SELECT MAX(date_scraping) FROM arnaques") stats['derniere_maj'] = cursor.fetchone()[0] # Par type d'arnaque cursor.execute(""" SELECT type_arnaque, COUNT(*) FROM arnaques WHERE is_active = TRUE GROUP BY type_arnaque ORDER BY COUNT(*) DESC """) stats['par_type'] = dict(cursor.fetchall()) # Top marques cursor.execute(""" SELECT marque, COUNT(*) FROM arnaques WHERE marque IS NOT NULL AND is_active = TRUE GROUP BY marque ORDER BY COUNT(*) DESC LIMIT 15 """) stats['top_marques'] = dict(cursor.fetchall()) # Top supermarchés cursor.execute(""" SELECT supermarche, COUNT(*) FROM arnaques WHERE supermarche IS NOT NULL AND is_active = TRUE GROUP BY supermarche ORDER BY COUNT(*) DESC LIMIT 15 """) stats['top_supermarches'] = dict(cursor.fetchall()) # Additifs fréquents cursor.execute(""" SELECT ingredients_problematiques, COUNT(*) FROM arnaques WHERE ingredients_problematiques IS NOT NULL AND ingredients_problematiques != '' AND is_active = TRUE GROUP BY ingredients_problematiques ORDER BY COUNT(*) DESC LIMIT 15 """) stats['additifs_frequents'] = dict(cursor.fetchall()) # Taille images (simulation) stats['total_images'] = stats['produits_avec_images'] stats['taille_images_mb'] = round(stats['produits_avec_images'] * 0.15, 2) # Estimation conn.close() return stats except Exception as e: st.error(f"Erreur calcul statistiques: {e}") return {} def load_data_from_db(self, limit: Optional[int] = None) -> pd.DataFrame: """Charge les données avec option de limite""" try: conn = sqlite3.connect(self.db_path) query = "SELECT * FROM arnaques WHERE is_active = TRUE ORDER BY date_scraping DESC" if limit: query += f" LIMIT {limit}" df = pd.read_sql_query(query, conn) conn.close() return df except Exception as e: st.error(f"Erreur chargement données: {e}") return pd.DataFrame() def get_scraping_history(self) -> pd.DataFrame: """Récupère l'historique des sessions de scraping""" try: conn = sqlite3.connect(self.db_path) df = pd.read_sql_query(""" SELECT * FROM scraping_sessions ORDER BY start_time DESC """, conn) conn.close() return df except: return pd.DataFrame() def main(): """Interface principale de l'application""" st.markdown("""
Base de données complète avec images des arnaques alimentaires
Version démo avec données réalistes