"""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"(? "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": []}