""" 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']}")