74 lines
2.5 KiB
Python
74 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
SUPPORTED_LANGUAGES = ("nl-BE", "en-GB", "fr-BE")
|
|
|
|
# Stable across runs (and across which language ships first) so a document's RAGcore
|
|
# source_id never changes just because the sync ran on a different day or in a
|
|
# different order -- required for RAGcore's upload idempotency to work per document.
|
|
_SOURCE_ID_NAMESPACE = uuid.uuid5(
|
|
uuid.NAMESPACE_URL, "https://mobilityops.internal/knowledge/procedures"
|
|
)
|
|
|
|
|
|
def parse_frontmatter(raw: str) -> tuple[dict[str, str], str]:
|
|
if not raw.startswith("---"):
|
|
return {}, raw
|
|
end = raw.find("\n---", 3)
|
|
if end == -1:
|
|
return {}, raw
|
|
block = raw[3:end].strip()
|
|
body = raw[end + 4 :].lstrip("\n")
|
|
meta: dict[str, str] = {}
|
|
for line in block.splitlines():
|
|
if ":" not in line:
|
|
continue
|
|
key, _, value = line.partition(":")
|
|
meta[key.strip()] = value.strip().strip('"')
|
|
return meta, body
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ProcedureDocument:
|
|
source_id: str
|
|
language: str
|
|
document_id: str
|
|
title: str
|
|
version: str
|
|
content: str
|
|
content_hash: str
|
|
|
|
|
|
def iter_procedure_documents(knowledge_dir: Path) -> list[ProcedureDocument]:
|
|
"""Read every procedure Markdown file Fleet Ops ships, across every supported
|
|
language, as a flat list ready for external sync (e.g. into RAGcore). Frontmatter
|
|
fields (title, version) come from the same files the demo knowledge provider
|
|
already reads -- see parse_frontmatter -- so the two never drift apart."""
|
|
|
|
documents: list[ProcedureDocument] = []
|
|
for language in SUPPORTED_LANGUAGES:
|
|
language_dir = knowledge_dir / language
|
|
if not language_dir.is_dir():
|
|
continue
|
|
for path in sorted(language_dir.glob("*.md")):
|
|
raw = path.read_text(encoding="utf-8")
|
|
meta, body = parse_frontmatter(raw)
|
|
document_id = meta.get("document_id", path.stem)
|
|
content = body.strip()
|
|
documents.append(
|
|
ProcedureDocument(
|
|
source_id=str(uuid.uuid5(_SOURCE_ID_NAMESPACE, f"{language}:{document_id}")),
|
|
language=language,
|
|
document_id=document_id,
|
|
title=meta.get("title", path.stem),
|
|
version=meta.get("version", "1.0"),
|
|
content=content,
|
|
content_hash=hashlib.sha256(content.encode("utf-8")).hexdigest(),
|
|
)
|
|
)
|
|
return documents
|