FamilyBuddy

cat testbot.py

cat neonize_dh_version.py | wc -l

import time, requests, os, re, json, litellm
import datetime as dt
from zoneinfo import ZoneInfo
from threading import Thread
from fastcore.utils import *
from dialoghelper.core import _add_msg_unsafe
from lisette.core import mk_msg
from neonize.client import NewClient
from neonize.events import MessageEv, ConnectedEv, event
from neonize.utils.jid import build_jid
from neonize.utils.enum import ChatPresence, ChatPresenceMedia, ReceiptType

model = 'claude-sonnet-4-5-20250929'
wa_cfg = Path('whatsapp_group.json').read_json()
group_id = wa_cfg['group_id']
participants = wa_cfg['participants']

HIST_DNAME = 'dlgs/hist'
MEDIA_DIR = Path('dlgs/media')

def wait_for_completion(msg_id, dname=HIST_DNAME, timeout=120, poll_interval=0.5):
    "Poll until prompt completes. Returns the output text."
    start = time.time()
    while time.time() - start < timeout:
        msg = read_msg(id=msg_id, n=0, relative=True, dname=dname)
        if not msg.get('run'): return msg.get('output', '')
        time.sleep(poll_interval)
    raise TimeoutError(f"Prompt {msg_id} did not complete within {timeout}s")

def hide_section(header_text, dname=HIST_DNAME):
    "Skip all messages in a section and collapse the header."
    msgs = find_msgs(header_section=header_text, dname=dname)
    for m in msgs: update_msg(id=m['id'], skipped=1, dname=dname)
    if msgs: update_msg(id=msgs[0]['id'], heading_collapsed=1, dname=dname)

def ensure_today_section(dname=HIST_DNAME):
    "Create today's section header if needed, hide previous day's section."
    today = dt.datetime.now(ZoneInfo('Europe/Amsterdam')).date().strftime('%a %d %b %Y')
    if (existing:=find_msgs(re_pattern=f"^## {today}$", dname=dname)): return existing[0]['id']
    for sec in find_msgs(re_pattern=r"^## [A-Z][a-z]{2} \d{2} [A-Z][a-z]{2} \d{4}$", dname=dname): hide_section(sec['content'], dname=dname)
    return add_msg(f"## {today}", placement='at_end', dname=dname)

client = NewClient("familybuddy")
group_jid = build_jid(group_id)

def start_typing(chat_jid): client.send_chat_presence(chat_jid, ChatPresence.CHAT_PRESENCE_COMPOSING, ChatPresenceMedia.CHAT_PRESENCE_MEDIA_TEXT)
def stop_typing(chat_jid): client.send_chat_presence(chat_jid, ChatPresence.CHAT_PRESENCE_PAUSED, ChatPresenceMedia.CHAT_PRESENCE_MEDIA_TEXT)
def mark_read(chat_jid, sender_jid, msg_ids): client.mark_read(*msg_ids, chat=chat_jid, sender=sender_jid, receipt=ReceiptType.READ)
def send_reply(chat_jid, text): client.send_message(chat_jid, text)

def transcribe_audio(audio_bytes):
    msg = mk_msg(['Please transcribe the following message, only return the text they said, nothing else.', audio_bytes])
    resp = litellm.completion('gemini/gemini-3-flash-preview', [msg])
    return resp.choices[0].message.content

def jid_str(jid): return f"{jid.User}@{jid.Server}"

def save_image(img_bytes, ext='jpg'):
    "Save image to media dir and return markdown reference."
    fname = f"{unqid()}.{ext}"
    (MEDIA_DIR / fname).write_bytes(img_bytes)
    return f" ![](./media/{fname}#ai) "

def format_batch(events: list, participants: dict) -> str | None:
    """Convert Neonize MessageEv list to prompt text for dialoghelper."""
    lines = []
    
    for event in events:
        msg,info = event.Message,event.Info
        sender_jid = jid_str(info.MessageSource.Sender)
        name = participants.get(sender_jid, 'unknown')
        time_str = dt.datetime.fromtimestamp(info.Timestamp / 1000, AMS).strftime('%a %Y-%m-%d %H:%M')
        
        body = msg.conversation or ''
        if msg.extendedTextMessage and msg.extendedTextMessage.text: body = msg.extendedTextMessage.text
        
        if msg.HasField('imageMessage'):
            img_bytes = client.download_any(msg)
            img_ref = save_image(img_bytes)
            caption = msg.imageMessage.caption or ''
            body = f'{caption}\n{img_ref}'.strip() if caption else img_ref        
        elif msg.HasField('audioMessage'):
            audio_bytes = client.download_any(msg)
            text = transcribe_audio(audio_bytes)
            body = f'{body} [voice: {text}]'.strip() if body else f'[voice: {text}]'       
        elif msg.HasField('videoMessage') or msg.HasField('documentMessage'): body = f'[media] {body}' if body else '[media]'

        if not body: continue
        lines.append(f'<{name} time="{time_str}">{body}</{name}>')
    
    if not lines: return None
    return '\n'.join(lines)

def format_output_for_wa(output):
    "Convert tool call details to concise format for WhatsApp"
    def format_tool(match): return f"🔧`{match.group(1)}`"    
    output = re.sub(r"<details class='tool-usage-details'>\s*<summary>(.*?)</summary>.*?</details>", format_tool, output, flags=re.DOTALL)
    output = re.sub(r'\n{3,}', '\n\n', output)
    return output.strip('🧠\n')

pending_events = []

@client.event(MessageEv)
def on_message(c: NewClient, event: MessageEv):
    chat_jid = event.Info.MessageSource.Chat
    if jid_str(chat_jid) != group_id: return
    if event.Info.MessageSource.IsFromMe: return
    pending_events.append(event)
    msg = event.Message
    body = msg.conversation or (msg.extendedTextMessage.text if msg.extendedTextMessage and msg.extendedTextMessage.text else '')
    if body and body[0] == '.': return
    mark_read(chat_jid, event.Info.MessageSource.Sender, [event.Info.ID])
    start_typing(chat_jid)
    try:
        ensure_today_section()
        if prompt_text:=format_batch(pending_events, participants):
            msg_id = _add_msg_unsafe(prompt_text, msg_type='prompt', run=True, placement='at_end', dname=HIST_DNAME)
            if reply:=wait_for_completion(msg_id): send_reply(chat_jid, format_output_for_wa(reply))
        pending_events.clear()
    except Exception as e: send_reply(chat_jid, f"🚨 Faby error:\n{type(e).__name__}: {e}")
    finally: stop_typing(chat_jid)

Thread(target=client.connect, daemon=True).start()