167 lines
7.3 KiB
Python
167 lines
7.3 KiB
Python
"""Render only validated structured facts; never infer facts or read gold."""
|
|
|
|
from copy import deepcopy
|
|
import re
|
|
|
|
from tools.graphrag.graphrag import WORK_TYPE_LABELS
|
|
from tools.answers.presentation import fact_sentences, no_answer_text, clean_answer
|
|
|
|
|
|
def render_graph_answer(answer: dict, context: dict) -> dict:
|
|
out = deepcopy(answer)
|
|
lines = []
|
|
if out["status"] == "ok":
|
|
shared_fields = set()
|
|
shared_people = []
|
|
labels = {
|
|
"keyword": "kľúčové slová",
|
|
"category": "kategórie",
|
|
"document_author": "autori dokumentov",
|
|
}
|
|
for field, label in labels.items():
|
|
values = sorted(
|
|
{
|
|
row["value"]
|
|
for row in context.get("shared_properties", [])
|
|
if row["field"] == field
|
|
and all(
|
|
any(
|
|
c.get("person") == person and c.get(field) == row["value"]
|
|
for c in out["claims"]
|
|
)
|
|
for person in row["people"]
|
|
)
|
|
}
|
|
)
|
|
if values:
|
|
shared_fields.add(field)
|
|
full = context.get("complete") is True and not context.get("truncated")
|
|
prefix = (
|
|
("Spoloční " if full else "Doložení spoloční ")
|
|
if field == "document_author"
|
|
else ("Spoločné " if full else "Doložené spoločné ")
|
|
)
|
|
lines.append(
|
|
prefix + label + ": " + ", ".join("**" + v + "**" for v in values) + "."
|
|
)
|
|
for row in context.get("shared_properties", []):
|
|
if row["field"] == field and row["value"] in values:
|
|
shared_people.extend(row["people"])
|
|
if shared_people:
|
|
lines.extend(
|
|
[
|
|
"Doložené pri: "
|
|
+ ", ".join("**" + p + "**" for p in dict.fromkeys(shared_people))
|
|
+ ".",
|
|
"",
|
|
]
|
|
)
|
|
comparison = context.get("academic_year_comparison")
|
|
if comparison:
|
|
people = comparison.get("people", [])
|
|
witnessed = [
|
|
{
|
|
claim["year"]
|
|
for claim in out["claims"]
|
|
if claim.get("person") == person and claim.get("year") is not None
|
|
}
|
|
for person in people
|
|
]
|
|
visible = set.intersection(*witnessed) if witnessed else set()
|
|
years = sorted(visible & set(comparison.get("shared_years", [])))
|
|
if years:
|
|
label = (
|
|
"Spoločný evidovaný rok je" if len(years) == 1 else "Spoločné evidované roky sú"
|
|
)
|
|
if context.get("complete") is not True or context.get("truncated"):
|
|
label = "Medzi doložené spoločné roky patria"
|
|
lines.extend([label + " **" + ", ".join(map(str, years)) + "**.", ""])
|
|
records = {}
|
|
for claim in out["claims"]:
|
|
if claim.get("record_id"):
|
|
records.setdefault(claim["person"], []).append(claim)
|
|
else:
|
|
shown = {key: value for key, value in claim.items() if key not in shared_fields}
|
|
if any(key != "person" and value is not None for key, value in shown.items()):
|
|
lines.append(fact_sentences(shown, context.get("question", "")))
|
|
index = 0
|
|
for person, claims in records.items():
|
|
lines.extend([f"{person} má v dostupných podkladoch tieto práce a projekty:", ""])
|
|
# Stable sorting retains the source order of same-year projects.
|
|
for claim in sorted(
|
|
claims, key=lambda row: (row.get("year") is None, row.get("year") or 0)
|
|
):
|
|
index += 1
|
|
kind = WORK_TYPE_LABELS.get(claim.get("work_type"), "Práca")
|
|
label = claim.get("work_label") or kind
|
|
details = []
|
|
if kind.casefold() not in label.casefold():
|
|
details.append(kind)
|
|
if claim.get("year") is not None and not re.search(
|
|
r"\b" + str(claim["year"]) + r"\b", label
|
|
):
|
|
details.append(f'rok práce: {claim["year"]}')
|
|
if claim.get("year") is None:
|
|
details.append("rok v zdroji nie je uvedený")
|
|
title = (
|
|
"**" + claim["title"].rstrip(".") + "**"
|
|
if claim.get("title")
|
|
else "názov v zdroji nie je uvedený"
|
|
)
|
|
lines.append(
|
|
f"{index}. **{label}**"
|
|
+ (" (" + "; ".join(details) + ")" if details else "")
|
|
+ f" — {title}."
|
|
)
|
|
extra = {
|
|
key: claim[key]
|
|
for key in ("person", "keyword", "category", "document_author", "start_year")
|
|
if claim.get(key) is not None
|
|
}
|
|
if len(extra) > 1:
|
|
lines.append(fact_sentences(extra))
|
|
lines.append("")
|
|
if out.get("count") is not None:
|
|
aggregation = context.get("aggregation") or {}
|
|
label='Počet evidovaných prác a projektov' if aggregation.get('count_domain')=='academic_records' else 'Počet vyhovujúcich osôb'
|
|
lines.append(label+f': {out["count"]}.')
|
|
if aggregation.get("complete") is True and out["count"] == aggregation.get("count"):
|
|
# A full total is backed by the inspected candidate set, not
|
|
# just the one person left visible after a presentation cap.
|
|
out["source_urls"] = list(
|
|
dict.fromkeys(
|
|
[
|
|
*out["source_urls"],
|
|
*(
|
|
s["source_url"]
|
|
for s in context.get("sources", [])
|
|
if s.get("source_url")
|
|
),
|
|
]
|
|
)
|
|
)
|
|
if context.get("complete") is False or context.get("truncated"):
|
|
lines.append("Výpis je neúplný.")
|
|
lines.append("Výpis môžeš zúžiť uvedením osoby, typu práce alebo konkrétneho roku.")
|
|
elif context.get("complete") is None:
|
|
lines.append("Úplnosť vyhľadania nemožno zaručiť.")
|
|
elif out["status"] == "ambiguous":
|
|
lines.append("Požiadavka je nejednoznačná; údaje nemožno spoľahlivo určiť.")
|
|
elif out["status"] == "error":
|
|
lines.append("Vyhľadanie zlyhalo; odpoveď nemožno pripraviť.")
|
|
else:
|
|
if context.get("academic_year_comparison") is not None:
|
|
lines.append("V dostupných podkladoch sa pre zadané podmienky spoločný rok nenašiel.")
|
|
else:
|
|
lines.append(no_answer_text(context))
|
|
if out["source_urls"]:
|
|
lines.extend(
|
|
[
|
|
"",
|
|
("Zdroj: " if len(out["source_urls"]) == 1 else "Zdroje: ")
|
|
+ ", ".join(out["source_urls"]),
|
|
]
|
|
)
|
|
out["answer"] = clean_answer(lines)
|
|
return out
|