from __future__ import annotations
import base64
import html
import json
import os
from copy import deepcopy
from io import BytesIO
from pathlib import Path
from typing import Any
import gradio as gr
import requests
from PIL import Image
DEFAULT_VLM_URL = os.getenv("VLM_API_URL", "https://vlm.smartbiblia.fr/v1/chat/completions")
DEFAULT_VLM_MODEL = os.getenv("VLM_MODEL", "Qwen3-VL-8B-Instruct-GGUF")
#DEFAULT_VLM_URL = os.getenv("VLM_API_URL", "https://mqt7w4m4abb63whh.eu-west-1.aws.endpoints.huggingface.cloud/v1/chat/completions")
#DEFAULT_VLM_MODEL = os.getenv("VLM_MODEL", "unsloth/Qwen3-VL-8B-Instruct-GGUF")
DEFAULT_OPENAI_URL = os.getenv("OPENAI_API_URL", "https://api.openai.com/v1/chat/completions")
DEFAULT_OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o")
DEFAULT_HF_URL = os.getenv("HF_API_URL", "https://router.huggingface.co/v1/chat/completions")
HF_MODEL_CHOICES = [
"google/gemma-3-12b-it:featherless-ai",
"Qwen/Qwen3.6-35B-A3B:featherless-ai",
"moonshotai/Kimi-K2.6:fireworks-ai",
]
DEFAULT_HF_MODEL = os.getenv("HF_MODEL", HF_MODEL_CHOICES[0])
DEFAULT_ALBERT_URL = os.getenv("ALBERT_API_URL", "https://albert.api.etalab.gouv.fr/v1/chat/completions")
DEFAULT_ALBERT_MODEL = os.getenv("ALBERT_MODEL", "mistralai/Ministral-3-8B-Instruct-2512")
# datalab-to/lift, quantized Q8 (GGUF), served by llama.cpp on a HF dedicated endpoint.
# Strict JSON-schema model: uses its own prompt template (build_lift_prompt) and
# returns arrays for multivalued fields, coerced to pipe-strings downstream.
DEFAULT_LIFT_URL = os.getenv("LIFT_API_URL", "https://jpectntdw48b1jip.eu-west-1.aws.endpoints.huggingface.cloud/v1/chat/completions")
DEFAULT_LIFT_MODEL = os.getenv("LIFT_MODEL", "prithivMLmods/lift-GGUF")
#DEFAULT_LIFT_URL = os.getenv("LIFT_API_URL", "https://vlm.smartbiblia.fr/v1/chat/completions")
#DEFAULT_LIFT_MODEL = os.getenv("LIFT_MODEL", "lift-GGUF")
DEFAULT_SUDOC_URL = os.getenv("SUDOC_CHECKER_API_URL", "https://sudoc-checker.smartbiblia.fr")
DEFAULT_IDREF_URL = os.getenv("IDREF_QUALINKA_API_URL", "https://idref-linker.smartbiblia.fr")
DEFAULT_IDREF_API_KEY = os.getenv("IDREF_QUALINKA_API_KEY", "")
DEFAULT_DEWEY_URL = os.getenv("DEWEY_CLASSIFICATION_API_URL", "https://dewey-classifier.smartbiblia.fr")
DEFAULT_DEWEY_API_KEY = os.getenv("CLASSIFICATION_API_KEY", "")
EMBEDDING_MODEL = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
# Poids de scoring du service idref-linker (voir README de humatheque-idref-qualinka-api).
DEFAULT_WEIGHT_NAME = float(os.getenv("IDREF_WEIGHT_NAME", "0.40"))
DEFAULT_WEIGHT_ATTRRA_SOURCE = float(os.getenv("IDREF_WEIGHT_ATTRRA_SOURCE", "0.25"))
DEFAULT_WEIGHT_ATTRRA_NOTE = float(os.getenv("IDREF_WEIGHT_ATTRRA_NOTE", "0.15"))
DEFAULT_WEIGHT_REFERENCES = float(os.getenv("IDREF_WEIGHT_REFERENCES", "0.15"))
DEFAULT_WEIGHT_INSTITUTION_YEAR = float(os.getenv("IDREF_WEIGHT_INSTITUTION_YEAR", "0.05"))
# Title-page scans can be huge (e.g. 2597x3670), which makes base64 payloads slow
# to upload and slow for the on-premise llama.cpp VLM to process. Downscale the
# longest side and re-encode as JPEG before sending.
VLM_IMAGE_MAX_SIZE = int(os.getenv("VLM_IMAGE_MAX_SIZE", "768")) #512, 1024,1248
VLM_IMAGE_JPEG_QUALITY = int(os.getenv("VLM_IMAGE_JPEG_QUALITY", "85"))
THESIS_DEGREE_TYPE_VALUES = [
"Thèse d'État",
"Thèse de doctorat",
"Thèse de 3e cycle",
"Thèse d'université",
"Thèse de docteur-ingénieur",
"Thèse d'exercice",
]
DISSERTATION_DEGREE_TYPE_VALUES = [
"Habilitation à diriger des recherches",
"Mémoire de DEA",
"Mémoire de DES",
"Mémoire de DESS",
"Mémoire de DU",
"Mémoire de DIU",
"Mémoire de DUT",
"Mémoire de maîtrise",
"Mémoire de master professionnel 1re année",
"Mémoire de master professionnel 2e année",
"Mémoire de master recherche 1re année",
"Mémoire de master recherche 2e année",
]
METADATA_FIELDS = [
"title",
"subtitle",
"author",
"degree_type",
"discipline",
"granting_institution",
"co_tutelle_institutions",
"doctoral_school",
"defense_year",
"advisor",
"jury_president",
"reviewers",
"committee_members",
"language",
"confidence",
]
PERSON_FIELDS = ["author", "advisor", "jury_president", "reviewers", "committee_members"]
ROLE_LABELS = {
"author": "Auteur",
"advisor": "Directeur",
"jury_president": "President du jury",
"reviewers": "Rapporteur",
"committee_members": "Membre du jury",
}
PERSON_ROLE_CODES = {
"author": "070",
"advisor_thesis": "727",
"advisor_dissertation": "003",
"jury_president": "956",
"reviewers": "958",
"committee_members": "555",
}
CORPORATE_ROLE_CODES = {
"granting_institution": "295",
"co_tutelle_institutions": "995",
"doctoral_school": "996",
"partner_institutions": "985",
}
STRONG_IDREF_STATUSES = {"accepted"}
EXAMPLE_IMAGE_URL = "https://minio.smartbiblia.fr/images/theses/theses.fr/2015EPHE4076/p1.png"
def empty_state() -> dict[str, Any]:
return {
"input": {"doc_type": "these", "image_url": "", "image_filename": ""},
"vlm": {"request": None, "raw_response": None, "extracted": None, "corrected": None, "error": None},
"sudoc": {"request": None, "response": None, "error": None},
"idref": {"persons": [], "settings": {}, "error": None},
"dewey": {"request": None, "response": None, "classes": [], "selected": None, "error": None},
"record_draft": None,
"events": [],
}
def add_event(state: dict[str, Any], step: str, status: str, message: str) -> dict[str, Any]:
state = deepcopy(state or empty_state())
state.setdefault("events", []).append({"step": step, "status": status, "message": message})
return state
def build_eval_prompt(doc_type: str) -> str:
doc_type_norm = str(doc_type or "").strip().lower()
if doc_type_norm == "memoire":
degree_type_values = DISSERTATION_DEGREE_TYPE_VALUES
document_label = "graduate dissertation"
else:
degree_type_values = THESIS_DEGREE_TYPE_VALUES
document_label = "graduate thesis"
degree_type_str = " | ".join(degree_type_values)
return f"""You are extracting metadata from a French {document_label} title page.
General rules:
- Return ONLY valid JSON.
- Do not add markdown.
- Do not add explanations.
- Do not invent missing data.
- Use null when a field is absent.
- Use an empty array [] for co_tutelle_institutions when none are present.
- If source text is written in all-caps, transform it in sentence case with french accents.
- For person names, apply sentence case (e.g. Jean-Daniel DUBOIS → Jean-Daniel Dubois, MARIE-FRANCE Dupont → Marie-France Dupond) and remove all civility and title prefixes (e.g. Monsieur, Madame, M., Mme, Professeur, Pr, Dr, etc.) before returning the name.
Role attribution rules:
- A role is explicit only when a textual role label is directly associated with the person.
- Assign a person to a specific role ONLY if that role is explicitly written on the title page.
- Do NOT infer roles from position, ordering, typography, academic conventions, or prior knowledge.
- Do not infer role equivalence from document structure or visual grouping.
- If people are listed without explicit role labels, place them ONLY in committee_members.
- If no jury or committee members are explicitly listed, use null.
- Do NOT guess who is advisor, reviewer, or jury president.
{{
"title": "Main title as it appears on the title page",
"subtitle": "Subtitle or remainder of the title, usually following a colon; null if not present",
"author": "Full name of the author (student) who wrote the {document_label}",
"degree_type": "Academic degree sought by the author. Possible values are {degree_type_str}",
"discipline": "Academic field or discipline of the {document_label}.",
"granting_institution": "Institution where the {document_label} was submitted and the degree is granted",
"co_tutelle_institutions": "List of institutions involved in a joint supervision or co-tutelle agreement; empty list if none",
"doctoral_school": "Doctoral school or graduate program, if explicitly mentioned",
"defense_year": "Year the {document_label} was defended. Format yyyy",
"advisor": "ONLY persons explicitly identified as {document_label} advisor/supervisor/director. Use | as separator. Null if the role is not explicitly stated.",
"jury_president": "ONLY the person explicitly identified as president/chair of the jury. Null if not explicitly stated.",
"reviewers": "ONLY official reviewers/rapporteurs explicitly identified as such. Use | as separator. Null if not explicitly stated.",
"committee_members": "Persons explicitly listed as jury or committee members whose role is not explicitly identified as president/chair or reviewer/rapporteur. Include unlabeled persons here. Use | as separator.",
"language": "Language in ISO 639-3 codes. Example: fre, eng, ita...",
"confidence": "Confidence score between 0.0 and 1.0 reflecting certainty based only on explicitly visible evidence on the page"
}}"""
def build_lift_prompt(doc_type: str) -> str:
"""Prompt for the datalab-to/lift model, which enforces a strict JSON schema and
is prompted differently from the on-prem VLM: a schema-driven extraction template
rather than the free-form role-attribution instructions of build_eval_prompt."""
doc_type_norm = str(doc_type or "").strip().lower()
if doc_type_norm == "memoire":
degree_type_values = DISSERTATION_DEGREE_TYPE_VALUES
document_label = "graduate dissertation"
else:
degree_type_values = THESIS_DEGREE_TYPE_VALUES
document_label = "graduate thesis"
degree_type_str = " | ".join(degree_type_values)
schema = {
"type": "object",
"properties": {
"title": {"type": "string", "description": "Main title as it appears on the title page. Apply sentence case."},
"subtitle": {"type": "string", "description": "Subtitle or remainder of the title, usually following a colon; null if not present. Apply sentence case."},
"author": {"type": "string", "description": f"Full name of the author (student) who wrote the {document_label}. Apply sentence case and remove all civility/title prefixes (Monsieur, Madame, M., Mme, Professeur, Pr, Dr, etc.)."},
"degree_type": {"type": "string", "enum": list(degree_type_values), "description": f"Academic degree sought by the author. Possible values are {degree_type_str}"},
"discipline": {"type": "string", "description": f"Academic field or discipline of the {document_label}. Apply sentence case."},
"granting_institution": {"type": "string", "description": f"Institution where the {document_label} was submitted and the degree is granted"},
"co_tutelle_institutions": {"type": "array", "items": {"type": "string"}, "description": "List of institutions involved in a joint supervision or co-tutelle agreement; empty list if none"},
"doctoral_school": {"type": "string", "description": "Doctoral school or graduate program, if explicitly mentioned"},
"defense_year": {"type": "integer", "description": f"Year the {document_label} was defended. Format yyyy"},
"advisor": {"type": "array", "items": {"type": "string"}, "description": f"ONLY persons explicitly identified as {document_label} advisor/supervisor/director. Apply sentence case and remove all civility/title prefixes (Monsieur, Madame, M., Mme, Professeur, Pr, Dr, etc.). Empty if the role is not explicitly stated."},
"jury_president": {"type": "string", "description": "ONLY the person explicitly identified as president/chair of the jury. Apply sentence case and remove all civility/title prefixes (Monsieur, Madame, M., Mme, Professeur, Pr, Dr, etc.). Null if not explicitly stated."},
"reviewers": {"type": "array", "items": {"type": "string"}, "description": "ONLY official reviewers/rapporteurs explicitly identified as such. Apply sentence case and remove all civility/title prefixes (Monsieur, Madame, M., Mme, Professeur, Pr, Dr, etc.). Empty if not explicitly stated."},
"committee_members": {"type": "array", "items": {"type": "string"}, "description": "Persons explicitly listed as jury or committee members whose role is not explicitly identified as president/chair or reviewer/rapporteur. Apply sentence case and remove all civility/title prefixes (Monsieur, Madame, M., Mme, Professeur, Pr, Dr, etc.). Include unlabeled persons here."},
"language": {"type": "string", "description": "Language in ISO 639-3 codes. Example: fre, eng, ita..."},
},
}
schema_json = json.dumps(schema, ensure_ascii=False, indent=2)
return f"""Extract structured data from this document according to the provided JSON schema. The document is provided as images, in page order.
## JSON Schema
```json
{schema_json}
```
## Instructions
- Return a JSON object matching the schema
- Use the correct type for each field (string, number, array)"""
def coerce_lift_response(extracted: dict[str, Any]) -> dict[str, Any]:
"""The lift model returns arrays for multivalued person fields; tab 2 and the rest
of the pipeline expect pipe-separated strings. Join those, leaving
co_tutelle_institutions as a list (normalize_metadata handles that one)."""
coerced = dict(extracted)
for field in ("advisor", "reviewers", "committee_members"):
value = coerced.get(field)
if isinstance(value, list):
coerced[field] = " | ".join(str(item).strip() for item in value if str(item).strip())
return coerced
def as_json(value: Any) -> str:
return json.dumps(value, ensure_ascii=False, indent=2)
def load_json_object(text: str, label: str = "JSON") -> dict[str, Any]:
try:
value = json.loads(text)
except json.JSONDecodeError as exc:
raise gr.Error(f"{label} invalide: {exc}") from exc
if not isinstance(value, dict):
raise gr.Error(f"{label} doit être un objet.")
return value
def compress_image_to_data_url(data: bytes, max_size: int = VLM_IMAGE_MAX_SIZE) -> str:
"""Downscale and JPEG-encode image bytes, returning a base64 data URI.
Keeps the base64 path (llama.cpp may not support image URLs) but shrinks the
payload: large scans dominate upload and inference time. Falls back to the raw
bytes if Pillow cannot decode the image.
"""
try:
img = Image.open(BytesIO(data))
img.load()
except Exception:
# Not a decodable image — pass the original bytes through unchanged.
encoded = base64.b64encode(data).decode("ascii")
return f"data:image/jpeg;base64,{encoded}"
if img.mode not in ("RGB", "L"):
img = img.convert("RGB")
w, h = img.size
ratio = min(max_size / w, max_size / h, 1.0)
if ratio < 1:
img = img.resize((int(w * ratio), int(h * ratio)), Image.LANCZOS)
buf = BytesIO()
img.save(buf, format="JPEG", quality=VLM_IMAGE_JPEG_QUALITY, optimize=True)
encoded = base64.b64encode(buf.getvalue()).decode("ascii")
return f"data:image/jpeg;base64,{encoded}"
def image_file_to_data_url(path: str | None, max_size: int = VLM_IMAGE_MAX_SIZE) -> str:
if not path:
return ""
return compress_image_to_data_url(Path(path).read_bytes(), max_size)
def resolve_image_to_data_url(
image_url: str | None,
image_path: str | None,
max_size: int = VLM_IMAGE_MAX_SIZE,
) -> str:
raw = (image_url or "").strip()
if not raw:
# Fall back to local file
return image_file_to_data_url(image_path, max_size)
if raw.startswith("data:"):
# Already a data URI: decode, recompress, and re-encode.
try:
header, _, b64 = raw.partition(",")
if "base64" in header and b64:
return compress_image_to_data_url(base64.b64decode(b64), max_size)
except Exception:
pass
return raw
if raw.startswith("http://") or raw.startswith("https://"):
# Fetch on our side with a hard timeout — never let llama.cpp do this
resp = requests.get(raw, timeout=10, allow_redirects=True)
resp.raise_for_status()
return compress_image_to_data_url(resp.content, max_size)
# Assume raw base64 string without the data URI prefix
try:
return compress_image_to_data_url(base64.b64decode(raw), max_size)
except Exception:
return f"data:image/jpeg;base64,{raw}"
def resolve_vlm_provider(
provider: str,
vlm_endpoint: str,
vlm_model: str,
openai_endpoint: str,
openai_model: str,
hf_endpoint: str,
hf_model: str,
albert_endpoint: str,
albert_model: str,
lift_endpoint: str,
lift_model: str,
) -> tuple[str, str, dict[str, str]]:
"""Return (endpoint, model, auth_headers) for the selected OpenAI-compatible provider."""
provider = str(provider or "vlm").strip().lower()
if provider == "openai":
api_key = os.getenv("OPENAI_API_KEY", "").strip()
if not api_key:
raise gr.Error("Définissez la variable d'environnement OPENAI_API_KEY.")
endpoint = openai_endpoint.strip() or DEFAULT_OPENAI_URL
model = openai_model.strip() or DEFAULT_OPENAI_MODEL
return endpoint, model, {"Authorization": f"Bearer {api_key}"}
if provider in {"huggingface", "hf"}:
api_key = os.getenv("HF_TOKEN", "").strip()
if not api_key:
raise gr.Error("Définissez la variable d'environnement HF_TOKEN.")
endpoint = hf_endpoint.strip() or DEFAULT_HF_URL
model = hf_model.strip() or DEFAULT_HF_MODEL
return endpoint, model, {"Authorization": f"Bearer {api_key}"}
if provider == "albert":
api_key = os.getenv("ALBERT_API_KEY", "").strip()
if not api_key:
raise gr.Error("Définissez la variable d'environnement ALBERT_API_KEY.")
endpoint = albert_endpoint.strip() or DEFAULT_ALBERT_URL
model = albert_model.strip() or DEFAULT_ALBERT_MODEL
return endpoint, model, {"Authorization": f"Bearer {api_key}"}
if provider == "lift":
# HF dedicated endpoint: authenticate with the same HF token.
api_key = os.getenv("HF_TOKEN", "").strip()
if not api_key:
raise gr.Error("Définissez la variable d'environnement HF_TOKEN.")
endpoint = lift_endpoint.strip() or DEFAULT_LIFT_URL
model = lift_model.strip() or DEFAULT_LIFT_MODEL
return endpoint, model, {"Authorization": f"Bearer {api_key}"}
endpoint = vlm_endpoint.strip() or DEFAULT_VLM_URL
model = vlm_model.strip() or DEFAULT_VLM_MODEL
return endpoint, model, {}
def call_vlm(
state: dict[str, Any] | None,
doc_type: str,
image_path: str | None,
image_url: str,
prompt: str,
provider: str,
vlm_endpoint: str,
vlm_model: str,
openai_endpoint: str,
openai_model: str,
hf_endpoint: str,
hf_model: str,
albert_endpoint: str,
albert_model: str,
lift_endpoint: str,
lift_model: str,
image_max_size: int,
) -> tuple[dict[str, Any], str, str, str, str, str, str, str, str, str, str, str, str, str, str, str, str]:
state = deepcopy(state or empty_state())
provider_norm = str(provider or "vlm").strip().lower()
resolved_image_url = resolve_image_to_data_url(image_url, image_path, int(image_max_size))
if not resolved_image_url:
raise gr.Error("Ajoutez une image locale ou une URL d'image.")
if not prompt.strip():
prompt = build_lift_prompt(doc_type) if provider_norm == "lift" else build_eval_prompt(doc_type)
endpoint, model, auth_headers = resolve_vlm_provider(
provider, vlm_endpoint, vlm_model, openai_endpoint, openai_model,
hf_endpoint, hf_model, albert_endpoint, albert_model, lift_endpoint, lift_model
)
payload = {
"model": model,
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{"type": "image_url", "image_url": {"url": resolved_image_url}},
],
}
],
"response_format": {"type": "json_object"},
}
try:
response = requests.post(
endpoint.strip(),
headers={"Content-Type": "application/json", **auth_headers},
json=payload,
timeout=None,
)
response.raise_for_status()
raw_response = response.json()
except requests.RequestException as exc:
state["vlm"]["error"] = str(exc)
state = add_event(state, "vlm", "error", str(exc))
raise gr.Error(f"Erreur VLM: {exc}") from exc
except ValueError as exc:
state["vlm"]["error"] = "Réponse VLM non JSON"
state = add_event(state, "vlm", "error", "Réponse VLM non JSON")
raise gr.Error("La réponse VLM n'est pas un JSON valide.") from exc
content = (((raw_response.get("choices") or [{}])[0].get("message") or {}).get("content") or "").strip()
extracted = load_json_object(content, "Contenu JSON du VLM")
if provider_norm == "lift":
extracted = coerce_lift_response(extracted)
normalized = normalize_metadata(extracted)
state["input"] = {
"doc_type": doc_type,
"image_url": image_url.strip(),
"image_filename": str(image_path or ""),
"provider": str(provider or "vlm").strip().lower(),
"endpoint": endpoint,
"model": model,
}
state["vlm"] = {"request": payload, "raw_response": raw_response, "extracted": extracted, "corrected": normalized, "error": None}
state = add_event(state, "vlm", "ok", "Extraction VLM terminée.")
return state, as_json(extracted), as_json(raw_response), status_card("Extraction VLM", "ok", "Métadonnées extraites."), *metadata_to_outputs(normalized)
def normalize_metadata(payload: dict[str, Any]) -> dict[str, Any]:
normalized = {field: payload.get(field) for field in METADATA_FIELDS}
if normalized.get("co_tutelle_institutions") in (None, ""):
normalized["co_tutelle_institutions"] = []
if not isinstance(normalized.get("co_tutelle_institutions"), list):
normalized["co_tutelle_institutions"] = [str(normalized["co_tutelle_institutions"])]
return normalized
def metadata_to_outputs(payload: dict[str, Any]) -> tuple[str, ...]:
values = []
for field in METADATA_FIELDS:
value = payload.get(field)
if field == "co_tutelle_institutions":
values.append(" | ".join(str(item) for item in (value or [])))
elif value is None:
values.append("")
else:
values.append(str(value))
return tuple(values)
def parse_year(value: str) -> int | None:
value = str(value or "").strip()
if not value:
return None
try:
return int(value)
except ValueError:
return None
def parse_float(value: str) -> float | None:
value = str(value or "").strip()
if not value:
return None
try:
return float(value)
except ValueError:
return None
def corrected_from_fields(*values: str) -> dict[str, Any]:
payload = dict(zip(METADATA_FIELDS, values))
payload["co_tutelle_institutions"] = [item.strip() for item in str(payload.get("co_tutelle_institutions") or "").split("|") if item.strip()]
payload["defense_year"] = parse_year(str(payload.get("defense_year") or ""))
payload["confidence"] = parse_float(str(payload.get("confidence") or ""))
for key, value in list(payload.items()):
if value == "":
payload[key] = None if key not in {"co_tutelle_institutions"} else []
return payload
def string_or_empty(value: Any) -> str:
if value is None:
return ""
return str(value)
def sudoc_request_payload(corrected: dict[str, Any]) -> dict[str, Any]:
"""Match the Sudoc checker schema, which expects strings for optional text fields.
Document-kind specifics (TDO filter, NTH content filter) are not in the body:
they are selected server-side by routing to /check/thesis or /check/dissertation.
"""
payload = {
"title": string_or_empty(corrected.get("title")),
"subtitle": string_or_empty(corrected.get("subtitle")),
"author": string_or_empty(corrected.get("author")),
"degree_type": string_or_empty(corrected.get("degree_type")),
"discipline": string_or_empty(corrected.get("discipline")),
"granting_institution": string_or_empty(corrected.get("granting_institution")),
"co_tutelle_institutions": corrected.get("co_tutelle_institutions") or [],
"doctoral_school": string_or_empty(corrected.get("doctoral_school")),
"defense_year": corrected.get("defense_year"),
"advisor": string_or_empty(corrected.get("advisor")),
"jury_president": string_or_empty(corrected.get("jury_president")),
"reviewers": string_or_empty(corrected.get("reviewers")),
"committee_members": string_or_empty(corrected.get("committee_members")),
"language": string_or_empty(corrected.get("language")),
"confidence": corrected.get("confidence"),
}
if payload["defense_year"] is None:
payload["defense_year"] = ""
if payload["confidence"] is None:
payload.pop("confidence")
return payload
def sudoc_route_for(doc_type: str) -> str:
return "/check/dissertation" if str(doc_type or "").strip().lower() == "memoire" else "/check/thesis"
def update_corrected_metadata(state: dict[str, Any] | None, *values: str) -> tuple[dict[str, Any], str, str, list[list[str]], str]:
state = deepcopy(state or empty_state())
corrected = corrected_from_fields(*values)
state.setdefault("vlm", {})["corrected"] = corrected
state = add_event(state, "metadata", "ok", "Métadonnées corrigées mises à jour.")
return state, as_json(corrected), status_card("Métadonnées corrigées", "ok", "Ces valeurs alimenteront Sudoc et IdRef."), people_rows(corrected), pipeline_summary(state)
def call_sudoc(state: dict[str, Any] | None, doc_type: str, api_url: str) -> tuple[dict[str, Any], str, str, list[list[Any]], str]:
state = deepcopy(state or empty_state())
corrected = state.get("vlm", {}).get("corrected")
if not corrected:
raise gr.Error("Validez d'abord les métadonnées corrigées.")
request_payload = sudoc_request_payload(corrected)
base_url = api_url.rstrip("/")
route = sudoc_route_for(doc_type)
try:
response = requests.post(f"{base_url}{route}", headers={"Content-Type": "application/json"}, json=request_payload, timeout=None)
response.raise_for_status()
result = response.json()
except requests.RequestException as exc:
state["sudoc"]["error"] = str(exc)
state = add_event(state, "sudoc", "error", str(exc))
raise gr.Error(f"Erreur Sudoc: {exc}") from exc
except ValueError as exc:
state["sudoc"]["error"] = "Réponse Sudoc non JSON"
state = add_event(state, "sudoc", "error", "Réponse Sudoc non JSON")
raise gr.Error("La réponse Sudoc n'est pas un JSON valide.") from exc
state["sudoc"] = {"request": request_payload, "route": route, "response": result, "error": None}
state = add_event(state, "sudoc", "ok", f"Sudoc ({route}): {result.get('status')}")
return state, render_sudoc(result), as_json(result), sudoc_candidate_rows(result), pipeline_summary(state)
def split_people(value: Any) -> list[str]:
if value is None:
return []
if isinstance(value, list):
parts = value
else:
parts = str(value).split("|")
return [str(part).strip() for part in parts if str(part).strip()]
def normalize_name_key(name: str) -> str:
return " ".join(str(name).lower().split())
def extract_people(payload: dict[str, Any]) -> list[dict[str, Any]]:
people: dict[str, dict[str, Any]] = {}
for field in PERSON_FIELDS:
for name in split_people(payload.get(field)):
key = normalize_name_key(name)
if key not in people:
people[key] = {"name": name, "roles": []}
if field not in people[key]["roles"]:
people[key]["roles"].append(field)
return list(people.values())
def people_rows(payload: dict[str, Any]) -> list[list[str]]:
rows = []
for person in extract_people(payload):
rows.append([person["name"], " | ".join(ROLE_LABELS.get(role, role) for role in person["roles"])])
return rows
def build_align_payload(
extraction: dict[str, Any],
name: str,
use_embeddings: bool,
weights: dict[str, float],
) -> dict[str, Any]:
return {
"name": name,
"title": str(extraction.get("title") or ""),
"subtitle": str(extraction.get("subtitle") or ""),
"discipline": str(extraction.get("discipline") or ""),
"institution": str(extraction.get("granting_institution") or ""),
"doctoral_school": str(extraction.get("doctoral_school") or ""),
"degree_type": str(extraction.get("degree_type") or ""),
"year": str(extraction.get("defense_year") or ""),
"max_candidates": 20,
"max_docs_per_role": 20,
"reference_top_k": 3,
"embedding_model": EMBEDDING_MODEL if use_embeddings else "",
"weight_name": float(weights["weight_name"]),
"weight_attrra_source": float(weights["weight_attrra_source"]),
"weight_attrra_note": float(weights["weight_attrra_note"]),
"weight_references": float(weights["weight_references"]),
"weight_institution_year": float(weights["weight_institution_year"]),
}
def call_idref_for_all(
state: dict[str, Any] | None,
api_url: str,
api_key: str,
use_embeddings: bool,
weight_name: float,
weight_attrra_source: float,
weight_attrra_note: float,
weight_references: float,
weight_institution_year: float,
) -> tuple[dict[str, Any], str, str, list[list[Any]], str]:
state = deepcopy(state or empty_state())
corrected = state.get("vlm", {}).get("corrected")
if not corrected:
raise gr.Error("Validez d'abord les métadonnées corrigées.")
people = extract_people(corrected)
if not people:
raise gr.Error("Aucune personne à aligner dans les métadonnées corrigées.")
base_url = api_url.rstrip("/")
headers = {"Content-Type": "application/json"}
if api_key.strip():
headers["X-API-Key"] = api_key.strip()
weights = {
"weight_name": weight_name,
"weight_attrra_source": weight_attrra_source,
"weight_attrra_note": weight_attrra_note,
"weight_references": weight_references,
"weight_institution_year": weight_institution_year,
}
aligned = []
for person in people:
payload = build_align_payload(corrected, person["name"], use_embeddings, weights)
try:
response = requests.post(f"{base_url}/align/person", headers=headers, json=payload, timeout=None)
response.raise_for_status()
result = response.json()
except requests.RequestException as exc:
result = {"status": "error", "error": str(exc), "query": {"name": person["name"]}, "candidates": []}
except ValueError:
result = {"status": "error", "error": "Réponse non JSON", "query": {"name": person["name"]}, "candidates": []}
aligned.append({"name": person["name"], "roles": person["roles"], "request": payload, "response": result})
state["idref"] = {
"persons": aligned,
"settings": {"api_url": base_url, "use_embeddings": use_embeddings, "weights": weights},
"error": None,
}
state = add_event(state, "idref", "ok", f"{len(aligned)} personne(s) alignée(s).")
return state, render_idref(aligned), as_json(aligned), idref_rows(aligned), pipeline_summary(state)
def build_dewey_text(corrected: dict[str, Any]) -> str:
"""Concatenate title + subtitle (when present) + discipline for the classifier."""
parts = [
string_or_empty(corrected.get("title")).strip(),
string_or_empty(corrected.get("subtitle")).strip(),
string_or_empty(corrected.get("discipline")).strip(),
]
return " ".join(part for part in parts if part)
def dewey_choice_label(cls: dict[str, Any]) -> str:
code = str(cls.get("dewey") or "")
label = str(cls.get("label") or "")
score = cls.get("score")
score_text = f" — {float(score):.4f}" if isinstance(score, int | float) else ""
return f"{code} · {label}{score_text}".strip(" ·")
def dewey_radio_update(classes: list[dict[str, Any]], selected: dict[str, Any] | None):
choices = [(dewey_choice_label(cls), str(cls.get("dewey") or "")) for cls in classes]
value = str(selected.get("code")) if selected and selected.get("code") is not None else None
return gr.update(choices=choices, value=value)
def selected_from_class(cls: dict[str, Any]) -> dict[str, Any]:
return {"code": cls.get("dewey"), "label": cls.get("label"), "score": cls.get("score")}
def call_dewey(
state: dict[str, Any] | None,
api_url: str,
api_key: str,
top_k: int,
threshold: float,
method: str,
) -> tuple[dict[str, Any], str, str, Any, str]:
state = deepcopy(state or empty_state())
corrected = state.get("vlm", {}).get("corrected")
if not corrected:
raise gr.Error("Validez d'abord les métadonnées corrigées.")
text = build_dewey_text(corrected)
if not text:
raise gr.Error("Titre et discipline manquants pour la classification Dewey.")
method = (method or "local").strip().lower()
if method not in ("local", "albert"):
raise gr.Error("Méthode de classification invalide (local ou albert).")
request_payload = {
"text": text,
"classification_type": "multi-label",
"top_k": int(top_k),
"threshold": float(threshold),
"method": method,
}
base_url = api_url.rstrip("/")
headers = {"Content-Type": "application/json"}
if api_key.strip():
headers["X-API-Key"] = api_key.strip()
try:
response = requests.post(f"{base_url}/classify", headers=headers, json=request_payload, timeout=None)
response.raise_for_status()
result = response.json()
except requests.RequestException as exc:
state["dewey"]["error"] = str(exc)
state = add_event(state, "dewey", "error", str(exc))
raise gr.Error(f"Erreur Dewey: {exc}") from exc
except ValueError as exc:
state["dewey"]["error"] = "Réponse Dewey non JSON"
state = add_event(state, "dewey", "error", "Réponse Dewey non JSON")
raise gr.Error("La réponse Dewey n'est pas un JSON valide.") from exc
classes = (((result.get("results") or [{}])[0]).get("classes")) or []
selected = selected_from_class(classes[0]) if classes else None
state["dewey"] = {
"request": request_payload,
"response": result,
"classes": classes,
"selected": selected,
"error": None,
}
result_method = result.get("method") or method
result_model = result.get("model")
state = add_event(
state, "dewey", "ok",
f"{len(classes)} classe(s) Dewey proposée(s) (méthode {result_method}).",
)
return (
state,
render_dewey(classes, selected, result_method, result_model),
as_json(result),
dewey_radio_update(classes, selected),
pipeline_summary(state),
)
def select_dewey_class(state: dict[str, Any] | None, code: str | None) -> tuple[dict[str, Any], str, str]:
state = deepcopy(state or empty_state())
classes = state.get("dewey", {}).get("classes") or []
chosen = next((cls for cls in classes if str(cls.get("dewey")) == str(code)), None)
if not chosen:
raise gr.Error("Sélectionnez une classe Dewey valide.")
selected = selected_from_class(chosen)
state.setdefault("dewey", {})["selected"] = selected
state = add_event(state, "dewey", "ok", f"Classe Dewey retenue: {selected['code']} {selected['label']}")
message = f"Classe retenue : {selected['code']} — {selected['label']}"
return state, status_card("Classification Dewey", "ok", message), pipeline_summary(state)
def make_record_draft(state: dict[str, Any] | None) -> tuple[dict[str, Any], str, str, str]:
state = deepcopy(state or empty_state())
corrected = state.get("vlm", {}).get("corrected") or {}
sudoc_response = state.get("sudoc", {}).get("response") or {}
aligned = state.get("idref", {}).get("persons") or []
doc_type = (state.get("input") or {}).get("doc_type") or "these"
persons_by_name: dict[str, dict[str, Any]] = {}
for item in aligned:
response = item.get("response") or {}
best = response.get("best_candidate") or {}
name = item.get("name")
if not name:
continue
entry = persons_by_name.setdefault(
normalize_name_key(name),
{
"name": name,
"roles": [],
"role_labels": [],
"idref_ppn": response.get("best_ppn") if response.get("status") in STRONG_IDREF_STATUSES else None,
"status": response.get("status"),
"score": (best.get("score") or {}).get("final"),
"idref_url": best.get("url"),
"strong_idref": response.get("status") in STRONG_IDREF_STATUSES,
},
)
for role in item.get("roles") or []:
if role not in entry["roles"]:
entry["roles"].append(role)
entry["role_labels"].append(ROLE_LABELS.get(role, role))
if response.get("status") in STRONG_IDREF_STATUSES:
entry["idref_ppn"] = response.get("best_ppn")
entry["strong_idref"] = True
entry["idref_url"] = best.get("url")
entry["score"] = (best.get("score") or {}).get("final")
entry["status"] = response.get("status")
persons = list(persons_by_name.values())
draft = {
"profile": "memoire_original_imprime" if doc_type == "memoire" else "these_originale_imprimee",
"work_metadata": {
"title": corrected.get("title"),
"subtitle": corrected.get("subtitle"),
"language": corrected.get("language"),
"defense_year": corrected.get("defense_year"),
},
"thesis_metadata": {
"degree_type": corrected.get("degree_type"),
"discipline": corrected.get("discipline"),
"granting_institution": corrected.get("granting_institution"),
"co_tutelle_institutions": corrected.get("co_tutelle_institutions") or [],
"doctoral_school": corrected.get("doctoral_school"),
"partner_institutions": corrected.get("partner_institutions") or [],
"abstract": corrected.get("abstract"),
},
"persons": persons,
"dewey_classification": state.get("dewey", {}).get("selected") or None,
"sudoc_check": {
"status": sudoc_response.get("status"),
"duplicate_score": sudoc_response.get("duplicate_score"),
"printed_duplicate_ppn": (sudoc_response.get("best_print_candidate") or {}).get("ppn"),
"best_electronic_ppn": (sudoc_response.get("best_electronic_candidate") or {}).get("ppn"),
},
}
state["record_draft"] = draft
state = add_event(state, "draft", "ok", "Brouillon consolidé généré.")
return state, render_draft(draft), as_json(draft), as_json(state)
def score_value(candidate: dict[str, Any] | None, key: str = "final") -> str:
if not candidate:
return ""
value = (candidate.get("score") or {}).get(key)
if value is None:
return ""
return f"{float(value):.4f}"
def status_card(title: str, status: str, message: str) -> str:
colors = {
"ok": ("#dcfce7", "#166534"),
"warn": ("#fef9c3", "#854d0e"),
"error": ("#fee2e2", "#991b1b"),
"idle": ("#e5e7eb", "#374151"),
}
bg, fg = colors.get(status, colors["idle"])
return (
f""
f"{html.escape(title)}
{html.escape(message)}
"
)
def badge(value: Any) -> str:
text = str(value or "non exécuté")
colors = {
"duplicate_found": ("#fee2e2", "#991b1b"),
"ambiguous_print_candidate": ("#fef9c3", "#854d0e"),
"electronic_only": ("#dbeafe", "#1d4ed8"),
"no_print_duplicate_found": ("#dcfce7", "#166534"),
"accepted": ("#dcfce7", "#166534"),
"ambiguous": ("#fef9c3", "#854d0e"),
"low_confidence": ("#fee2e2", "#991b1b"),
"not_found": ("#e5e7eb", "#374151"),
"error": ("#fee2e2", "#991b1b"),
}
bg, fg = colors.get(text, ("#e5e7eb", "#374151"))
return f"{html.escape(text)}"
def pipeline_summary(state: dict[str, Any] | None) -> str:
state = state or empty_state()
vlm_status = "ok" if state.get("vlm", {}).get("corrected") else "idle"
sudoc_status = (state.get("sudoc", {}).get("response") or {}).get("status") or "idle"
idref_people = state.get("idref", {}).get("persons") or []
accepted = sum(1 for item in idref_people if (item.get("response") or {}).get("status") == "accepted")
idref_label = f"{accepted}/{len(idref_people)} accepted" if idref_people else "idle"
dewey_selected = state.get("dewey", {}).get("selected") or {}
dewey_label = str(dewey_selected.get("code")) if dewey_selected.get("code") else "idle"
draft_status = "ok" if state.get("record_draft") else "idle"
return f"""
VLM
{badge(vlm_status)}
Sudoc
{badge(sudoc_status)}
IdRef
{badge(idref_label)}
Dewey
{badge(dewey_label)}
Brouillon
{badge(draft_status)}
"""
def render_sudoc(result: dict[str, Any]) -> str:
best_print = result.get("best_print_candidate")
best_elec = result.get("best_electronic_candidate")
profile = result.get("profile") or {}
profile_name = profile.get("name") or "thesis"
sru_filter = profile.get("sru_type_filter")
note_subs = profile.get("note_required_substrings") or []
excluded = (result.get("sru") or {}).get("excluded_by_profile_filter")
filter_parts = []
if sru_filter:
filter_parts.append(f"SRU {html.escape(str(sru_filter))}")
else:
filter_parts.append("aucun filtre tdo")
if note_subs:
filter_parts.append("NTH contient " + ", ".join(f'{html.escape(str(sub))}' for sub in note_subs))
filter_label = " · ".join(filter_parts)
excluded_html = (
f"Candidats écartés par le filtre profil : {int(excluded)}
"
if isinstance(excluded, int) and excluded > 0
else ""
)
def candidate_panel(title: str, candidate: dict[str, Any] | None) -> str:
if not candidate:
return f"{html.escape(title)}
Aucun candidat.
"
evidence = candidate.get("evidence") or {}
return f"""
{html.escape(title)}
PPN : {html.escape(candidate.get('ppn') or '')}
Titre : {html.escape(str(candidate.get('title') or ''))}
Support : {badge(candidate.get('carrier'))}
Compte comme doublon imprimé : {html.escape(str(candidate.get('counts_as_print_duplicate')))}
{score_table(candidate.get('score') or {})}
Indices de support
{list_html(evidence.get('carrier_evidence') or [])}
Requêtes SRU correspondantes
{list_html(evidence.get('matched_queries') or [])}
"""
return f"""
Décision Sudoc
{badge(result.get('status'))}
Profil documentaire : {html.escape(profile_name)}
Score doublon imprimé : {html.escape(str(result.get('duplicate_score')))}
Stratégie SRU : {filter_label}
{excluded_html}
{candidate_panel('Meilleur candidat imprimé/physique', best_print)}
{candidate_panel('Meilleur candidat électronique', best_elec)}
"""
def score_table(score: dict[str, Any]) -> str:
rows = "".join(
f"| {html.escape(str(key))} | {float(value):.4f} |
"
for key, value in score.items()
if isinstance(value, int | float)
)
return f""
def list_html(values: list[Any]) -> str:
if not values:
return "Aucun indice.
"
return "" + "".join(f"- {html.escape(str(value))}
" for value in values) + "
"
def sudoc_candidate_rows(result: dict[str, Any]) -> list[list[Any]]:
rows = []
for candidate in result.get("candidates") or []:
rows.append(
[
candidate.get("ppn"),
score_value(candidate),
candidate.get("carrier"),
candidate.get("counts_as_print_duplicate"),
candidate.get("title"),
" | ".join(candidate.get("authors") or []),
candidate.get("year"),
candidate.get("nnt"),
candidate.get("url"),
]
)
return rows
def idref_rows(aligned: list[dict[str, Any]]) -> list[list[Any]]:
rows = []
for item in aligned:
response = item.get("response") or {}
best = response.get("best_candidate") or {}
rows.append(
[
item.get("name"),
" | ".join(ROLE_LABELS.get(role, role) for role in item.get("roles") or []),
response.get("status"),
response.get("best_ppn"),
score_value(best),
" | ".join((best.get("evidence") or {}).get("preferred_forms") or []),
best.get("url"),
]
)
return rows
def render_idref(aligned: list[dict[str, Any]]) -> str:
if not aligned:
return "Aucun alignement.
"
panels = []
for item in aligned:
response = item.get("response") or {}
best = response.get("best_candidate") or {}
candidates = response.get("candidates") or []
evidence = best.get("evidence") or {}
rows = "".join(
""
f"| {html.escape(candidate.get('ppn') or '')} | "
f"{score_value(candidate)} | "
f"{html.escape(' | '.join((candidate.get('evidence') or {}).get('preferred_forms') or []))} | "
"
"
for candidate in candidates[:8]
)
panels.append(
f"""
{html.escape(str(item.get('name') or ''))}
Rôles : {html.escape(' | '.join(ROLE_LABELS.get(role, role) for role in item.get('roles') or []))}
Décision : {badge(response.get('status'))}
PPN accepté : {html.escape(str(response.get('best_ppn') or 'aucun'))}
{score_table(best.get('score') or {})}
Formes préférées
{list_html(evidence.get('preferred_forms') or [])}
Meilleure source attrra
{html.escape(str(evidence.get('best_attrra_source') or ''))}
Meilleure note attrra
{html.escape(str(evidence.get('best_attrra_note') or ''))}
Meilleures références
{list_html(evidence.get('best_references') or [])}
Candidats
"""
)
return "\n".join(panels)
def render_dewey(
classes: list[dict[str, Any]],
selected: dict[str, Any] | None,
method: str | None = None,
model: str | None = None,
) -> str:
if not classes:
return "Aucune classe Dewey proposée.
"
selected_code = str((selected or {}).get("code")) if selected else None
def fmt_score(value: Any) -> str:
return f"{float(value):.4f}" if isinstance(value, int | float) else ""
rows = "".join(
""
f"| {html.escape(str(cls.get('dewey') or ''))} | "
f"{html.escape(str(cls.get('label') or ''))} | "
f"{fmt_score(cls.get('score'))} | "
f"{'✓' if str(cls.get('dewey')) == selected_code else ''} | "
"
"
for cls in classes
)
meta_bits = []
if method:
meta_bits.append(f"méthode {html.escape(str(method))}")
if model:
meta_bits.append(f"modèle {html.escape(str(model))}")
meta = f"{' · '.join(meta_bits)}
" if meta_bits else ""
return f"""
Classes Dewey proposées (par score décroissant)
{meta}
| Code | Label | Score | Retenu |
|---|
{rows}
"""
def render_draft(draft: dict[str, Any]) -> str:
persons = draft.get("persons") or []
person_rows = "".join(
""
f"| {html.escape(' | '.join(person.get('role_labels') or []))} | "
f"{html.escape(str(person.get('name') or ''))} | "
f"{html.escape(str(person.get('idref_ppn') or ''))} | "
f"{badge(person.get('status'))} | "
f"{html.escape(str(person.get('score') or ''))} | "
"
"
for person in persons
)
return f"""
Brouillon bibliographique consolidé
Profil : {html.escape(str(draft.get('profile') or ''))}
Titre : {html.escape(str((draft.get('work_metadata') or {}).get('title') or ''))}
Diplôme : {html.escape(str((draft.get('thesis_metadata') or {}).get('degree_type') or ''))}
Discipline : {html.escape(str((draft.get('thesis_metadata') or {}).get('discipline') or ''))}
Sudoc : {badge((draft.get('sudoc_check') or {}).get('status'))}
| Rôle | Nom | PPN IdRef | Statut | Score |
|---|
{person_rows}
Aperçu pseudo-UNIMARC
{html.escape(pseudo_unimarc(draft))}
Aperçu de travail uniquement. Ce n'est pas encore une sortie UNIMARC conforme.
"""
def pseudo_unimarc(draft: dict[str, Any]) -> str:
work = draft.get("work_metadata") or {}
thesis = draft.get("thesis_metadata") or {}
profile = draft.get("profile") or "these_originale_imprimee"
year = work.get("defense_year") or ""
language = work.get("language") or "fre"
institution = thesis.get("granting_institution") or ""
discipline = thesis.get("discipline") or ""
degree = thesis.get("degree_type") or ""
dewey = draft.get("dewey_classification") or {}
dewey_code = dewey.get("code")
dewey_label = dewey.get("label")
# The validated Dewey class label drives the discipline subfield in 328.
classification_label = dewey_label or discipline
advisors = people_with_role(draft, "advisor")
author = first_person_with_role(draft, "author")
lines = []
lines.append("001 [à attribuer]")
lines.append("008 Aax3")
lines.append(f"029 ##$aFR")
lines.append(f"100 ##$a{coded_100_value(year, language)}")
lines.append(f"101 0#$a{language}")
lines.append("102 ##$aFR")
lines.append("106 ##$ar")
if profile == "these_originale_imprimee":
lines.append("104 ##$ak$by$cy$dba$e0$ffre")
lines.append("105 ##$bm$c0$d0$fy")
else:
lines.append("105 ##$bv$c0$d0$fy")
lines.append("181 ##$P01$ctxt")
lines.append("182 ##$P01$cn")
lines.append("183 ##$P01$anga")
lines.append(f"200 1#$a{work.get('title') or ''}{sf('e', work.get('subtitle'))}{sf('f', author_statement(author))}{sf('g', advisor_statement(advisors))}")
lines.append(f"214 #1$d{year}")
lines.append(f"328 #0$b{degree}$c{classification_label}$e{institution}$d{year}")
if thesis.get("abstract"):
lines.append(f"330 ##$a{thesis.get('abstract')}")
lines.append("608 ##$3027253139$aThèses et écrits académiques$2rameau")
if dewey_code:
lines.append(f"686 ##$a{dewey_code}$2TEF")
for person in draft.get("persons") or []:
tag = "700" if "author" in (person.get("roles") or []) else "701"
role_codes = person_role_codes(person.get("roles") or [], profile)
if tag == "701" and any(role in person.get("roles", []) for role in ["advisor", "jury_president", "reviewers"]):
role_codes.append("555")
role_codes = dedupe(role_codes)
lines.append(f"{tag} #1{ppn_subfield(person)}{name_subfields(person.get('name'))}{''.join(f'$4{code}' for code in role_codes)}")
for line in corporate_lines(thesis):
lines.append(line)
return "\n".join(lines)
def sf(code: str, value: Any) -> str:
return f"${code}{value}" if value else ""
def dedupe(values: list[str]) -> list[str]:
seen = set()
output = []
for value in values:
if value and value not in seen:
seen.add(value)
output.append(value)
return output
def coded_100_value(year: Any, language: str) -> str:
year_text = str(year or "uuuu")
if len(year_text) != 4 or not year_text.isdigit():
year_text = "uuuu"
return f"{year_text} k y0{language or 'fre'}y50 ba"
def people_with_role(draft: dict[str, Any], role: str) -> list[dict[str, Any]]:
return [person for person in draft.get("persons") or [] if role in (person.get("roles") or [])]
def first_person_with_role(draft: dict[str, Any], role: str) -> dict[str, Any] | None:
people = people_with_role(draft, role)
return people[0] if people else None
def author_statement(person: dict[str, Any] | None) -> str:
return person.get("name") if person else ""
def advisor_statement(advisors: list[dict[str, Any]]) -> str:
names = [person.get("name") for person in advisors if person.get("name")]
if not names:
return ""
return "sous la direction de " + " ; ".join(names)
def person_role_codes(roles: list[str], profile: str) -> list[str]:
advisor_code = PERSON_ROLE_CODES["advisor_dissertation"] if profile == "memoire_original_imprime" else PERSON_ROLE_CODES["advisor_thesis"]
mapping = {
"author": PERSON_ROLE_CODES["author"],
"advisor": advisor_code,
"jury_president": PERSON_ROLE_CODES["jury_president"],
"reviewers": PERSON_ROLE_CODES["reviewers"],
"committee_members": PERSON_ROLE_CODES["committee_members"],
}
return [mapping[role] for role in roles if role in mapping]
def ppn_subfield(person: dict[str, Any]) -> str:
if person.get("strong_idref") and person.get("idref_ppn"):
return f"$3{person.get('idref_ppn')}"
return ""
def name_subfields(name: Any) -> str:
text = str(name or "").strip()
if not text:
return "$a[Nom manquant]"
parts = text.split()
if len(parts) >= 2:
return f"$a{' '.join(parts[1:])}$b{parts[0]}"
return f"$a{text}"
def corporate_lines(thesis: dict[str, Any]) -> list[str]:
corporates: dict[str, dict[str, Any]] = {}
def add(name: Any, role: str) -> None:
text = str(name or "").strip()
if not text:
return
key = normalize_name_key(text)
entry = corporates.setdefault(key, {"name": text, "roles": []})
if role not in entry["roles"]:
entry["roles"].append(role)
add(thesis.get("granting_institution"), "granting_institution")
for institution in thesis.get("co_tutelle_institutions") or []:
add(institution, "co_tutelle_institutions")
add(thesis.get("doctoral_school"), "doctoral_school")
for institution in thesis.get("partner_institutions") or []:
add(institution, "partner_institutions")
lines = []
for corporate in corporates.values():
codes = [CORPORATE_ROLE_CODES[role] for role in corporate["roles"] if role in CORPORATE_ROLE_CODES]
lines.append(f"711 02$a{corporate['name']}{''.join(f'$4{code}' for code in dedupe(codes))}")
return lines
def CSS() -> str:
return """
.gradio-container { max-width: 1320px !important; }
.status-card { border-radius: 8px; padding: 12px 14px; margin: 8px 0; }
.summary-grid { display:grid; grid-template-columns: repeat(5, minmax(0, 1fr)); gap:10px; margin: 8px 0 14px; }
.summary-grid > div, .panel { border:1px solid #e5e7eb; border-radius:8px; padding:12px; background:#fff; }
.badge { display:inline-block; padding:4px 9px; border-radius:999px; font-weight:700; font-size:13px; }
.muted { color:#6b7280; }
.data-table { width:100%; border-collapse:collapse; margin-top:10px; }
.data-table th, .data-table td { border-bottom:1px solid #e5e7eb; padding:7px 8px; text-align:left; vertical-align:top; }
.data-table th { background:#f9fafb; font-weight:700; }
.evidence-grid { display:grid; grid-template-columns: repeat(2, minmax(0, 1fr)); gap:10px; }
pre { white-space: pre-wrap; background:#f9fafb; border:1px solid #e5e7eb; border-radius:8px; padding:12px; }
@media (max-width: 900px) {
.summary-grid, .evidence-grid { grid-template-columns: 1fr; }
}
"""
with gr.Blocks(title="Humatheque Pipeline POC") as demo:
state = gr.State(empty_state())
gr.Markdown("# Humatheque Pipeline POC\nImage de page de titre -> extraction VLM -> vérification Sudoc -> alignement IdRef -> classification Dewey -> brouillon bibliographique.")
summary = gr.HTML(pipeline_summary(empty_state()))
with gr.Tab("1. Extraction VLM"):
with gr.Row():
with gr.Column(scale=2):
doc_type = gr.Radio([("Thèse", "these"), ("Mémoire / dissertation", "memoire")], value="these", label="Type de document")
image = gr.Image(type="filepath", label="Image locale de la page de titre")
image_url = gr.Textbox(value=EXAMPLE_IMAGE_URL, label="Ou URL d'image")
prompt = gr.Code(value=build_eval_prompt("these"), language="markdown", label="Prompt VLM", lines=18)
with gr.Column(scale=1):
vlm_provider = gr.Radio(
[
("VLM on-premise", "vlm"),
("OpenAI (gpt-4o)", "openai"),
("Hugging Face", "huggingface"),
("Albert API (Etalab)", "albert"),
("LIFT Q8 (datalab-to, HF endpoint)", "lift"),
],
value="vlm",
label="Fournisseur d'inférence",
)
with gr.Group(visible=True) as vlm_group:
vlm_endpoint = gr.Textbox(value=DEFAULT_VLM_URL, label="Endpoint VLM on-premise")
vlm_model = gr.Textbox(value=DEFAULT_VLM_MODEL, label="Modèle VLM on-premise")
with gr.Group(visible=False) as openai_group:
openai_endpoint = gr.Textbox(value=DEFAULT_OPENAI_URL, label="Endpoint OpenAI")
openai_model = gr.Textbox(value=DEFAULT_OPENAI_MODEL, label="Modèle OpenAI")
with gr.Group(visible=False) as hf_group:
hf_endpoint = gr.Textbox(value=DEFAULT_HF_URL, label="Endpoint Hugging Face")
hf_model = gr.Dropdown(
choices=HF_MODEL_CHOICES,
value=DEFAULT_HF_MODEL,
label="Modèle Hugging Face",
)
with gr.Group(visible=False) as albert_group:
albert_endpoint = gr.Textbox(value=DEFAULT_ALBERT_URL, label="Endpoint Albert API")
albert_model = gr.Textbox(value=DEFAULT_ALBERT_MODEL, label="Modèle Albert API")
with gr.Group(visible=False) as lift_group:
lift_endpoint = gr.Textbox(value=DEFAULT_LIFT_URL, label="Endpoint LIFT")
lift_model = gr.Textbox(value=DEFAULT_LIFT_MODEL, label="Modèle LIFT")
gr.Markdown(
"Clés API via variables d'environnement : `OPENAI_API_KEY` (OpenAI), "
"`HF_TOKEN` (Hugging Face et LIFT), `ALBERT_API_KEY` (Albert API). "
"Le VLM on-premise ne requiert pas de clé. LIFT impose son propre "
"prompt à schéma JSON strict, sélectionné automatiquement."
)
image_max_size = gr.Slider(
256, 2048, value=VLM_IMAGE_MAX_SIZE, step=64,
label="Taille max de l'image (px, plus grand côté)",
info="Redimensionnement avant envoi au VLM. Plus petit = envoi/inférence plus rapides.",
)
extract_btn = gr.Button("Extraire les métadonnées", variant="primary")
vlm_status = gr.HTML(status_card("Extraction VLM", "idle", "En attente."))
with gr.Row():
extracted_json = gr.Code(language="json", label="JSON extrait", lines=20)
vlm_raw_json = gr.Code(language="json", label="Réponse VLM brute", lines=20)
with gr.Tab("2. Métadonnées corrigées"):
gr.Markdown("Corrigez les champs avant de lancer Sudoc et IdRef. Les listes de personnes utilisent `|` comme séparateur.")
with gr.Row():
title = gr.Textbox(label="Titre")
subtitle = gr.Textbox(label="Sous-titre")
with gr.Row():
author = gr.Textbox(label="Auteur")
degree_type = gr.Textbox(label="Type de diplôme")
with gr.Row():
discipline = gr.Textbox(label="Discipline")
granting_institution = gr.Textbox(label="Établissement de soutenance")
with gr.Row():
co_tutelle_institutions = gr.Textbox(label="Cotutelles, séparées par |")
doctoral_school = gr.Textbox(label="École doctorale")
with gr.Row():
defense_year = gr.Textbox(label="Année de soutenance")
language = gr.Textbox(label="Langue ISO 639-3")
confidence = gr.Textbox(label="Confiance VLM")
with gr.Row():
advisor = gr.Textbox(label="Directeur(s), séparés par |")
jury_president = gr.Textbox(label="Président du jury")
with gr.Row():
reviewers = gr.Textbox(label="Rapporteurs, séparés par |")
committee_members = gr.Textbox(label="Membres du jury, séparés par |")
update_metadata_btn = gr.Button("Valider les métadonnées corrigées", variant="primary")
metadata_status = gr.HTML(status_card("Métadonnées corrigées", "idle", "En attente."))
corrected_json = gr.Code(language="json", label="JSON corrigé utilisé par les modules", lines=18)
people_table = gr.Dataframe(headers=["Nom", "Rôles"], datatype=["str", "str"], label="Personnes détectées", interactive=False, wrap=True)
metadata_inputs = [
title,
subtitle,
author,
degree_type,
discipline,
granting_institution,
co_tutelle_institutions,
doctoral_school,
defense_year,
advisor,
jury_president,
reviewers,
committee_members,
language,
confidence,
]
with gr.Tab("3. Vérification Sudoc"):
sudoc_api_url = gr.Textbox(value=DEFAULT_SUDOC_URL, label="API Sudoc checker")
sudoc_btn = gr.Button("Lancer la vérification Sudoc", variant="primary")
sudoc_html = gr.HTML()
sudoc_candidates = gr.Dataframe(
headers=["PPN", "Score", "Support", "Compte doublon imprimé", "Titre", "Auteur", "Année", "NNT", "URL"],
datatype=["str", "str", "str", "bool", "str", "str", "str", "str", "str"],
label="Candidats Sudoc",
interactive=False,
wrap=True,
)
sudoc_raw_json = gr.Code(language="json", label="Réponse Sudoc brute", lines=22)
with gr.Tab("4. Alignement IdRef"):
with gr.Row():
idref_api_url = gr.Textbox(value=DEFAULT_IDREF_URL, label="API IdRef Qualinka")
idref_api_key = gr.Textbox(value=DEFAULT_IDREF_API_KEY, label="API key IdRef", type="password")
with gr.Row():
use_embeddings = gr.Checkbox(value=True, label=f"Utiliser le mode embedding ({EMBEDDING_MODEL})")
with gr.Accordion("Pondération du score de désambiguïsation", open=False):
gr.Markdown(
"Poids relatifs des indices combinés par le service idref-linker pour scorer chaque candidat. "
"Idéalement leur somme vaut 1."
)
with gr.Row():
weight_name = gr.Slider(
0.0, 1.0, value=DEFAULT_WEIGHT_NAME, step=0.05, label="Nom (weight_name)"
)
weight_attrra_source = gr.Slider(
0.0, 1.0, value=DEFAULT_WEIGHT_ATTRRA_SOURCE, step=0.05,
label="Sources attrRA (weight_attrra_source)",
)
weight_attrra_note = gr.Slider(
0.0, 1.0, value=DEFAULT_WEIGHT_ATTRRA_NOTE, step=0.05,
label="Notes attrRA (weight_attrra_note)",
)
with gr.Row():
weight_references = gr.Slider(
0.0, 1.0, value=DEFAULT_WEIGHT_REFERENCES, step=0.05,
label="Références (weight_references)",
)
weight_institution_year = gr.Slider(
0.0, 1.0, value=DEFAULT_WEIGHT_INSTITUTION_YEAR, step=0.05,
label="Institution / année (weight_institution_year)",
)
idref_btn = gr.Button("Aligner toutes les personnes", variant="primary")
idref_table = gr.Dataframe(
headers=["Nom", "Rôles", "Statut", "PPN accepté", "Score", "Formes préférées", "URL"],
datatype=["str", "str", "str", "str", "str", "str", "str"],
label="Résumé des alignements",
interactive=False,
wrap=True,
)
idref_html = gr.HTML()
idref_raw_json = gr.Code(language="json", label="Réponses IdRef brutes", lines=24)
with gr.Tab("5. Classification Dewey"):
gr.Markdown(
"Le texte classé est la concaténation `titre + sous-titre + discipline` des "
"métadonnées corrigées. Les classes sont triées par score décroissant ; la "
"première est présélectionnée. Validez-la ou choisissez-en une autre.\n\n"
"**Méthode** : `local` (bi-encodeur local) ou `albert` (récupération bge-m3 + "
"rerank bge-reranker-v2-m3 via l'API Albert). Les scores `albert` sont des scores "
"de pertinence du reranker, non comparables aux scores de similarité `local`."
)
with gr.Row():
dewey_api_url = gr.Textbox(value=DEFAULT_DEWEY_URL, label="API classification Dewey")
dewey_api_key = gr.Textbox(value=DEFAULT_DEWEY_API_KEY, label="API key classification", type="password")
with gr.Row():
dewey_method = gr.Radio(
choices=[("Local (bi-encodeur)", "local"), ("Albert (rerank)", "albert")],
value="local",
label="Méthode de classification",
)
dewey_top_k = gr.Slider(1, 15, value=5, step=1, label="Nombre de classes (top_k)")
dewey_threshold = gr.Slider(-1.0, 1.0, value=0.0, step=0.05, label="Seuil de similarité")
dewey_btn = gr.Button("Classer (Dewey)", variant="primary")
dewey_status = gr.HTML(status_card("Classification Dewey", "idle", "En attente."))
dewey_choice = gr.Radio(choices=[], label="Classe Dewey retenue")
dewey_select_btn = gr.Button("Valider la classe sélectionnée")
dewey_html = gr.HTML()
dewey_raw_json = gr.Code(language="json", label="Réponse Dewey brute", lines=20)
with gr.Tab("6. Brouillon / export"):
draft_btn = gr.Button("Générer le brouillon consolidé", variant="primary")
draft_html = gr.HTML()
with gr.Row():
draft_json = gr.Code(language="json", label="Brouillon JSON", lines=24)
pipeline_json = gr.Code(language="json", label="État complet du pipeline", lines=24)
def toggle_provider_fields(provider: str, doc_type: str, current_prompt: str):
p = str(provider or "vlm").strip().lower()
is_lift = p == "lift"
# Swap the prompt template only when crossing the lift/non-lift boundary,
# so manual edits within the same template family are preserved.
prompt_is_lift = str(current_prompt or "").lstrip().startswith("Extract structured data")
if is_lift and not prompt_is_lift:
prompt_update = gr.update(value=build_lift_prompt(doc_type))
elif not is_lift and prompt_is_lift:
prompt_update = gr.update(value=build_eval_prompt(doc_type))
else:
prompt_update = gr.update()
return (
gr.update(visible=p == "vlm"),
gr.update(visible=p == "openai"),
gr.update(visible=p in {"huggingface", "hf"}),
gr.update(visible=p == "albert"),
gr.update(visible=is_lift),
prompt_update,
)
vlm_provider.change(
toggle_provider_fields,
inputs=[vlm_provider, doc_type, prompt],
outputs=[vlm_group, openai_group, hf_group, albert_group, lift_group, prompt],
)
def rebuild_prompt(doc_type: str, provider: str):
p = str(provider or "vlm").strip().lower()
return build_lift_prompt(doc_type) if p == "lift" else build_eval_prompt(doc_type)
doc_type.change(rebuild_prompt, inputs=[doc_type, vlm_provider], outputs=[prompt])
extract_btn.click(
call_vlm,
inputs=[
state,
doc_type,
image,
image_url,
prompt,
vlm_provider,
vlm_endpoint,
vlm_model,
openai_endpoint,
openai_model,
hf_endpoint,
hf_model,
albert_endpoint,
albert_model,
lift_endpoint,
lift_model,
image_max_size,
],
outputs=[state, extracted_json, vlm_raw_json, vlm_status, *metadata_inputs],
).then(
update_corrected_metadata,
inputs=[state, *metadata_inputs],
outputs=[state, corrected_json, metadata_status, people_table, summary],
)
update_metadata_btn.click(
update_corrected_metadata,
inputs=[state, *metadata_inputs],
outputs=[state, corrected_json, metadata_status, people_table, summary],
)
sudoc_btn.click(
call_sudoc,
inputs=[state, doc_type, sudoc_api_url],
outputs=[state, sudoc_html, sudoc_raw_json, sudoc_candidates, summary],
)
idref_btn.click(
call_idref_for_all,
inputs=[
state,
idref_api_url,
idref_api_key,
use_embeddings,
weight_name,
weight_attrra_source,
weight_attrra_note,
weight_references,
weight_institution_year,
],
outputs=[state, idref_html, idref_raw_json, idref_table, summary],
)
dewey_btn.click(
call_dewey,
inputs=[state, dewey_api_url, dewey_api_key, dewey_top_k, dewey_threshold, dewey_method],
outputs=[state, dewey_html, dewey_raw_json, dewey_choice, summary],
)
dewey_select_btn.click(
select_dewey_class,
inputs=[state, dewey_choice],
outputs=[state, dewey_status, summary],
)
dewey_choice.change(
select_dewey_class,
inputs=[state, dewey_choice],
outputs=[state, dewey_status, summary],
)
draft_btn.click(
make_record_draft,
inputs=[state],
outputs=[state, draft_html, draft_json, pipeline_json],
).then(
pipeline_summary,
inputs=[state],
outputs=[summary],
)
demo.launch()