Conversion thesaurus SKOS (ttl) en ThesX

Introduction

Le thesaurus matière, ou thesaurus W, doit être mis à jour lors des publication de nouvelles versions par le SIAF.

Voir ici : http://data.culture.fr/thesaurus/ et pour le theaurus matières : http://data.culture.fr/thesaurus/page/ark:/67717/Matiere

ThesX utilise un format relativement proche du SKOS Turtle. La conversion avec un script Python est assez simple.

Le script Python suivant est a adapter en fonction des prochaines versions du thesaurus matières SKOS Turtle.

(Voir aussi : Import-export_de_thesaurus_dans_ThesX)

Conversion

Lancement du script

python3 convert_ttl_to_exist_thesaurus_xml.py \
  thesaurus-skos-turtle.ttl \
  thesaurus-matiere-pour-thesX.xml

Par exemple :

python3 convert_ttl_to_exist_thesaurus_xml.py \
  Thesaurus_matieres_pour_l_indexation_des_archives_locales.ttl \
  thesaurus-matiere-exist.xml

script convert_ttl_to_exist_thesaurus_xml.py

#!/usr/bin/env python3
"""Convert a SKOS Turtle (.ttl) thesaurus into the hierarchical XML format used by the eXist 2.1 application.

This targets the simplified schema observed in thesaurus-w-minuscules.xml:
<thesaurus id="...">
  <informations>...</informations>
  <concept id="T00001"> ... nested <concept> ... </concept>
</thesaurus>

Mapping kept (others ignored):
- skos:prefLabel (French) -> <nom>
- skos:altLabel (French)  -> <ep>
- skos:scopeNote (French) -> <ne>
- skos:related            -> <va ref="ID">label</va>
Hierarchy:
- skos:hasTopConcept gives roots
- skos:narrower gives children

IDs:
- generated as T + 5-digit sequential number, deterministically assigned by traversing the tree.

Usage:
  ./convert_ttl_to_exist_thesaurus_xml.py input.ttl output.xml [scheme_uri]
"""

from __future__ import annotations

import sys
from collections import defaultdict
from typing import Dict, List, Optional, Set, Tuple

from rdflib import Graph, Literal, URIRef
from rdflib.namespace import RDF

SKOS = URIRef("http://www.w3.org/2004/02/skos/core#")
SKOSXL = URIRef("http://www.w3.org/2008/05/skos-xl#")
DCT = URIRef("http://purl.org/dc/terms/")
DC = URIRef("http://purl.org/dc/elements/1.1/")

SKOS_Concept = URIRef(str(SKOS) + "Concept")
SKOS_ConceptScheme = URIRef(str(SKOS) + "ConceptScheme")

PREF_LABEL = URIRef(str(SKOS) + "prefLabel")
ALT_LABEL = URIRef(str(SKOS) + "altLabel")
SCOPE_NOTE = URIRef(str(SKOS) + "scopeNote")
HAS_TOP = URIRef(str(SKOS) + "hasTopConcept")
NARROWER = URIRef(str(SKOS) + "narrower")
RELATED = URIRef(str(SKOS) + "related")
IN_SCHEME = URIRef(str(SKOS) + "inScheme")

TITLE = URIRef(str(DC) + "title")
DESCRIPTION = URIRef(str(DC) + "description")
MODIFIED = URIRef(str(DCT) + "modified")


def _lit_str(lit: Literal) -> str:
    return str(lit)


def pick_fr_literal(lits: List[Literal]) -> Optional[str]:
    """Pick a French literal if present, else first literal."""
    if not lits:
        return None

    def lang_key(l: Literal) -> Tuple[int, str]:
        lang = (l.language or "").lower()
        # Prefer fr-fr, then fr, then anything else
        if lang == "fr-fr":
            return (0, lang)
        if lang.startswith("fr"):
            return (1, lang)
        if lang:
            return (2, lang)
        return (3, lang)

    best = sorted(lits, key=lang_key)[0]
    return _lit_str(best)


def as_minuscule(s: str) -> str:
    # The sample file is fully lowercased. Keep accents, just lower().
    return s.lower()


def safe_date_lower_t(s: str) -> str:
    # Sample uses lowercase 't' in the datetime.
    return s.replace("T", "t")


def find_scheme(g: Graph, scheme_uri: Optional[str]) -> URIRef:
    if scheme_uri:
        return URIRef(scheme_uri)

    schemes = sorted({s for s in g.subjects(RDF.type, SKOS_ConceptScheme)}, key=str)
    if not schemes:
        raise SystemExit("No skos:ConceptScheme found in TTL.")
    if len(schemes) > 1:
        # Deterministic fallback: pick the first by URI
        return schemes[0]
    return schemes[0]


def collect_labels(g: Graph) -> Tuple[Dict[URIRef, str], Dict[URIRef, List[str]], Dict[URIRef, List[str]]]:
    pref: Dict[URIRef, str] = {}
    alt: Dict[URIRef, List[str]] = defaultdict(list)
    scope: Dict[URIRef, List[str]] = defaultdict(list)

    for c in g.subjects(RDF.type, SKOS_Concept):
        c = URIRef(str(c))
        pref_lits = [o for o in g.objects(c, PREF_LABEL) if isinstance(o, Literal)]
        pref_s = pick_fr_literal(pref_lits)
        if pref_s:
            pref[c] = as_minuscule(pref_s)

        alt_lits = [o for o in g.objects(c, ALT_LABEL) if isinstance(o, Literal)]
        # keep all fr-ish alt labels
        for l in alt_lits:
            lang = (l.language or "").lower()
            if lang.startswith("fr") or not lang:
                alt[c].append(as_minuscule(_lit_str(l)))

        scope_lits = [o for o in g.objects(c, SCOPE_NOTE) if isinstance(o, Literal)]
        for l in scope_lits:
            lang = (l.language or "").lower()
            if lang.startswith("fr") or not lang:
                scope[c].append(as_minuscule(_lit_str(l)))

    # de-dup alt/scope while preserving order
    for d in (alt, scope):
        for k, vals in list(d.items()):
            seen = set()
            out = []
            for v in vals:
                if v not in seen:
                    out.append(v)
                    seen.add(v)
            d[k] = out

    return pref, alt, scope


def build_hierarchy(g: Graph, scheme: URIRef) -> Tuple[List[URIRef], Dict[URIRef, List[URIRef]]]:
    # roots from hasTopConcept (if absent, infer as concepts in scheme with no broader)
    roots = sorted({URIRef(str(o)) for o in g.objects(scheme, HAS_TOP) if isinstance(o, URIRef)}, key=str)

    children: Dict[URIRef, List[URIRef]] = defaultdict(list)
    for parent, child in g.subject_objects(NARROWER):
        if not isinstance(parent, URIRef) or not isinstance(child, URIRef):
            continue
        parent = URIRef(str(parent))
        child = URIRef(str(child))
        children[parent].append(child)

    # Keep deterministic order for children lists
    for p in list(children.keys()):
        children[p] = sorted(set(children[p]), key=str)

    if not roots:
        # infer roots for this scheme
        in_scheme = {URIRef(str(c)) for c in g.subjects(IN_SCHEME, scheme) if isinstance(c, URIRef)}
        broader = URIRef(str(SKOS) + "broader")
        has_broader = {URIRef(str(c)) for c in g.subjects(broader, None) if isinstance(c, URIRef)}
        roots = sorted(in_scheme - has_broader, key=str)

    return roots, children


def assign_ids(roots: List[URIRef], children: Dict[URIRef, List[URIRef]]) -> Tuple[Dict[URIRef, str], List[URIRef]]:
    """Assign sequential IDs in a stable pre-order traversal."""
    mapping: Dict[URIRef, str] = {}
    ordered: List[URIRef] = []
    counter = 1

    def visit(n: URIRef, stack: Set[URIRef]):
        nonlocal counter
        if n in mapping:
            return
        if n in stack:
            return  # cycle protection
        stack.add(n)
        mapping[n] = f"T{counter:05d}"
        ordered.append(n)
        counter += 1
        for ch in children.get(n, []):
            visit(ch, stack)
        stack.remove(n)

    for r in roots:
        visit(r, set())

    # Some concepts might be disconnected; keep them after roots
    all_concepts = sorted(set(children.keys()) | {c for lst in children.values() for c in lst}, key=str)
    for c in all_concepts:
        if c not in mapping:
            visit(c, set())

    return mapping, ordered


def xml_escape_text(s: str) -> str:
    # lxml will escape; we keep as is.
    return s


def concept_to_xml(
    g: Graph,
    node: URIRef,
    id_map: Dict[URIRef, str],
    children: Dict[URIRef, List[URIRef]],
    pref: Dict[URIRef, str],
    alt: Dict[URIRef, List[str]],
    scope: Dict[URIRef, List[str]],
) -> "etree._Element":
    from lxml import etree

    el = etree.Element("concept")
    el.set("id", id_map[node])

    rel = etree.SubElement(el, "relations")
    nom = etree.SubElement(rel, "nom")
    nom.text = xml_escape_text(pref.get(node, str(node)))

    for ep_val in alt.get(node, []):
        ep = etree.SubElement(rel, "ep")
        ep.text = xml_escape_text(ep_val)

    for ne_val in scope.get(node, []):
        ne = etree.SubElement(rel, "ne")
        ne.text = xml_escape_text(ne_val)

    # related -> va
    for o in sorted({URIRef(str(o)) for o in g.objects(node, RELATED) if isinstance(o, URIRef)}, key=str):
        if o not in id_map:
            continue
        va = etree.SubElement(rel, "va")
        va.set("ref", id_map[o])
        va.text = xml_escape_text(pref.get(o, str(o)))

    for ch in children.get(node, []):
        if ch in id_map:
            el.append(concept_to_xml(g, ch, id_map, children, pref, alt, scope))

    return el


def main(argv: List[str]) -> int:
    if len(argv) < 3:
        print("Usage: convert_ttl_to_exist_thesaurus_xml.py input.ttl output.xml [scheme_uri]", file=sys.stderr)
        return 2

    in_path = argv[1]
    out_path = argv[2]
    scheme_uri = argv[3] if len(argv) >= 4 else None

    g = Graph()
    g.parse(in_path, format="turtle")

    scheme = find_scheme(g, scheme_uri)

    # scheme metadata
    title = pick_fr_literal([o for o in g.objects(scheme, TITLE) if isinstance(o, Literal)])
    desc = pick_fr_literal([o for o in g.objects(scheme, DESCRIPTION) if isinstance(o, Literal)])
    mod = next((str(o) for o in g.objects(scheme, MODIFIED) if isinstance(o, Literal)), None)

    pref, alt, scope = collect_labels(g)
    roots, children = build_hierarchy(g, scheme)
    id_map, _ordered = assign_ids(roots, children)

    from lxml import etree

    root = etree.Element("thesaurus")
    root.set("id", "mat")

    info = etree.SubElement(root, "informations")
    info_nom = etree.SubElement(info, "nom")
    info_nom.text = as_minuscule(title) if title else "thesaurus"
    info_desc = etree.SubElement(info, "description")
    info_desc.text = as_minuscule(desc) if desc else ""

    mod_el = etree.SubElement(info, "modification")
    mod_date = etree.SubElement(mod_el, "date")
    mod_date.text = safe_date_lower_t(mod) if mod else ""
    agent = etree.SubElement(mod_el, "agent")
    agent.set("code", "auto")
    agent.text = "conversion skos"

    # Build XML tree from roots
    for r in roots:
        if r in id_map:
            root.append(concept_to_xml(g, r, id_map, children, pref, alt, scope))

    # Pretty print
    xml_bytes = etree.tostring(
        root,
        xml_declaration=True,
        encoding="UTF-8",
        pretty_print=True,
    )

    with open(out_path, "wb") as f:
        f.write(xml_bytes)

    # Small report
    print(f"Wrote {out_path} with {len(id_map)} concepts and {len(roots)} top concepts.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))