neckfireandClaude Opus 4.8
3d007c7013
Build & Deploy / build (push) Successful in 23s
- SECRET_KEY et DATABASE_URL externalisés (Vault -> env), plus de secret en dur - SQLite -> postgres-shared (psycopg2), schéma créé au boot (entrypoint.sh) - Dockerfile + docker-compose (Watchtower) + .gitea/workflows/build.yml Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
276 lines
9.3 KiB
Python
276 lines
9.3 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
|
|
|
|
|
|
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() |