#!/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("""

🛡️ Foodwatch Arnaques Database

Base de données complète avec images des arnaques alimentaires

Version démo avec données réalistes

""", unsafe_allow_html=True) # Initialisation de l'application try: app = FoodwatchAdvancedDB() except Exception as e: st.error(f"Erreur initialisation: {e}") st.stop() # Sidebar avec informations de la base st.sidebar.title("📊 État de la Base") stats = app.get_database_stats() if stats: st.sidebar.metric("📦 Produits", stats.get('total_produits', 0)) st.sidebar.metric("📸 Avec images", stats.get('produits_avec_images', 0)) st.sidebar.metric("🏭 Marques", stats.get('total_marques', 0)) st.sidebar.metric("💾 Taille images", f"{stats.get('taille_images_mb', 0)} MB") if stats.get('derniere_maj'): try: derniere_maj = pd.to_datetime(stats['derniere_maj']).strftime('%d/%m/%Y %H:%M') st.sidebar.info(f"🕒 Dernière MAJ: {derniere_maj}") except: st.sidebar.info("🕒 Dernière MAJ: Récente") st.sidebar.markdown("---") # Navigation page = st.sidebar.selectbox( "🔧 Navigation", ["🏠 Dashboard", "🕷️ Scraping Démo", "📊 Analyses", "🔍 Base de Données"] ) # PAGE DASHBOARD if page == "🏠 Dashboard": st.header("📈 Dashboard de la Base de Données") if stats: # Métriques principales col1, col2, col3, col4 = st.columns(4) with col1: st.metric("🏷️ Total Produits", stats['total_produits']) with col2: st.metric("📸 Avec Images", stats['produits_avec_images']) with col3: st.metric("🏭 Marques", stats['total_marques']) with col4: st.metric("🏪 Supermarchés", stats['total_supermarches']) st.divider() # Graphiques col1, col2 = st.columns(2) with col1: st.subheader("📊 Types d'Arnaques") if stats['par_type']: fig_pie = px.pie( values=list(stats['par_type'].values()), names=list(stats['par_type'].keys()), color_discrete_sequence=px.colors.qualitative.Set3 ) fig_pie.update_layout(height=400) st.plotly_chart(fig_pie, use_container_width=True) with col2: st.subheader("🏭 Top 10 Marques") if stats['top_marques']: marques_data = list(stats['top_marques'].items())[:10] fig_bar = px.bar( x=[item[1] for item in marques_data], y=[item[0] for item in marques_data], orientation='h', color=[item[1] for item in marques_data], color_continuous_scale="Reds" ) fig_bar.update_layout(height=400, yaxis_title="Marques", xaxis_title="Nombre d'arnaques") st.plotly_chart(fig_bar, use_container_width=True) # Derniers produits ajoutés st.subheader("🆕 Derniers Produits Ajoutés") df_recent = app.load_data_from_db(limit=5) if not df_recent.empty: for idx, row in df_recent.iterrows(): with st.expander(f"📦 {row['nom_produit']} - {row.get('marque', 'N/A')}"): col1, col2 = st.columns([1, 2]) with col1: if row.get('url_image_original'): try: st.image(row['url_image_original'], caption="Image du produit", use_column_width=True) except: st.info("📷 Image non disponible") else: st.info("📷 Pas d'image") with col2: st.write(f"**Marque:** {row.get('marque', 'N/A')}") st.write(f"**Supermarché:** {row.get('supermarche', 'N/A')}") st.write(f"**Type:** {row.get('type_arnaque', 'N/A')}") st.write(f"**Prix:** {row.get('prix', 'N/A')}") if row.get('ingredients_problematiques'): st.error(f"⚠️ **Additifs:** {row['ingredients_problematiques']}") if row.get('description'): st.write(f"**Description:** {row['description'][:200]}...") else: st.info("Aucun produit en base. Lancez un scraping pour commencer.") else: st.warning("Base de données vide. Effectuez un premier scraping.") # PAGE SCRAPING DÉMO elif page == "🕷️ Scraping Démo": st.header("🕷️ Scraping de Démonstration") # Status de la base st.markdown(f"""
📊 État actuel de la base:
• {stats.get('total_produits', 0)} produits en base
• {stats.get('produits_avec_images', 0)} produits avec images
• {stats.get('sessions_scraping', 0)} sessions de scraping effectuées
""", unsafe_allow_html=True) st.info(""" 🔔 **Mode Démonstration** Cette version utilise des données de démonstration réalistes basées sur de vraies arnaques documentées par Foodwatch. Les images sont des placeholders mais la structure de données est identique au scraping réel. """) # Configuration du scraping st.subheader("⚙️ Configuration") col1, col2 = st.columns(2) with col1: max_pages = st.slider("Nombre de produits à ajouter", 1, 5, 3) incremental_mode = st.checkbox("Mode incrémental (éviter doublons)", True) with col2: st.info(""" **🔄 Mode incrémental:** - Évite les doublons - Met à jour les produits existants - Optimise la base de données **📊 Données de démo:** - Basées sur vraies arnaques Foodwatch - Images placeholder réalistes - Structure complète identique au vrai scraping """) # Historique des scrapings st.subheader("📈 Historique des Sessions") df_history = app.get_scraping_history() if not df_history.empty: df_display = df_history[['id', 'start_time', 'products_new', 'products_updated', 'images_downloaded', 'status']].head(5) df_display.columns = ['Session ID', 'Date', 'Nouveaux', 'MAJ', 'Images', 'Statut'] st.dataframe(df_display, use_container_width=True) st.divider() # Lancement du scraping col1, col2, col3 = st.columns([1, 2, 1]) with col2: if st.button("🚀 LANCER LE SCRAPING DÉMO", type="primary", use_container_width=True): st.markdown("""
🔄 SCRAPING DÉMO EN COURS
Ajout de données réalistes avec images placeholder...
""", unsafe_allow_html=True) # Lancement du scraping de démo results = app.scrape_with_demo_data(max_pages) # Affichage des résultats if results: st.success("✅ **Scraping démo terminé avec succès !**") col1, col2, col3, col4 = st.columns(4) with col1: st.metric("📦 Produits traités", results['produits_extraits']) with col2: st.metric("🆕 Nouveaux", results['produits_nouveaux']) with col3: st.metric("📝 Mis à jour", results['produits_maj']) with col4: st.metric("📸 Images", results['images_telechargees']) st.info(f"🆔 Session ID: {results['session_id']}") st.experimental_rerun() # PAGE ANALYSES elif page == "📊 Analyses": st.header("📊 Analyses des Données") df = app.load_data_from_db() if df.empty: st.warning("⚠️ Aucune donnée en base. Effectuez un scraping d'abord.") return st.success(f"📊 Analyse de **{len(df)} produits** en base de données") analyse_type = st.selectbox( "Type d'analyse", ["🧪 Additifs Controversés", "🏭 Analyse par Marque", "🏪 Analyse par Supermarché", "📸 Analyse des Images"] ) if analyse_type == "🧪 Additifs Controversés": st.subheader("🧪 Analyse des Additifs Controversés") # Produits avec additifs df_additifs = df[df['ingredients_problematiques'].notna() & (df['ingredients_problematiques'] != '')] if not df_additifs.empty: col1, col2 = st.columns(2) with col1: # Additifs les plus fréquents additifs_list = [] for ingredients in df_additifs['ingredients_problematiques']: additifs_list.extend([x.strip() for x in str(ingredients).split(',') if x.strip()]) if additifs_list: additifs_count = pd.Series(additifs_list).value_counts().head(10) fig = px.bar( x=additifs_count.values, y=additifs_count.index, orientation='h', title="Top 10 des additifs problématiques", color=additifs_count.values, color_continuous_scale="Reds" ) fig.update_layout(height=400) st.plotly_chart(fig, use_container_width=True) with col2: # Marques avec le plus d'additifs marques_additifs = df_additifs.groupby('marque').size().sort_values(ascending=False).head(8) fig = px.pie( values=marques_additifs.values, names=marques_additifs.index, title="Marques avec additifs problématiques" ) fig.update_layout(height=400) st.plotly_chart(fig, use_container_width=True) # Table de référence des additifs st.subheader("📚 Base de Référence des Additifs") try: conn = sqlite3.connect(app.db_path) df_additifs_ref = pd.read_sql_query(""" SELECT code_additif, nom_additif, categorie, risques_sante, niveau_risque FROM additifs_references ORDER BY niveau_risque DESC, code_additif """, conn) conn.close() if not df_additifs_ref.empty: st.dataframe(df_additifs_ref, use_container_width=True) except Exception as e: st.error(f"Erreur chargement référentiel additifs: {e}") else: st.info("Aucun produit avec additifs problématiques identifiés.") elif analyse_type == "🏭 Analyse par Marque": st.subheader("🏭 Analyse par Marque") marques_count = df['marque'].value_counts().head(10) if not marques_count.empty: fig = px.bar( x=marques_count.index, y=marques_count.values, title="Top 10 des marques les plus signalées", color=marques_count.values, color_continuous_scale="Oranges" ) fig.update_xaxes(tickangle=45) st.plotly_chart(fig, use_container_width=True) # Analyse détaillée st.subheader("Analyse détaillée par marque") marque_selected = st.selectbox("Sélectionner une marque", df['marque'].unique()) if marque_selected: df_marque = df[df['marque'] == marque_selected] types_count = df_marque['type_arnaque'].value_counts() col1, col2 = st.columns(2) with col1: fig = px.pie( values=types_count.values, names=types_count.index, title=f"Types d'arnaques - {marque_selected}" ) st.plotly_chart(fig, use_container_width=True) with col2: st.write("**Détails des signalements:**") df_display = df_marque[['nom_produit', 'type_arnaque', 'supermarche', 'date_signalement']].copy() df_display.columns = ['Produit', 'Type', 'Supermarché', 'Date'] st.dataframe(df_display, use_container_width=True) elif analyse_type == "🏪 Analyse par Supermarché": st.subheader("🏪 Analyse par Supermarché") super_count = df['supermarche'].value_counts().head(10) if not super_count.empty: fig = px.bar( x=super_count.values, y=super_count.index, orientation='h', title="Signalements par supermarché", color=super_count.values, color_continuous_scale="Reds" ) st.plotly_chart(fig, use_container_width=True) elif analyse_type == "📸 Analyse des Images": st.subheader("📸 Analyse des Images") # Statistiques images total_avec_images = len(df[df['url_image_original'].notna() & (df['url_image_original'] != '')]) total_sans_images = len(df) - total_avec_images col1, col2, col3 = st.columns(3) with col1: st.metric("📸 Avec images", total_avec_images) with col2: st.metric("❌ Sans images", total_sans_images) with col3: couverture = round((total_avec_images / len(df)) * 100, 1) if len(df) > 0 else 0 st.metric("📊 Couverture", f"{couverture}%") # Galerie d'images if total_avec_images > 0: st.subheader("🖼️ Galerie des Images") df_with_images = df[df['url_image_original'].notna() & (df['url_image_original'] != '')].head(6) cols = st.columns(3) for idx, (_, product) in enumerate(df_with_images.iterrows()): with cols[idx % 3]: try: st.image(product['url_image_original'], caption=f"{product['nom_produit'][:30]}...", use_column_width=True) # Badge type d'arnaque type_colors = { "Ingrédients masqués": "#f44336", "Arnaque au prix": "#ff9800", "Arnaque à l'origine": "#ffc107", "Plein de vide": "#9c27b0", "Arnaque au visuel": "#2196f3", "Intox détox": "#4caf50" } color = type_colors.get(product['type_arnaque'], "#757575") st.markdown(f"""
{product['type_arnaque']}
""", unsafe_allow_html=True) except: st.error("Erreur chargement image") else: st.info("Aucune image disponible.") # PAGE BASE DE DONNÉES elif page == "🔍 Base de Données": st.header("🔍 Exploration de la Base de Données") df = app.load_data_from_db() if df.empty: st.warning("⚠️ Base de données vide.") return st.success(f"📊 **{len(df)} produits** en base de données") # Filtres st.subheader("🔎 Filtres de Recherche") col1, col2 = st.columns(2) with col1: marques_filter = st.multiselect( "Filtrer par marque", options=sorted(df['marque'].dropna().unique()) ) types_filter = st.multiselect( "Filtrer par type d'arnaque", options=sorted(df['type_arnaque'].dropna().unique()) ) with col2: super_filter = st.multiselect( "Filtrer par supermarché", options=sorted(df['supermarche'].dropna().unique()) ) avec_additifs = st.checkbox("Seulement produits avec additifs") # Recherche textuelle search = st.text_input("🔍 Recherche textuelle") # Application des filtres df_filtered = df.copy() if marques_filter: df_filtered = df_filtered[df_filtered['marque'].isin(marques_filter)] if types_filter: df_filtered = df_filtered[df_filtered['type_arnaque'].isin(types_filter)] if super_filter: df_filtered = df_filtered[df_filtered['supermarche'].isin(super_filter)] if avec_additifs: df_filtered = df_filtered[df_filtered['ingredients_problematiques'].notna() & (df_filtered['ingredients_problematiques'] != '')] if search: df_filtered = df_filtered[ df_filtered['nom_produit'].str.contains(search, case=False, na=False) | df_filtered['description'].str.contains(search, case=False, na=False) ] st.divider() # Résultats col1, col2 = st.columns([3, 1]) with col1: st.subheader(f"📋 Résultats ({len(df_filtered)} produits)") with col2: if not df_filtered.empty: csv_buffer = io.StringIO() df_filtered.to_csv(csv_buffer, index=False) st.download_button( "📥 Export CSV", csv_buffer.getvalue(), f"foodwatch_export_{datetime.now().strftime('%Y%m%d_%H%M')}.csv", "text/csv", use_container_width=True ) if not df_filtered.empty: # Tableau cols_display = ['nom_produit', 'marque', 'supermarche', 'type_arnaque', 'prix'] cols_labels = ['Produit', 'Marque', 'Supermarché', 'Type arnaque', 'Prix'] df_display = df_filtered[cols_display].copy() df_display.columns = cols_labels st.dataframe(df_display, use_container_width=True, height=400) # Détail d'un produit if len(df_filtered) > 0: st.subheader("🔍 Détail d'un produit") idx = st.selectbox( "Sélectionner un produit", range(len(df_filtered)), format_func=lambda x: f"{df_filtered.iloc[x]['nom_produit']} - {df_filtered.iloc[x].get('marque', 'N/A')}" ) if idx is not None: product = df_filtered.iloc[idx] col1, col2 = st.columns([1, 2]) with col1: if product.get('url_image_original'): try: st.image(product['url_image_original'], caption="Image du produit", use_column_width=True) except: st.info("📷 Image non disponible") else: st.info("📷 Pas d'image") with col2: st.markdown(f"**🏷️ {product['nom_produit']}**") st.write(f"**Marque:** {product.get('marque', 'N/A')}") st.write(f"**Supermarché:** {product.get('supermarche', 'N/A')}") st.write(f"**Type:** {product.get('type_arnaque', 'N/A')}") st.write(f"**Prix:** {product.get('prix', 'N/A')}") if product.get('ingredients_problematiques'): st.error(f"⚠️ **Additifs:** {product['ingredients_problematiques']}") else: st.success("✅ Aucun additif problématique") if product.get('description'): with st.expander("📝 Description complète"): st.write(product['description']) else: st.info("🔍 Aucun résultat pour ces filtres.") # Footer st.markdown("---") st.markdown("""
🛡️ Foodwatch Arnaques Database v2.0 | Version démo avec données réalistes | Source: Foodwatch.org
""", unsafe_allow_html=True) if __name__ == "__main__": main()