"""
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:
- What I tried: Checked uvicorn health with curl
What happened: Running but bound to 127.0.0.1 - 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']}")