367 lines
18 KiB
Python
367 lines
18 KiB
Python
"""Deterministic roster: which works (topics) and students belong to given years.
|
||
|
||
"Aké témy sa riešili v roku 2024 a akí študenti?" asks for a list, not for prose.
|
||
Every listed work is one unchanged source row (its title/topic when available)
|
||
of the same student, work and year; no model call is needed.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from collections import defaultdict
|
||
import re
|
||
|
||
from tools.answers.context_policy import normalized
|
||
from tools.knowledge.evidence import claim_eligible
|
||
from tools.answers.content_selection import administrative_only
|
||
|
||
MAX_WORKS = 40
|
||
_RANK = {"topic": 0, "assignment": 1}
|
||
|
||
|
||
def roster_question(spec) -> bool:
|
||
n = normalized(spec.question)
|
||
years = spec.years or (spec.period and spec.period.field == "work_year")
|
||
return bool(
|
||
spec.intent == "metadata"
|
||
and not spec.people
|
||
and years
|
||
and re.search(r"\b(?:tem\w*|prac\w*|projekt\w*|na com sa|co sa|ries\w*)\b", n)
|
||
and not re.search(r"\b(?:tag\w*|klucov\w*|kategor\w*|kolko|pocet|nazov\w*|autor\w*|vedu\w*|vedie)\b", n)
|
||
)
|
||
|
||
|
||
def _year_label(spec):
|
||
if spec.period and spec.period.field == "work_year":
|
||
first, last = spec.period.start[:4], spec.period.end[:4]
|
||
return f"z rokov {first} – {last}" if first != last else f"z roku {first}"
|
||
years = sorted(set(spec.years))
|
||
return ("z roku " + str(years[0])) if len(years) == 1 else "z rokov " + ", ".join(map(str, years))
|
||
|
||
|
||
def answer_roster(spec, index, *, engine: str) -> dict:
|
||
best = {}
|
||
for row in index["units"]:
|
||
if (not claim_eligible(row) or not row.get("work_label") or not row.get("work_year")
|
||
or not spec.matches_evidence(row) or administrative_only(row["text"])):
|
||
continue
|
||
if spec.years and row["work_year"] not in spec.years:
|
||
continue
|
||
key = (row["person"], row["source_path"], row["work_label"])
|
||
# A bare heading ("## Diplomový projekt 2024") names the work, not its topic.
|
||
heading = bool(re.fullmatch(r"\s*#+\s.*", row["text"])) or normalized(row["text"]) == normalized(row["work_label"])
|
||
rank = 9 if heading else _RANK.get(row["kind"], 2)
|
||
if key not in best or rank < best[key][0]:
|
||
best[key] = (rank, row)
|
||
works = sorted((row for _, row in best.values()),
|
||
key=lambda r: (r["work_year"], normalized(r["person"]), r["work_label"]))
|
||
truncated = len(works) > MAX_WORKS
|
||
shown = works[:MAX_WORKS]
|
||
base = {
|
||
"engine": engine, "index_version": index["index_version"], "knowledge_version": index["version"],
|
||
"accepted_for_delivery": True, "claims": [],
|
||
}
|
||
if not shown:
|
||
return {**base, "status": "no_answer", "reason": "no_content_evidence",
|
||
"answer": f"V dostupných podkladoch sa nenašli práce {_year_label(spec)} zodpovedajúce otázke.",
|
||
"evidence": [], "source_urls": [], "presentation_notes": [],
|
||
"complete": False, "truncated": False, "truncation_reasons": []}
|
||
by_person = defaultdict(list)
|
||
for row in shown:
|
||
by_person[row["person"]].append(row)
|
||
students = len({r["person"] for r in works})
|
||
lines = [f"Práce {_year_label(spec)} v ZP Wiki: {len(works)} prác, {students} študentov."]
|
||
urls = []
|
||
for person, rows in by_person.items():
|
||
lines.append(f"**{person}**")
|
||
for row in rows:
|
||
if re.fullmatch(r"\s*#+\s.*", row["text"]) or normalized(row["text"]) == normalized(row["work_label"]):
|
||
lines.append(f"- {row['work_label']} (téma v profile neuvedená)")
|
||
continue
|
||
label = "téma alebo názov" if row["kind"] == "topic" else "zadanie" if row["kind"] == "assignment" else "podklad"
|
||
lines.append(f"- {row['work_label']}; {label}: {row['text'].strip()}")
|
||
url = rows[0].get("source_url")
|
||
if url:
|
||
lines.append(f"Zdroj: {url}")
|
||
urls.append(url)
|
||
notes = []
|
||
if truncated:
|
||
notes.append(f"Zobrazených je prvých {MAX_WORKS} prác; zúž otázku typom práce alebo rokom.")
|
||
lines.append(notes[-1])
|
||
return {**base, "status": "ok", "reason": None, "answer": "\n\n".join(lines),
|
||
"evidence": shown, "source_urls": list(dict.fromkeys(urls)), "presentation_notes": notes,
|
||
"complete": not truncated, "truncated": truncated,
|
||
"truncation_reasons": ["result_limit"] if truncated else []}
|
||
|
||
|
||
# --- Students having several work types ("bakalársku aj diplomovú prácu") ---
|
||
|
||
_ADJECTIVES = [
|
||
(r"bakalar\w*", "bachelor_thesis", "bachelor_project"),
|
||
(r"diplomov\w*", "diploma_thesis", "diploma_project"),
|
||
(r"dizertacn\w*", "doctoral_thesis", "doctoral_thesis"),
|
||
(r"timov\w*", "team_project", "team_project"),
|
||
(r"(?:vyskumn|vedeck)\w*", "research_project", "research_project"),
|
||
]
|
||
TYPE_NAMES = {
|
||
"bachelor_thesis": "bakalárska práca", "bachelor_project": "bakalársky projekt",
|
||
"diploma_thesis": "diplomová práca", "diploma_project": "diplomový projekt",
|
||
"doctoral_thesis": "dizertačná práca", "team_project": "tímový projekt",
|
||
"research_project": "vedecký projekt",
|
||
}
|
||
|
||
|
||
TYPE_ACCUSATIVE = {
|
||
"bachelor_thesis": "bakalársku prácu", "bachelor_project": "bakalársky projekt",
|
||
"diploma_thesis": "diplomovú prácu", "diploma_project": "diplomový projekt",
|
||
"doctoral_thesis": "dizertačnú prácu", "team_project": "tímový projekt",
|
||
"research_project": "vedecký projekt",
|
||
}
|
||
|
||
|
||
def students_word(count: int) -> str:
|
||
return "študent" if count == 1 else "študenti" if 2 <= count <= 4 else "študentov"
|
||
|
||
|
||
def conjunctive_types(question: str) -> list[str]:
|
||
"""Work types joined by "aj"/"a zároveň": each adjective takes the next noun."""
|
||
n = normalized(question)
|
||
if not re.search(r"\b(?:aj|zaroven|sucasne|tiez|i)\b", n):
|
||
return []
|
||
found = []
|
||
for match in re.finditer(r"\b(" + "|".join(a for a, _, _ in _ADJECTIVES) + r")\b", n):
|
||
noun = re.compile(r"\b(prac\w*|projekt\w*)\b").search(n, match.end())
|
||
for pattern, thesis, project in _ADJECTIVES:
|
||
if re.fullmatch(pattern, match[1]):
|
||
kind = project if noun and noun[1].startswith("projekt") else thesis
|
||
if kind not in found:
|
||
found.append(kind)
|
||
return found if len(found) >= 2 else []
|
||
|
||
|
||
def requested_conjunction(spec) -> list[str]:
|
||
if getattr(spec, "route", None) == "students_with_work_types":
|
||
kinds = list(dict.fromkeys(w.work_type for w in spec.works if w.work_type))
|
||
if len(kinds) >= 2:
|
||
return kinds
|
||
return conjunctive_types(spec.question)
|
||
|
||
|
||
def both_types_question(spec) -> bool:
|
||
if spec.intent != "metadata" or spec.people:
|
||
return False
|
||
if getattr(spec, "route", None) == "students_with_work_types":
|
||
return len(requested_conjunction(spec)) >= 2
|
||
return bool(conjunctive_types(spec.question)
|
||
and re.search(r"\b(?:kto|ktor\w*|ak\w*|kolko|pocet|vypis\w*|vymenuj\w*|zoznam\w*)\b",
|
||
normalized(spec.question)))
|
||
|
||
|
||
def _best_rows(rows):
|
||
best = {}
|
||
for row in rows:
|
||
key = (row["person"], row["source_path"], row["work_label"])
|
||
heading = bool(re.fullmatch(r"\s*#+\s.*", row["text"])) or normalized(row["text"]) == normalized(row["work_label"])
|
||
# "Návrh na tému:" introduces a list; it is not the work's topic.
|
||
label_only = row["text"].strip().endswith(":") or len(row["text"].strip()) < 12
|
||
rank = 9 if heading else 8 if label_only else _RANK.get(row["kind"], 2)
|
||
if key not in best or rank < best[key][0]:
|
||
best[key] = (rank, row)
|
||
return {key: row for key, (_, row) in best.items()}
|
||
|
||
|
||
def _work_line(row):
|
||
if re.fullmatch(r"\s*#+\s.*", row["text"]) or normalized(row["text"]) == normalized(row["work_label"]):
|
||
return f"- {row['work_label']}"
|
||
label = "téma alebo názov" if row["kind"] == "topic" else "zadanie" if row["kind"] == "assignment" else "podklad"
|
||
return f"- {row['work_label']}; {label}: {row['text'].strip()}"
|
||
|
||
|
||
def _base(index, engine):
|
||
return {"engine": engine, "index_version": index["index_version"], "knowledge_version": index["version"],
|
||
"accepted_for_delivery": True, "claims": [], "presentation_notes": []}
|
||
|
||
|
||
def answer_work_types(spec, index, *, engine: str) -> dict:
|
||
kinds = requested_conjunction(spec)
|
||
rows = [r for r in index["units"] if claim_eligible(r) and r.get("work_label") and r.get("work_type") in kinds
|
||
and r["source_path"].startswith(("pages/students/", "students/")) and not administrative_only(r["text"])]
|
||
works = _best_rows(rows)
|
||
by_profile = defaultdict(dict)
|
||
for (person, path, label), row in works.items():
|
||
by_profile[(person, path)].setdefault(row["work_type"], []).append(row)
|
||
chosen = sorted((key for key, types in by_profile.items() if all(k in types for k in kinds)),
|
||
key=lambda k: normalized(k[0]))
|
||
names = " aj ".join(TYPE_ACCUSATIVE[k] for k in kinds)
|
||
base = _base(index, engine)
|
||
if not chosen:
|
||
return {**base, "status": "no_answer", "reason": "no_content_evidence",
|
||
"answer": f"V profiloch ZP Wiki sa nenašiel študent, ktorý má {names}.",
|
||
"evidence": [], "source_urls": [], "complete": False, "truncated": False, "truncation_reasons": []}
|
||
lines = [f"Študenti, ktorí majú v ZP Wiki {names}: {len(chosen)}."]
|
||
evidence, urls = [], []
|
||
for person, path in chosen:
|
||
lines.append(f"**{person}**")
|
||
for kind in kinds:
|
||
for row in sorted(by_profile[(person, path)][kind], key=lambda r: (r.get("work_year") or 0, r["work_label"])):
|
||
lines.append(_work_line(row))
|
||
evidence.append(row)
|
||
url = by_profile[(person, path)][kinds[0]][0].get("source_url")
|
||
if url:
|
||
lines.append(f"Zdroj: {url}")
|
||
urls.append(url)
|
||
note = "Ide o práce evidované v profiloch; evidovaná práca sama nepotvrdzuje jej obhajobu."
|
||
lines.append(note)
|
||
return {**base, "status": "ok", "reason": None, "answer": "\n\n".join(lines), "evidence": evidence,
|
||
"source_urls": list(dict.fromkeys(urls)), "presentation_notes": [note],
|
||
"complete": True, "truncated": False, "truncation_reasons": []}
|
||
|
||
|
||
# --- Topics related to a topic ("Ktoré témy súvisia s RAG?") ---
|
||
|
||
_RELATED = re.compile(
|
||
r"\b(?:suvis\w*|pribuzn\w*|spojen\w*|prepojen\w*|sprevadza\w*)\s+(?:s|so|k)\s+(?:temou\s+|pojmom\s+|tagom\s+)?"
|
||
r"([^?.!,;]{2,40}?)\s*(?:[?.!,;]|$)"
|
||
r"|\bs\s+(?:cim|ktorymi\s+temami)\s+(?:suvisi|suvisia|sa\s+spaja)\s+(?:tema\s+)?([^?.!,;]{2,40}?)\s*(?:[?.!,;]|$)")
|
||
_NOISE = {"poznamky", "zdroje", "tutorial", "zlepsit", "prestudovat", "google scholar", "slovenskom jazyku",
|
||
"trenovaciu cast", "zero", "spracovanie tejto databazy", "evaluation", "retrieval", "vyhladavanie",
|
||
"python", "nlp", "system", "model", "student", "praca", "projekt", "git", "data", "text", "clanky"}
|
||
|
||
|
||
_CASE_ENDINGS = ("ami", "ach", "och", "ovi", "om", "ov", "ou", "a", "u", "e", "y", "i")
|
||
|
||
|
||
def topic_stem(key: str) -> str:
|
||
""""chatbotom" -> "chatbot"; short words and acronyms ("rag") stay whole."""
|
||
if " " in key or len(key) < 5:
|
||
return key
|
||
for ending in _CASE_ENDINGS:
|
||
if key.endswith(ending) and len(key) - len(ending) >= 4:
|
||
return key[: -len(ending)]
|
||
return key
|
||
|
||
|
||
def original_phrase(question: str, key: str) -> str:
|
||
"""The user's own spelling ("RAG") of a phrase recognized in normalized form."""
|
||
words = re.findall(r"[^\W_][\w-]*", question)
|
||
return next((" ".join(words[i:i + k]) for k in (1, 2, 3) for i in range(len(words))
|
||
if normalized(" ".join(words[i:i + k])) == key), key)
|
||
|
||
|
||
def related_topic(spec) -> str | None:
|
||
if spec.intent != "metadata" or spec.people:
|
||
return None
|
||
if getattr(spec, "route", None) == "related_topics" and getattr(spec, "topic", None):
|
||
return normalized(spec.topic).strip()
|
||
n = normalized(spec.question)
|
||
if not re.search(r"\b(?:tem\w*|oblast\w*|pojm\w*|tag\w*|znack\w*|co|cim)\b", n):
|
||
return None
|
||
match = _RELATED.search(n)
|
||
return (match[1] or match[2]).strip() if match else None
|
||
|
||
|
||
def profile_tags(db_file=None) -> dict:
|
||
"""Profile path -> set of its taxonomy tags, read from the published index."""
|
||
import sqlite3
|
||
from contextlib import closing
|
||
from pathlib import Path
|
||
from tools.rag.common import DB_FILE
|
||
path = Path(db_file or DB_FILE)
|
||
try:
|
||
with closing(sqlite3.connect(f"file:{path}?mode=ro", uri=True)) as conn:
|
||
pairs = conn.execute(
|
||
"SELECT DISTINCT c.document_path, t.tag FROM chunk_tags t JOIN chunks c ON c.chunk_id=t.chunk_id"
|
||
" WHERE COALESCE(c.published,1)=1").fetchall()
|
||
except sqlite3.Error:
|
||
return {}
|
||
tags = defaultdict(set)
|
||
for doc, tag in pairs:
|
||
tags[doc].add(normalized(tag))
|
||
return dict(tags)
|
||
|
||
|
||
def answer_related(spec, index, *, engine: str, db_file=None) -> dict:
|
||
from collections import Counter
|
||
asked = related_topic(spec)
|
||
stem = topic_stem(asked)
|
||
# "súvisia s chatbotom": every case form of the topic, not only the literal.
|
||
word = re.compile(r"(?<![\w-])" + re.escape(stem) + r"\w{0,4}(?![\w-])")
|
||
tags = profile_tags(db_file)
|
||
tag_names = {t for values in tags.values() for t in values if word.fullmatch(t)}
|
||
key = min(tag_names, key=len) if tag_names else asked
|
||
# The user's spelling when it is the base form ("RAG"), else the base form
|
||
# of the matching tag ("chatbotom" -> "chatbot").
|
||
spelled = original_phrase(spec.question, asked)
|
||
topic = spelled if normalized(spelled) == key else key
|
||
units = [r for r in index["units"] if claim_eligible(r) and r.get("work_label") and not administrative_only(r["text"])]
|
||
mentions = [r for r in units if word.search(normalized(r["text"]))
|
||
or any(word.fullmatch(normalized(c)) for c in r.get("concepts", []))]
|
||
tagged = {p for p, values in tags.items() if values & tag_names}
|
||
paths = {r["source_path"] for r in mentions} | tagged
|
||
base = _base(index, engine)
|
||
if not paths:
|
||
return {**base, "status": "no_answer", "reason": "no_content_evidence",
|
||
"answer": f"V dostupných podkladoch sa téma „{topic}“ nenašla.",
|
||
"evidence": [], "source_urls": [], "complete": False, "truncated": False, "truncation_reasons": []}
|
||
tag_counts = Counter(t for p in tagged for t in tags[p] if t != key)
|
||
concept_people = defaultdict(set)
|
||
concept_row = {}
|
||
for row in mentions:
|
||
for concept in row.get("concepts", []):
|
||
c = normalized(concept)
|
||
if (word.fullmatch(c) or any(word.fullmatch(part) for part in c.split()) or c in _NOISE or len(c) < 3 or len(c.split()) > 3
|
||
or re.search(r"\d|\bcom\b|\bwww\b|https?", c)):
|
||
continue
|
||
concept_people[c].add(row["person"])
|
||
concept_row.setdefault(c, (concept, row))
|
||
# A concept shared by at least two students; one mention is an anecdote.
|
||
concepts = sorted((c for c in concept_people if len(concept_people[c]) >= 2),
|
||
key=lambda c: (-len(concept_people[c]), c))[:10]
|
||
# The activity graph unifies names ("RAG" = "retrieval augmented
|
||
# generation"): its co-occurring concepts replace the raw unit concepts.
|
||
from tools.knowledge.activities import load_graph, related_concepts
|
||
from pathlib import Path
|
||
from tools.rag.common import DATA_DIR
|
||
graph = load_graph(Path(db_file).parent if db_file else DATA_DIR, index)
|
||
if graph is not None:
|
||
gids, _ = graph.resolve(spelled, getattr(spec, "topic_canonical", None))
|
||
if gids:
|
||
related, graph_units = related_concepts(graph, gids)
|
||
concept_people = {g: set(range(count)) for g, count, _ in related}
|
||
concept_row = {g: (graph.concept_label(g), row) for g, _, row in related}
|
||
concepts = [g for g, _, _ in related]
|
||
mentions = mentions + [graph.units[u] for u in graph_units if graph.units[u] not in mentions]
|
||
paths |= {graph.units[u]["source_path"] for u in graph_units}
|
||
lines = [f"Téma „{topic}“ sa v ZP Wiki spomína v profiloch {len(paths)} {'študenta' if len(paths) == 1 else 'študentov'}."]
|
||
evidence, notes = [], []
|
||
if tag_counts:
|
||
profiles = "profil" if len(tagged) == 1 else "profily" if 2 <= len(tagged) <= 4 else "profilov"
|
||
lines.append(f"Ďalšie značky profilov so značkou „{key}“ ({len(tagged)} {profiles}): "
|
||
+ ", ".join(f"{t} ({c})" for t, c in tag_counts.most_common(10)) + ".")
|
||
if concepts:
|
||
lines.append(f"Pojmy, ktoré sa spomínajú v tých istých záznamoch ako „{topic}“:")
|
||
for c in concepts:
|
||
label, row = concept_row[c]
|
||
count = len(concept_people[c])
|
||
lines.append(f"- {label} ({count} {students_word(count)}) — napríklad {row['person']}, {row['work_label']}: {row['text'].strip()}")
|
||
evidence.append(row)
|
||
lines.append("**Študenti a práce s touto témou:**")
|
||
works = _best_rows([r for r in units if r["source_path"] in paths])
|
||
shown_people = set()
|
||
for (person, path, label), row in sorted(works.items(), key=lambda kv: (normalized(kv[0][0]), -(kv[1].get("work_year") or 0))):
|
||
if (person, path) in shown_people:
|
||
continue
|
||
own = sorted((r for r in mentions if r["source_path"] == path
|
||
and not re.fullmatch(r"\s*[-*+]?\s*\[[^\]]*\]\([^)]*\)\s*", r["text"])),
|
||
key=lambda r: (r["phase"] not in {"result", "progress"}, r["kind"] != "topic", r["line"] if "line" in r else 0))
|
||
quote = own[0] if own else row
|
||
shown_people.add((person, path))
|
||
lines.append(f"- {person} — {quote['work_label']}: {quote['text'].strip()}")
|
||
evidence.append(quote)
|
||
urls = list(dict.fromkeys(r["source_url"] for r in evidence if r.get("source_url")))
|
||
note = "Súvislosť znamená spoločný výskyt v profiloch alebo záznamoch, nie potvrdený vecný vzťah tém."
|
||
lines.append(note)
|
||
notes.append(note)
|
||
lines.append("Zdroje: " + ", ".join(urls))
|
||
return {**base, "status": "ok", "reason": None, "answer": "\n\n".join(lines), "evidence": evidence,
|
||
"source_urls": urls, "presentation_notes": notes,
|
||
"complete": True, "truncated": False, "truncation_reasons": []}
|