This commit is contained in:
2026-06-12 15:32:22 +02:00
parent 51ee67c7f3
commit bed5576efa
9 changed files with 1505 additions and 21 deletions
+140 -13
View File
@@ -207,30 +207,112 @@ def get_patient(idpac: int) -> dict:
raise
def _strip_diacritics(s: str) -> str:
"""Bez diakritiky, velkými písmeny, sjednocené mezery."""
import re
import unicodedata
s = unicodedata.normalize('NFKD', s or '')
s = ''.join(c for c in s if not unicodedata.combining(c))
return re.sub(r'\s+', ' ', s).strip().upper()
@mcp.tool()
def search_patients(query: str) -> list:
"""Vyhledá pacienty podle příjmení, jména nebo rodného čísla (částečná shoda).
Vrátí max. 50 výsledků: idpac, jmeno, prijmeni, rc, pojistovna.
def search_patients(query: str, datum_narozeni: Optional[str] = None) -> list:
"""Vyhledá pacienty podle jména/příjmení (bez ohledu na diakritiku a pořadí slov,
např. "Mateju Petr" najde "Petr Matějů") nebo rodného čísla (částečná shoda, jen číslice).
datum_narozeni: volitelný filtr YYYY-MM-DD.
Vrátí max. 50 výsledků: idpac, jmeno, prijmeni, rc, datnar, pojistovna, vyrazen.
"""
try:
import datetime
import re
cur = conn.cursor()
q = query.strip().upper()
cur.execute("""
SELECT FIRST 50 IDPAC, JMENO, PRIJMENI, RODCIS, POJ
SELECT IDPAC, JMENO, PRIJMENI, RODCIS, DATNAR, POJ, VYRAZEN
FROM KAR
WHERE UPPER(PRIJMENI) LIKE ? OR UPPER(JMENO) LIKE ? OR RODCIS LIKE ?
ORDER BY PRIJMENI, JMENO
""", [f'%{q}%', f'%{q}%', f'%{q}%'])
rows = cur.fetchall()
return [
{'idpac': r[0], 'jmeno': r[1], 'prijmeni': r[2], 'rc': r[3], 'pojistovna': r[4]}
for r in rows
]
WHERE PRIJMENI IS NOT NULL AND PRIJMENI <> ''
""")
q_digits = re.sub(r'\D', '', query)
q_tokens = _strip_diacritics(query).split()
results = []
for r in cur.fetchall():
idpac, jmeno, prijmeni, rc, datnar, poj, vyrazen = r
datnar_iso = datnar.isoformat() if isinstance(datnar, datetime.date) else datnar
if datum_narozeni and str(datnar_iso or '')[:10] != datum_narozeni:
continue
if q_digits and len(q_digits) >= 4:
if q_digits not in (rc or ''):
continue
elif q_tokens:
name_norm = _strip_diacritics(f"{jmeno or ''} {prijmeni or ''}")
if not all(t in name_norm for t in q_tokens):
continue
elif not datum_narozeni:
continue # prázdný dotaz bez filtru data — nevracet celou kartotéku
results.append({
'idpac': idpac, 'jmeno': jmeno, 'prijmeni': prijmeni, 'rc': rc,
'datnar': datnar_iso, 'pojistovna': poj, 'vyrazen': vyrazen == 'A',
})
results.sort(key=lambda p: (p['prijmeni'] or '', p['jmeno'] or ''))
return results[:50]
except Exception:
log(f"search_patients chyba: {traceback.format_exc()}")
raise
@mcp.tool()
def search_patient_by_contact(kontakt: str) -> list:
"""Vyhledá pacienty podle kontaktu (e-mail nebo telefon) v tabulce KARKONTAKT.
Částečná shoda bez ohledu na velikost písmen; u telefonů se porovnávají jen
číslice (ignoruje mezery a +420). Vrátí max. 50 výsledků: pacient (idpac,
jmeno, prijmeni, rc, datnar, pojistovna, vyrazen) + kontakt (typ, popis, vztah).
"""
try:
import datetime
import re
cur = conn.cursor()
cur.execute("""
SELECT kk.IDPAC, kk.KONTAKT, kk.TYP, kk.POPIS, kk.VZTAH,
k.JMENO, k.PRIJMENI, k.RODCIS, k.DATNAR, k.POJ, k.VYRAZEN
FROM KARKONTAKT kk
JOIN KAR k ON k.IDPAC = kk.IDPAC
WHERE kk.KONTAKT IS NOT NULL AND kk.KONTAKT <> ''
""")
q = kontakt.strip().lower()
q_digits = re.sub(r'\D', '', kontakt)
if q_digits.startswith('420'):
q_digits = q_digits[3:]
results = []
for r in cur.fetchall():
(idpac, kont, typ, popis, vztah,
jmeno, prijmeni, rc, datnar, poj, vyrazen) = r
kont = (kont or '').strip()
hit = q and q in kont.lower()
if not hit and len(q_digits) >= 6:
kont_digits = re.sub(r'\D', '', kont)
if kont_digits.startswith('420'):
kont_digits = kont_digits[3:]
hit = q_digits in kont_digits
if not hit:
continue
results.append({
'idpac': idpac, 'jmeno': jmeno, 'prijmeni': prijmeni, 'rc': rc,
'datnar': datnar.isoformat() if isinstance(datnar, datetime.date) else datnar,
'pojistovna': poj, 'vyrazen': vyrazen == 'A',
'kontakt': kont, 'typ': typ, 'popis': popis or '', 'vztah': vztah or '',
})
return results[:50]
except Exception:
log(f"search_patient_by_contact chyba: {traceback.format_exc()}")
raise
@mcp.tool()
def get_patient_timeline(idpac: int, datum_od: Optional[str] = None, datum_do: Optional[str] = None) -> dict:
"""Chronologický přehled všech záznamů pacienta z DOCLIST.
@@ -377,6 +459,51 @@ def get_table_info(table_name: str) -> dict:
raise
@mcp.tool()
def get_columns_overview(table_name: str, sample_rows: int = 1000) -> dict:
"""Přehled obsahu sloupců tabulky pro pochopení sémantiky (např. co znamená
KARKONTAKT.TYP=3 nebo RECEPT.STORNO='T'). Ze vzorku posledních N řádků
(výchozí 1000) vrátí pro každý sloupec: počet vyplněných, počet distinct
hodnot a top 5 nejčastějších hodnot s četností. Hodnoty zkráceny na 80 znaků.
"""
try:
from collections import Counter
cur = conn.cursor()
cur.execute(f'SELECT FIRST {int(sample_rows)} * FROM {table_name.upper()}')
rows = rows_to_json(cur.fetchall(), cur.description or [])
if not rows:
return {'tabulka': table_name.upper(), 'sample': 0, 'sloupce': {}}
overview = {}
for col in rows[0].keys():
values = [r[col] for r in rows if r[col] is not None and r[col] != '']
counter = Counter(
str(v)[:80] for v in values
)
overview[col] = {
'vyplneno': len(values),
'distinct': len(counter),
'top': [
{'hodnota': v, 'pocet': n} for v, n in counter.most_common(5)
],
}
result = {
'tabulka': table_name.upper(),
'sample': len(rows),
'sloupce': overview,
}
if table_name.upper() in LARGE_TABLES:
result['warning'] = (
f'Tabulka {table_name.upper()} je velká — přehled je jen '
f'ze vzorku prvních {len(rows)} řádků.'
)
return result
except Exception:
log(f"get_columns_overview chyba: {traceback.format_exc()}")
raise
@mcp.tool()
def safe_query(sql: str, params: Optional[list] = None) -> dict:
"""Bezpečný SELECT s ochranou před timeoutem na velkých tabulkách.