380 lines
12 KiB
Python
380 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Classe les PDF déjà indexés dans extractions_pdf_indexed_all/.
|
||
|
||
Sorties :
|
||
- pdf_collecteur/state/classification.csv
|
||
- pdf_collecteur/state/sources_prioritaires.csv
|
||
|
||
Le classement est heuristique et traçable : chaque ligne contient un score et
|
||
des indices. Il sert à piloter la génération pédagogique, pas à remplacer la
|
||
validation humaine.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import csv
|
||
import re
|
||
import sys
|
||
import unicodedata
|
||
from collections import defaultdict
|
||
from pathlib import Path
|
||
|
||
|
||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||
PROGRAMME_DIR = SCRIPT_DIR.parent
|
||
DEFAULT_INDEX = PROGRAMME_DIR / "extractions_pdf_indexed_all"
|
||
DEFAULT_OUT = SCRIPT_DIR / "state"
|
||
|
||
|
||
CLASSIFICATION_FIELDS = [
|
||
"Id",
|
||
"RelativePath",
|
||
"Cycle",
|
||
"Matiere",
|
||
"Niveau",
|
||
"TypeDocument",
|
||
"UsagePrioritaire",
|
||
"Score",
|
||
"Indices",
|
||
"TextFile",
|
||
"InfoFile",
|
||
"SHA256",
|
||
]
|
||
|
||
PRIORITY_FIELDS = [
|
||
"PriorityRank",
|
||
"Cycle",
|
||
"Matiere",
|
||
"Niveau",
|
||
"TypeDocument",
|
||
"UsagePrioritaire",
|
||
"Score",
|
||
"Id",
|
||
"RelativePath",
|
||
"Indices",
|
||
]
|
||
|
||
|
||
LEVEL_TERMS = {
|
||
"CM1": ["cm1"],
|
||
"CM2": ["cm2"],
|
||
"6e": ["6e", "6eme", "sixième", "sixieme"],
|
||
"5e": ["5e", "5eme", "cinquième", "cinquieme"],
|
||
"4e": ["4e", "4eme", "quatrième", "quatrieme"],
|
||
"3e": ["3e", "3eme", "troisième", "troisieme"],
|
||
}
|
||
|
||
DOC_TYPE_TERMS = {
|
||
"programme_officiel": [
|
||
"programme d'enseignement",
|
||
"programme de ",
|
||
"annexe",
|
||
"boen",
|
||
"bulletin officiel",
|
||
],
|
||
"attendus_reperes": [
|
||
"attendus de fin d'année",
|
||
"attendus de fin d’annee",
|
||
"attendus de fin de cycle",
|
||
"repères annuels",
|
||
"reperes annuels",
|
||
"repères de progression",
|
||
"reperes de progression",
|
||
],
|
||
"guide_enseignant": [
|
||
"guide",
|
||
"vademecum",
|
||
"document maître",
|
||
"doc maitre",
|
||
"document maitre",
|
||
"ressource d'accompagnement",
|
||
"ressources d'accompagnement",
|
||
"mise en œuvre",
|
||
"mise en oeuvre",
|
||
"l'enseignant",
|
||
"enseignant",
|
||
],
|
||
"activite_sequence": [
|
||
"séquence",
|
||
"sequence",
|
||
"situation d'apprentissage",
|
||
"situation-problème",
|
||
"situation probleme",
|
||
"activité",
|
||
"activite",
|
||
"exemples de mise en œuvre",
|
||
"exemplesmiseenoeuvre",
|
||
],
|
||
"exercice": [
|
||
"exercice",
|
||
"questions flash",
|
||
"question flash",
|
||
"tâche",
|
||
"tache",
|
||
"calcul mental",
|
||
"compléter",
|
||
"complete",
|
||
"calcule",
|
||
"réponds",
|
||
"reponds",
|
||
],
|
||
"evaluation": [
|
||
"évaluation",
|
||
"evaluation",
|
||
"évaluer",
|
||
"evaluer",
|
||
"critères de réussite",
|
||
"criteres de reussite",
|
||
"niveau de maîtrise",
|
||
"niveau de maitrise",
|
||
"situations d'évaluation",
|
||
"situationsdevaluation",
|
||
],
|
||
}
|
||
|
||
USAGE_BY_TYPE = {
|
||
"programme_officiel": "cadrage_programme",
|
||
"attendus_reperes": "progression_niveau",
|
||
"guide_enseignant": "source_pedagogique",
|
||
"activite_sequence": "activite_transformable",
|
||
"exercice": "exercices",
|
||
"evaluation": "diagnostic_evaluation",
|
||
}
|
||
|
||
|
||
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 read_csv(path: Path) -> list[dict[str, str]]:
|
||
with path.open("r", encoding="utf-8-sig", newline="") as handle:
|
||
return list(csv.DictReader(handle))
|
||
|
||
|
||
def write_csv(path: Path, fields: list[str], rows: list[dict[str, str]]) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
with path.open("w", encoding="utf-8-sig", newline="") as handle:
|
||
writer = csv.DictWriter(handle, fieldnames=fields, extrasaction="ignore")
|
||
writer.writeheader()
|
||
writer.writerows(rows)
|
||
|
||
|
||
def infer_cycle(relative_path: str, text: str) -> tuple[str, list[str], int]:
|
||
folded = fold(relative_path + " " + text[:3000])
|
||
if relative_path.startswith("Cycle 3"):
|
||
return "cycle_3", ["path:Cycle 3"], 95
|
||
if relative_path.startswith("Cycle 4"):
|
||
return "cycle_4", ["path:Cycle 4"], 95
|
||
if "cycle 3" in folded or "cycle_3" in folded:
|
||
return "cycle_3", ["text:cycle 3"], 70
|
||
if "cycle 4" in folded or "cycle_4" in folded:
|
||
return "cycle_4", ["text:cycle 4"], 70
|
||
return "inconnu", ["cycle:non_detecte"], 10
|
||
|
||
|
||
def infer_subject(relative_path: str) -> tuple[str, list[str], int]:
|
||
parts = relative_path.split("\\")
|
||
if len(parts) >= 3 and parts[0] in {"Cycle 3", "Cycle 4"}:
|
||
return normalize_subject(parts[1]), [f"path:{parts[1]}"], 95
|
||
folded = fold(relative_path)
|
||
subject_terms = {
|
||
"mathematiques": ["math", "mathematiques", "mathématiques"],
|
||
"francais": ["francais", "français"],
|
||
"histoire_geographie": ["histoire", "geographie", "géographie"],
|
||
"histoire_des_arts": ["histoire des arts", "hart"],
|
||
"emc": ["enseignement moral", "civique", "emc"],
|
||
"physique_chimie": ["physique", "chimie", "phch"],
|
||
"svt": ["svt", "sciences de la vie"],
|
||
"eps": ["eps", "education physique", "éducation physique"],
|
||
"langues": ["anglais", "allemand", "langues"],
|
||
}
|
||
for subject, terms in subject_terms.items():
|
||
if any(fold(term) in folded for term in terms):
|
||
return subject, [f"path_term:{subject}"], 65
|
||
return "inconnu", ["matiere:non_detectee"], 10
|
||
|
||
|
||
def normalize_subject(value: str) -> str:
|
||
mapping = {
|
||
"Math": "mathematiques",
|
||
"Français": "francais",
|
||
"Histoire-Geographie": "histoire_geographie",
|
||
"Histoire des arts": "histoire_des_arts",
|
||
"Moral&Civisme": "emc",
|
||
"Physique-Chimie": "physique_chimie",
|
||
"SVT": "svt",
|
||
"EPS": "eps",
|
||
"Langues": "langues",
|
||
"Relationnel": "relationnel",
|
||
"Autre": "autre",
|
||
}
|
||
return mapping.get(value, fold(value).replace(" ", "_").replace("&", "_"))
|
||
|
||
|
||
def infer_level(relative_path: str, text: str, cycle: str) -> tuple[str, list[str], int]:
|
||
primary = fold(relative_path)
|
||
secondary = fold(text[:6000])
|
||
primary_scores: dict[str, int] = {}
|
||
secondary_scores: dict[str, int] = {}
|
||
indices: list[str] = []
|
||
for level, terms in LEVEL_TERMS.items():
|
||
for term in terms:
|
||
pattern = rf"(^|[^a-z0-9]){re.escape(fold(term))}([^a-z0-9]|$)"
|
||
if fold(term) in primary:
|
||
primary_scores[level] = primary_scores.get(level, 0) + 1
|
||
if re.search(pattern, secondary):
|
||
secondary_scores[level] = secondary_scores.get(level, 0) + 1
|
||
if primary_scores:
|
||
level, score = sorted(primary_scores.items(), key=lambda item: item[1], reverse=True)[0]
|
||
indices.append(f"niveau_path:{level}")
|
||
return level, indices, min(55 + score * 15, 95)
|
||
scores = secondary_scores
|
||
if scores:
|
||
level, score = sorted(scores.items(), key=lambda item: item[1], reverse=True)[0]
|
||
indices.append(f"niveau:{level}")
|
||
return level, indices, min(40 + score * 15, 90)
|
||
if cycle == "cycle_3":
|
||
return "cycle_3_general", ["niveau:cycle_3_general"], 35
|
||
if cycle == "cycle_4":
|
||
return "cycle_4_general", ["niveau:cycle_4_general"], 35
|
||
return "inconnu", ["niveau:non_detecte"], 10
|
||
|
||
|
||
def infer_doc_type(relative_path: str, text: str) -> tuple[str, list[str], int]:
|
||
primary = fold(relative_path)
|
||
secondary = fold(text[:12000])
|
||
scores: dict[str, int] = {}
|
||
indices: list[str] = []
|
||
for doc_type, terms in DOC_TYPE_TERMS.items():
|
||
for term in terms:
|
||
folded_term = fold(term)
|
||
if folded_term in primary:
|
||
scores[doc_type] = scores.get(doc_type, 0) + 3
|
||
if folded_term in secondary:
|
||
scores[doc_type] = scores.get(doc_type, 0) + 1
|
||
if not scores:
|
||
return "ressource_non_classee", ["type:non_classe"], 20
|
||
doc_type, raw_score = sorted(scores.items(), key=lambda item: item[1], reverse=True)[0]
|
||
for term in DOC_TYPE_TERMS[doc_type]:
|
||
if fold(term) in primary or fold(term) in secondary:
|
||
indices.append(f"type:{term}")
|
||
if len(indices) >= 3:
|
||
break
|
||
return doc_type, indices, min(35 + raw_score * 12, 95)
|
||
|
||
|
||
def classify_row(row: dict[str, str], index_dir: Path) -> dict[str, str]:
|
||
text_path = index_dir / row["TextFile"]
|
||
text = text_path.read_text(encoding="utf-8", errors="replace") if text_path.exists() else ""
|
||
relative_path = row["RelativePath"]
|
||
|
||
cycle, cycle_indices, cycle_score = infer_cycle(relative_path, text)
|
||
subject, subject_indices, subject_score = infer_subject(relative_path)
|
||
level, level_indices, level_score = infer_level(relative_path, text, cycle)
|
||
doc_type, type_indices, type_score = infer_doc_type(relative_path, text)
|
||
|
||
score = round((cycle_score * 0.25) + (subject_score * 0.25) + (level_score * 0.2) + (type_score * 0.3))
|
||
indices = cycle_indices + subject_indices + level_indices + type_indices
|
||
|
||
return {
|
||
"Id": row["Id"],
|
||
"RelativePath": relative_path,
|
||
"Cycle": cycle,
|
||
"Matiere": subject,
|
||
"Niveau": level,
|
||
"TypeDocument": doc_type,
|
||
"UsagePrioritaire": USAGE_BY_TYPE.get(doc_type, "a_trier"),
|
||
"Score": str(score),
|
||
"Indices": " | ".join(indices),
|
||
"TextFile": row["TextFile"],
|
||
"InfoFile": row["InfoFile"],
|
||
"SHA256": row.get("SHA256", ""),
|
||
}
|
||
|
||
|
||
def priority_rows(classified: list[dict[str, str]]) -> list[dict[str, str]]:
|
||
type_weight = {
|
||
"programme_officiel": 100,
|
||
"attendus_reperes": 90,
|
||
"guide_enseignant": 75,
|
||
"activite_sequence": 65,
|
||
"evaluation": 55,
|
||
"exercice": 50,
|
||
"ressource_non_classee": 10,
|
||
}
|
||
best_by_bucket: dict[tuple[str, str, str, str], list[dict[str, str]]] = defaultdict(list)
|
||
for row in classified:
|
||
bucket = (row["Cycle"], row["Matiere"], row["Niveau"], row["TypeDocument"])
|
||
best_by_bucket[bucket].append(row)
|
||
|
||
selected: list[dict[str, str]] = []
|
||
for rows in best_by_bucket.values():
|
||
rows.sort(
|
||
key=lambda row: (
|
||
type_weight.get(row["TypeDocument"], 0),
|
||
int(row["Score"] or 0),
|
||
),
|
||
reverse=True,
|
||
)
|
||
selected.extend(rows[:3])
|
||
|
||
selected.sort(
|
||
key=lambda row: (
|
||
row["Cycle"],
|
||
row["Matiere"],
|
||
row["Niveau"],
|
||
-type_weight.get(row["TypeDocument"], 0),
|
||
-int(row["Score"] or 0),
|
||
)
|
||
)
|
||
|
||
out: list[dict[str, str]] = []
|
||
for index, row in enumerate(selected, start=1):
|
||
out.append(
|
||
{
|
||
"PriorityRank": str(index),
|
||
"Cycle": row["Cycle"],
|
||
"Matiere": row["Matiere"],
|
||
"Niveau": row["Niveau"],
|
||
"TypeDocument": row["TypeDocument"],
|
||
"UsagePrioritaire": row["UsagePrioritaire"],
|
||
"Score": row["Score"],
|
||
"Id": row["Id"],
|
||
"RelativePath": row["RelativePath"],
|
||
"Indices": row["Indices"],
|
||
}
|
||
)
|
||
return out
|
||
|
||
|
||
def parse_args(argv: list[str]) -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description="Classe l'index PDF global.")
|
||
parser.add_argument("--index-dir", type=Path, default=DEFAULT_INDEX)
|
||
parser.add_argument("--out-dir", type=Path, default=DEFAULT_OUT)
|
||
return parser.parse_args(argv)
|
||
|
||
|
||
def main(argv: list[str]) -> int:
|
||
args = parse_args(argv)
|
||
manifest = args.index_dir / "manifest.csv"
|
||
rows = read_csv(manifest)
|
||
classified = [classify_row(row, args.index_dir) for row in rows]
|
||
priorities = priority_rows(classified)
|
||
|
||
write_csv(args.out_dir / "classification.csv", CLASSIFICATION_FIELDS, classified)
|
||
write_csv(args.out_dir / "sources_prioritaires.csv", PRIORITY_FIELDS, priorities)
|
||
|
||
print(f"PDF classés: {len(classified)}")
|
||
print(f"Sources prioritaires: {len(priorities)}")
|
||
print(f"Classification: {args.out_dir / 'classification.csv'}")
|
||
print(f"Sources: {args.out_dir / 'sources_prioritaires.csv'}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main(sys.argv[1:]))
|