Sécurité : validation EmailStr + tests complets 57/57 + RGPD
This commit is contained in:
Mouignihazi
@@ -0,0 +1,276 @@
|
||||
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
|
||||
|
||||
|
||||
headers = {
|
||||
"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"
|
||||
}
|
||||
|
||||
session = requests.Session()
|
||||
session.headers.update(headers)
|
||||
|
||||
BASE_URL = "https://www.cir-safety.org"
|
||||
MEETING_NUMBERS = range(115, 175)
|
||||
PDF_LIMIT = 100
|
||||
|
||||
url_templates = [
|
||||
"https://www.cir-safety.org/meeting/{num}th-expert-panel-meeting",
|
||||
"https://www.cir-safety.org/meeting/{num}st-expert-panel-meeting",
|
||||
"https://www.cir-safety.org/meeting/{num}nd-expert-panel-meeting",
|
||||
"https://www.cir-safety.org/meeting/{num}rd-expert-panel-meeting",
|
||||
"https://www.cir-safety.org/meeting/{num}th-cir-expert-panel-meeting",
|
||||
"https://www.cir-safety.org/meeting/{num}st-cir-expert-panel-meeting",
|
||||
]
|
||||
|
||||
PDF_RE = re.compile(r"\.pdf", re.I)
|
||||
SKIP_KEYWORDS = ["Agenda", "Minutes", "Status Report"]
|
||||
|
||||
|
||||
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) -> 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": "document",
|
||||
"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()
|
||||
|
||||
# Extrait le JSON même s'il y a du texte autour
|
||||
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 "document").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": "document",
|
||||
"date": None,
|
||||
}
|
||||
|
||||
|
||||
def get_working_url(num):
|
||||
for template in url_templates:
|
||||
url = template.format(num=num)
|
||||
try:
|
||||
r = session.get(url, timeout=20)
|
||||
if r.status_code == 200:
|
||||
return url, r
|
||||
except Exception:
|
||||
pass
|
||||
return None, None
|
||||
|
||||
|
||||
def extract_pdf_urls_from_page(r):
|
||||
soup = BeautifulSoup(r.text, "html.parser")
|
||||
all_pdf_links = soup.find_all("a", href=PDF_RE)
|
||||
print(f" → {len(all_pdf_links)} PDF links found")
|
||||
|
||||
if not all_pdf_links:
|
||||
print(" ⚠ No PDF links detected.")
|
||||
return []
|
||||
|
||||
urls = []
|
||||
seen = set()
|
||||
|
||||
for link in all_pdf_links:
|
||||
href = link.get("href", "")
|
||||
pdf_url = urljoin(BASE_URL, href)
|
||||
|
||||
if pdf_url in seen:
|
||||
continue
|
||||
seen.add(pdf_url)
|
||||
|
||||
context_el = None
|
||||
for tag in ("tr", "li", "div", "p"):
|
||||
context_el = link.find_parent(tag)
|
||||
if context_el:
|
||||
break
|
||||
text = context_el.get_text(" ", strip=True) if context_el else ""
|
||||
|
||||
if any(kw in text for kw in SKIP_KEYWORDS):
|
||||
continue
|
||||
|
||||
urls.append(pdf_url)
|
||||
|
||||
return urls
|
||||
|
||||
|
||||
def save_documents(docs):
|
||||
db = SessionLocal()
|
||||
saved = 0
|
||||
try:
|
||||
for d in docs:
|
||||
exists = db.query(Document).filter(Document.pdf_url == d["pdf_url"]).first()
|
||||
if not exists:
|
||||
db.add(Document(**d))
|
||||
saved += 1
|
||||
db.commit()
|
||||
print(f" ✓ {saved} new document(s) saved")
|
||||
except Exception as e:
|
||||
print(f" DB error: {e}")
|
||||
db.rollback()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def scrape_cir(limit=PDF_LIMIT):
|
||||
Base.metadata.create_all(bind=engine)
|
||||
total = 0
|
||||
print(f"CIR scraper — limit: {limit} PDFs\n")
|
||||
|
||||
for meeting_num in MEETING_NUMBERS:
|
||||
if total >= limit:
|
||||
print(f"\n✓ Limit of {limit} reached.")
|
||||
break
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print(f"Meeting: {meeting_num} ({total}/{limit})")
|
||||
|
||||
url, r = get_working_url(meeting_num)
|
||||
if not url:
|
||||
print(" ✗ No page found")
|
||||
continue
|
||||
|
||||
print(f" ✓ {url}")
|
||||
|
||||
try:
|
||||
pdf_urls = extract_pdf_urls_from_page(r)
|
||||
docs_to_save = []
|
||||
|
||||
for pdf_url in pdf_urls:
|
||||
if total >= limit:
|
||||
break
|
||||
|
||||
print(f"\n Downloading: {pdf_url.split('/')[-1]}")
|
||||
|
||||
try:
|
||||
pdf_resp = session.get(pdf_url, timeout=30)
|
||||
if pdf_resp.status_code != 200:
|
||||
print(f" ✗ HTTP {pdf_resp.status_code}")
|
||||
continue
|
||||
|
||||
text = extract_pages_text(pdf_resp.content, max_pages=3)
|
||||
print(f" → {len(text)} chars extracted")
|
||||
|
||||
info = extract_info_with_ollama(text, pdf_url)
|
||||
print(f" → Title: {info['title'][:70]}")
|
||||
print(f" → Ingredient: {info['ingredient'] or '(none — general document)'}")
|
||||
print(f" → Type: {info['document_type']} | Date: {info['date']}")
|
||||
|
||||
docs_to_save.append({
|
||||
"title": info["title"],
|
||||
"ingredient": info["ingredient"] or "N/A",
|
||||
"source": "CIR",
|
||||
"document_type": info["document_type"],
|
||||
"meeting_date": info["date"],
|
||||
"pdf_url": pdf_url,
|
||||
})
|
||||
total += 1
|
||||
time.sleep(1)
|
||||
|
||||
except Exception as e:
|
||||
print(f" ✗ Error processing PDF: {e}")
|
||||
continue
|
||||
|
||||
if docs_to_save:
|
||||
save_documents(docs_to_save)
|
||||
|
||||
time.sleep(2)
|
||||
|
||||
except Exception as e:
|
||||
print(f" ✗ Failed: {e}")
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print(f"CIR scraping complete: {total}/{limit} PDFs")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
scrape_cir()
|
||||
Reference in New Issue
Block a user