Vladimir Buzalka
867 lines
35 KiB
Python
867 lines
35 KiB
Python
"""
|
||
recepty_agent.py
|
||
----------------
|
||
Agent, který ve schránce ordinace@buzalkova.cz hledá ŽÁDOSTI O PŘEDPIS
|
||
(recept) od pacientů a vytěžuje z nich pacienta a požadované léky.
|
||
|
||
TESTOVACÍ REŽIM: čte N nejnovějších mailů z Inboxu (read-only, ve schránce
|
||
nic nemění) a vypíše report do konzole + logu.
|
||
|
||
Tok:
|
||
1. Microsoft Graph: načti N nejnovějších mailů z Inboxu (bez ohledu na přílohy).
|
||
2. AI KLASIFIKACE + VYTĚŽENÍ (Claude): u každého mailu rozhodne, zda jde
|
||
o žádost o předpis, a vytěží jméno pacienta (může se lišit od odesílatele),
|
||
rodné číslo (pokud je v textu) a seznam požadovaných léků s dávkováním.
|
||
3. OVĚŘENÍ V MEDICUSU: pacienta dohledá v kartotéce (KAR + KARKONTAKT)
|
||
v pořadí rodné číslo > e-mail odesílatele > jméno.
|
||
4. NEJEDNOZNAČNOST (více pacientů stejného jména): načte historii receptů
|
||
kandidátů (tabulka RECEPT) a rozhodne podle shody s požadovanými léky —
|
||
nejdřív deterministicky (název léku v historii), sporné případy dořeší
|
||
Claude nad seznamy předepsaných léků.
|
||
5. Vypíše report.
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import unicodedata
|
||
from datetime import date, datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
|
||
try:
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
except Exception:
|
||
pass
|
||
|
||
import requests
|
||
|
||
# graph_mail.py sdílíme s EmailAgent (stejná app registrace, Mail.Read).
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "EmailAgent"))
|
||
import graph_mail # noqa: E402
|
||
|
||
# medicus_db.py z Knihoven (Firebird, DSN podle názvu počítače).
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||
from Knihovny.medicus_db import get_medicus_db # noqa: E402
|
||
import mcp_medevio as _medevio # noqa: E402 GraphQL API + zaloz_pozadavek_recept
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
import recept_pending as _pending # noqa: E402 fronta dotazů (nejistá identifikace)
|
||
|
||
# =========================
|
||
# NASTAVENÍ
|
||
# =========================
|
||
MAILBOX = "ordinace@buzalkova.cz"
|
||
|
||
# DELTA REŽIM: zpracují se všechny maily s receivedDateTime > vodoznak
|
||
# (čas posledního zpracovaného mailu, uložen v _last_processed.txt). Při prvním
|
||
# běhu (vodoznak chybí) se vodoznak nastaví na nejnovější mail v Inboxu a od
|
||
# příště se zpracuje jen to, co přijde POTÉ (historie se nedohání).
|
||
MAX_PER_RUN = 200 # pojistka: max mailů na jeden běh (kdyby byl vodoznak hodně zpět)
|
||
|
||
# Kategorie (štítek na mailu), kterou agent označí mail po úspěšném založení
|
||
# požadavku v Medeviu. Při dalším běhu se takto označené maily přeskočí
|
||
# → idempotence, nezakládá duplicitní požadavky.
|
||
PROCESSED_CATEGORY = "ClaudeZpracovalRecept"
|
||
# Kategorie pro maily, které nešlo vyřídit automaticky (k ruční kontrole).
|
||
MANUAL_CATEGORY = "ReceptRucne"
|
||
|
||
# Práh jistoty pro PLNĚ automatické založení požadavku. Vytvoření požadavku je
|
||
# nevratné a hned viditelné pacientovi → pod tímto prahem agent NIC nezaloží
|
||
# a místo toho se zeptá člověka přes Telegram (přes pending frontu, viz
|
||
# recept_pending.py / recept_resolver.py).
|
||
SCORE_AUTO = 85
|
||
|
||
# Claude model pro klasifikaci + vytěžení.
|
||
ANTHROPIC_MODEL = "claude-haiku-4-5"
|
||
|
||
# Kolik měsíců historie receptů načíst při rozhodování nejednoznačnosti.
|
||
RECEPT_MONTHS = 24
|
||
|
||
# Cena Claude API — USD za 1M tokenů (input, output). Kurz pro přepočet.
|
||
USD_TO_CZK = 25.0
|
||
PRICING = {
|
||
"claude-haiku-4-5": (1.00, 5.00),
|
||
"claude-sonnet-4-6": (3.00, 15.00),
|
||
"claude-opus-4-8": (5.00, 25.00),
|
||
}
|
||
_cost = {"input_tokens": 0, "output_tokens": 0, "usd": 0.0, "calls": 0}
|
||
|
||
HERE = Path(__file__).resolve().parent
|
||
LOG_FILE = HERE / "_log_recepty.txt"
|
||
WATERMARK_FILE = HERE / "_last_processed.txt" # ISO čas (receivedDateTime) posledního zpracovaného mailu
|
||
|
||
|
||
# =========================
|
||
# ENV (Anthropic klíč)
|
||
# =========================
|
||
def _load_env():
|
||
env_path = Path(__file__).resolve().parent.parent / "Medevio" / ".env"
|
||
if env_path.exists():
|
||
for line in env_path.read_text(encoding="utf-8").splitlines():
|
||
line = line.strip()
|
||
if "=" in line and not line.startswith("#"):
|
||
k, v = line.split("=", 1)
|
||
os.environ[k.strip()] = v.strip().strip('"').strip("'")
|
||
|
||
|
||
_load_env()
|
||
|
||
|
||
def log(msg: str) -> None:
|
||
print(msg)
|
||
with LOG_FILE.open("a", encoding="utf-8") as f:
|
||
f.write(msg + "\n")
|
||
|
||
|
||
# =========================
|
||
# ČTENÍ MAILŮ (Graph, read-only)
|
||
# =========================
|
||
_SELECT = "id,subject,from,receivedDateTime,bodyPreview,body,categories"
|
||
|
||
|
||
def _load_watermark() -> tuple[str | None, str | None]:
|
||
"""Vrátí (receivedDateTime, id) posledního zpracovaného mailu (řádek 1 a 2
|
||
v _last_processed.txt). id slouží k odfiltrování hraničního mailu, který se
|
||
kvůli sub-sekundové přesnosti vrací i při filtru `gt` na oříznutý čas."""
|
||
try:
|
||
lines = WATERMARK_FILE.read_text(encoding="utf-8").splitlines()
|
||
t = (lines[0].strip() if lines else "") or None
|
||
i = (lines[1].strip() if len(lines) > 1 else "") or None
|
||
return t, i
|
||
except Exception:
|
||
return None, None
|
||
|
||
|
||
def _save_watermark(iso: str, msg_id: str = "") -> None:
|
||
WATERMARK_FILE.write_text(f"{iso}\n{msg_id}\n", encoding="utf-8")
|
||
|
||
|
||
def newest_received(mailbox: str) -> tuple[str, str]:
|
||
"""(receivedDateTime, id) nejnovějšího mailu v Inboxu — seed vodoznaku při
|
||
prvním běhu. Když je schránka prázdná, vrátí (aktuální UTC, '')."""
|
||
url = f"{graph_mail.GRAPH}/users/{mailbox}/mailFolders/inbox/messages"
|
||
params = {"$orderby": "receivedDateTime desc", "$select": "id,receivedDateTime", "$top": 1}
|
||
r = requests.get(url, headers=graph_mail._headers(), params=params, timeout=60)
|
||
r.raise_for_status()
|
||
vals = r.json().get("value", [])
|
||
if vals:
|
||
return vals[0]["receivedDateTime"], vals[0]["id"]
|
||
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), ""
|
||
|
||
|
||
def nove_inbox_messages(mailbox: str, since_iso: str) -> list[dict]:
|
||
"""Všechny maily z Inboxu s receivedDateTime > since_iso, od NEJSTARŠÍHO.
|
||
Stránkuje přes @odata.nextLink, max MAX_PER_RUN za jeden běh."""
|
||
url = f"{graph_mail.GRAPH}/users/{mailbox}/mailFolders/inbox/messages"
|
||
params = {
|
||
"$filter": f"receivedDateTime gt {since_iso}",
|
||
"$orderby": "receivedDateTime asc",
|
||
"$select": _SELECT,
|
||
"$top": 50,
|
||
}
|
||
headers = {**graph_mail._headers(), "Prefer": 'outlook.body-content-type="text"'}
|
||
out: list[dict] = []
|
||
while url and len(out) < MAX_PER_RUN:
|
||
r = requests.get(url, headers=headers, params=params, timeout=60)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
out.extend(data.get("value", []))
|
||
url = data.get("@odata.nextLink")
|
||
params = None # nextLink už nese všechny parametry
|
||
return out[:MAX_PER_RUN]
|
||
|
||
|
||
# =========================
|
||
# AI KLASIFIKACE + VYTĚŽENÍ (Claude)
|
||
# =========================
|
||
PROMPT = """Jsi asistent ordinace praktického lékaře (MUDr. Michaela Buzalková). \
|
||
Rozhoduješ, zda e-mail obsahuje ŽÁDOST O PŘEDPIS LÉKU (recept), a pokud ano, \
|
||
vytěžíš detaily.
|
||
|
||
Pravidla:
|
||
- "je_zadost_o_recept": true POUZE pokud pisatel žádá o předepsání léku / \
|
||
vystavení receptu (i opakovaného, i "prosím o léky jako obvykle").
|
||
- NENÍ žádost o recept: objednání na vyšetření, dotaz na výsledky, omluva, \
|
||
faktura, newsletter, zdravotní zpráva z nemocnice, žádanka, potvrzení.
|
||
- "pacient": celé jméno pacienta, pro kterého má být lék předepsán. POZOR: \
|
||
může se lišit od odesílatele (rodič píše za dítě, manžel za manželku). \
|
||
Pokud jméno z mailu neplyne, použij jméno odesílatele.
|
||
- "rodne_cislo": rodné číslo pacienta PŘESNĚ jak je v textu napsané (jen číslice, \
|
||
příp. s lomítkem — NEPŘEVÁDĚJ na datum narození), jinak null.
|
||
- "datum_narozeni": datum narození ve formátu YYYY-MM-DD, pokud je v textu \
|
||
uvedeno (a není to rodné číslo), jinak null.
|
||
- "leky": seznam požadovaných léků; u každého "nazev" a "poznamka" \
|
||
(síla/dávkování/množství/„jako obvykle", pokud je uvedeno, jinak null). \
|
||
Pokud pacient žádá o "své obvyklé léky" bez konkrét, vrať jeden záznam \
|
||
{"nazev": "obvyklé léky", "poznamka": "bez upřesnění"}.
|
||
- "telefon": telefonní číslo uvedené v mailu (jen číslice, jak je napsané), jinak null.
|
||
- "poznamka": cokoliv důležitého navíc (spěchá, vyzvedne osobně...), jinak null.
|
||
|
||
Vrať POUZE JSON:
|
||
{"je_zadost_o_recept": true/false, "pacient": "..."|null, "rodne_cislo": "..."|null,
|
||
"datum_narozeni": "YYYY-MM-DD"|null, "telefon": "..."|null,
|
||
"leky": [{"nazev": "...", "poznamka": "..."|null}], "poznamka": "..."|null,
|
||
"duvod": "krátké zdůvodnění rozhodnutí"}
|
||
|
||
E-MAIL:
|
||
Odesílatel: %(sender)s
|
||
Předmět: %(subject)s
|
||
Přijato: %(received)s
|
||
|
||
Tělo (zkráceno):
|
||
%(body)s
|
||
"""
|
||
|
||
|
||
def _claude_json(prompt: str, model: str, max_tokens: int) -> dict:
|
||
"""Zavolá Claude a vrátí JSON objekt z odpovědi."""
|
||
r = requests.post(
|
||
"https://api.anthropic.com/v1/messages",
|
||
headers={
|
||
"x-api-key": os.environ["ANTHROPIC_API_KEY"],
|
||
"anthropic-version": "2023-06-01",
|
||
"content-type": "application/json",
|
||
},
|
||
json={
|
||
"model": model,
|
||
"max_tokens": max_tokens,
|
||
"messages": [{"role": "user", "content": prompt}],
|
||
},
|
||
timeout=60,
|
||
)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
|
||
usage = data.get("usage", {})
|
||
in_tok = usage.get("input_tokens", 0)
|
||
out_tok = usage.get("output_tokens", 0)
|
||
price_in, price_out = PRICING.get(model, PRICING["claude-haiku-4-5"])
|
||
_cost["input_tokens"] += in_tok
|
||
_cost["output_tokens"] += out_tok
|
||
_cost["usd"] += in_tok / 1_000_000 * price_in + out_tok / 1_000_000 * price_out
|
||
_cost["calls"] += 1
|
||
|
||
text = data["content"][0]["text"].strip()
|
||
m = re.search(r"\{.*\}", text, re.DOTALL)
|
||
if not m:
|
||
raise ValueError(f"Claude nevrátil JSON: {text}")
|
||
return json.loads(m.group(0))
|
||
|
||
|
||
def classify(msg: dict) -> dict:
|
||
sender = (msg.get("from") or {}).get("emailAddress", {})
|
||
body = (msg.get("body") or {}).get("content") or msg.get("bodyPreview") or ""
|
||
prompt = PROMPT % {
|
||
"sender": f"{sender.get('name', '')} <{sender.get('address', '')}>",
|
||
"subject": msg.get("subject") or "",
|
||
"received": msg.get("receivedDateTime") or "",
|
||
"body": body[:6000],
|
||
}
|
||
return _claude_json(prompt, ANTHROPIC_MODEL, 500)
|
||
|
||
|
||
# Rozhodnutí nejednoznačného pacienta podle historie předepsaných léků.
|
||
AMBIG_PROMPT = """V kartotéce je více pacientů stejného jména. Podle požadovaných léků \
|
||
z e-mailu a historie předepsaných léků jednotlivých kandidátů rozhodni, který pacient \
|
||
o recept žádá.
|
||
|
||
Požadované léky z e-mailu: %(leky)s
|
||
|
||
Kandidáti a jejich léky předepsané v minulosti:
|
||
%(kandidati)s
|
||
|
||
Pravidla:
|
||
- Vyber kandidáta, jehož historie odpovídá požadovaným lékům (stejný lék, stejná \
|
||
účinná látka, ekvivalentní generikum/originál, lék na stejnou diagnózu).
|
||
- Pokud nelze spolehlivě rozhodnout (žádná smysluplná vazba), vrať idpac: null.
|
||
|
||
Vrať POUZE JSON:
|
||
{"idpac": 1234|null, "duvod": "krátké zdůvodnění"}
|
||
"""
|
||
|
||
|
||
# =========================
|
||
# OVĚŘENÍ PACIENTA V MEDICUSU
|
||
# =========================
|
||
def _norm_text(s: str) -> str:
|
||
"""Bez diakritiky, velkými písmeny, sjednocené mezery."""
|
||
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()
|
||
|
||
|
||
def _norm_rc(s: str) -> str:
|
||
"""Z rodného čísla nechá jen číslice (Medicus ukládá RČ bez lomítka)."""
|
||
return re.sub(r"\D", "", s or "")
|
||
|
||
|
||
def _norm_phone(s: str) -> str:
|
||
"""Z telefonu nechá jen číslice, bez předvolby 420."""
|
||
digits = re.sub(r"\D", "", s or "")
|
||
if digits.startswith("420"):
|
||
digits = digits[3:]
|
||
return digits
|
||
|
||
|
||
def _rc_to_birthdate(rc: str) -> str | None:
|
||
"""Z RČ odvodí datum narození YYYY-MM-DD (ženy mají měsíc +50)."""
|
||
rc = _norm_rc(rc)
|
||
if len(rc) not in (9, 10):
|
||
return None
|
||
yy, mm, dd = int(rc[0:2]), int(rc[2:4]), int(rc[4:6])
|
||
if mm > 50:
|
||
mm -= 50
|
||
year = yy + (2000 if len(rc) == 10 and yy < 54 else 1900)
|
||
try:
|
||
from datetime import date
|
||
return date(year, mm, dd).isoformat()
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _drug_matches(requested: str, prescribed: str) -> bool:
|
||
"""
|
||
Shoda názvu léku z mailu s názvem z receptu: substring oběma směry
|
||
("tadalafil" ~ "TADALAFIL ACCORD") + prefix prvních slov od 5 znaků
|
||
("Concord" ~ "CONCOR COR" — překlepy/varianty názvu).
|
||
"""
|
||
a, b = _norm_text(requested), _norm_text(prescribed)
|
||
if not a or not b:
|
||
return False
|
||
if a in b or b in a:
|
||
return True
|
||
ta, tb = a.split()[0], b.split()[0]
|
||
shorter, longer = sorted((ta, tb), key=len)
|
||
return len(shorter) >= 5 and longer.startswith(shorter)
|
||
|
||
|
||
class MedicusLookup:
|
||
"""Kartotéka v paměti: pacienti z KAR + e-maily z KARKONTAKT (TYP=3).
|
||
Drží otevřené spojení pro dotazy na historii receptů (RECEPT)."""
|
||
|
||
def __init__(self):
|
||
self.db = get_medicus_db()
|
||
self.patients = self.db.query_dict(
|
||
"SELECT k.IDPAC, k.RODCIS, k.PRIJMENI, k.JMENO, k.DATNAR, "
|
||
"k.POJ, k.VYRAZEN FROM KAR k "
|
||
"WHERE k.PRIJMENI IS NOT NULL AND k.PRIJMENI <> ''"
|
||
)
|
||
# TYP: 1 = pevná linka, 2 = mobil, 3 = e-mail.
|
||
contacts = self.db.query_dict(
|
||
"SELECT kk.IDPAC, kk.KONTAKT, kk.TYP FROM KARKONTAKT kk "
|
||
"WHERE kk.KONTAKT IS NOT NULL AND kk.KONTAKT <> ''"
|
||
)
|
||
|
||
self.by_rc = {}
|
||
self.by_name = {}
|
||
by_id = {}
|
||
for p in self.patients:
|
||
by_id[p["idpac"]] = p
|
||
rc = _norm_rc(p.get("rodcis") or "")
|
||
if rc:
|
||
self.by_rc[rc] = p
|
||
key = _norm_text(f"{p.get('jmeno') or ''} {p.get('prijmeni') or ''}")
|
||
if key:
|
||
self.by_name.setdefault(frozenset(key.split()), []).append(p)
|
||
|
||
self.by_email = {}
|
||
self.by_phone = {}
|
||
for c in contacts:
|
||
p = by_id.get(c["idpac"])
|
||
if not p:
|
||
continue
|
||
kontakt = (c["kontakt"] or "").strip()
|
||
if "@" in kontakt:
|
||
self.by_email.setdefault(kontakt.lower(), []).append(p)
|
||
else:
|
||
phone = _norm_phone(kontakt)
|
||
if len(phone) >= 9:
|
||
self.by_phone.setdefault(phone, []).append(p)
|
||
|
||
@staticmethod
|
||
def describe(p: dict) -> str:
|
||
rc = p.get("rodcis") or "?"
|
||
datnar = p.get("datnar")
|
||
vyrazen = " [VYŘAZEN]" if (p.get("vyrazen") or "") == "A" else ""
|
||
return (f"{p.get('prijmeni','')} {p.get('jmeno','')}, RČ {rc}, "
|
||
f"nar. {datnar}, poj. {p.get('poj','?')}, idpac {p.get('idpac')}{vyrazen}")
|
||
|
||
def match(self, verdict: dict, sender_email: str) -> tuple[str, list[dict]]:
|
||
"""
|
||
Vrátí (typ_shody, kandidáti). Pořadí spolehlivosti: RČ (jednoznačné) >
|
||
e-mail odesílatele > telefon z mailu > jméno (+ příp. datum narození).
|
||
"""
|
||
# 1) Rodné číslo z textu mailu — nejspolehlivější.
|
||
rc = _norm_rc(verdict.get("rodne_cislo") or "")
|
||
if rc and rc in self.by_rc:
|
||
return "RČ", [self.by_rc[rc]]
|
||
|
||
# 2) E-mail odesílatele v kartotéce.
|
||
hits = self.by_email.get((sender_email or "").strip().lower(), [])
|
||
if hits:
|
||
return "E-MAIL", hits
|
||
|
||
# 3) Telefon z textu mailu v kartotéce.
|
||
phone = _norm_phone(verdict.get("telefon") or "")
|
||
if len(phone) >= 9:
|
||
hits = self.by_phone.get(phone, [])
|
||
if hits:
|
||
return "TELEFON", hits
|
||
|
||
# 4) Jméno (bez diakritiky, bez ohledu na pořadí slov).
|
||
name_key = frozenset(_norm_text(verdict.get("pacient") or "").split())
|
||
candidates = self.by_name.get(name_key, []) if name_key else []
|
||
|
||
# Zúžení datem narození (z pole datum_narozeni nebo z RČ, které nesedlo).
|
||
birth = verdict.get("datum_narozeni") or (_rc_to_birthdate(rc) if rc else None)
|
||
if len(candidates) > 1 and birth:
|
||
narrowed = [p for p in candidates if str(p.get("datnar") or "")[:10] == birth]
|
||
if narrowed:
|
||
return "JMÉNO+DATUM", narrowed
|
||
|
||
if candidates:
|
||
return "JMÉNO", candidates
|
||
return "NENALEZEN", []
|
||
|
||
def close(self) -> None:
|
||
self.db.close()
|
||
|
||
def prescriptions(self, idpac: int, months: int = RECEPT_MONTHS) -> list[dict]:
|
||
"""Nestornované recepty pacienta za posledních N měsíců, nejnovější první."""
|
||
since = (date.today() - timedelta(days=months * 30)).isoformat()
|
||
return self.db.query_dict(
|
||
"SELECT r.DATUM, r.LEK, r.DSIG FROM RECEPT r "
|
||
"WHERE r.IDPAC = ? AND r.DATUM >= ? AND r.STORNO <> 'T' "
|
||
"ORDER BY r.DATUM DESC",
|
||
(idpac, since),
|
||
)
|
||
|
||
def resolve_by_prescriptions(
|
||
self, candidates: list[dict], leky: list[dict]
|
||
) -> tuple[dict | None, str, list[str]]:
|
||
"""
|
||
Rozhodne nejednoznačnost podle historie receptů kandidátů.
|
||
Vrátí (vítěz|None, popis_metody, řádky_detailu pro log).
|
||
"""
|
||
requested = [(lek.get("nazev") or "").strip() for lek in leky or []]
|
||
requested = [r for r in requested if r]
|
||
detail: list[str] = []
|
||
|
||
# Historie + skóre (kolik požadovaných léků má kandidát v historii).
|
||
infos = []
|
||
for p in candidates:
|
||
history = self.prescriptions(p["idpac"])
|
||
drugs = sorted({(h.get("lek") or "").strip() for h in history if h.get("lek")})
|
||
matched = sorted(
|
||
{req for req in requested if any(_drug_matches(req, d) for d in drugs)}
|
||
)
|
||
infos.append({"p": p, "drugs": drugs, "matched": matched})
|
||
detail.append(
|
||
f"idpac {p['idpac']}: {len(history)} receptů/{RECEPT_MONTHS} měs., "
|
||
f"shoda léků: {', '.join(matched) if matched else 'žádná'} "
|
||
f"(historie: {', '.join(drugs) if drugs else 'prázdná'})"
|
||
)
|
||
|
||
# Deterministicky: jediný kandidát s nejvyšším nenulovým skóre vyhrává.
|
||
best = max(len(i["matched"]) for i in infos)
|
||
winners = [i for i in infos if len(i["matched"]) == best]
|
||
if best > 0 and len(winners) == 1:
|
||
return winners[0]["p"], "LÉKY V HISTORII", detail
|
||
|
||
# Sporné (nikdo/více se shodou) → Claude nad seznamy léků.
|
||
if any(i["drugs"] for i in infos):
|
||
try:
|
||
prompt = AMBIG_PROMPT % {
|
||
"leky": ", ".join(requested) or "(neuvedeno)",
|
||
"kandidati": "\n".join(
|
||
f"- idpac {i['p']['idpac']} "
|
||
f"({self.describe(i['p'])}): "
|
||
f"{', '.join(i['drugs']) if i['drugs'] else 'žádné recepty'}"
|
||
for i in infos
|
||
),
|
||
}
|
||
v = _claude_json(prompt, ANTHROPIC_MODEL, 300)
|
||
idpac = v.get("idpac")
|
||
winner = next((i["p"] for i in infos if i["p"]["idpac"] == idpac), None)
|
||
if winner:
|
||
detail.append(f"AI rozhodnutí: {v.get('duvod', '')}")
|
||
return winner, "LÉKY+AI", detail
|
||
detail.append(f"AI nerozhodlo: {v.get('duvod', '')}")
|
||
except Exception as e:
|
||
detail.append(f"AI rozhodování selhalo: {type(e).__name__}: {e}")
|
||
|
||
return None, "", detail
|
||
|
||
|
||
# =========================
|
||
# SKÓRE JISTOTY IDENTIFIKACE PACIENTA
|
||
# =========================
|
||
def skore_jistoty(verdict: dict, patient: dict, sender_email: str,
|
||
lookup: "MedicusLookup") -> tuple[int, list[str]]:
|
||
"""Kvantifikuje jistotu (0–100), že `patient` z kartotéky je opravdu pacient
|
||
z e-mailu. Vrací (skóre, důvody). Více nezávislých shod = vyšší jistota;
|
||
rozpor (jiné jméno / datum / RČ) skóre tvrdě srazí a označí ⚠.
|
||
Tím se ošetří díra, kdy shoda na RČ (např. překlep) trefí jiného pacienta —
|
||
bez souhlasu jména spadne z 'jisté' do pásma 'nutná kontrola'."""
|
||
body = 0
|
||
duvody: list[str] = []
|
||
|
||
e_rc = _norm_rc(verdict.get("rodne_cislo") or "")
|
||
p_rc = _norm_rc(patient.get("rodcis") or "")
|
||
e_name = frozenset(_norm_text(verdict.get("pacient") or "").split())
|
||
p_name = frozenset(
|
||
_norm_text(f"{patient.get('jmeno') or ''} {patient.get('prijmeni') or ''}").split()
|
||
)
|
||
p_surname = _norm_text(patient.get("prijmeni") or "")
|
||
e_dob = verdict.get("datum_narozeni") or (_rc_to_birthdate(e_rc) if e_rc else None)
|
||
p_dob = (str(patient.get("datnar") or "")[:10]) or None
|
||
idpac = patient.get("idpac")
|
||
|
||
# Rodné číslo
|
||
if e_rc and p_rc:
|
||
if e_rc == p_rc:
|
||
body += 55; duvody.append("RČ sedí (+55)")
|
||
else:
|
||
body -= 35; duvody.append("⚠ RČ z mailu NESEDÍ na pacienta (−35)")
|
||
|
||
# Jméno
|
||
if e_name and p_name:
|
||
if e_name == p_name:
|
||
body += 30; duvody.append("jméno přesně (+30)")
|
||
elif p_surname and p_surname in e_name:
|
||
body += 15; duvody.append("příjmení sedí (+15)")
|
||
elif e_name & p_name:
|
||
body += 8; duvody.append("částečná shoda jména (+8)")
|
||
else:
|
||
body -= 45; duvody.append("⚠ jméno NESOUHLASÍ (−45)")
|
||
|
||
# Datum narození (z pole nebo odvozené z RČ)
|
||
if e_dob and p_dob:
|
||
if e_dob == p_dob:
|
||
body += 20; duvody.append("datum narození sedí (+20)")
|
||
else:
|
||
body -= 35; duvody.append("⚠ datum narození NESEDÍ (−35)")
|
||
|
||
# E-mail odesílatele v kartotéce pacienta
|
||
em = (sender_email or "").strip().lower()
|
||
if em and any(p.get("idpac") == idpac for p in lookup.by_email.get(em, [])):
|
||
body += 30; duvody.append("e-mail odesílatele v kartotéce (+30)")
|
||
|
||
# Telefon z textu mailu v kartotéce pacienta
|
||
ph = _norm_phone(verdict.get("telefon") or "")
|
||
if len(ph) >= 9 and any(p.get("idpac") == idpac for p in lookup.by_phone.get(ph, [])):
|
||
body += 20; duvody.append("telefon v kartotéce (+20)")
|
||
|
||
# Požadovaný lék v historii receptů pacienta
|
||
try:
|
||
requested = [(l.get("nazev") or "").strip() for l in (verdict.get("leky") or [])]
|
||
requested = [r for r in requested if r]
|
||
if requested and idpac is not None:
|
||
drugs = {(h.get("lek") or "").strip()
|
||
for h in lookup.prescriptions(idpac) if h.get("lek")}
|
||
if any(_drug_matches(req, d) for req in requested for d in drugs):
|
||
body += 10; duvody.append("lék v historii receptů (+10)")
|
||
except Exception:
|
||
pass
|
||
|
||
return max(0, min(100, body)), duvody
|
||
|
||
|
||
# =========================
|
||
# MEDEVIO — ZÁPIS POŽADAVKU
|
||
# =========================
|
||
|
||
# Markery oddělující forward/citaci v těle mailu (Outlook CZ/EN, Gmail > styl).
|
||
_FORWARD_MARKERS_RE = re.compile(
|
||
r"^-{3,}\s*(original message|forwarded message|původní zpráva|pův\.?\s*zpráva"
|
||
r"|weitergeleitete nachricht)",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
|
||
def _compress_body(body: str) -> str:
|
||
"""Odstraní forwardovanou/citovanou část a smrskne dvojité prázdné řádky."""
|
||
lines = (body or "").splitlines()
|
||
cut_at = None
|
||
for i, line in enumerate(lines):
|
||
s = line.strip()
|
||
sl = s.lower()
|
||
# Oddělovač Outlooku (--- Original Message --- apod.)
|
||
if _FORWARD_MARKERS_RE.match(s):
|
||
cut_at = i
|
||
break
|
||
# Citované řádky > (Gmail/Thunderbird)
|
||
if s.startswith(">"):
|
||
cut_at = i
|
||
break
|
||
# Outlook CZ: "Od: Jméno <email>" + "Odesláno:" do 5 řádků
|
||
if re.match(r"^od:\s*.+@", sl):
|
||
lookahead = " ".join(lines[i + 1 : i + 6]).lower()
|
||
if "odesláno:" in lookahead or "odeslano:" in lookahead:
|
||
cut_at = i
|
||
break
|
||
# Outlook EN: "From: Name <email>" + "Sent:" do 5 řádků
|
||
if re.match(r"^from:\s*.+@", sl):
|
||
lookahead = " ".join(lines[i + 1 : i + 6]).lower()
|
||
if "sent:" in lookahead:
|
||
cut_at = i
|
||
break
|
||
|
||
if cut_at is not None:
|
||
lines = lines[:cut_at]
|
||
|
||
# Odstraň trailing prázdné řádky
|
||
while lines and not lines[-1].strip():
|
||
lines.pop()
|
||
|
||
text = "\n".join(lines)
|
||
# Dva a více prázdných řádků → jeden prázdný řádek
|
||
text = re.sub(r"\n{3,}", "\n\n", text)
|
||
return text.strip()
|
||
|
||
|
||
def _kand_info(p: dict) -> dict:
|
||
"""Z Medicus pacienta udělá lehký dict kandidáta pro Telegram dotaz."""
|
||
return {
|
||
"idpac": p.get("idpac"),
|
||
"rc": _norm_rc(p.get("rodcis") or ""),
|
||
"jmeno": p.get("jmeno") or "",
|
||
"prijmeni": p.get("prijmeni") or "",
|
||
"datnar": str(p.get("datnar") or "")[:10],
|
||
"poj": p.get("poj") or "",
|
||
}
|
||
|
||
|
||
def _format_leky(leky: list) -> str:
|
||
"""Formátuje seznam léků pro pole 'Název léků' — čárkami oddělený výčet."""
|
||
parts = []
|
||
for lek in leky or []:
|
||
nazev = (lek.get("nazev") or "").strip()
|
||
if not nazev:
|
||
continue
|
||
pozn = (lek.get("poznamka") or "").strip()
|
||
parts.append(f"{nazev} ({pozn})" if pozn else nazev)
|
||
return ", ".join(parts)
|
||
|
||
|
||
def _format_poznamka(msg: dict) -> str:
|
||
"""Sestaví userNote pro Medevio: hlavička + zkomprimované tělo mailu."""
|
||
sender = (msg.get("from") or {}).get("emailAddress", {})
|
||
name = sender.get("name", "").strip()
|
||
email_addr = sender.get("address", "").strip()
|
||
received_raw = msg.get("receivedDateTime") or ""
|
||
|
||
try:
|
||
from dateutil import parser as _dtparser, tz as _tz
|
||
dt = _dtparser.isoparse(received_raw).astimezone(_tz.gettz("Europe/Prague"))
|
||
header_date = f"{dt.day}.{dt.month}.{dt.year} {dt.strftime('%H:%M')}"
|
||
except Exception:
|
||
header_date = received_raw
|
||
|
||
header = f"{header_date} | {name} <{email_addr}>"
|
||
body = (msg.get("body") or {}).get("content") or msg.get("bodyPreview") or ""
|
||
compressed = _compress_body(body)
|
||
return f"{header}\n\n{compressed}"
|
||
|
||
|
||
def _medevio_find_patient(rc_normalized: str) -> str | None:
|
||
"""Najde UUID pacienta v Medeviu podle normalizovaného RČ (jen číslice).
|
||
Používá MySQL zrcadlo medevio_pacient — patient_id je identické s GraphQL API."""
|
||
try:
|
||
from Knihovny.mysql_db import connect_mysql
|
||
conn = connect_mysql()
|
||
conn.ping(reconnect=True)
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"SELECT patient_id FROM medevio_pacient "
|
||
"WHERE REPLACE(identification_number,'/','') = %s LIMIT 1",
|
||
[rc_normalized],
|
||
)
|
||
row = cur.fetchone()
|
||
return row[0] if row else None
|
||
except Exception as e:
|
||
log(f" [Medevio hledání pacienta selhalo] {type(e).__name__}: {e}")
|
||
return None
|
||
|
||
|
||
# =========================
|
||
# HLAVNÍ BĚH
|
||
# =========================
|
||
def main() -> None:
|
||
log("\n" + "=" * 70)
|
||
log(f"START — schránka={MAILBOX}, DELTA režim (vše po posledním zpracovaném)")
|
||
log(f"REŽIM: zakládá požadavky v Medeviu; zpracované maily značí štítkem "
|
||
f"'{PROCESSED_CATEGORY}' (a příště přeskakuje)")
|
||
|
||
# Zajisti kategorii v master-listu schránky (s barvou). Best-effort.
|
||
try:
|
||
graph_mail.ensure_category(MAILBOX, PROCESSED_CATEGORY)
|
||
except Exception as e:
|
||
log(f"[POZOR] kategorii '{PROCESSED_CATEGORY}' nelze zajistit "
|
||
f"({type(e).__name__}: {e}) — chybí asi Mail.ReadWrite oprávnění")
|
||
|
||
watermark, last_id = _load_watermark()
|
||
if watermark is None:
|
||
# první běh — neznáme „poslední zpracovaný"; nastav vodoznak na nejnovější
|
||
# mail a od příště zpracovávej jen to, co přijde potom (historie se nedohání).
|
||
watermark, seed_id = newest_received(MAILBOX)
|
||
_save_watermark(watermark, seed_id)
|
||
log(f"První běh — vodoznak nastaven na {watermark}. "
|
||
f"Příští běh zpracuje maily přijaté po tomto čase.")
|
||
return
|
||
|
||
msgs = nove_inbox_messages(MAILBOX, watermark)
|
||
# `gt` na oříznutý (sekundový) čas vrací i hraniční už zpracovaný mail
|
||
# (Graph má sub-sekundovou přesnost) → odfiltruj ho podle ID.
|
||
if last_id:
|
||
msgs = [m for m in msgs if m.get("id") != last_id]
|
||
cap = " (dosažen strop MAX_PER_RUN)" if len(msgs) >= MAX_PER_RUN else ""
|
||
log(f"Vodoznak: {watermark} → nových mailů: {len(msgs)}{cap}")
|
||
if not msgs:
|
||
log("Nic nového — končím.")
|
||
return
|
||
|
||
lookup = MedicusLookup()
|
||
log(f"Medicus: kartotéka načtena ({len(lookup.patients)} pacientů, "
|
||
f"{len(lookup.by_email)} e-mailů, {len(lookup.by_phone)} telefonů).\n")
|
||
|
||
requests_found = 0
|
||
for i, msg in enumerate(msgs, 1):
|
||
sender = (msg.get("from") or {}).get("emailAddress", {})
|
||
subj = msg.get("subject") or "(bez předmětu)"
|
||
log(f"--- [{i}/{len(msgs)}] {msg.get('receivedDateTime', '')} ---")
|
||
log(f" Od: {sender.get('name', '')} <{sender.get('address', '')}>")
|
||
log(f" Předmět: {subj}")
|
||
|
||
# Idempotence: mail už agent jednou zpracoval → přeskoč (žádný duplicitní požadavek).
|
||
if PROCESSED_CATEGORY in (msg.get("categories") or []):
|
||
log(f" => PŘESKOČENO — již zpracováno (štítek {PROCESSED_CATEGORY})\n")
|
||
continue
|
||
# Už čeká na potvrzení člověka přes Telegram → znovu se neptej.
|
||
if _pending.je_mail_pending(msg["id"]):
|
||
log(" => PŘESKOČENO — čeká na odpověď přes Telegram\n")
|
||
continue
|
||
|
||
try:
|
||
v = classify(msg)
|
||
except Exception as e:
|
||
log(f" [AI CHYBA] {type(e).__name__}: {e}\n")
|
||
continue
|
||
|
||
if not v.get("je_zadost_o_recept"):
|
||
log(f" => NENÍ žádost o recept — {v.get('duvod', '')}\n")
|
||
continue
|
||
|
||
requests_found += 1
|
||
log(f" => ŽÁDOST O RECEPT — {v.get('duvod', '')}")
|
||
log(f" Pacient: {v.get('pacient') or '(neuvedeno)'}")
|
||
if v.get("rodne_cislo"):
|
||
log(f" Rodné číslo: {v['rodne_cislo']}")
|
||
if v.get("datum_narozeni"):
|
||
log(f" Narozen: {v['datum_narozeni']}")
|
||
if v.get("telefon"):
|
||
log(f" Telefon: {v['telefon']}")
|
||
for lek in v.get("leky") or []:
|
||
pozn = f" — {lek['poznamka']}" if lek.get("poznamka") else ""
|
||
log(f" Lék: {lek.get('nazev', '?')}{pozn}")
|
||
if v.get("poznamka"):
|
||
log(f" Poznámka: {v['poznamka']}")
|
||
|
||
# Ověření v Medicusu.
|
||
identified_patient = None
|
||
match_type, candidates = lookup.match(v, sender.get("address", ""))
|
||
if match_type == "NENALEZEN":
|
||
log(" Medicus: [NENALEZEN] pacient v kartotéce nedohledán")
|
||
elif len(candidates) == 1:
|
||
log(f" Medicus: [SHODA {match_type}] {lookup.describe(candidates[0])}")
|
||
identified_patient = candidates[0]
|
||
else:
|
||
log(f" Medicus: [NEJEDNOZNAČNÉ — {match_type}, "
|
||
f"{len(candidates)} kandidátů] — rozhoduji podle historie receptů:")
|
||
winner, method, detail = lookup.resolve_by_prescriptions(
|
||
candidates, v.get("leky") or []
|
||
)
|
||
for line in detail:
|
||
log(f" - {line}")
|
||
if winner:
|
||
log(f" Medicus: [SHODA {match_type}+{method}] "
|
||
f"{lookup.describe(winner)}")
|
||
identified_patient = winner
|
||
else:
|
||
log(" Medicus: [NEROZHODNUTO] historie receptů "
|
||
"nejednoznačnost nevyřešila — nutná ruční kontrola")
|
||
for p in candidates:
|
||
log(f" - {lookup.describe(p)}")
|
||
|
||
# Skóre jistoty identifikace → rozhodnutí: založit / zeptat se člověka.
|
||
if identified_patient:
|
||
skore, duvody = skore_jistoty(
|
||
v, identified_patient, sender.get("address", ""), lookup
|
||
)
|
||
else:
|
||
skore, duvody = 0, ["pacient nedohledán v kartotéce"]
|
||
log(f" Jistota: {skore}/100 — {'; '.join(duvody) or 'bez signálů'}")
|
||
|
||
leky_str = _format_leky(v.get("leky") or [])
|
||
pozn_str = _format_poznamka(msg)
|
||
|
||
if identified_patient and skore >= SCORE_AUTO:
|
||
# Vysoká jistota → založ rovnou.
|
||
rc = _norm_rc(identified_patient.get("rodcis") or "")
|
||
patient_uuid = _medevio_find_patient(rc)
|
||
if not patient_uuid:
|
||
log(f" Medevio: [NENÍ V MEDEVIU] RČ {rc} — k ruční kontrole")
|
||
try:
|
||
graph_mail.add_category(MAILBOX, msg["id"], MANUAL_CATEGORY)
|
||
log(f" Mail: [OZNAČEN] {MANUAL_CATEGORY}")
|
||
except Exception as e:
|
||
log(f" Mail: [POZOR] štítek nenastaven "
|
||
f"({type(e).__name__}: {e})")
|
||
else:
|
||
try:
|
||
result = _medevio.zaloz_pozadavek_recept(
|
||
patient_uuid, leky_str, pozn_str
|
||
)
|
||
log(f" Medevio: [ZALOZENO] požadavek "
|
||
f"{result['request_id']} [{skore}/100] | léky: {leky_str}")
|
||
# Označ mail jako zpracovaný → příště se přeskočí (idempotence).
|
||
try:
|
||
graph_mail.add_category(MAILBOX, msg["id"], PROCESSED_CATEGORY)
|
||
log(f" Mail: [OZNAČEN] štítek {PROCESSED_CATEGORY}")
|
||
except Exception as e:
|
||
log(f" Mail: [POZOR] štítek nenastaven "
|
||
f"({type(e).__name__}: {e}) — riziko duplicity při dalším běhu!")
|
||
except Exception as e:
|
||
log(f" Medevio: [CHYBA] {type(e).__name__}: {e}")
|
||
else:
|
||
# Nejistá identifikace → NEZAKLÁDAT, zeptat se člověka přes Telegram.
|
||
kandidati = [_kand_info(p) for p in candidates]
|
||
_pending.pridej(
|
||
email_message_id=msg["id"],
|
||
email_subject=subj,
|
||
sender=f"{sender.get('name', '')} <{sender.get('address', '')}>",
|
||
leky_str=leky_str,
|
||
pozn_str=pozn_str,
|
||
skore=skore,
|
||
duvody=duvody,
|
||
kandidati=kandidati,
|
||
)
|
||
log(f" Rozhodnutí: [DOTAZ] jistota {skore} < {SCORE_AUTO} "
|
||
f"— čeká na potvrzení přes Telegram ({len(kandidati)} kandidátů)")
|
||
|
||
log("")
|
||
|
||
lookup.close()
|
||
_save_watermark(msgs[-1]["receivedDateTime"], msgs[-1].get("id", "")) # posun na nejnovější zpracovaný
|
||
log(f"HOTOVO: {len(msgs)} mailů, žádostí o recept: {requests_found}. "
|
||
f"Nový vodoznak: {msgs[-1]['receivedDateTime']}")
|
||
log(
|
||
f"CENA AI: {_cost['calls']} volání, "
|
||
f"tokeny input={_cost['input_tokens']} output={_cost['output_tokens']}, "
|
||
f"${_cost['usd']:.4f} ≈ {_cost['usd'] * USD_TO_CZK:.2f} Kč"
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|