📄 telegram_incident.py 6,400 bytes Apr 23, 2026 📋 Raw

"""
Telegram Bot Command — /logincident

Usage:
/logincident "Title" system1,system2 "Error message"

Flow:
1. Parse command
2. Collect attempts (interactive prompts)
3. Collect fix and reflection
4. Save incident file
5. Confirm with preview
"""

import asyncio
import json
import sys
from pathlib import Path
from typing import Dict, List, Optional
from datetime import datetime, timezone

sys.path.insert(0, '/home/hoffmann_admin/.openclaw/workspace-socrates/hoffdesk-api')
from incidents import incident

async def log_incident_handler(message_text: str, chat_id: int) -> str:
"""
Handle /logincident command from Telegram.

Expected format:
    /logincident "Title" system1,system2 "Error message"
"""
# Parse command
parts = message_text.split('"')
if len(parts) < 5:
    return (
        "Usage:\n"
        '/logincident "Title of incident" system1,system2 "Error message"\n\n'
        'Example:\n'
        '/logincident "Cloudflare Error 1033" cloudflared,uvicorn,systemd "Error 1033: Cannot reach origin"'
    )

title = parts[1].strip()
systems = [s.strip() for s in parts[2].strip().split(',')]
error = parts[3].strip()

# For Telegram, we'd use inline keyboards or conversation state
# For now, return a prompt asking for structured input
return f"""📝 Logging incident: {title}

📋 Please provide details in this format:

Attempts:
1. What I tried: ...
What happened: ...
2. What I tried: ...
What happened: ...

Fix: ...
Reflection: ...
Hours spent: ...
Sleep lost: yes/no"""

def parse_incident_response(response_text: str) -> Optional[Dict]:
"""
Parse user response into incident structure.

Expected format:
    Attempts:
    1. What I tried: Checked uvicorn health
       What happened: Running but bound to 127.0.0.1
    2. What I tried: Added systemd drop-in
       What happened: Still Error 1033

    Fix: Removed stale override.conf
    Reflection: Should have run systemctl cat first
    Hours spent: 2.5
    Sleep lost: yes
"""
lines = response_text.strip().split('\n')

attempts = []
current_attempt = {}
fix = ""
reflection = ""
hours = 0.0
sleep_lost = False

in_attempts = False

for line in lines:
    line = line.strip()
    if not line:
        continue

    if line.startswith('Attempts:'):
        in_attempts = True
        continue

    if line.startswith('Fix:'):
        in_attempts = False
        fix = line[4:].strip()
        continue

    if line.startswith('Reflection:'):
        reflection = line[11:].strip()
        continue

    if line.startswith('Hours spent:'):
        try:
            hours = float(line[12:].strip())
        except ValueError:
            hours = 0.0
        continue

    if line.startswith('Sleep lost:'):
        sleep_lost = line[11:].strip().lower() in ['yes', 'true', 'y']
        continue

    if in_attempts:
        if line[0].isdigit() and '. What I tried:' in line:
            if current_attempt:
                attempts.append(current_attempt)
            current_attempt = {
                'attempt': line.split('. What I tried:', 1)[1].strip()
            }
        elif line.startswith('What happened:') and current_attempt:
            current_attempt['result'] = line[14:].strip()

if current_attempt:
    attempts.append(current_attempt)

if not attempts:
    return None

return {
    'attempts': attempts,
    'fix': fix,
    'reflection': reflection,
    'hours': hours,
    'sleep_lost': sleep_lost
}

def preview_incident(incident_data: Dict) -> str:
"""Generate preview of incident for Telegram."""
lines = [
f"{incident_data['title']}",
f"📅 {incident_data['date'][:10]}",
f"🔧 {', '.join(incident_data['systems'])}",
f"❌ {incident_data['error']}",
"",
f"Attempts: {len(incident_data['attempts'])}",
]

for i, attempt in enumerate(incident_data['attempts'], 1):
    lines.append(f"  {i}. {attempt['attempt'][:40]}...")
    if 'result' in attempt:
        lines.append(f"     → {attempt['result'][:40]}...")

lines.extend([
    "",
    f"✅ Fix: {incident_data['fix'][:60]}..." if incident_data.get('fix') else "",
    f"💭 {incident_data['reflection'][:80]}..." if incident_data.get('reflection') else "",
    f"⏱️ {incident_data.get('hours', 0)} hours",
    f"😴 Sleep lost: {'yes' if incident_data.get('sleep_lost') else 'no'}",
])

return '\n'.join(lines)

Telegram bot integration stub

async def handle_telegram_command(update: Dict) -> str:
"""
Entry point for Telegram bot webhook.

Args:
    update: Telegram update object

Returns:
    Response message
"""
message = update.get("message", {})
text = message.get("text", "")
chat_id = message.get("chat", {}).get("id")

if text.startswith("/logincident"):
    return await log_incident_handler(text, chat_id)

return "Unknown command"

if name == "main":
# Test with sample data
test_message = '/logincident "Cloudflare Error 1033" cloudflared,uvicorn,systemd "Error 1033: Cannot reach origin"'

result = asyncio.run(log_incident_handler(test_message, 12345))
print(result)
print()

# Test response parsing
test_response = """Attempts:
  1. What I tried: Checked uvicorn health with curl
    What happened: Running but bound to 127.0.0.1
  2. What I tried: Added systemd drop-in for 0.0.0.0
    What happened: Still Error 1033

Fix: Removed stale override.conf
Reflection: Should have run systemctl cat first
Hours spent: 2.5
Sleep lost: yes"""

parsed = parse_incident_response(test_response)
if parsed:
    print("Parsed successfully:")
    print(f"  Attempts: {len(parsed['attempts'])}")
    print(f"  Fix: {parsed['fix']}")
    print(f"  Reflection: {parsed['reflection']}")
    print(f"  Hours: {parsed['hours']}")
    print(f"  Sleep lost: {parsed['sleep_lost']}")