administrator/ordinaceprojekt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

notebookvb

Browse Source
This commit is contained in:
Vladimir Buzalka
2026-06-14 12:07:35 +02:00
parent 9133fe9497
commit 2bdac59676
16 changed files with 1484 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Knihovny/medicus_db.py
+8 -3
View File
@@ -2,14 +2,18 @@
# Připojení k Firebird databázi Medicus (medicus.fdb). Volí DSN podle názvu počítače.
# Obsahuje třídu MedicusDB s metodami pro dotazy na pacienty, registrace a faktury.
import os
import socket
import fdb
def get_medicus_connection():
"""
Připojí se k Firebird medicus.fdb podle názvu počítače.
Vrátí fdb.Connection nebo vyhodí RuntimeError pro neznámý počítač.
Připojí se k Firebird medicus.fdb. DSN se vybere takto:
1) env MEDICUS_FDB_DSN (má přednost — nutné v dockeru, kde hostname = ID kontejneru),
2) podle názvu počítače (dsn_map),
3) default.
Vrátí fdb.Connection nebo vyhodí RuntimeError.
"""
computer_name = socket.gethostname().upper()
dsn_map = {
@@ -20,7 +24,8 @@ def get_medicus_connection():
"Z230": r"192.168.1.76:/firebird/data/medicus.fdb", # přepnuto z reporteru na tower 2026-06-14
"TOWER": r"192.168.1.76:/firebird/data/medicus.fdb", # Firebird 2.5 docker kontejner na toweru
}
dsn = dsn_map.get(computer_name, r"localhost:c:\medicus 3\data\medicus.fdb")
dsn = (os.environ.get("MEDICUS_FDB_DSN")
or dsn_map.get(computer_name, r"localhost:c:\medicus 3\data\medicus.fdb"))
import sys
print(f"[medicus_db] Pripojuji se jako {computer_name} -> {dsn}", file=sys.stderr, flush=True)
return fdb.connect(dsn=dsn, user="SYSDBA", password="masterkey", charset="win1250")
Knihovny/telegram_notify.py
+184
View File
@@ -0,0 +1,184 @@
"""
telegram_notify.py
------------------
Notifikace a obousměrná komunikace přes Telegram Bot API
(bot ClaudeBot @Vlado_Claude_Bot).
Token a výchozí chat_id se načítají z `Medevio/.env`:
TELEGRAM_BOT_TOKEN=123456789:AAE...
TELEGRAM_CHAT_ID=6639316354
Použití ze skriptu:
from Knihovny.telegram_notify import posli_telegram, zeptej_se_telegram
posli_telegram("Pipeline 08 hotová, 142 záznamů")
odpoved = zeptej_se_telegram("Mám reimportovat i archiv? (ano/ne)")
if odpoved and odpoved.strip().lower() == "ano":
...
Použití z příkazové řádky:
python -m Knihovny.telegram_notify "Hotovo"
python -m Knihovny.telegram_notify --ask "Pokracovat? (ano/ne)"
POZN.: getUpdates smí v jednu chvíli pollovat jen JEDEN proces. Pokud běží
víc skriptů naráz, které čekají na odpověď, kradou si navzájem zprávy —
v praxi se ptá vždy jen jeden agent.
"""
import os
import sys
import time
from pathlib import Path
import requests
# =========================
# Načtení .env (Medevio/.env)
# =========================
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.setdefault(k.strip(), v.strip())
_load_env()
API_BASE = "https://api.telegram.org/bot{token}/{method}"
def _token() -> str:
token = os.environ.get("TELEGRAM_BOT_TOKEN")
if not token:
raise RuntimeError("Chybí TELEGRAM_BOT_TOKEN v Medevio/.env")
return token
def _resolve_chat_id(chat_id: str | None) -> str:
chat_id = chat_id or os.environ.get("TELEGRAM_CHAT_ID")
if not chat_id:
raise RuntimeError("Chybí TELEGRAM_CHAT_ID (zadej argumentem nebo v Medevio/.env)")
return str(chat_id)
def _call(method: str, *, http_timeout: int = 15, **params):
"""Zavolá Telegram Bot API metodu a vrátí pole `result`."""
url = API_BASE.format(token=_token(), method=method)
r = requests.post(url, json=params, timeout=http_timeout)
data = r.json()
if not data.get("ok"):
raise RuntimeError(f"Telegram {method} selhal [{r.status_code}]: {data}")
return data["result"]
def posli_telegram(
text: str,
*,
chat_id: str | None = None,
parse_mode: str | None = None,
disable_notification: bool = False,
) -> dict:
"""
Pošle zprávu přes Telegram bota.
:param text: text zprávy (max 4096 znaků)
:param chat_id: cílový chat; výchozí z TELEGRAM_CHAT_ID
:param parse_mode: None | "Markdown" | "MarkdownV2" | "HTML"
:param disable_notification: True = tichá zpráva (bez upozornění)
:return: odeslaná zpráva (dict z Telegram API)
"""
params = {
"chat_id": _resolve_chat_id(chat_id),
"text": text,
"disable_notification": disable_notification,
}
if parse_mode:
params["parse_mode"] = parse_mode
return _call("sendMessage", **params)
def zeptej_se_telegram(
otazka: str,
*,
chat_id: str | None = None,
timeout: int = 300,
poll_timeout: int = 30,
parse_mode: str | None = None,
) -> str | None:
"""
Pošle otázku a BLOKUJÍCÍ čeká na textovou odpověď uživatele.
Zahodí starší zprávy a bere jen tu, která přijde PO odeslání otázky.
:param otazka: text otázky
:param chat_id: cílový chat; výchozí z TELEGRAM_CHAT_ID
:param timeout: celkové čekání na odpověď v sekundách (pak vrátí None)
:param poll_timeout: délka jednoho long-poll cyklu v sekundách
:param parse_mode: formátování otázky (None | "HTML" | "Markdown")
:return: text odpovědi, nebo None když nikdo neodpoví do timeoutu
"""
cid = _resolve_chat_id(chat_id)
# Zjisti poslední update_id, ať bereme jen NOVÉ zprávy po otázce.
existujici = _call("getUpdates", http_timeout=15)
offset = (existujici[-1]["update_id"] + 1) if existujici else 0
posli_telegram(otazka, chat_id=cid, parse_mode=parse_mode)
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
zbyva = int(deadline - time.monotonic())
if zbyva <= 0:
break
lp = max(1, min(poll_timeout, zbyva))
updates = _call("getUpdates", http_timeout=lp + 10, offset=offset, timeout=lp)
for u in updates:
offset = u["update_id"] + 1
msg = u.get("message") or {}
if str(msg.get("chat", {}).get("id")) != cid:
continue # zpráva z jiného chatu — ignoruj
text = msg.get("text")
if text:
return text
return None
def _safe_print(text: str):
"""Výpis odolný vůči kódování Windows konzole (cp1252)."""
try:
print(text)
except UnicodeEncodeError:
print(text.encode("ascii", "replace").decode("ascii"))
if __name__ == "__main__":
# Ať projdou i diakritika/emoji na Windows konzoli.
try:
sys.stdout.reconfigure(encoding="utf-8")
except Exception:
pass
args = sys.argv[1:]
if not args:
print('Použití:')
print(' python -m Knihovny.telegram_notify "text zprávy"')
print(' python -m Knihovny.telegram_notify --ask "otázka?"')
sys.exit(1)
if args[0] == "--ask":
otazka = " ".join(args[1:]) or "?"
odpoved = zeptej_se_telegram(otazka, timeout=240)
if odpoved is None:
_safe_print("(bez odpovědi — vypršel timeout)")
sys.exit(2)
_safe_print(odpoved)
else:
posli_telegram(" ".join(args))
_safe_print("Odesláno OK")
Knihovny/telegram_user.py
+302
View File
@@ -0,0 +1,302 @@
"""
telegram_user.py
----------------
Ovládání PLNOHODNOTNÉHO Telegram účtu (ne bota) přes user API (MTProto / Telethon).
Na rozdíl od bota umí napsat komukoli a unese VÍCE souběžných agentů na jednom účtu
(jako Telegram otevřený zároveň na PC, tabletu i mobilu).
⚠️ Jedná JMÉNEM přihlášeného účtu. Session soubor = plný přístup k účtu.
⚠️ Nové účty na automatizaci Telegram rychle banuje (zvlášť VoIP čísla — použij reálnou SIM).
────────────────────────────────────────────────────────────────────────
VÍCE AGENTŮ NA JEDNOM ÚČTU
────────────────────────────────────────────────────────────────────────
- api_id/api_hash se SDÍLÍ (identifikují „aplikaci", ne zařízení).
- Každý agent musí mít VLASTNÍ session soubor (= vlastní autorizace / „zařízení").
Sdílet jednu session mezi procesy NELZE (database is locked / AUTH_KEY_DUPLICATED).
→ každý agent se přihlásí zvlášť: `login --jako <jmeno>` (jeden SMS kód na agenta).
- Všechny sessions vidí stejný chat, proto se odpovědi směrují přes Telegram **Reply**:
agent pošle označenou otázku a bere jen tu odpověď, která je Reply na *jeho* zprávu
(shoda `reply_to_msg_id`). Tím se odpovědi více agentů nepomíchají.
Konfigurace v `Medevio/.env` (api_id/api_hash z https://my.telegram.org):
TELEGRAM_API_ID=1234567
TELEGRAM_API_HASH=abcdef0123456789abcdef0123456789
TELEGRAM_PHONE=+420... # nepovinné (jinak se zeptá při loginu)
Session soubory: `Medevio/agent_telegram_<jmeno>.session` (gitignored).
────────────────────────────────────────────────────────────────────────
CLI
────────────────────────────────────────────────────────────────────────
Jednorázové přihlášení agenta (spusť ve svém terminálu — čeká na kód z SMS):
python -m Knihovny.telegram_user login --jako recepty
python -m Knihovny.telegram_user login --jako kalendar
Poslání zprávy ("me" = Uložené zprávy / Saved Messages):
python -m Knihovny.telegram_user send me "Test" --jako recepty
Otázka + čekání na Reply odpověď (vypíše odpověď na stdout):
python -m Knihovny.telegram_user ask recepty "Mam pokracovat? (ano/ne)"
────────────────────────────────────────────────────────────────────────
ZE SKRIPTU
────────────────────────────────────────────────────────────────────────
from Knihovny.telegram_user import posli_jako_ja, zeptej_se_jako
posli_jako_ja("me", "Pipeline 08 hotová", session="recepty")
odp = zeptej_se_jako("recepty", "Našel jsem 3 sporné záznamy. Pokračovat?")
if odp and odp.strip().lower() == "ano":
...
"""
import argparse
import os
import sys
import time
from pathlib import Path
# telethon.sync zpřístupní metody synchronně (bez async/await)
from telethon.sync import TelegramClient
from telethon.errors import SessionPasswordNeededError, PhoneNumberUnoccupiedError
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.setdefault(k.strip(), v.strip())
_load_env()
def _api_id() -> int:
val = os.environ.get("TELEGRAM_API_ID")
if not val:
raise RuntimeError("Chybí TELEGRAM_API_ID v Medevio/.env (z https://my.telegram.org)")
return int(val)
def _api_hash() -> str:
val = os.environ.get("TELEGRAM_API_HASH")
if not val:
raise RuntimeError("Chybí TELEGRAM_API_HASH v Medevio/.env (z https://my.telegram.org)")
return val
def _session_path(jmeno: str | None) -> Path:
base = f"agent_telegram_{jmeno}" if jmeno else "agent_telegram"
return Path(__file__).resolve().parent.parent / "Medevio" / base
def _new_client(session: str | None = None) -> TelegramClient:
return TelegramClient(str(_session_path(session)), _api_id(), _api_hash())
def prihlas(session: str | None = None) -> None:
"""
Jednorázové přihlášení dané session. Interaktivně se zeptá na kód z SMS
a případně na heslo dvoufázového ověření. Vytvoří session soubor.
SPOUŠTĚJ V TERMINÁLU (potřebuje input).
"""
client = _new_client(session)
client.start(phone=os.environ.get("TELEGRAM_PHONE") or (lambda: input("Telefon (+420...): ")))
me = client.get_me()
print(f"Session '{session or 'default'}' přihlášena jako "
f"{me.first_name or ''} (@{me.username}) id={me.id}")
client.disconnect()
def _phone() -> str:
val = os.environ.get("TELEGRAM_PHONE")
if not val:
raise RuntimeError("Chybí TELEGRAM_PHONE v Medevio/.env")
return val
def login_posli_kod(session: str | None = None) -> None:
"""
1. krok přihlášení (řízeného na dálku): vyžádá si od Telegramu kód.
Vytiskne `PHONE_CODE_HASH=...`, který je potřeba pro 2. krok.
"""
client = _new_client(session)
client.connect()
try:
sent = client.send_code_request(_phone())
print("PHONE_CODE_HASH=" + sent.phone_code_hash)
finally:
client.disconnect()
def login_dokonci(code, phone_code_hash: str, session: str | None = None,
password: str | None = None) -> None:
"""
2. krok přihlášení: dokončí login zadaným kódem (a případně heslem 2FA).
Při úspěchu uloží session soubor.
"""
client = _new_client(session)
client.connect()
try:
try:
client.sign_in(phone=_phone(), code=str(code), phone_code_hash=phone_code_hash)
except SessionPasswordNeededError:
if not password:
print("NEED_PASSWORD")
return
client.sign_in(password=password)
except PhoneNumberUnoccupiedError:
print("UCET_NEEXISTUJE - nejdriv zaregistruj cislo v aplikaci Telegram")
return
me = client.get_me()
print(f"OK prihlaseno jako {me.first_name or ''} (@{me.username}) id={me.id}")
finally:
client.disconnect()
def posli_jako_ja(komu, text: str, *, session: str | None = None):
"""
Pošle zprávu jménem přihlášeného účtu z dané session.
:param komu: "me" (Saved Messages) | "@username" | telefon | int id
:param text: text zprávy
:param session: jméno session (které přihlášení použít)
:return: odeslaná zpráva (Telethon Message)
"""
with _new_client(session) as client:
if not client.is_user_authorized():
raise RuntimeError(
f"Session '{session or 'default'}' není přihlášena — "
f"spusť: python -m Knihovny.telegram_user login"
+ (f" --jako {session}" if session else "")
)
return client.send_message(komu, text)
def precti_zpravy(komu, limit: int = 10, *, session: str | None = None):
"""
Vrátí posledních `limit` zpráv z daného chatu.
:return: list dictů {"id", "text", "odeslal_ja", "reply_na", "datum"}
"""
out = []
with _new_client(session) as client:
if not client.is_user_authorized():
raise RuntimeError(f"Session '{session or 'default'}' není přihlášena.")
for msg in client.iter_messages(komu, limit=limit):
out.append({
"id": msg.id,
"text": msg.text or "",
"odeslal_ja": bool(msg.out),
"reply_na": msg.reply_to_msg_id,
"datum": msg.date,
})
return out
def zeptej_se_jako(
agent: str,
otazka: str,
*,
komu="me",
session: str | None = None,
timeout: int = 300,
poll_interval: int = 3,
vyzaduj_reply: bool = True,
) -> str | None:
"""
Pošle označenou otázku ("[agent] otázka") a BLOKUJÍCÍ čeká na odpověď.
Při více agentech naráz se odpovědi rozlišují přes Telegram **Reply**:
bere jen tu příchozí zprávu, která je Reply na právě odeslanou otázku.
:param agent: jméno agenta (objeví se v textu otázky jako štítek)
:param otazka: text otázky
:param komu: kam poslat ("me" = Saved Messages | "@username" | id)
:param session: jméno session; výchozí = `agent` (každý agent svůj soubor)
:param timeout: celkové čekání v sekundách (pak vrátí None)
:param poll_interval: jak často kontrolovat nové zprávy (s)
:param vyzaduj_reply: True = bere jen Reply na svou otázku (bezpečné pro víc agentů);
False = vezme první příchozí zprávu (jen pro 1 agenta)
:return: text odpovědi, nebo None při timeoutu
"""
session = session or agent
with _new_client(session) as client:
if not client.is_user_authorized():
raise RuntimeError(
f"Session '{session}' není přihlášena — "
f"spusť: python -m Knihovny.telegram_user login --jako {session}"
)
sent = client.send_message(komu, f"[{agent}] {otazka}")
qid = sent.id
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
# jen zprávy novější než naše otázka, od nejstarší
for msg in client.iter_messages(komu, min_id=qid, reverse=True):
if msg.out:
continue # naše vlastní zpráva
if vyzaduj_reply:
if msg.reply_to_msg_id == qid:
return msg.text
else:
return msg.text
zbyva = deadline - time.monotonic()
if zbyva <= 0:
break
time.sleep(min(poll_interval, max(1, zbyva)))
return None
def _safe_print(text: str):
try:
print(text)
except UnicodeEncodeError:
print(text.encode("ascii", "replace").decode("ascii"))
def _main():
try:
sys.stdout.reconfigure(encoding="utf-8")
except Exception:
pass
parser = argparse.ArgumentParser(prog="telegram_user", description="Telegram user účet (Telethon)")
sub = parser.add_subparsers(dest="cmd", required=True)
p_login = sub.add_parser("login", help="jednorázové přihlášení session")
p_login.add_argument("--jako", dest="jako", default=None, help="jméno session/agenta")
p_send = sub.add_parser("send", help="poslat zprávu")
p_send.add_argument("komu", help='"me" | "@username" | telefon | id')
p_send.add_argument("text", help="text zprávy")
p_send.add_argument("--jako", dest="jako", default=None, help="jméno session")
p_ask = sub.add_parser("ask", help="poslat otázku a počkat na Reply odpověď")
p_ask.add_argument("agent", help="jméno agenta (štítek + výchozí session)")
p_ask.add_argument("text", help="text otázky")
p_ask.add_argument("--komu", dest="komu", default="me", help='kam (výchozí "me")')
p_ask.add_argument("--timeout", dest="timeout", type=int, default=240, help="čekání v s")
args = parser.parse_args()
if args.cmd == "login":
prihlas(args.jako)
elif args.cmd == "send":
posli_jako_ja(args.komu, args.text, session=args.jako)
_safe_print("Odesláno OK")
elif args.cmd == "ask":
odp = zeptej_se_jako(args.agent, args.text, komu=args.komu, timeout=args.timeout)
if odp is None:
_safe_print("(bez odpovědi — vypršel timeout)")
sys.exit(2)
_safe_print(odp)
if __name__ == "__main__":
_main()