271 lines
8.6 KiB
Python
271 lines
8.6 KiB
Python
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() |