import sys import os import requests import time import re import io import ollama import json import pdfplumber from bs4 import BeautifulSoup from urllib.parse import urljoin sys.path.append( os.path.abspath( os.path.join(os.path.dirname(__file__), "..") ) ) from database.database import SessionLocal, engine from database.models import Document, Base BASE_URL = "https://health.ec.europa.eu" START_URL = ( "https://health.ec.europa.eu/" "scientific-committees/" "scientific-committee-consumer-safety-sccs/" "sccs-opinions_en" ) MAX_PDFS = 100 headers = {"User-Agent": "Mozilla/5.0"} session = requests.Session() session.headers.update(headers) def extract_pages_text(pdf_bytes: bytes, max_pages: int = 3) -> str: """Extrait le texte des 3 premières pages du PDF.""" try: with pdfplumber.open(io.BytesIO(pdf_bytes)) as pdf: if not pdf.pages: return "" texts = [] for i, page in enumerate(pdf.pages[:max_pages]): text = page.extract_text() or "" if text.strip(): texts.append(f"--- PAGE {i+1} ---\n{text}") return "\n\n".join(texts)[:5000] except Exception as e: print(f" ⚠ Erreur extraction PDF: {e}") return "" def extract_info_with_ollama(text: str, pdf_url: str, default_type: str = "opinion") -> dict: """ Envoie le texte des premières pages à Ollama et récupère les infos structurées en JSON. """ if not text.strip(): filename = pdf_url.split("/")[-1].replace("_", " ").replace(".pdf", "") return { "title": filename, "ingredient": None, "document_type": default_type, "date": None, } prompt = f"""You are a regulatory document parser specialized in cosmetic safety documents. Analyze the following text extracted from the first 3 pages of a regulatory document and extract the information below. Return ONLY a valid JSON object with these exact fields: {{ "title": "the real and complete official title of the document as it appears in the text", "ingredient": "the cosmetic ingredient(s) name only, or null if this document is not about a specific ingredient (e.g. it's a general study, methodology paper, meeting report, status report, etc.)", "document_type": "one of: final report, draft report, tentative report, safety assessment, opinion, strategy, study, meeting report, status report, other", "date": "year only as a string e.g. '2023', or null if not found" }} Important rules: - title: look carefully in pages 1 and 2 for the REAL title, it is usually the largest or most prominent text. Do NOT use a generic description. - ingredient: extract ONLY the ingredient name(s). Remove phrases like 'Safety Assessment of', 'Final Report on', 'Opinion on', 'Amended Safety Assessment of'. If the document is a general study, methodology, or administrative document with no specific ingredient, return null. - document_type: detect from keywords. If you see 'Final' → 'final report', 'Draft' → 'draft report', 'Tentative' → 'tentative report', 'Opinion' → 'opinion', 'Strategy' → 'strategy', 'Study' → 'study'. - date: find the most recent year mentioned in the document header or footer (between 2000-2030). - Return ONLY the JSON object, no markdown, no explanation. Document text (first 3 pages): {text}""" try: response = ollama.chat( model="llama3.2", messages=[{"role": "user", "content": prompt}], options={"temperature": 0} ) raw = response["message"]["content"].strip() raw = re.sub(r"```json|```", "", raw).strip() match = re.search(r"\{.*\}", raw, re.DOTALL) if match: raw = match.group(0) data = json.loads(raw) ingredient = data.get("ingredient") if ingredient and str(ingredient).strip().lower() in ("null", "none", "n/a", ""): ingredient = None return { "title": str(data.get("title") or "").strip()[:200] or "Unknown", "ingredient": str(ingredient).strip()[:200] if ingredient else None, "document_type": str(data.get("document_type") or default_type).strip().lower(), "date": str(data.get("date")).strip() if data.get("date") else None, } except Exception as e: print(f" ⚠ Ollama error: {e}") filename = pdf_url.split("/")[-1].replace("_", " ").replace(".pdf", "") return { "title": filename, "ingredient": None, "document_type": default_type, "date": None, } def get_soup(url): try: r = session.get(url, timeout=25) if r.status_code == 429: print("429 → waiting 60s") time.sleep(60) r = session.get(url, timeout=25) r.raise_for_status() return BeautifulSoup(r.text, "html.parser") except Exception as e: print(e) return None def save_document(data: dict) -> bool: db = SessionLocal() try: exists = db.query(Document).filter(Document.pdf_url == data["pdf_url"]).first() if exists: return False db.add(Document(**data)) db.commit() return True except Exception as e: print(f" DB error: {e}") db.rollback() return False finally: db.close() def process_pdf(pdf_url: str) -> dict | None: print(f"\n Downloading: {pdf_url.split('/')[-1]}") try: r = session.get(pdf_url, timeout=30) if r.status_code != 200: print(f" ✗ HTTP {r.status_code}") return None text = extract_pages_text(r.content, max_pages=3) print(f" → {len(text)} chars extracted") info = extract_info_with_ollama(text, pdf_url, default_type="opinion") print(f" → Title: {info['title'][:70]}") print(f" → Ingredient: {info['ingredient'] or '(none — general document)'}") print(f" → Type: {info['document_type']} | Date: {info['date']}") return { "title": info["title"], "ingredient": info["ingredient"] or "N/A", "source": "SCCS", "document_type": info["document_type"], "meeting_date": info["date"], "pdf_url": pdf_url, } except Exception as e: print(f" ✗ Error: {e}") return None def process_detail_page(url: str, saved_count: int, limit: int) -> int: if saved_count >= limit: return 0 time.sleep(2) print(f"\n Page: {url.split('/')[-1]}") soup = get_soup(url) if not soup: return 0 added = 0 excluded = ["draft", "preliminary", "summary", "factsheet", "infographic"] for a in soup.find_all("a", href=re.compile(r"\.pdf$", re.I)): if saved_count + added >= limit: break link_text = a.get_text(strip=True).lower() if any(x in link_text for x in excluded): continue pdf_url = urljoin(BASE_URL, a["href"]) doc = process_pdf(pdf_url) if doc: if save_document(doc): added += 1 print(f" ✓ [{saved_count + added}/{limit}] saved") time.sleep(1) return added def scrape_sccs(limit=MAX_PDFS): Base.metadata.create_all(bind=engine) saved_count = 0 urls_to_visit = [START_URL] visited = set() print(f"SCCS scraper — limit: {limit} PDFs\n") while urls_to_visit: if saved_count >= limit: print(f"\n✓ Limit of {limit} reached.") break current = urls_to_visit.pop(0) if current in visited: continue visited.add(current) print(f"\n{'='*60}") print(f"Exploring: {current.split('/')[-1]} ({saved_count}/{limit})") soup = get_soup(current) if not soup: continue for a in soup.find_all("a", href=True): if saved_count >= limit: break href = a["href"].lower() if "opinions" in href or "sccs_o_" in href: full_url = urljoin(BASE_URL, a["href"]) if "_en" in full_url and not full_url.endswith(".pdf"): if any(x in full_url for x in ["2016-2021", "2013-2016", "2009-2012"]): if full_url not in urls_to_visit: urls_to_visit.append(full_url) else: added = process_detail_page(full_url, saved_count, limit) saved_count += added print(f"\n{'='*60}") print(f"SCCS scraping complete: {saved_count}/{limit} PDFs") if __name__ == "__main__": scrape_sccs()