717 lines
22 KiB
Python
717 lines
22 KiB
Python
#!/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"<!doctype html", b"<html")):
|
|
raise ValueError(f"Non-HTML content skipped: {content_type or 'unknown content-type'}")
|
|
charset = "utf-8"
|
|
match = re.search(r"charset=([\w.-]+)", content_type, re.I)
|
|
if match:
|
|
charset = match.group(1)
|
|
text = raw.decode(charset, errors="replace")
|
|
parser = LinkParser()
|
|
parser.feed(text)
|
|
links = [Link(normalize_url(urljoin(url, link.url)), link.text) for link in parser.links]
|
|
return PageResult(url=url, title=parser.title, links=links)
|
|
|
|
|
|
def head_pdf(url: str) -> 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:]))
|