import os import csv import io import json import shutil import psutil from datetime import datetime from flask import Flask, render_template, request, redirect, url_for, session, flash, jsonify, send_file, Response from modules.config import ConfigManager, DEFAULT_CONFIG from modules.logger import logger from modules.database import get_db, init_db, query_db, execute_db from modules.auth import hash_password, verify_password, log_login_attempt, is_ip_blocked, generate_csrf_token, csrf_protect, login_required, sanitize_input from modules.telegram import TelegramManager, SESSION_DIR # Initializing Flask app app = Flask(__name__, template_folder='../templates', static_folder='../static') app.secret_key = os.urandom(24) # Random session key # Ensure CSRF token is injected in all template context @app.context_processor def inject_csrf_token(): return dict(csrf_token=generate_csrf_token()) @app.context_processor def inject_site_config(): config = ConfigManager() return dict( site_name=config.get('site_name', 'TG Auto Reply Pro'), theme=config.get('theme', 'dark'), primary_color=config.get('primary_color', '#0d6efd'), footer_text=config.get('footer_text', '© 2026 TG Auto Reply Pro. All Rights Reserved.'), installed=config.get('installed', False) ) # --- Installer Router --- @app.route('/install', methods=['GET', 'POST']) def install(): config = ConfigManager() if config.get('installed', False): # Already installed, redirect to dashboard or login return redirect(url_for('login')) step = request.args.get('step', '1') if step == '1': # Step 1: System Checks checks = { "python": True, # Running python 3 "write_config": os.access(os.path.dirname(config.get('favicon_path') or './'), os.W_OK) or True, "write_sessions": os.access(SESSION_DIR, os.W_OK) or True, "sqlite": True } try: import sqlite3 except ImportError: checks["sqlite"] = False system_ready = all(checks.values()) return render_template('install.html', step=1, checks=checks, system_ready=system_ready) elif step == '2': # Step 2: Telegram Credentials Setup if request.method == 'POST': api_id = request.form.get('api_id', '').strip() api_hash = request.form.get('api_hash', '').strip() phone = request.form.get('phone', '').strip() if not api_id or not api_hash or not phone: flash("All fields are required.", "danger") return render_template('install.html', step=2) try: tg = TelegramManager() # Connect client and request code tg.send_otp(api_id, api_hash, phone) # Store credentials temporarily in session session['temp_api_id'] = api_id session['temp_api_hash'] = api_hash session['temp_phone'] = phone return redirect(url_for('install', step='2_otp')) except Exception as e: flash(f"Telegram connection error: {e}", "danger") return render_template('install.html', step=2) return render_template('install.html', step=2) elif step == '2_otp': # Step 2 OTP Verification if 'temp_phone' not in session: return redirect(url_for('install', step='2')) if request.method == 'POST': otp_code = request.form.get('otp', '').strip() if not otp_code: flash("Please enter the OTP code.", "danger") return render_template('install.html', step='2_otp') tg = TelegramManager() res = tg.verify_otp(otp_code) if res.get('status') == 'success': # OTP authenticated successfully session['temp_auth_complete'] = True return redirect(url_for('install', step='3')) elif res.get('status') == '2fa_required': # 2FA password is required return redirect(url_for('install', step='2_2fa')) else: flash(f"Verification failed: {res.get('message', 'Invalid Code')}", "danger") return render_template('install.html', step='2_otp') return render_template('install.html', step='2_otp') elif step == '2_2fa': # Step 2 2FA verification if 'temp_phone' not in session: return redirect(url_for('install', step='2')) if request.method == 'POST': password_2fa = request.form.get('password_2fa', '').strip() if not password_2fa: flash("Please enter your 2FA password.", "danger") return render_template('install.html', step='2_2fa') tg = TelegramManager() res = tg.verify_2fa(password_2fa) if res.get('status') == 'success': session['temp_auth_complete'] = True return redirect(url_for('install', step='3')) else: flash(f"2FA Authentication failed: {res.get('message', 'Incorrect Password')}", "danger") return render_template('install.html', step='2_2fa') return render_template('install.html', step='2_2fa') elif step == '3': # Step 3: Admin Configuration if not session.get('temp_auth_complete'): return redirect(url_for('install', step='2')) if request.method == 'POST': username = request.form.get('username', '').strip() password = request.form.get('password', '').strip() email = request.form.get('email', '').strip() timezone = request.form.get('timezone', 'Asia/Dhaka').strip() if not username or not password or not email: flash("All fields are required.", "danger") return render_template('install.html', step=3) # Initialize SQLite DB try: init_db() except Exception as e: flash(f"Database initialization failed: {e}", "danger") return render_template('install.html', step=3) # Create Admin Account pass_hash = hash_password(password) try: # Clear existing admins just in case execute_db("DELETE FROM admin") execute_db( "INSERT INTO admin (username, password_hash, email, timezone) VALUES (?, ?, ?, ?)", (username, pass_hash, email, timezone) ) except Exception as e: flash(f"Could not create admin account: {e}", "danger") return render_template('install.html', step=3) # Store timezone and complete installer session['temp_admin_tz'] = timezone return redirect(url_for('install', step='4')) return render_template('install.html', step=3) elif step == '4': # Step 4: Finish setup if 'temp_phone' not in session or 'temp_admin_tz' not in session: return redirect(url_for('install', step='1')) # Write config.json config_data = { "tg_api_id": session.get('temp_api_id'), "tg_api_hash": session.get('temp_api_hash'), "tg_phone": session.get('temp_phone'), "timezone": session.get('temp_admin_tz'), "installed": True } config.update(config_data) # Clear temporary installer sessions session.pop('temp_api_id', None) session.pop('temp_api_hash', None) session.pop('temp_phone', None) session.pop('temp_auth_complete', None) session.pop('temp_admin_tz', None) # Start Telegram bot tg = TelegramManager() tg.start_bot() flash("TG Auto Reply Pro successfully installed!", "success") return render_template('install.html', step=4) return redirect(url_for('install', step='1')) # --- Login and Admin Auth --- @app.route('/login', methods=['GET', 'POST']) @csrf_protect def login(): config = ConfigManager() if not config.get('installed', False): return redirect(url_for('install')) if session.get('logged_in'): return redirect(url_for('dashboard')) ip = request.remote_addr if is_ip_blocked(ip): return "Your IP is temporarily blocked due to multiple failed login attempts. Try again in 15 minutes.", 403 if request.method == 'POST': username = request.form.get('username', '').strip() password = request.form.get('password', '').strip() # Query database for admin admin = query_db("SELECT * FROM admin WHERE username = ?", (username,), one=True) if admin and verify_password(password, admin['password_hash']): # Login successful session.clear() session['logged_in'] = True session['username'] = admin['username'] session['email'] = admin['email'] session['timezone'] = admin['timezone'] # Log login audit trail log_login_attempt(username, ip, "Success") # Start TG client if not running tg = TelegramManager() if not tg.is_connected(): tg.start_bot() return redirect(url_for('dashboard')) else: # Login failed log_login_attempt(username, ip, "Failed") flash("Invalid username or password.", "danger") return render_template('login.html') return render_template('login.html') @app.route('/logout') def logout(): session.clear() flash("You have been logged out.", "success") return redirect(url_for('login')) # --- Dashboard View --- @app.route('/') @app.route('/dashboard') @login_required def dashboard(): config = ConfigManager() tg = TelegramManager() # 1. Bot & Client Status bot_connected = tg.is_connected() # 2. SQLite Statistics today_str = datetime.now().strftime('%Y-%m-%d') try: replies_today = query_db( "SELECT COUNT(*) as count FROM autoreply_logs WHERE status = 'Replied' AND timestamp LIKE ?", (f"{today_str}%",), one=True )['count'] messages_today = query_db( "SELECT COUNT(*) as count FROM autoreply_logs WHERE timestamp LIKE ?", (f"{today_str}%",), one=True )['count'] total_replies = query_db( "SELECT COUNT(*) as count FROM autoreply_logs WHERE status = 'Replied'", one=True )['count'] total_users = query_db( "SELECT COUNT(DISTINCT user_id) as count FROM autoreply_logs", one=True )['count'] recent_logs = query_db( "SELECT * FROM autoreply_logs ORDER BY timestamp DESC LIMIT 5" ) except Exception as e: logger.error(f"Error fetching dashboard statistics: {e}") replies_today = messages_today = total_replies = total_users = 0 recent_logs = [] # 3. System Usage Metrics cpu_usage = psutil.cpu_percent() ram_usage = psutil.virtual_memory().percent # Storage Usage total, used, free = shutil.disk_usage(os.path.dirname(os.path.abspath(__file__))) storage_usage = round((used / total) * 100, 1) # System Time & Timezone tz = config.get('timezone', 'Asia/Dhaka') import pytz local_time = datetime.now(pytz.timezone(tz)).strftime('%Y-%m-%d %I:%M:%S %p') # Profile info if connected profile = tg.get_me() if bot_connected else None return render_template( 'dashboard.html', bot_connected=bot_connected, replies_today=replies_today, messages_today=messages_today, total_replies=total_replies, total_users=total_users, cpu_usage=cpu_usage, ram_usage=ram_usage, storage_usage=storage_usage, local_time=local_time, timezone=tz, recent_logs=recent_logs, profile=profile ) # API Endpoint to fetch dynamic statistics for chart @app.route('/api/stats') @login_required def api_stats(): # Return count of replies for the last 7 days dates = [] replies = [] ignores = [] for i in range(6, -1, -1): day = (datetime.now() - timedelta(days=i)).strftime('%Y-%m-%d') dates.append(day) rep_count = query_db( "SELECT COUNT(*) as count FROM autoreply_logs WHERE status = 'Replied' AND timestamp LIKE ?", (f"{day}%",), one=True )['count'] replies.append(rep_count) ign_count = query_db( "SELECT COUNT(*) as count FROM autoreply_logs WHERE status LIKE 'Ignored%' AND timestamp LIKE ?", (f"{day}%",), one=True )['count'] ignores.append(ign_count) return jsonify({ "labels": dates, "replies": replies, "ignores": ignores }) # --- Settings View --- @app.route('/settings', methods=['GET', 'POST']) @csrf_protect @login_required def settings(): config = ConfigManager() tg = TelegramManager() if request.method == 'POST': action = request.form.get('action') if action == 'general': # Save General Auto Reply Settings enabled = 'enabled' in request.form reply_once = 'reply_once' in request.form config.update({ "enabled": enabled, "reply_once": reply_once, "reply_after_hours": int(request.form.get('reply_after_hours', 24)), "typing_delay": int(request.form.get('typing_delay', 3)), "random_delay_min": int(request.form.get('random_delay_min', 1)), "random_delay_max": int(request.form.get('random_delay_max', 3)), "reply_message": request.form.get('reply_message', ''), "timezone": request.form.get('timezone', 'Asia/Dhaka') }) # If enabled toggle has changed, sync Telegram client status if enabled: if not tg.is_connected(): tg.start_bot() else: tg.stop_bot() flash("General settings updated successfully.", "success") elif action == 'schedule_add': # Add off-time schedule start_time = request.form.get('start_time') end_time = request.form.get('end_time') timezone = request.form.get('timezone', 'Asia/Dhaka') days = request.form.getlist('days') if not start_time or not end_time or not days: flash("All schedule parameters are required.", "danger") else: days_str = ",".join(days) try: execute_db( "INSERT INTO schedules (start_time, end_time, timezone, days, enabled) VALUES (?, ?, ?, ?, 1)", (start_time, end_time, timezone, days_str) ) flash("Off-time schedule added successfully.", "success") except Exception as e: flash(f"Error saving schedule: {e}", "danger") elif action == 'schedule_delete': # Delete schedule sid = request.form.get('schedule_id') execute_db("DELETE FROM schedules WHERE id = ?", (sid,)) flash("Schedule removed.", "success") elif action == 'vip_add': # Add to VIP ignore list list_type = request.form.get('list_type') identifier = request.form.get('identifier', '').strip() note = request.form.get('note', '').strip() if not identifier or list_type not in ['whitelist', 'blacklist']: flash("Invalid VIP inputs.", "danger") else: try: execute_db( "INSERT INTO vip_list (list_type, identifier, note) VALUES (?, ?, ?)", (list_type, identifier, note) ) flash(f"Identifier added to {list_type} successfully.", "success") except Exception as e: flash(f"Error adding VIP: {e}. Identifier may already exist.", "danger") elif action == 'vip_delete': vid = request.form.get('vip_id') execute_db("DELETE FROM vip_list WHERE id = ?", (vid,)) flash("VIP list entry removed.", "success") elif action == 'keyword_add': # Add keyword to ignore list keyword = request.form.get('keyword', '').strip() if not keyword: flash("Keyword cannot be empty.", "danger") else: try: execute_db("INSERT INTO keywords_ignore (keyword) VALUES (?)", (keyword,)) flash("Ignore keyword added.", "success") except Exception as e: flash("Keyword already exists in the database.", "danger") elif action == 'keyword_delete': kid = request.form.get('keyword_id') execute_db("DELETE FROM keywords_ignore WHERE id = ?", (kid,)) flash("Ignore keyword removed.", "success") elif action == 'forwarding': # Save forwarding settings forward_enabled = 'forward_enabled' in request.form config.update({ "forward_enabled": forward_enabled, "forward_target": request.form.get('forward_target', 'saved_messages').strip() }) flash("Forwarding configurations updated.", "success") elif action == 'notification': # Save notifications SMTP parameters email_enabled = 'email_notification_enabled' in request.form desktop_enabled = 'desktop_notification_enabled' in request.form config.update({ "email_notification_enabled": email_enabled, "desktop_notification_enabled": desktop_enabled, "smtp_server": request.form.get('smtp_server', '').strip(), "smtp_port": int(request.form.get('smtp_port', 587)), "smtp_user": request.form.get('smtp_user', '').strip(), "smtp_pass": request.form.get('smtp_pass', ''), "smtp_recipient": request.form.get('smtp_recipient', '').strip() }) flash("Notification settings saved successfully.", "success") elif action == 'appearance': # Save site details & design theme tokens config.update({ "site_name": sanitize_input(request.form.get('site_name', 'TG Auto Reply Pro')), "theme": request.form.get('theme', 'dark'), "primary_color": request.form.get('primary_color', '#0d6efd'), "footer_text": sanitize_input(request.form.get('footer_text', '')) }) flash("Site appearance configurations updated successfully.", "success") elif action == 'telegram_logout': # Log out client and delete local sessions tg.logout() flash("Telegram session closed. Please run the installer or setup the credentials again.", "warning") return redirect(url_for('install', step='2')) elif action == 'change_password': # Update web panel credentials old_pass = request.form.get('old_password', '') new_pass = request.form.get('new_password', '') admin = query_db("SELECT * FROM admin WHERE username = ?", (session.get('username'),), one=True) if admin and verify_password(old_pass, admin['password_hash']): new_hash = hash_password(new_pass) execute_db("UPDATE admin SET password_hash = ? WHERE id = ?", (new_hash, admin['id'])) flash("Admin password successfully updated.", "success") else: flash("Incorrect old password.", "danger") return redirect(url_for('settings')) # Load dynamic configurations for render schedules = query_db("SELECT * FROM schedules") vip_blacklist = query_db("SELECT * FROM vip_list WHERE list_type = 'blacklist'") vip_whitelist = query_db("SELECT * FROM vip_list WHERE list_type = 'whitelist'") keywords = query_db("SELECT * FROM keywords_ignore") # Client login status bot_connected = tg.is_connected() return render_template( 'settings.html', config=config.get_all(), schedules=schedules, vip_blacklist=vip_blacklist, vip_whitelist=vip_whitelist, keywords=keywords, bot_connected=bot_connected ) # --- Logs Router --- @app.route('/logs') @login_required def logs(): status_filter = request.args.get('status', 'all') search_query = request.args.get('search', '').strip() page = int(request.args.get('page', 1)) per_page = 20 offset = (page - 1) * per_page # Base query construction query_parts = ["SELECT * FROM autoreply_logs"] count_parts = ["SELECT COUNT(*) as count FROM autoreply_logs"] params = [] where_clauses = [] if status_filter != 'all': where_clauses.append("status = ?") params.append(status_filter) if search_query: where_clauses.append("(username LIKE ? OR full_name LIKE ? OR original_message LIKE ? OR user_id LIKE ?)") search_param = f"%{search_query}%" params.extend([search_param, search_param, search_param, search_param]) if where_clauses: clause_str = " WHERE " + " AND ".join(where_clauses) query_parts.append(clause_str) count_parts.append(clause_str) query_parts.append(" ORDER BY timestamp DESC LIMIT ? OFFSET ?") # Calculate totals try: total_rows = query_db(" ".join(count_parts), params, one=True)['count'] params.extend([per_page, offset]) logs_list = query_db(" ".join(query_parts), params) except Exception as e: logger.error(f"Error querying logs: {e}") total_rows = 0 logs_list = [] total_pages = max(1, (total_rows + per_page - 1) // per_page) # Get distinct statuses for filter selector statuses = ["Replied", "Ignored: Blacklisted", "Ignored: Non-Whitelisted", "Ignored: Keyword", "Ignored: Reply Once Rule", "Error"] return render_template( 'logs.html', logs=logs_list, status_filter=status_filter, search_query=search_query, page=page, total_pages=total_pages, total_rows=total_rows, statuses=statuses ) # --- CSV & JSON Exports --- @app.route('/logs/export/csv') @login_required def export_logs_csv(): try: logs_list = query_db("SELECT timestamp, user_id, username, full_name, original_message, reply_message, status FROM autoreply_logs ORDER BY timestamp DESC") except Exception: logs_list = [] output = io.StringIO() writer = csv.writer(output) # Write header writer.writerow(['Timestamp', 'User ID', 'Username', 'Full Name', 'Original Message', 'Reply Message', 'Status']) for row in logs_list: writer.writerow([row['timestamp'], row['user_id'], row['username'], row['full_name'], row['original_message'], row['reply_message'], row['status']]) output.seek(0) return Response( output.getvalue(), mimetype="text/csv", headers={"Content-disposition": "attachment; filename=autoreply_logs.csv"} ) @app.route('/logs/export/json') @login_required def export_logs_json(): try: logs_list = query_db("SELECT timestamp, user_id, username, full_name, original_message, reply_message, status FROM autoreply_logs ORDER BY timestamp DESC") logs_data = [dict(row) for row in logs_list] except Exception: logs_data = [] return jsonify(logs_data) # --- Backup Center --- @app.route('/backup') @login_required def backup(): return render_template('backup.html') @app.route('/backup/download/') @login_required def download_backup(file_type): from modules.database import DB_FILE from modules.config import CONFIG_FILE if file_type == 'database': if os.path.exists(DB_FILE): return send_file(DB_FILE, as_attachment=True, download_name="database.db") else: flash("Database file not found.", "danger") elif file_type == 'config': if os.path.exists(CONFIG_FILE): return send_file(CONFIG_FILE, as_attachment=True, download_name="config.json") else: flash("Config file not found.", "danger") elif file_type == 'session': session_file_path = f"{SESSION_FILE}.session" if os.path.exists(session_file_path): return send_file(session_file_path, as_attachment=True, download_name="telegram_session.session") else: flash("Telegram session file not found. Have you logged in?", "danger") return redirect(url_for('backup'))