262 lines
14 KiB
Python
262 lines
14 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)
|
||
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*(20\d{2})', 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 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
|