customized_farm_planner / telegram_service.py
pranit144's picture
Upload 56 files
429a26d verified
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