generated from Templates/Docker_Image
152 lines
5.6 KiB
Python
152 lines
5.6 KiB
Python
"""Telegram bot integration for n8n webhook."""
|
|
|
|
import asyncio
|
|
import json
|
|
import re
|
|
from typing import Optional
|
|
|
|
|
|
class TelegramHandler:
|
|
"""Handles Telegram messages via n8n webhook."""
|
|
|
|
def __init__(self, webhook_url: str):
|
|
self.webhook_url = webhook_url
|
|
self.api_url = "https://api.telegram.org/bot"
|
|
|
|
async def handle_message(self, message_data: dict) -> dict:
|
|
"""Handle incoming Telegram message."""
|
|
text = message_data.get("text", "")
|
|
chat_id = message_data.get("chat", {}).get("id", "")
|
|
|
|
# Extract software request from message
|
|
request = self._parse_request(text)
|
|
|
|
if request:
|
|
# Forward to backend API
|
|
async def fetch_software():
|
|
try:
|
|
import aiohttp
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(
|
|
"http://localhost:8000/generate",
|
|
json=request
|
|
) as resp:
|
|
return await resp.json()
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
result = await fetch_software()
|
|
return {
|
|
"status": "success",
|
|
"data": result,
|
|
"response": self._format_response(result)
|
|
}
|
|
else:
|
|
return {
|
|
"status": "error",
|
|
"message": "Could not parse software request"
|
|
}
|
|
|
|
def _parse_request(self, text: str) -> Optional[dict]:
|
|
"""Parse software request from user message."""
|
|
# Simple parser - in production, use LLM to extract
|
|
request = {
|
|
"name": None,
|
|
"description": None,
|
|
"features": []
|
|
}
|
|
|
|
lines = text.split("\n")
|
|
|
|
# Parse name
|
|
name_idx = -1
|
|
for i, line in enumerate(lines):
|
|
line_stripped = line.strip()
|
|
if line_stripped.lower().startswith("name:"):
|
|
request["name"] = line_stripped.split(":", 1)[1].strip()
|
|
name_idx = i
|
|
break
|
|
|
|
if not request["name"]:
|
|
return None
|
|
|
|
# Parse description (everything after name until features section)
|
|
# First, find where features section starts
|
|
features_idx = -1
|
|
for i in range(name_idx + 1, len(lines)):
|
|
line_stripped = lines[i].strip()
|
|
if line_stripped.lower().startswith("features:"):
|
|
features_idx = i
|
|
break
|
|
|
|
if features_idx > name_idx:
|
|
# Description is between name and features
|
|
request["description"] = "\n".join(lines[name_idx + 1:features_idx]).strip()
|
|
else:
|
|
# Description is everything after name
|
|
request["description"] = "\n".join(lines[name_idx + 1:]).strip()
|
|
|
|
# Strip description prefix if present
|
|
if request["description"]:
|
|
request["description"] = request["description"].strip()
|
|
if request["description"].lower().startswith("description:"):
|
|
request["description"] = request["description"][len("description:") + 1:].strip()
|
|
|
|
# Parse features
|
|
if features_idx > 0:
|
|
features_line = lines[features_idx]
|
|
# Parse inline features after "Features:"
|
|
if ":" in features_line:
|
|
inline_part = features_line.split(":", 1)[1].strip()
|
|
|
|
# Skip if it starts with dash (it's a multiline list marker)
|
|
if inline_part and not inline_part.startswith("-"):
|
|
# Remove any leading dashes or asterisks
|
|
if inline_part.startswith("-"):
|
|
inline_part = inline_part[1:].strip()
|
|
elif inline_part.startswith("*"):
|
|
inline_part = inline_part[1:].strip()
|
|
|
|
if inline_part:
|
|
# Split by comma for inline features
|
|
request["features"].extend([f.strip() for f in inline_part.split(",") if f.strip()])
|
|
|
|
# Parse multiline features (dash lines after features:)
|
|
for line in lines[features_idx + 1:]:
|
|
line_stripped = line.strip()
|
|
if not line_stripped:
|
|
continue
|
|
if line_stripped.startswith("-"):
|
|
feature_text = line_stripped[1:].strip()
|
|
if feature_text:
|
|
request["features"].append(feature_text)
|
|
elif line_stripped.startswith("*"):
|
|
feature_text = line_stripped[1:].strip()
|
|
if feature_text:
|
|
request["features"].append(feature_text)
|
|
elif ":" in line_stripped:
|
|
# Non-feature line with colon
|
|
break
|
|
|
|
# MUST have features
|
|
if not request["features"]:
|
|
return None
|
|
|
|
return request
|
|
|
|
def _format_response(self, result: dict) -> dict:
|
|
"""Format response for Telegram."""
|
|
status = result.get("status", "error")
|
|
message = result.get("message", result.get("detail", ""))
|
|
progress = result.get("progress", 0)
|
|
|
|
response_data = {
|
|
"status": status,
|
|
"message": message,
|
|
"progress": progress,
|
|
"project_name": result.get("name", "N/A"),
|
|
"logs": result.get("logs", [])
|
|
}
|
|
|
|
return response_data
|