116 lines
3.7 KiB
Python
116 lines
3.7 KiB
Python
"""
|
|
telegram_notify.py — PŘIBALENÁ kopie pro běh na serveru (python-runner)
|
|
=======================================================================
|
|
Na toweru není balík `Knihovny/` ani `Medevio/.env`, proto má watcher tuto
|
|
soběstačnou kopii. Funkce jsou shodné s `Knihovny/telegram_notify.py`.
|
|
|
|
Token a chat_id se hledají v `.env` na víc místech (první nalezené vyhrává):
|
|
1) `.env` ve stejném adresáři jako tento soubor (server: /scripts/Webinare/.env)
|
|
2) `../Medevio/.env` (lokální vývoj)
|
|
3) `../../Medevio/.env` (kořen projektu)
|
|
|
|
TELEGRAM_BOT_TOKEN=123456789:AAE...
|
|
TELEGRAM_CHAT_ID=6639316354
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
|
|
|
|
def _load_env():
|
|
here = Path(__file__).resolve().parent
|
|
kandidati = [
|
|
here / ".env",
|
|
here.parent / "Medevio" / ".env",
|
|
here.parent.parent / "Medevio" / ".env",
|
|
]
|
|
for env_path in kandidati:
|
|
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 (.env)")
|
|
return token
|
|
|
|
|
|
def _resolve_chat_id(chat_id):
|
|
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 .env)")
|
|
return str(chat_id)
|
|
|
|
|
|
def _call(method, *, http_timeout=15, **params):
|
|
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, *, chat_id=None, parse_mode=None, disable_notification=False):
|
|
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, *, chat_id=None, timeout=300, poll_timeout=30, parse_mode=None):
|
|
cid = _resolve_chat_id(chat_id)
|
|
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
|
|
text = msg.get("text")
|
|
if text:
|
|
return text
|
|
return None
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
except Exception:
|
|
pass
|
|
args = sys.argv[1:]
|
|
if args and args[0] == "--ask":
|
|
print(zeptej_se_telegram(" ".join(args[1:]) or "?", timeout=240) or "(bez odpovědi)")
|
|
elif args:
|
|
posli_telegram(" ".join(args))
|
|
print("Odesláno OK")
|
|
else:
|
|
print('Použití: python telegram_notify.py "text" | --ask "otázka?"')
|