Bot de monitoratge de benestar amb Telegram i Python

python3 --version

pip install python-telegram-bot apscheduler

https://api.telegram.org/botTOKEN/getUpdates

monitor-bot/
├── config.json      ← token, persones, horaris
├── bot.py           ← bot principal (arrencar aquest)
├── logger.py        ← guarda converses i anomalies
└── analyzer.py      ← detecta anomalies en el text

{
  "telegram_bot_token": "EL_TEU_TOKEN",
  "horaris": {
    "mati":   "08:00",
    "migdia": "15:00",
    "nit":    "21:00"
  },
  "timeouts": {
    "primer_avis_minuts": 60,
    "segon_avis_minuts":  5
  },
  "llindar_anomalia_dies": 3,
  "usuaris": [
    {
      "id":  "usuari_1",
      "nom": "Maria",
      "chat_id": "111111111",
      "contacte_emergencia": {
        "nom":     "Joan",
        "chat_id": "222222222"
      }
    }
  ]
}

import json, csv, os
from datetime import datetime, timedelta

def _path_json(uid): return os.path.join("log", uid, "conversations.json")
def _path_csv(uid):  return os.path.join("log", uid, "anomalies.csv")
def _path_global():  return os.path.join("log", "anomalies_global.csv")

def init_logs(uid):
    directori = os.path.join("log", uid)
    os.makedirs(directori, exist_ok=True)
    if not os.path.exists(_path_json(uid)):
        with open(_path_json(uid), "w", encoding="utf-8") as f:
            json.dump([], f)
    if not os.path.exists(_path_csv(uid)):
        with open(_path_csv(uid), "w", newline="", encoding="utf-8") as f:
            csv.writer(f).writerow(["timestamp","persona","tipus_anomalia",
                                    "missatge_original","resposta","escalat","contacte_avisat"])
    if not os.path.exists(_path_global()):
        os.makedirs("log", exist_ok=True)
        with open(_path_global(), "w", newline="", encoding="utf-8") as f:
            csv.writer(f).writerow(["timestamp","usuari_id","persona","tipus_anomalia",
                                    "missatge_original","resposta","escalat","contacte_avisat"])

def guardar_conversa(uid, nom, hora, missatge, resposta, temps_min, anomalies, escalat):
    with open(_path_json(uid), "r", encoding="utf-8") as f:
        data = json.load(f)
    data.append({"timestamp": datetime.now().isoformat(), "usuari_id": uid,
                  "persona": nom, "hora_programada": hora, "missatge_enviat": missatge,
                  "resposta": resposta, "temps_resposta_min": temps_min,
                  "anomalies_detectades": anomalies, "escalat": escalat})
    with open(_path_json(uid), "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)

def guardar_anomalia(uid, nom, tipus, missatge, resposta, escalat, contacte_avisat):
    ts = datetime.now().strftime("%Y-%m-%d %H:%M")
    with open(_path_csv(uid), "a", newline="", encoding="utf-8") as f:
        csv.writer(f).writerow([ts, nom, tipus, missatge, resposta or "", escalat, contacte_avisat])
    with open(_path_global(), "a", newline="", encoding="utf-8") as f:
        csv.writer(f).writerow([ts, uid, nom, tipus, missatge, resposta or "", escalat, contacte_avisat])

def llegir_historial_recent(uid, dies=7):
    path = _path_json(uid)
    if not os.path.exists(path): return []
    with open(path, "r", encoding="utf-8") as f:
        data = json.load(f)
    llindar = datetime.now() - timedelta(days=dies)
    return [e for e in data if datetime.fromisoformat(e["timestamp"]) > llindar]

from logger import llegir_historial_recent

KEYWORDS = {
    "caiguda": [
        "he caigut", "m'he caigut", "caiguda", "no puc aixecar",
        "estic a terra", "no em puc moure",
    ],
    "dolor_malaltia": [
        "em fa mal", "no em trobo bé", "estic malalt", "estic malalta",
        "febre", "em trobo fatal", "mal de cap", "no puc respirar",
    ],
    "tristesa": [
        "estic trist", "estic trista", "no tinc ganes",
        "estic cansat de tot", "no vull res", "em sento sola",
        "tot va malament", "estic deprimit",
    ],
}

POSITIVES = ["bé", "molt bé", "perfecte", "genial", "fenomenal",
             "content", "contenta", "alegre", "feliç", "descansat", "ok"]

def detectar_anomalies_text(text):
    if not text: return []
    tl = text.lower().strip()
    anomalies = [t for t, kws in KEYWORDS.items() if any(k in tl for k in kws)]
    if len(tl.split()) <= 3 and not any(p in tl for p in POSITIVES):
        anomalies.append("resposta_curta")
    return anomalies

def detectar_anomalia_patro(uid):
    hist = llegir_historial_recent(uid, dies=7)
    if len(hist) < 3: return {"alertar": False, "motiu": ""}
    recent = hist[-9:]
    curtes = sum(1 for e in recent if "resposta_curta" in e.get("anomalies_detectades", []))
    if curtes >= 6:
        return {"alertar": True, "motiu": f"Respostes curtes: {curtes} dels últims {len(recent)} missatges."}
    dies_tristes = set(e["timestamp"][:10] for e in hist if "tristesa" in e.get("anomalies_detectades", []))
    if len(dies_tristes) >= 3:
        return {"alertar": True, "motiu": f"Tristesa {len(dies_tristes)} dies seguits."}
    return {"alertar": False, "motiu": ""}

def anomalia_requereix_escalat_immediat(anomalies):
    return "caiguda" in anomalies or "dolor_malaltia" in anomalies

import asyncio, json, random, logging
from datetime import datetime
from telegram import Bot
from telegram.ext import Application, MessageHandler, filters, ContextTypes
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from logger   import init_logs, guardar_conversa, guardar_anomalia
from analyzer import detectar_anomalies_text, detectar_anomalia_patro, anomalia_requereix_escalat_immediat

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.FileHandler("log/bot.log", encoding="utf-8"), logging.StreamHandler()])
log = logging.getLogger(__name__)

with open("config.json", encoding="utf-8") as f: CONFIG = json.load(f)
TOKEN    = CONFIG["telegram_bot_token"]
USUARIS  = CONFIG["usuaris"]
HORARIS  = CONFIG["horaris"]
TIMEOUT_1 = CONFIG["timeouts"]["primer_avis_minuts"] * 60
TIMEOUT_2 = CONFIG["timeouts"]["segon_avis_minuts"]  * 60
CHAT_ID_A_USUARI = {u["chat_id"]: u for u in USUARIS}

def missatges_hora(nom):
    return {
        HORARIS["mati"]:   [f"Bon dia, {nom}! 🌅 Com has dormit avui?",
                            f"Bon dia, {nom}! ☀️ Que tens de fer avui?"],
        HORARIS["migdia"]: [f"Hola, {nom}! 🍽️ Ja has dinat?",
                            f"Hola, {nom}! 😄 Que has dinat? Estava bo?"],
        HORARIS["nit"]:    [f"Bona nit, {nom}! 🌙 Ja has sopat?",
                            f"Hola, {nom}! 🌆 Com ha anat el dia?"],
    }

def missatges_seguiment(nom):
    return [f"Estàs bé, {nom}? 🙏 Necessites quelcom?",
            f"Hola {nom}! Volia saber si estàs bé. 💙"]

estats = {u["id"]: {"esperant_resposta": False, "hora_enviat": None,
                    "missatge_enviat": None, "hora_programada": None,
                    "task_timeout": None} for u in USUARIS}
bot = Bot(token=TOKEN)

async def enviar_missatge(text, chat_id):
    try:
        await bot.send_message(chat_id=chat_id, text=text)
    except Exception as e: log.error(f"Error: {e}")

async def avisar_contacte(usuari, motiu):
    nom = usuari["nom"]; c = usuari["contacte_emergencia"]
    text = (f"⚠️ AVÍS — {nom} no respon o necessita ajuda.\n"
            f"Motiu: {motiu}\n"
            f"Hora: {datetime.now().strftime('%H:%M del %d/%m/%Y')}\n"
            f"Sisplau comprova com està {nom}.")
    await enviar_missatge(text, c["chat_id"])

async def gestionar_timeout(usuari):
    uid = usuari["id"]; nom = usuari["nom"]; estat = estats[uid]
    await asyncio.sleep(TIMEOUT_1)
    if not estat["esperant_resposta"]: return
    await enviar_missatge(random.choice(missatges_seguiment(nom)), usuari["chat_id"])
    guardar_anomalia(uid, nom, "no_resposta_1h", estat["missatge_enviat"], None, False, False)
    await asyncio.sleep(TIMEOUT_2)
    if not estat["esperant_resposta"]: return
    await avisar_contacte(usuari, f"{nom} no respon des de les {estat['hora_programada']}.")
    guardar_conversa(uid, nom, estat["hora_programada"], estat["missatge_enviat"],
                     None, None, ["no_resposta_total"], True)
    guardar_anomalia(uid, nom, "no_resposta_total", estat["missatge_enviat"], None, True, True)
    estat["esperant_resposta"] = False

async def fer_checkin_usuari(usuari, hora):
    uid = usuari["id"]; estat = estats[uid]
    missatge = random.choice(missatges_hora(usuari["nom"])[hora])
    await enviar_missatge(missatge, usuari["chat_id"])
    estat.update({"esperant_resposta": True, "hora_enviat": datetime.now(),
                  "missatge_enviat": missatge, "hora_programada": hora})
    if estat["task_timeout"] and not estat["task_timeout"].done():
        estat["task_timeout"].cancel()
    estat["task_timeout"] = asyncio.create_task(gestionar_timeout(usuari))

async def fer_checkin_tots(hora):
    await asyncio.gather(*[fer_checkin_usuari(u, hora) for u in USUARIS])

async def gestionar_resposta(update, context: ContextTypes.DEFAULT_TYPE):
    chat_id = str(update.message.chat_id)
    usuari  = CHAT_ID_A_USUARI.get(chat_id)
    if not usuari: return
    uid = usuari["id"]; estat = estats[uid]
    text = update.message.text
    temps_min = int((datetime.now() - estat["hora_enviat"]).total_seconds() / 60) if estat["hora_enviat"] else None
    estat["esperant_resposta"] = False
    anomalies = detectar_anomalies_text(text)
    patro = detectar_anomalia_patro(uid)
    if patro["alertar"]: anomalies.append("canvi_patro")
    guardar_conversa(uid, usuari["nom"], estat.get("hora_programada","?"),
                     estat.get("missatge_enviat",""), text, temps_min, anomalies, bool(anomalies))
    if anomalia_requereix_escalat_immediat(anomalies):
        for t in ["caiguda", "dolor_malaltia"]:
            if t in anomalies:
                await avisar_contacte(usuari, f"{usuari['nom']} ha mencionat '{t}': \"{text}\"")
                guardar_anomalia(uid, usuari["nom"], t, estat.get("missatge_enviat",""), text, True, True)
        return
    if patro["alertar"]:
        await avisar_contacte(usuari, patro["motiu"])
        guardar_anomalia(uid, usuari["nom"], "canvi_patro", estat.get("missatge_enviat",""), text, True, True)
    for t in anomalies:
        if t not in ["caiguda","dolor_malaltia","canvi_patro"]:
            guardar_anomalia(uid, usuari["nom"], t, estat.get("missatge_enviat",""), text, False, False)

async def main():
    for u in USUARIS: init_logs(u["id"])
    app = Application.builder().token(TOKEN).build()
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, gestionar_resposta))
    scheduler = AsyncIOScheduler()
    for clau in ["mati", "migdia", "nit"]:
        h, m = [int(x) for x in HORARIS[clau].split(":")]
        scheduler.add_job(fer_checkin_tots, "cron", hour=h, minute=m, args=[HORARIS[clau]])
    scheduler.start()
    log.info(f"✅ Bot actiu! {len(USUARIS)} usuaris.")
    await app.run_polling()

if __name__ == "__main__": asyncio.run(main())

python3 bot.py

🤖 Bot iniciant amb 1 usuaris configurats:
   → Maria (contacte: Joan)
✅ Bot actiu! Check-ins programats: 08:00 / 15:00 / 21:00

[Unit]
Description=OpenClaw Monitor Bot
After=network.target

[Service]
WorkingDirectory=/ruta/completa/monitor-bot
ExecStart=/usr/bin/python3 bot.py
Restart=always
User=el_teu_usuari

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable monitor-bot
sudo systemctl start monitor-bot