Spaces:
Sleeping
Sleeping
import requests | |
import logging | |
import json | |
from typing import Optional | |
from datetime import datetime | |
# Configure logging | |
logging.basicConfig(level=logging.INFO) | |
logger = logging.getLogger(__name__) | |
class TelegramBotService: | |
"""Service class for Telegram Bot integration""" | |
def __init__(self, bot_token: str): | |
"""Initialize Telegram Bot service""" | |
self.bot_token = bot_token | |
self.base_url = f"https://api.telegram.org/bot{bot_token}" | |
def send_message(self, chat_id: str, message: str, parse_mode: str = 'HTML') -> dict: | |
""" | |
Send message via Telegram Bot | |
Args: | |
chat_id: Telegram chat ID or username | |
message: Message content to send | |
parse_mode: Message formatting (HTML, Markdown, or None) | |
Returns: | |
Dictionary with status and message details | |
""" | |
try: | |
url = f"{self.base_url}/sendMessage" | |
payload = { | |
'chat_id': chat_id, | |
'text': message, | |
'parse_mode': parse_mode | |
} | |
response = requests.post(url, json=payload, timeout=10) | |
if response.status_code == 200: | |
result = response.json() | |
if result.get('ok'): | |
logger.info(f"Telegram message sent successfully to {chat_id}") | |
return { | |
'status': 'sent', | |
'message_id': result['result']['message_id'], | |
'chat_id': chat_id, | |
'message': message, | |
'sent_at': datetime.utcnow(), | |
'error': None | |
} | |
else: | |
error_msg = result.get('description', 'Unknown Telegram API error') | |
logger.error(f"Telegram API error: {error_msg}") | |
return { | |
'status': 'failed', | |
'message_id': None, | |
'chat_id': chat_id, | |
'message': message, | |
'sent_at': datetime.utcnow(), | |
'error': error_msg | |
} | |
else: | |
error_msg = f"HTTP {response.status_code}: {response.text}" | |
logger.error(f"Failed to send Telegram message: {error_msg}") | |
return { | |
'status': 'failed', | |
'message_id': None, | |
'chat_id': chat_id, | |
'message': message, | |
'sent_at': datetime.utcnow(), | |
'error': error_msg | |
} | |
except Exception as e: | |
logger.error(f"Failed to send Telegram message to {chat_id}: {str(e)}") | |
return { | |
'status': 'failed', | |
'message_id': None, | |
'chat_id': chat_id, | |
'message': message, | |
'sent_at': datetime.utcnow(), | |
'error': str(e) | |
} | |
def get_me(self) -> Optional[dict]: | |
""" | |
Get bot information | |
Returns: | |
Bot info dictionary or None if failed | |
""" | |
try: | |
url = f"{self.base_url}/getMe" | |
response = requests.get(url, timeout=10) | |
if response.status_code == 200: | |
result = response.json() | |
if result.get('ok'): | |
return result['result'] | |
return None | |
except Exception as e: | |
logger.error(f"Failed to get bot info: {str(e)}") | |
return None | |
def send_document(self, chat_id: str, document_path: str, caption: str = None, parse_mode: str = 'HTML') -> dict: | |
""" | |
Send document (PDF, Word, etc.) via Telegram Bot | |
Args: | |
chat_id: Telegram chat ID or username | |
document_path: Path to the document file | |
caption: Optional caption for the document | |
parse_mode: Caption formatting (HTML, Markdown, or None) | |
Returns: | |
Dictionary with status and message details | |
""" | |
try: | |
import os | |
if not os.path.exists(document_path): | |
return { | |
'status': 'failed', | |
'error': 'Document file not found' | |
} | |
url = f"{self.base_url}/sendDocument" | |
with open(document_path, 'rb') as document: | |
files = {'document': document} | |
data = {'chat_id': chat_id} | |
if caption: | |
data['caption'] = caption | |
data['parse_mode'] = parse_mode | |
response = requests.post(url, files=files, data=data, timeout=30) | |
if response.status_code == 200: | |
result = response.json() | |
if result.get('ok'): | |
logger.info(f"Telegram document sent successfully to {chat_id}") | |
return { | |
'status': 'sent', | |
'message_id': result['result']['message_id'], | |
'chat_id': chat_id, | |
'document_path': document_path, | |
'sent_at': datetime.utcnow(), | |
'error': None | |
} | |
else: | |
error_msg = result.get('description', 'Unknown Telegram API error') | |
logger.error(f"Telegram API error: {error_msg}") | |
return { | |
'status': 'failed', | |
'error': error_msg | |
} | |
else: | |
error_msg = f"HTTP {response.status_code}: {response.text}" | |
logger.error(f"Failed to send Telegram document: {error_msg}") | |
return { | |
'status': 'failed', | |
'error': error_msg | |
} | |
except Exception as e: | |
logger.error(f"Failed to send Telegram document to {chat_id}: {str(e)}") | |
return { | |
'status': 'failed', | |
'error': str(e) | |
} | |
def send_photo(self, chat_id: str, photo_path: str, caption: str = None, parse_mode: str = 'HTML') -> dict: | |
""" | |
Send photo/image via Telegram Bot | |
Args: | |
chat_id: Telegram chat ID or username | |
photo_path: Path to the photo file | |
caption: Optional caption for the photo | |
parse_mode: Caption formatting (HTML, Markdown, or None) | |
Returns: | |
Dictionary with status and message details | |
""" | |
try: | |
import os | |
if not os.path.exists(photo_path): | |
return { | |
'status': 'failed', | |
'error': 'Photo file not found' | |
} | |
url = f"{self.base_url}/sendPhoto" | |
with open(photo_path, 'rb') as photo: | |
files = {'photo': photo} | |
data = {'chat_id': chat_id} | |
if caption: | |
data['caption'] = caption | |
data['parse_mode'] = parse_mode | |
response = requests.post(url, files=files, data=data, timeout=30) | |
if response.status_code == 200: | |
result = response.json() | |
if result.get('ok'): | |
logger.info(f"Telegram photo sent successfully to {chat_id}") | |
return { | |
'status': 'sent', | |
'message_id': result['result']['message_id'], | |
'chat_id': chat_id, | |
'photo_path': photo_path, | |
'sent_at': datetime.utcnow(), | |
'error': None | |
} | |
else: | |
error_msg = result.get('description', 'Unknown Telegram API error') | |
logger.error(f"Telegram API error: {error_msg}") | |
return { | |
'status': 'failed', | |
'error': error_msg | |
} | |
else: | |
error_msg = f"HTTP {response.status_code}: {response.text}" | |
logger.error(f"Failed to send Telegram photo: {error_msg}") | |
return { | |
'status': 'failed', | |
'error': error_msg | |
} | |
except Exception as e: | |
logger.error(f"Failed to send Telegram photo to {chat_id}: {str(e)}") | |
return { | |
'status': 'failed', | |
'error': str(e) | |
} | |
def send_bulk_messages(self, recipients: list, message: str) -> list: | |
""" | |
Send message to multiple recipients | |
Args: | |
recipients: List of chat IDs | |
message: Message content | |
Returns: | |
List of send results | |
""" | |
results = [] | |
for recipient in recipients: | |
result = self.send_message(recipient, message) | |
results.append(result) | |
return results | |
def format_daily_advisory_telegram(farmer_name: str, advisory_data: dict) -> str: | |
""" | |
Format daily advisory as Telegram message with HTML formatting | |
Args: | |
farmer_name: Name of the farmer | |
advisory_data: Advisory data from AI | |
Returns: | |
Formatted Telegram message with HTML | |
""" | |
message = f"🌱 <b>नमस्ते {farmer_name} जी!</b>\n\n" | |
message += f"📅 <b>आज की सलाह ({datetime.now().strftime('%d/%m/%Y')})</b>\n\n" | |
# Task to do | |
if advisory_data.get('task_to_do'): | |
message += f"✅ <b>आज करें:</b>\n{advisory_data['task_to_do']}\n\n" | |
# Task to avoid | |
if advisory_data.get('task_to_avoid'): | |
message += f"❌ <b>आज न करें:</b>\n{advisory_data['task_to_avoid']}\n\n" | |
# Reason/explanation | |
if advisory_data.get('reason_explanation'): | |
message += f"💡 <b>कारण:</b>\n{advisory_data['reason_explanation']}\n\n" | |
message += "🙏 <b>शुभ दिन!</b>\n" | |
message += "<i>- कृषि मित्र बॉट</i>" | |
return message | |
def format_weather_alert_telegram(farmer_name: str, weather_alert: str) -> str: | |
""" | |
Format weather alert as Telegram message | |
Args: | |
farmer_name: Name of the farmer | |
weather_alert: Weather alert message | |
Returns: | |
Formatted Telegram message | |
""" | |
message = f"🌦️ <b>मौसम अलर्ट!</b>\n\n" | |
message += f"<b>{farmer_name} जी,</b>\n\n" | |
message += f"{weather_alert}\n\n" | |
message += "⚠️ <b>कृपया अपनी फसल की सुरक्षा करें।</b>\n" | |
message += "<i>- कृषि मित्र बॉट</i>" | |
return message | |
def format_yearly_plan_summary(farmer_name: str, plan_summary) -> str: | |
""" | |
Format a yearly plan summary for Telegram - handles both dict and string inputs. | |
""" | |
message = f"🌾 <b>नमस्ते {farmer_name} जी!</b>\n\n" | |
# Handle string input (simple summary) | |
if isinstance(plan_summary, str): | |
message += f"📅 <b>आपकी वार्षिक खेती योजना तैयार है!</b>\n\n" | |
message += plan_summary + "\n\n" | |
message += "👉 पूरी विस्तृत योजना देखने के लिए अपने डैशबोर्ड पर जाएं।\n" | |
message += "📊 यह AI तकनीक से बनाई गई विस्तृत योजना है।\n" | |
message += "<i>- कृषि मित्र बॉट</i>" | |
return message | |
# Handle dict input (comprehensive plan) | |
year = plan_summary.get('year', 'Current Year') | |
message += f"📅 <b>आपकी वार्षिक खेती योजना ({year}) तैयार है!</b>\n\n" | |
message += "🤖 <b>AI तकनीक से बनाई गई विस्तृत योजना</b>\n\n" | |
farms = plan_summary.get('farms', []) | |
for f in farms: | |
farm_name = f.get('farm_name', 'Unknown Farm') | |
message += f"🏡 <b>{farm_name}</b>\n" | |
# Check if AI generated | |
if f.get('ai_generated', False): | |
message += "✅ AI द्वारा विश्लेषण किया गया\n" | |
message += "📋 मिली है विस्तृत जानकारी:\n" | |
message += "• मासिक गतिविधि कैलेंडर\n" | |
message += "• फसल-वार रणनीति\n" | |
message += "• वित्तीय अनुमान\n" | |
message += "• मिट्टी प्रबंधन योजना\n" | |
message += "• जोखिम प्रबंधन\n\n" | |
message += "� <b>मुख्य विशेषताएं:</b>\n" | |
message += "📊 विस्तृत तालिकाओं के साथ\n" | |
message += "📈 वित्तीय अनुमान के साथ\n" | |
message += "🌡️ मौसम आधारित सुझाव\n" | |
message += "🧪 मिट्टी परीक्षण आधारित सलाह\n\n" | |
message += "👆 <b>पूरी योजना देखने के लिए:</b>\n" | |
message += "1️⃣ अपने डैशबोर्ड पर जाएं\n" | |
message += "2️⃣ अपने खेत के पास 'View Plan' पर क्लिक करें\n" | |
message += "3️⃣ विस्तृत HTML रिपोर्ट देखें\n\n" | |
message += "🌟 <b>यह AI द्वारा बनाई गई पेशेवर खेती योजना है</b>\n" | |
message += "<i>- कृषि मित्र बॉट</i>" | |
return message | |
def format_complete_yearly_plan_telegram(farmer_name: str, farm_name: str, plan_data: dict) -> list: | |
""" | |
Format complete yearly plan for Telegram (split into multiple messages due to length) | |
Args: | |
farmer_name: Name of the farmer | |
farm_name: Name of the farm | |
plan_data: Complete plan data from gemini service | |
Returns: | |
List of formatted messages for Telegram | |
""" | |
messages = [] | |
# Message 1: Header and Farm Overview | |
msg1 = f"🌾 <b>Comprehensive Yearly Farming Plan 2025</b>\n" | |
msg1 += f"👨🌾 <b>Farmer:</b> {farmer_name}\n" | |
msg1 += f"🏡 <b>Farm:</b> {farm_name}\n\n" | |
# Extract farm overview from plan data | |
farms = plan_data.get('farms', []) | |
if farms: | |
farm = farms[0] # Get first farm | |
if 'plan' in farm: | |
# Try to extract key information from HTML plan | |
plan_content = farm['plan'] | |
# Extract farm size, crops, etc. from plan content | |
msg1 += "📊 <b>Farm Overview:</b>\n" | |
if 'farm_size' in str(plan_content).lower(): | |
msg1 += "• Farm details included in comprehensive plan\n" | |
if 'crop' in str(plan_content).lower(): | |
msg1 += "• Crop-wise strategies provided\n" | |
if 'month' in str(plan_content).lower(): | |
msg1 += "• Monthly calendar included\n" | |
msg1 += "\n" | |
msg1 += "📋 <b>Plan Components:</b>\n" | |
msg1 += "✅ Monthly Farming Calendar\n" | |
msg1 += "✅ Crop-wise Annual Strategy\n" | |
msg1 += "✅ Financial Projections\n" | |
msg1 += "✅ Soil Management Plan\n" | |
msg1 += "✅ Risk Management\n" | |
msg1 += "✅ Seasonal Recommendations\n\n" | |
messages.append(msg1) | |
# Message 2: Key Monthly Activities (simplified) | |
msg2 = "📅 <b>Key Monthly Activities Summary:</b>\n\n" | |
monthly_activities = { | |
"जनवरी (January)": "• रबी फसल की देखभाल\n• गेहूं/सरसों की निराई\n• सब्जियों की बुआई", | |
"फरवरी (February)": "• फसल में सिंचाई\n• उर्वरक का छिड़काव\n• बीज तैयारी", | |
"मार्च (March)": "• रबी फसल की कटाई\n• खरीफ की तैयारी\n• मिट्टी की जांच", | |
"अप्रैल (April)": "• खेत की जुताई\n• कम्पोस्ट तैयारी\n• सिंचाई व्यवस्था", | |
"मई (May)": "• खरीफ बीज तैयारी\n• मिट्टी उपचार\n• पानी की व्यवस्था", | |
"जून (June)": "• खरीफ बुआई\n• धान/मक्का रोपाई\n• कीट नियंत्रण" | |
} | |
for month, activities in list(monthly_activities.items())[:6]: | |
msg2 += f"<b>{month}:</b>\n{activities}\n\n" | |
messages.append(msg2) | |
# Message 3: Remaining months and recommendations | |
msg3 = "📅 <b>Remaining Months & Key Tips:</b>\n\n" | |
remaining_months = { | |
"जुलाई (July)": "• खरीफ फसल देखभाल\n• कीट-रोग नियंत्रण\n• निराई-गुड़ाई", | |
"अगस्त (August)": "• उर्वरक प्रबंधन\n• सिंचाई नियमित\n• फसल निगरानी", | |
"सितंबर (September)": "• फसल सुरक्षा\n• कटाई तैयारी\n• बाजार जानकारी", | |
"अक्टूबर (October)": "• खरीफ कटाई\n• रबी तैयारी\n• भंडारण व्यवस्था", | |
"नवंबर (November)": "• रबी बुआई\n• गेहूं-सरसों रोपाई\n• खाद प्रबंधन", | |
"दिसंबर (December)": "• फसल देखभाल\n• सिंचाई व्यवस्था\n• अगले वर्ष योजना" | |
} | |
for month, activities in remaining_months.items(): | |
msg3 += f"<b>{month}:</b>\n{activities}\n\n" | |
msg3 += "💡 <b>Important Tips:</b>\n" | |
msg3 += "• मौसम अपडेट नियमित देखें\n" | |
msg3 += "• मिट्टी परीक्षण साल में दो बार\n" | |
msg3 += "• कीट-रोग की निगरानी रखें\n" | |
msg3 += "• बाजार भाव की जानकारी रखें\n\n" | |
messages.append(msg3) | |
# Message 4: Financial and final recommendations | |
msg4 = "💰 <b>Financial Planning & Final Notes:</b>\n\n" | |
msg4 += "<b>वित्तीय योजना:</b>\n" | |
msg4 += "• बीज, खाद की लागत का बजट\n" | |
msg4 += "• अपेक्षित उत्पादन की गणना\n" | |
msg4 += "• बाजार मूल्य का अनुमान\n" | |
msg4 += "• मुनाफे का अनुमान\n\n" | |
msg4 += "<b>जोखिम प्रबंधन:</b>\n" | |
msg4 += "• फसल बीमा कराएं\n" | |
msg4 += "• मौसम आधारित बचाव\n" | |
msg4 += "• वैकल्पिक फसल योजना\n" | |
msg4 += "• पानी संरक्षण करें\n\n" | |
msg4 += "📄 <b>Complete detailed plan with tables and charts is available in the PDF file sent above.</b>\n\n" | |
msg4 += "🌟 <b>यह AI-powered comprehensive farming plan है जो आपकी मिट्टी, मौसम, और फसल के डेटा के आधार पर बनाई गई है।</b>\n\n" | |
msg4 += "<i>🤖 Generated by Agricultural AI Assistant</i>" | |
messages.append(msg4) | |
return messages | |
def extract_plan_content_from_html(html_content: str) -> dict: | |
""" | |
Extract key information from HTML plan content for Telegram formatting | |
Args: | |
html_content: HTML content of the yearly plan | |
Returns: | |
Dictionary with extracted content | |
""" | |
try: | |
# Remove HTML tags and extract text content | |
import re | |
# Clean HTML content | |
text_content = re.sub(r'<[^>]+>', '', html_content) | |
text_content = re.sub(r'\s+', ' ', text_content).strip() | |
extracted = { | |
'farm_overview': '', | |
'monthly_activities': '', | |
'crop_strategy': '', | |
'financial_info': '', | |
'soil_management': '' | |
} | |
# Look for specific sections | |
if 'Farm Overview' in text_content: | |
try: | |
start = text_content.find('Farm Overview') | |
end = text_content.find('Monthly Farming Calendar', start) if 'Monthly Farming Calendar' in text_content else start + 500 | |
extracted['farm_overview'] = text_content[start:end][:500] + "..." | |
except: | |
pass | |
if 'Monthly Farming Calendar' in text_content: | |
try: | |
start = text_content.find('Monthly Farming Calendar') | |
end = text_content.find('Crop-wise Annual Strategy', start) if 'Crop-wise Annual Strategy' in text_content else start + 800 | |
extracted['monthly_activities'] = text_content[start:end][:800] + "..." | |
except: | |
pass | |
if 'Crop-wise Annual Strategy' in text_content: | |
try: | |
start = text_content.find('Crop-wise Annual Strategy') | |
end = text_content.find('Financial', start) if 'Financial' in text_content else start + 600 | |
extracted['crop_strategy'] = text_content[start:end][:600] + "..." | |
except: | |
pass | |
return extracted | |
except Exception as e: | |
logger.error(f"Error extracting plan content from HTML: {str(e)}") | |
return {} | |
def format_enhanced_yearly_plan_telegram(farmer_name: str, farm_name: str, plan_data: dict) -> list: | |
""" | |
Enhanced yearly plan formatting that tries to extract actual plan content | |
Args: | |
farmer_name: Name of the farmer | |
farm_name: Name of the farm | |
plan_data: Complete plan data from gemini service | |
Returns: | |
List of formatted messages for Telegram | |
""" | |
messages = [] | |
# Try to extract actual content from the plan | |
actual_content = {} | |
farms = plan_data.get('farms', []) | |
if farms and 'plan' in farms[0]: | |
actual_content = extract_plan_content_from_html(farms[0]['plan']) | |
# Message 1: Header and Farm Overview | |
msg1 = f"🌾 <b>Complete Yearly Farming Plan 2025</b>\n" | |
msg1 += f"👨🌾 <b>Farmer:</b> {farmer_name}\n" | |
msg1 += f"🏡 <b>Farm:</b> {farm_name}\n\n" | |
if actual_content.get('farm_overview'): | |
msg1 += "📊 <b>Farm Overview (From Your AI Plan):</b>\n" | |
msg1 += f"{actual_content['farm_overview'][:500]}...\n\n" | |
else: | |
msg1 += "📊 <b>Plan Components Generated:</b>\n" | |
msg1 += "✅ Detailed Farm Analysis\n" | |
msg1 += "✅ Soil & Weather Assessment\n" | |
msg1 += "✅ Crop Performance Metrics\n" | |
msg1 += "✅ Monthly Activity Schedule\n\n" | |
messages.append(msg1) | |
# Message 2: Monthly Activities | |
msg2 = "📅 <b>Monthly Farming Calendar:</b>\n\n" | |
if actual_content.get('monthly_activities'): | |
msg2 += f"{actual_content['monthly_activities'][:800]}...\n\n" | |
else: | |
# Provide comprehensive monthly guide | |
msg2 += "<b>🌱 Kharif Season (June-October):</b>\n" | |
msg2 += "• June: Land preparation, seed sowing\n" | |
msg2 += "• July: Transplanting, weed management\n" | |
msg2 += "• August: Fertilizer application, pest control\n" | |
msg2 += "• September: Crop monitoring, irrigation\n" | |
msg2 += "• October: Harvesting preparation\n\n" | |
msg2 += "<b>❄️ Rabi Season (November-April):</b>\n" | |
msg2 += "• November: Rabi crop sowing\n" | |
msg2 += "• December: Crop establishment, irrigation\n" | |
msg2 += "• January: Mid-season care, fertilizing\n" | |
msg2 += "• February: Disease monitoring, spraying\n" | |
msg2 += "• March: Pre-harvest activities\n" | |
msg2 += "• April: Harvesting, storage\n\n" | |
messages.append(msg2) | |
# Message 3: Crop Strategy and Financial Info | |
msg3 = "🌾 <b>Crop Strategy & Financial Planning:</b>\n\n" | |
if actual_content.get('crop_strategy'): | |
msg3 += f"<b>Crop-wise Strategy:</b>\n{actual_content['crop_strategy'][:600]}...\n\n" | |
else: | |
msg3 += "<b>Strategic Recommendations:</b>\n" | |
msg3 += "• Diversified cropping for risk reduction\n" | |
msg3 += "• Optimal seed varieties for your soil\n" | |
msg3 += "• Integrated pest management\n" | |
msg3 += "• Water-efficient irrigation scheduling\n" | |
msg3 += "• Market-oriented crop selection\n\n" | |
msg3 += "💰 <b>Financial Planning:</b>\n" | |
msg3 += "• Input cost optimization\n" | |
msg3 += "• Expected yield calculations\n" | |
msg3 += "• Revenue projections per crop\n" | |
msg3 += "• Profit margin analysis\n" | |
msg3 += "• Cash flow management\n\n" | |
messages.append(msg3) | |
# Message 4: Final recommendations and notes | |
msg4 = "🎯 <b>Key Implementation Tips:</b>\n\n" | |
msg4 += "<b>🔬 Soil Management:</b>\n" | |
msg4 += "• Regular soil testing (twice yearly)\n" | |
msg4 += "• Organic matter enhancement\n" | |
msg4 += "• Balanced fertilizer application\n" | |
msg4 += "• pH level monitoring\n\n" | |
msg4 += "<b>⚠️ Risk Management:</b>\n" | |
msg4 += "• Weather-based crop insurance\n" | |
msg4 += "• Diversified income sources\n" | |
msg4 += "• Emergency fund planning\n" | |
msg4 += "• Alternative crop options\n\n" | |
msg4 += "📄 <b>The complete detailed plan with tables, charts, and specific calculations is available in the PDF sent above.</b>\n\n" | |
msg4 += "🤖 <b>This AI-generated plan is based on your specific soil data, weather patterns, and crop requirements.</b>\n\n" | |
msg4 += "<i>📱 For any questions, contact our agricultural support team!</i>" | |
messages.append(msg4) | |
return messages | |