dp-zp-agent/tools/crzp/answering/workplaces.py
2026-11-15 16:27:08 +01:00

181 lines
8.1 KiB
Python

"""Names of CRZP workplace codes, learned from CRZP records themselves.
CRZP writes newer works as "Technická univerzita v Košiciach 1040 104009" and
does not publish what 104009 means. Older records (about until 2015) write the
department by name ("1040 Katedra počítačov a informatiky"). A supervisor stays
at a department, so the name is learned per code:
1. search the code; the full supervisor facet of that result gives the people
who supervise works with this code;
2. for the most frequent supervisors read their older works (OLD_YEARS) at the
same university and collect the department names written there;
3. each supervisor votes once, for the department named most often in his or
her older works (spellings cut by CRZP, "… umelej inteligenci", count as the
full name);
4. accept a name only when at least MIN_VOTES supervisors agree, they are at
least MIN_SHARE of the voters and the name leads the next one by MARGIN votes;
5. a name that wins for several codes is kept only for the code with the
clearest vote (the others stay numbers); placeholder codes (999999) are skipped.
The result is stored in workplaces.json (committed, reviewable) with its votes,
so a code is named only with evidence; other codes stay "pracovisko 104009".
python -m tools.crzp.answering.workplaces learn # TUKE codes seen in the cache
python -m tools.crzp.answering.workplaces learn 104009 104004
"""
from __future__ import annotations
import argparse
import json
import re
from collections import Counter
from functools import lru_cache
from pathlib import Path
from tools.crzp.source.parse import display_name, fold
FILE = Path(__file__).with_name("workplaces.json")
TUKE = "Technická univerzita v Košiciach"
OLD_YEARS = (2005, 2015)
SUPERVISORS = 6
MIN_VOTES = 3
MIN_SHARE = 0.6
MARGIN = 2
PLACEHOLDERS = {"999999", "000000"}
SMALL_WORDS = {"a", "v", "vo", "pre", "na", "so", "s", "z", "zo", "k", "ku", "o", "a."}
@lru_cache(maxsize=1)
def learned() -> dict:
"""{university: {code: {"name", "short", "votes", "voters"}}} from workplaces.json."""
try:
return json.loads(FILE.read_text(encoding="utf-8"))
except (OSError, ValueError):
return {}
def name_of(university: str, code: str) -> dict | None:
return learned().get(university, {}).get(code)
def acronym(name: str) -> str:
""""Katedra počítačov a informatiky" -> "KPI"."""
return "".join(w[0].upper() for w in re.findall(r"[^\W\d_]+", name) if w.lower() not in SMALL_WORDS)
def by_acronym(university: str = TUKE) -> dict[str, tuple[str, dict]]:
"""Lower-case acronym -> (code, entry) for acronyms that name exactly one learned workplace."""
seen: dict[str, list] = {}
for code, entry in learned().get(university, {}).items():
seen.setdefault(entry["short"].lower(), []).append((code, entry))
return {k: v[0] for k, v in seen.items() if len(v) == 1 and len(k) >= 3}
CUT_WORDS = {"inteligenci": "inteligencie"} # CRZP cut old names at 40 characters
def _department(unit: str) -> str | None:
"""The written department of an old record: "1040 Katedra počítačov a informatiky"."""
text = re.sub(r"^\s*[\d\s]+", "", unit or "").strip(" ,")
text = re.sub(r"^(Katedra|Ústav)(?=[a-záäčďéíĺľňóôŕšťúýž])", r"\1 ", text) # "Katedraaplikovanej"
words = text.split()
if words and words[-1] in CUT_WORDS:
text = " ".join([*words[:-1], CUT_WORDS[words[-1]]])
return text if re.search(r"(?i)\b(katedr|ústav|ustav|centrum|pracovisk)\w*", text) else None
def codes_in_cache(university: str = TUKE) -> list[str]:
from tools.crzp.answering.answer import university as university_of
from tools.crzp.fetch import store
counts = Counter(code for row in store().cached_rows() if university_of(row) == university
for code in re.findall(r"\b\d{6}\b", row.get("school_unit") or ""))
return [code for code, _ in counts.most_common()]
def learn_code(code: str, university: str = TUKE, *, search=None) -> dict:
"""Votes for the department name of one code (see the module docstring)."""
from tools.crzp import fetch
from tools.crzp.answering.answer import university as university_of
search = search or fetch.search
result = search(code, links=False, depth=20)
supervisors = [display_name(f["label"]) for f in result["facets"].get("Školiteľ", [])][:SUPERVISORS]
votes, voters, spelled = Counter(), 0, {}
for name in supervisors:
old = search(name, years=OLD_YEARS, links=False, depth=200)
surname = fold(name.split()[-1])
own = Counter(_department(r.get("school_unit")) for r in old["rows"] if university_of(r) == university
and any(surname in fold(s) for s in r.get("supervisors", [])))
own.pop(None, None)
merged = Counter()
for department, n in own.items():
key = _canonical(department, spelled)
merged[key] += n
ranked = merged.most_common(2)
if ranked and (len(ranked) == 1 or ranked[0][1] > ranked[1][1]): # one clear department of this person
voters += 1
votes[ranked[0][0]] += 1
if not votes:
return {"code": code, "accepted": False, "voters": voters, "votes": {}}
final = Counter() # a key may have grown to a longer spelling after it got votes
for key, n in votes.items():
final[next((k for k in spelled if k.startswith(key)), key)] += n
votes = final
ranked = votes.most_common(2)
best, count = ranked[0]
second = ranked[1][1] if len(ranked) > 1 else 0
accepted = count >= MIN_VOTES and count / voters >= MIN_SHARE and count - second >= MARGIN
best = spelled[best]
return {"code": code, "accepted": accepted, "name": best, "short": acronym(best),
"votes": {spelled[k]: v for k, v in votes.items()}, "lead": count - second,
"voters": voters, "supervisors": supervisors}
def _canonical(department: str, spelled: dict) -> str:
"""Folded key shared by a name and its cut spellings; `spelled` keeps the longest form."""
folded = " ".join(fold(department).replace("katedraa", "katedra a").split())
for key in list(spelled):
if key.startswith(folded) or folded.startswith(key):
longer = folded if len(folded) > len(key) else key
text = department if len(folded) > len(key) else spelled[key]
if longer != key:
spelled[longer] = text
spelled.pop(key)
return longer
spelled[folded] = department
return folded
def learn(codes: list[str], university: str = TUKE) -> dict:
report = [learn_code(code, university) for code in codes if code not in PLACEHOLDERS]
winners: dict = {}
for outcome in report:
if outcome["accepted"] and outcome["code"] != "104005": # KEMT is verified by hand (units.py)
key = fold(outcome["name"])
if key not in winners or (outcome["lead"], outcome["voters"]) > (winners[key]["lead"], winners[key]["voters"]):
winners[key] = outcome
table = {o["code"]: {"name": o["name"], "short": o["short"], "votes": o["votes"][o["name"]],
"voters": o["voters"], "lead": o["lead"]} for o in winners.values()}
data = {**learned(), university: dict(sorted(table.items()))}
FILE.write_text(json.dumps(data, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
learned.cache_clear()
return {"learned": len(table), "report": report}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
sub = parser.add_subparsers(dest="command", required=True)
run = sub.add_parser("learn")
run.add_argument("codes", nargs="*")
run.add_argument("--limit", type=int, default=25)
args = parser.parse_args()
codes = args.codes or ["104005", *[c for c in codes_in_cache() if c != "104005"][: args.limit]]
result = learn(codes)
for line in result["report"]:
print(json.dumps({k: line.get(k) for k in ("code", "accepted", "name", "voters", "lead", "votes")}, ensure_ascii=False))
print(json.dumps({"learned": result["learned"]}))
return 0
if __name__ == "__main__":
raise SystemExit(main())