431 lines
16 KiB
Python
431 lines
16 KiB
Python
"""Independent benchmark oracle over original Markdown, not the GraphRAG parser.
|
||
|
||
Frontmatter defines tags/categories/document authors. Academic year comes from an
|
||
explicit academic heading, with a matching category used only as a fallback.
|
||
Every extracted value records the original file and source lines. No graph
|
||
payload, chunk JSON or retrieval result is used to construct this oracle.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import os
|
||
import json
|
||
import re
|
||
import unicodedata
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import frontmatter
|
||
|
||
ROOT = Path(__file__).resolve().parents[3]
|
||
DEFAULT_WIKI = Path(os.getenv("ZPWIKI_ROOT", str(ROOT.parent / "zpwiki")))
|
||
|
||
|
||
def norm(value: Any) -> str:
|
||
value = unicodedata.normalize("NFKD", str(value or ""))
|
||
value = "".join(c for c in value if not unicodedata.combining(c))
|
||
return " ".join(re.findall(r"[a-z0-9]+", value.casefold()))
|
||
|
||
|
||
def clean(value: str) -> str:
|
||
value = re.sub(r"\[([^\]]+)\]\([^)]*\)", r"\1", value)
|
||
return re.sub(r"[*_`#]", "", value).strip()
|
||
|
||
|
||
def values(value: Any) -> list[str]:
|
||
if value is None:
|
||
return []
|
||
if isinstance(value, str):
|
||
return [value.strip()] if value.strip() else []
|
||
return [str(item).strip() for item in value if str(item).strip()]
|
||
|
||
|
||
def profile_path(path: str) -> str | None:
|
||
match = re.match(r"^(pages/students/\d{4}/[^/]+|pages/interns/[^/]+)/", path)
|
||
return match.group(1) + "/README.md" if match else None
|
||
|
||
|
||
def source_url(path: str) -> str:
|
||
return "https://zp.kemt.fei.tuke.sk/" + path.removeprefix("pages/").removesuffix("/README.md")
|
||
|
||
|
||
def category_type(category: str) -> tuple[str | None, int | None]:
|
||
match = re.fullmatch(r"(bp|dp|phd|tp|vp)(20\d{2})", norm(category).replace(" ", ""))
|
||
if not match:
|
||
return None, None
|
||
return {
|
||
"bp": "bachelor_thesis",
|
||
"dp": "diploma_thesis",
|
||
"phd": "doctoral_thesis",
|
||
"tp": "team_project",
|
||
"vp": "research_project",
|
||
}[match[1]], int(match[2])
|
||
|
||
|
||
def academic_type(heading: str) -> str | None:
|
||
text = norm(heading)
|
||
if text.startswith(("priprava na ", "priprava k ")):
|
||
return None
|
||
for prefix, kind in [
|
||
("bakalarska praca", "bachelor_thesis"),
|
||
("bakalarsky projekt", "bachelor_project"),
|
||
("diplomova praca", "diploma_thesis"),
|
||
("dizertacna praca", "doctoral_thesis"),
|
||
("diplomovy projekt", "diploma_project"),
|
||
("timovy projekt", "team_project"),
|
||
("vyskumny projekt", "research_project"),
|
||
("vedecky projekt", "research_project"),
|
||
]:
|
||
if text.startswith(prefix):
|
||
return kind
|
||
return None
|
||
|
||
|
||
def acceptable_title(value: str, *, explicit: bool = False) -> bool:
|
||
text = norm(value)
|
||
# A bare deliverable label is not evidence of an academic title. Do not
|
||
# inherit a linked document's title/year into its parent academic record.
|
||
if text in {"resers", "tutorial", "projekt", "dokument", "repozitar", "pisomna praca"}:
|
||
return False
|
||
if not (
|
||
len(value) >= (6 if explicit else 10)
|
||
and len(value) <= 180
|
||
and len(text.split()) >= (1 if explicit else 2)
|
||
):
|
||
return False
|
||
if re.search(r"https?://", value):
|
||
return False
|
||
if value.endswith(":") or re.match(r"^(?:https?://|[-*+] |\d+[.)] )", value):
|
||
return False
|
||
return not any(
|
||
text.startswith(prefix)
|
||
for prefix in [
|
||
"stretnutie",
|
||
"stav",
|
||
"ciel",
|
||
"ulohy",
|
||
"uloha",
|
||
"zadanie",
|
||
"navrh na zadanie",
|
||
"navrh na zadamie",
|
||
"poznamka",
|
||
"praca uspesne",
|
||
"projekt uspesne",
|
||
"projekt zakoncen",
|
||
"bakalarska praca na crzp",
|
||
"git repozitar",
|
||
"zdrojove texty",
|
||
"praca na crzp",
|
||
"rok zaciatku",
|
||
"zaciatok studia",
|
||
"meno veduceho",
|
||
"veduci",
|
||
"crzp",
|
||
"pozriet si",
|
||
"prejst si",
|
||
"konzultacie",
|
||
"projekt odlozeny",
|
||
"napady na",
|
||
"virtualne stretnutie",
|
||
"chceme vytvorit",
|
||
"praca suvisi",
|
||
"pisomna praca",
|
||
"spolupraca",
|
||
"suvisiaca praca",
|
||
"odporucany cas konzult",
|
||
"odporucany termin konzult",
|
||
"cas konzult",
|
||
"termin konzult",
|
||
]
|
||
)
|
||
|
||
|
||
def title_candidates(lines: list[str], offset: int = 0) -> list[dict]:
|
||
found = []
|
||
for index, raw in enumerate(lines):
|
||
line = clean(raw)
|
||
if ":" not in line:
|
||
continue
|
||
label, inline = line.split(":", 1)
|
||
key = norm(label)
|
||
if not (
|
||
key
|
||
in {
|
||
"nazov",
|
||
"nazov prace",
|
||
"nazov bakalarskej prace",
|
||
"nazov diplomovej prace",
|
||
"pisomna praca",
|
||
"predbezny nazov",
|
||
"tema",
|
||
"navrh na temu",
|
||
}
|
||
or key.startswith("navrh na nazov")
|
||
):
|
||
continue
|
||
value = inline.strip()
|
||
value_index = index
|
||
if not value:
|
||
for next_index in range(index + 1, min(index + 7, len(lines))):
|
||
if lines[next_index].strip():
|
||
value = clean(lines[next_index])
|
||
value_index = next_index
|
||
break
|
||
if acceptable_title(value, explicit=True):
|
||
found.append(
|
||
{
|
||
"title": value,
|
||
"line": offset + value_index + 1,
|
||
"label": key,
|
||
"label_line": offset + index + 1,
|
||
"priority": (
|
||
0
|
||
if key.startswith("nazov")
|
||
else 1 if key.startswith("navrh na nazov") else 2
|
||
),
|
||
}
|
||
)
|
||
return sorted(found, key=lambda row: (row["priority"], row["line"]))
|
||
|
||
|
||
def extract_academics(document: dict) -> list[dict]:
|
||
lines = document["body_lines"]
|
||
offset = document["body_offset"]
|
||
headings = []
|
||
for index, line in enumerate(lines):
|
||
if re.match(r"^\s*#{1,6}\s+", line):
|
||
kind = academic_type(clean(line))
|
||
if kind:
|
||
headings.append((index, kind, clean(line)))
|
||
records = []
|
||
root_end = headings[0][0] if headings else len(lines)
|
||
root_candidates = title_candidates(lines[:root_end], offset)
|
||
cat_years: dict[str, set[int]] = {}
|
||
for cat in document["categories"]:
|
||
kind, year = category_type(cat)
|
||
if kind and year:
|
||
cat_years.setdefault(kind, set()).add(year)
|
||
for pos, (index, kind, heading) in enumerate(headings):
|
||
end = headings[pos + 1][0] if pos + 1 < len(headings) else len(lines)
|
||
section = lines[index + 1 : end]
|
||
# Academic headings can contain a range, e.g. 2023/24. Prefer the
|
||
# matching category year within that range; otherwise use its end.
|
||
heading_years = [int(value) for value in re.findall(r"\b20\d{2}\b", heading)]
|
||
for match in re.finditer(r"\b(20\d{2})\s*[/–-]\s*(\d{2})(?!\d)", heading):
|
||
heading_years.append(int(match[1]) // 100 * 100 + int(match[2]))
|
||
matching_years = set(heading_years) & cat_years.get(kind, set())
|
||
year = (
|
||
(max(matching_years) if matching_years else max(heading_years))
|
||
if heading_years
|
||
else (next(iter(cat_years[kind])) if len(cat_years.get(kind, [])) == 1 else None)
|
||
)
|
||
candidates = title_candidates(section, offset + index + 1)
|
||
chosen = candidates[0] if candidates else None
|
||
# A root title explicitly marked BP/DP belongs to the matching thesis.
|
||
if not chosen:
|
||
matching_root = [
|
||
row
|
||
for row in root_candidates
|
||
if (" dp" in row["label"] or "diplom" in row["label"])
|
||
and kind == "diploma_thesis"
|
||
or (" bp" in row["label"] or "bakalar" in row["label"])
|
||
and kind == "bachelor_thesis"
|
||
]
|
||
if matching_root:
|
||
chosen = matching_root[0]
|
||
if not chosen:
|
||
# A short leading plain paragraph before assignment lists is a title;
|
||
# task/status labels, links and prose after a task boundary are not.
|
||
leading = []
|
||
for n, raw in enumerate(section[:15]):
|
||
text = clean(raw)
|
||
if not text:
|
||
continue
|
||
if re.match(r"^\s*#{1,6}\s+", raw) or re.match(r"^\s*\d+[.)]\s+", raw):
|
||
break
|
||
if re.match(r"^\s*[-*+]\s+", raw):
|
||
if "](" in raw:
|
||
continue
|
||
break
|
||
if any(
|
||
norm(text).startswith(word)
|
||
for word in [
|
||
"zadanie",
|
||
"ciel",
|
||
"ulohy",
|
||
"navrh na zadanie",
|
||
"navrh na zadamie",
|
||
"stretnutie",
|
||
]
|
||
):
|
||
break
|
||
if raw.strip().startswith(("http", "[")) or not acceptable_title(text):
|
||
continue
|
||
leading.append(
|
||
{"title": text, "line": offset + index + n + 2, "label": "leading_title"}
|
||
)
|
||
if leading:
|
||
chosen = leading[-1]
|
||
records.append(
|
||
{
|
||
"academic_kind": "Thesis" if kind.endswith("_thesis") else "Work",
|
||
"academic_type": kind,
|
||
"year": year,
|
||
"title": chosen["title"] if chosen else None,
|
||
"document_path": document["path"],
|
||
"heading": heading,
|
||
"heading_line": offset + index + 1,
|
||
"title_line": chosen["line"] if chosen else None,
|
||
"year_basis": "heading" if heading_years else "frontmatter_category",
|
||
}
|
||
)
|
||
# Standalone root titles occur in profiles without an academic heading.
|
||
# Use an explicit type label, or one unambiguous latest thesis category.
|
||
if root_candidates:
|
||
latest = sorted(
|
||
(year, kind)
|
||
for kind, years in cat_years.items()
|
||
if kind.endswith("_thesis")
|
||
for year in years
|
||
)
|
||
if latest:
|
||
year, kind = latest[-1]
|
||
candidates = [
|
||
row
|
||
for row in root_candidates
|
||
if (
|
||
" dp" not in row["label"]
|
||
and "diplom" not in row["label"]
|
||
or kind == "diploma_thesis"
|
||
)
|
||
and (
|
||
" bp" not in row["label"]
|
||
and "bakalar" not in row["label"]
|
||
or kind == "bachelor_thesis"
|
||
)
|
||
]
|
||
if candidates:
|
||
existing = next(
|
||
(r for r in records if r["academic_type"] == kind and r["year"] == year), None
|
||
)
|
||
chosen = candidates[0]
|
||
if existing and not existing["title"]:
|
||
existing.update(title=chosen["title"], title_line=chosen["line"])
|
||
elif existing is None:
|
||
records.append(
|
||
{
|
||
"academic_kind": "Thesis",
|
||
"academic_type": kind,
|
||
"year": year,
|
||
"title": chosen["title"],
|
||
"document_path": document["path"],
|
||
"heading": None,
|
||
"heading_line": None,
|
||
"title_line": chosen["line"],
|
||
"year_basis": "frontmatter_category",
|
||
}
|
||
)
|
||
return records
|
||
|
||
|
||
def load_source_oracle(wiki_root: Path = DEFAULT_WIKI) -> dict:
|
||
documents = {}
|
||
corpus_hashes = {}
|
||
for file in sorted((wiki_root / "pages").rglob("README.md")):
|
||
raw = file.read_text(encoding="utf-8")
|
||
path = file.relative_to(wiki_root).as_posix()
|
||
corpus_hashes[path] = hashlib.sha256(file.read_bytes()).hexdigest()
|
||
page = frontmatter.loads(raw)
|
||
taxonomy = page.get("taxonomy") or {}
|
||
raw_lines = raw.splitlines()
|
||
body_offset = 0
|
||
if raw_lines and raw_lines[0].strip() == "---":
|
||
body_offset = next(
|
||
(
|
||
index + 1
|
||
for index in range(1, len(raw_lines))
|
||
if raw_lines[index].strip() == "---"
|
||
),
|
||
0,
|
||
)
|
||
documents[path] = {
|
||
"path": path,
|
||
"title": str(page.get("title") or "").strip(),
|
||
"published": page.get("published") is not False,
|
||
"tags": values(taxonomy.get("tag")),
|
||
"categories": values(taxonomy.get("category")),
|
||
"author": str(taxonomy.get("author") or page.get("author") or "").strip() or None,
|
||
"body_lines": raw_lines[body_offset:],
|
||
"body_offset": body_offset,
|
||
"sha256": corpus_hashes[path],
|
||
"source_url": source_url(path),
|
||
"raw_text": raw,
|
||
}
|
||
people = {}
|
||
for path, document in documents.items():
|
||
if profile_path(path) != path or not document["title"]:
|
||
continue
|
||
family = [doc for child, doc in documents.items() if profile_path(child) == path]
|
||
start = re.search(
|
||
r"(?:rok\s+)?zaciat\w*\s+studi\w*\s*:?\s*([12]\d{3})",
|
||
norm("\n".join(document["body_lines"])),
|
||
)
|
||
if not start:
|
||
start = re.search(
|
||
r"beginning of the study\s*(20\d{2})", norm("\n".join(document["body_lines"]))
|
||
)
|
||
people[document["title"]] = {
|
||
"person": document["title"],
|
||
"profile": path,
|
||
"documents": [d["path"] for d in family],
|
||
"tags": sorted({tag for d in family for tag in d["tags"]}, key=norm),
|
||
"categories": sorted({cat for d in family for cat in d["categories"]}, key=norm),
|
||
"start_year": int(start[1]) if start else None,
|
||
"start_year_basis": "explicit_text" if start else "missing",
|
||
"cohort_year": int(path.split("/")[2]) if path.startswith("pages/students/") else None,
|
||
"academics": extract_academics(document),
|
||
"year_conflicts": [
|
||
r
|
||
for r in extract_academics(document)
|
||
if r["year_basis"] == "heading"
|
||
and any(
|
||
category_type(cat)[0] == r["academic_type"]
|
||
and category_type(cat)[1] != r["year"]
|
||
for cat in document["categories"]
|
||
)
|
||
],
|
||
}
|
||
canonical = json.dumps(corpus_hashes, sort_keys=True, separators=(",", ":"))
|
||
return {
|
||
"wiki_root": str(wiki_root.resolve()),
|
||
"people": people,
|
||
"documents": documents,
|
||
"source_corpus": {
|
||
"sha256": hashlib.sha256(canonical.encode()).hexdigest(),
|
||
"document_count": len(corpus_hashes),
|
||
"documents": corpus_hashes,
|
||
},
|
||
}
|
||
|
||
|
||
def supporting_documents(oracle: dict, person: str, **constraints: Any) -> list[str]:
|
||
profile = oracle["people"].get(person)
|
||
if not profile:
|
||
return []
|
||
result = []
|
||
for path in profile["documents"]:
|
||
doc = oracle["documents"][path]
|
||
if constraints.get("keyword") and norm(constraints["keyword"]) not in {
|
||
norm(tag) for tag in doc["tags"]
|
||
}:
|
||
continue
|
||
if constraints.get("category") and constraints["category"] not in doc["categories"]:
|
||
continue
|
||
if constraints.get("document_author") and norm(doc["author"]) != norm(
|
||
constraints["document_author"]
|
||
):
|
||
continue
|
||
result.append(path)
|
||
return result
|