#!/usr/bin/env python3 """ Collecteur prudent de PDF officiels pour Programme/Cycle 3 et Programme/Cycle 4. Fonctionne sans dépendances externes. Par défaut, le script fait un dry-run : il explore les pages, classe les PDF probables et écrit le manifest, mais ne télécharge pas les fichiers finaux sans --download. """ from __future__ import annotations import argparse import csv import hashlib import html import json import os import re import shutil import ssl import sys import time import unicodedata from collections import deque from dataclasses import dataclass from datetime import datetime, timezone from html.parser import HTMLParser from pathlib import Path from typing import Iterable from urllib.error import HTTPError, URLError from urllib.parse import urldefrag, urljoin, urlparse, unquote from urllib.request import Request, urlopen SCRIPT_DIR = Path(__file__).resolve().parent PROGRAMME_DIR = SCRIPT_DIR.parent STATE_DIR = SCRIPT_DIR / "state" TMP_DIR = STATE_DIR / "downloads_tmp" DEFAULT_CONFIG = SCRIPT_DIR / "seeds.json" USER_AGENT = ( "OpenSchool-Programme-PDF-Collector/0.1 " "(local educational archive; contact: local-admin)" ) SSL_CONTEXT = None SUBJECT_DIRS = { "math": "Math", "mathematiques": "Math", "mathématiques": "Math", "francais": "Français", "français": "Français", "grammaire": "Français", "lecture": "Français", "comprehension": "Français", "compréhension": "Français", "langue": "Langues", "langues": "Langues", "anglais": "Langues", "allemand": "Langues", "espagnol": "Langues", "emc": "Moral&Civisme", "moral": "Moral&Civisme", "civique": "Moral&Civisme", "citoyennete": "Moral&Civisme", "citoyenneté": "Moral&Civisme", "histoire": "Histoire-Geographie", "geographie": "Histoire-Geographie", "géographie": "Histoire-Geographie", "arts": "Histoire des arts", "physique": "Physique-Chimie", "chimie": "Physique-Chimie", "svt": "SVT", "sciences de la vie": "SVT", "technologie": "Technologie", "eps": "EPS", "education physique": "EPS", "éducation physique": "EPS", "relationnelle": "Relationnel", "affective": "Relationnel", } EXCLUDED_CRAWL_TERMS = [ "lycee", "lycée", "voie professionnelle", "voie-professionnelle", "baccalaureat", "baccalauréat", "terminale", "premiere", "première", "seconde", ] RELEVANT_CRAWL_TERMS = [ "cycle 3", "cycle-3", "cycle 4", "cycle-4", "c3", "c4", "cm1", "cm2", "6e", "5e", "4e", "3e", "college", "collège", "mathematiques", "mathématiques", "francais", "français", "histoire", "geographie", "géographie", "svt", "technologie", "physique", "chimie", "langues", "anglais", "allemand", "emc", "moral", "civique", ] MANIFEST_FIELDS = [ "seen_at", "status", "cycle", "subject", "confidence", "local_path", "filename", "sha256", "source_url", "discovered_from", "link_text", "page_title", "content_length", "last_modified", "etag", "notes", ] @dataclass(frozen=True) class Seed: url: str cycle_hint: str = "" subject_hint: str = "" @dataclass class Link: url: str text: str @dataclass class PageResult: url: str title: str links: list[Link] @dataclass class PdfCandidate: url: str discovered_from: str page_title: str link_text: str cycle: str subject: str confidence: int filename: str content_length: str = "" last_modified: str = "" etag: str = "" class LinkParser(HTMLParser): def __init__(self) -> None: super().__init__() self.links: list[Link] = [] self._current_href: str | None = None self._current_text: list[str] = [] self._in_title = False self._title_parts: list[str] = [] def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: if tag.lower() == "a": href = dict(attrs).get("href") if href: self._current_href = href self._current_text = [] elif tag.lower() == "title": self._in_title = True def handle_endtag(self, tag: str) -> None: if tag.lower() == "a" and self._current_href: text = normalize_space(" ".join(self._current_text)) self.links.append(Link(self._current_href, text)) self._current_href = None self._current_text = [] elif tag.lower() == "title": self._in_title = False def handle_data(self, data: str) -> None: if self._current_href: self._current_text.append(data) if self._in_title: self._title_parts.append(data) @property def title(self) -> str: return normalize_space(" ".join(self._title_parts)) def normalize_space(value: str) -> str: return re.sub(r"\s+", " ", html.unescape(value or "")).strip() def fold(value: str) -> str: value = unicodedata.normalize("NFKD", value or "") value = "".join(ch for ch in value if not unicodedata.combining(ch)) return value.casefold() def is_allowed_url(url: str, allowed_domains: set[str]) -> bool: parsed = urlparse(url) if parsed.scheme not in {"http", "https"}: return False host = (parsed.hostname or "").lower() return any(host == domain or host.endswith("." + domain) for domain in allowed_domains) def normalize_url(url: str) -> str: clean, _fragment = urldefrag(url) return clean def looks_like_pdf(url: str) -> bool: parsed = urlparse(url) path = unquote(parsed.path).lower() return path.endswith(".pdf") or ".pdf/" in path or "pdf" in path.rsplit("/", 1)[-1] def should_enqueue_page(url: str, link_text: str, seed: Seed) -> bool: text = fold(" ".join([url, link_text, seed.cycle_hint, seed.subject_hint])) if any(term in text for term in map(fold, EXCLUDED_CRAWL_TERMS)): return False return any(term in text for term in map(fold, RELEVANT_CRAWL_TERMS)) def request_url(url: str, method: str = "GET", timeout: int = 25): request = Request(url, method=method, headers={"User-Agent": USER_AGENT}) return urlopen(request, timeout=timeout, context=SSL_CONTEXT) def fetch_page(url: str) -> PageResult: with request_url(url) as response: raw = response.read() content_type = response.headers.get("Content-Type", "") leading = raw[:200].lstrip().lower() if "html" not in content_type.casefold() and not leading.startswith((b" dict[str, str]: try: with request_url(url, method="HEAD", timeout=20) as response: return { "content_length": response.headers.get("Content-Length", ""), "last_modified": response.headers.get("Last-Modified", ""), "etag": response.headers.get("ETag", ""), "content_type": response.headers.get("Content-Type", ""), } except Exception: return {} def classify_candidate( url: str, link_text: str, page_title: str, seed: Seed, ) -> tuple[str, str, int]: confidence = 0 cycle_scores = {"Cycle 3": 0, "Cycle 4": 0} primary_text = fold(" ".join([url, link_text])) secondary_text = fold(page_title) cycle3_terms = ["cycle 3", "cycle-3", "c3", "cm1", "cm2", "6e", "6eme", "cours moyen"] cycle4_terms = ["cycle 4", "cycle-4", "c4", "5e", "5eme", "4e", "4eme", "3e", "3eme", "college"] for term in cycle3_terms: if term in primary_text: cycle_scores["Cycle 3"] += 3 if term in secondary_text: cycle_scores["Cycle 3"] += 1 for term in cycle4_terms: if term in primary_text: cycle_scores["Cycle 4"] += 3 if term in secondary_text: cycle_scores["Cycle 4"] += 1 if seed.cycle_hint in cycle_scores: cycle_scores[seed.cycle_hint] += 1 if cycle_scores["Cycle 3"] > cycle_scores["Cycle 4"]: cycle = "Cycle 3" confidence += min(cycle_scores["Cycle 3"] * 10, 45) elif cycle_scores["Cycle 4"] > cycle_scores["Cycle 3"]: cycle = "Cycle 4" confidence += min(cycle_scores["Cycle 4"] * 10, 45) else: cycle = seed.cycle_hint if seed.cycle_hint else "A_trier" confidence += 10 if seed.cycle_hint else 0 subject_scores: dict[str, int] = {} for term, directory in SUBJECT_DIRS.items(): folded_term = fold(term) if folded_term in primary_text: subject_scores[directory] = subject_scores.get(directory, 0) + 2 if folded_term in secondary_text: subject_scores[directory] = subject_scores.get(directory, 0) + 1 if seed.subject_hint: subject_scores[seed.subject_hint] = subject_scores.get(seed.subject_hint, 0) + 1 if subject_scores: subject = sorted(subject_scores.items(), key=lambda item: item[1], reverse=True)[0][0] confidence += min(subject_scores[subject] * 10, 35) else: subject = "A_trier" if looks_like_pdf(url): confidence += 15 if link_text: confidence += 5 return cycle, subject, min(confidence, 100) def safe_filename_from_url(url: str, fallback: str = "document.pdf") -> str: path = unquote(urlparse(url).path) name = Path(path).name or fallback if not name.lower().endswith(".pdf"): name = f"{name}.pdf" name = normalize_space(name) name = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "-", name) name = re.sub(r"\s+", " ", name).strip(" .") return name[:180] or fallback def unique_path(directory: Path, filename: str) -> Path: candidate = directory / filename if not candidate.exists(): return candidate stem = candidate.stem suffix = candidate.suffix index = 2 while True: next_candidate = directory / f"{stem} ({index}){suffix}" if not next_candidate.exists(): return next_candidate index += 1 def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest().upper() def scan_existing_pdfs(root: Path) -> dict[str, Path]: existing: dict[str, Path] = {} for pdf in root.rglob("*.pdf"): if SCRIPT_DIR in pdf.parents: continue try: existing[sha256_file(pdf)] = pdf except OSError: continue return existing def load_manifest_hashes(path: Path) -> set[str]: hashes: set[str] = set() if not path.exists(): return hashes with path.open("r", encoding="utf-8-sig", newline="") as handle: reader = csv.DictReader(handle) for row in reader: value = (row.get("sha256") or "").strip().upper() if value: hashes.add(value) return hashes def append_manifest(path: Path, row: dict[str, str]) -> None: path.parent.mkdir(parents=True, exist_ok=True) exists = path.exists() with path.open("a", encoding="utf-8-sig", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=MANIFEST_FIELDS, extrasaction="ignore") if not exists: writer.writeheader() writer.writerow({field: row.get(field, "") for field in MANIFEST_FIELDS}) def log_event(path: Path, event: str, **payload: object) -> None: path.parent.mkdir(parents=True, exist_ok=True) record = { "time": datetime.now(timezone.utc).isoformat(), "event": event, **payload, } with path.open("a", encoding="utf-8") as handle: handle.write(json.dumps(record, ensure_ascii=False) + "\n") def download_pdf(url: str, tmp_dir: Path) -> Path: tmp_dir.mkdir(parents=True, exist_ok=True) tmp_path = tmp_dir / (hashlib.sha256(url.encode("utf-8")).hexdigest() + ".pdf") with request_url(url, timeout=60) as response, tmp_path.open("wb") as handle: shutil.copyfileobj(response, handle) return tmp_path def target_directory(root: Path, cycle: str, subject: str) -> Path: if cycle not in {"Cycle 3", "Cycle 4"}: return root / "A_trier" / "cycle_inconnu" subject_dir = subject if subject != "A_trier" else "A_trier" return root / cycle / subject_dir def load_config(path: Path) -> tuple[set[str], list[Seed], int, float]: with path.open("r", encoding="utf-8") as handle: data = json.load(handle) allowed = set(data.get("allowed_domains", [])) seeds = [ Seed( url=item["url"], cycle_hint=item.get("cycle_hint", ""), subject_hint=item.get("subject_hint", ""), ) for item in data.get("seeds", []) ] return ( allowed, seeds, int(data.get("default_max_depth", 2)), float(data.get("request_delay_seconds", 0.6)), ) def crawl( seeds: list[Seed], allowed_domains: set[str], max_depth: int, delay: float, limit_pages: int | None, log_path: Path, ) -> list[PdfCandidate]: queue: deque[tuple[str, int, Seed]] = deque((seed.url, 0, seed) for seed in seeds) visited_pages: set[str] = set() seen_pdfs: set[str] = set() candidates: list[PdfCandidate] = [] while queue: if limit_pages is not None and len(visited_pages) >= limit_pages: log_event(log_path, "page_limit_reached", limit_pages=limit_pages) break url, depth, seed = queue.popleft() url = normalize_url(url) if url in visited_pages or not is_allowed_url(url, allowed_domains): continue if looks_like_pdf(url): continue visited_pages.add(url) log_event(log_path, "fetch_page", url=url, depth=depth) try: page = fetch_page(url) except (AssertionError, HTTPError, URLError, TimeoutError, UnicodeError, ValueError, OSError) as exc: log_event(log_path, "fetch_page_error", url=url, error=str(exc)) continue time.sleep(delay) for link in page.links: link_url = normalize_url(link.url) if not is_allowed_url(link_url, allowed_domains): continue if looks_like_pdf(link_url): if link_url in seen_pdfs: continue seen_pdfs.add(link_url) cycle, subject, confidence = classify_candidate( link_url, link.text, page.title, seed, ) headers = head_pdf(link_url) candidates.append( PdfCandidate( url=link_url, discovered_from=page.url, page_title=page.title, link_text=link.text, cycle=cycle, subject=subject, confidence=confidence, filename=safe_filename_from_url(link_url), content_length=headers.get("content_length", ""), last_modified=headers.get("last_modified", ""), etag=headers.get("etag", ""), ) ) log_event( log_path, "pdf_candidate", url=link_url, cycle=cycle, subject=subject, confidence=confidence, ) time.sleep(delay) elif depth < max_depth and link_url not in visited_pages and should_enqueue_page(link_url, link.text, seed): queue.append((link_url, depth + 1, seed)) return candidates def process_candidates( candidates: Iterable[PdfCandidate], programme_dir: Path, manifest_path: Path, log_path: Path, download: bool, ) -> tuple[int, int, int]: existing_hashes = scan_existing_pdfs(programme_dir) manifest_hashes = load_manifest_hashes(manifest_path) seen_urls: set[str] = set() new_count = 0 duplicate_count = 0 dry_count = 0 for candidate in candidates: if candidate.url in seen_urls: continue seen_urls.add(candidate.url) now = datetime.now(timezone.utc).isoformat() row = { "seen_at": now, "cycle": candidate.cycle, "subject": candidate.subject, "confidence": str(candidate.confidence), "filename": candidate.filename, "source_url": candidate.url, "discovered_from": candidate.discovered_from, "link_text": candidate.link_text, "page_title": candidate.page_title, "content_length": candidate.content_length, "last_modified": candidate.last_modified, "etag": candidate.etag, } if not download: row.update({"status": "dry_run", "notes": "Not downloaded. Re-run with --download."}) append_manifest(manifest_path, row) dry_count += 1 continue try: tmp_path = download_pdf(candidate.url, TMP_DIR) digest = sha256_file(tmp_path) except Exception as exc: row.update({"status": "download_error", "notes": str(exc)}) append_manifest(manifest_path, row) log_event(log_path, "download_error", url=candidate.url, error=str(exc)) continue row["sha256"] = digest if digest in existing_hashes: existing_path = existing_hashes[digest] row.update( { "status": "duplicate_existing_file", "local_path": str(existing_path.relative_to(programme_dir)), "notes": "SHA256 already present in Programme.", } ) tmp_path.unlink(missing_ok=True) duplicate_count += 1 elif digest in manifest_hashes: row.update({"status": "duplicate_manifest", "notes": "SHA256 already present in manifest."}) tmp_path.unlink(missing_ok=True) duplicate_count += 1 else: directory = target_directory(programme_dir, candidate.cycle, candidate.subject) directory.mkdir(parents=True, exist_ok=True) final_path = unique_path(directory, candidate.filename) shutil.move(str(tmp_path), final_path) existing_hashes[digest] = final_path manifest_hashes.add(digest) row.update( { "status": "downloaded", "local_path": str(final_path.relative_to(programme_dir)), "notes": "", } ) new_count += 1 append_manifest(manifest_path, row) return dry_count, new_count, duplicate_count def parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser(description="Collecte les PDF officiels cycles 3/4.") parser.add_argument("--config", type=Path, default=DEFAULT_CONFIG, help="Chemin vers seeds.json.") parser.add_argument("--download", action="store_true", help="Télécharge et range les nouveaux PDF.") parser.add_argument("--dry-run", action="store_true", help="Force le mode simulation.") parser.add_argument("--max-depth", type=int, default=None, help="Profondeur maximale de crawl.") parser.add_argument("--limit-pages", type=int, default=None, help="Limite de pages HTML à explorer.") parser.add_argument("--manifest", type=Path, default=STATE_DIR / "manifest.csv") parser.add_argument("--log", type=Path, default=STATE_DIR / "crawl_log.jsonl") parser.add_argument( "--fresh", action="store_true", help="Supprime manifest/log existants avant de lancer la collecte.", ) parser.add_argument( "--insecure", action="store_true", help="Désactive la vérification TLS si les certificats locaux sont mal configurés.", ) return parser.parse_args(argv) def main(argv: list[str]) -> int: global SSL_CONTEXT args = parse_args(argv) if args.insecure: SSL_CONTEXT = ssl._create_unverified_context() if args.fresh: args.manifest.unlink(missing_ok=True) args.log.unlink(missing_ok=True) allowed_domains, seeds, default_depth, delay = load_config(args.config) max_depth = args.max_depth if args.max_depth is not None else default_depth download = bool(args.download and not args.dry_run) if not seeds: print("Aucune seed configurée.", file=sys.stderr) return 2 STATE_DIR.mkdir(parents=True, exist_ok=True) log_event(args.log, "run_start", download=download, max_depth=max_depth, seed_count=len(seeds)) candidates = crawl( seeds=seeds, allowed_domains=allowed_domains, max_depth=max_depth, delay=delay, limit_pages=args.limit_pages, log_path=args.log, ) dry_count, new_count, duplicate_count = process_candidates( candidates=candidates, programme_dir=PROGRAMME_DIR, manifest_path=args.manifest, log_path=args.log, download=download, ) log_event( args.log, "run_end", candidates=len(candidates), dry_run_rows=dry_count, downloaded=new_count, duplicates=duplicate_count, ) mode = "download" if download else "dry-run" print(f"Mode: {mode}") print(f"PDF candidats: {len(candidates)}") print(f"Lignes dry-run: {dry_count}") print(f"Nouveaux téléchargés: {new_count}") print(f"Doublons ignorés: {duplicate_count}") print(f"Manifest: {args.manifest}") print(f"Log: {args.log}") return 0 if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))