help@rskworld.in +91 93305 39277
RSK World
  • Home
  • Development
    • Web Development
    • Mobile Apps
    • Software
    • Games
    • Project
  • Technologies
    • Data Science
    • AI Development
    • Cloud Development
    • Blockchain
    • Cyber Security
    • Dev Tools
    • Testing Tools
  • Blog
  • About
  • Contact

Theme Settings

Color Scheme
Display Options
Font Size
100%
Back to Project
RSK World
news-summary-bot
RSK World
news-summary-bot
News Summary Bot - Python + Flask + OpenAI + NewsAPI + AI Summarization + Real-time News + News Aggregation
news-summary-bot
  • __pycache__
  • static
  • templates
  • .env459 B
  • .gitignore761 B
  • GITHUB_RELEASE_SUMMARY.md3.7 KB
  • INSTALLATION.md2.7 KB
  • PROJECT_SUMMARY.md9.5 KB
  • README.md11.4 KB
  • RELEASE_NOTES_v1.0.0.md6.7 KB
  • admin.py7.6 KB
  • analytics.py11 KB
  • app.py14.1 KB
  • auth.py15 KB
  • cache.py11.1 KB
  • export.py14.2 KB
  • news_bot.py4.7 KB
  • requirements.txt286 B
  • search.py16.8 KB
  • security.py14.4 KB
app.pyauth.py
app.py
Raw Download
Find: Go to:
"""
Main Flask Application for News Summary Bot
Developer: Molla Samser
Design & Testing: Rima Khatun
Company: RSK World
Year: 2026
Website: https://rskworld.in
"""

from flask import Flask, request, jsonify, render_template, session, redirect, url_for, flash, Response
from flask_cors import CORS
from datetime import datetime
from news_bot import NewsBot
from analytics import NewsAnalytics, AdvancedNLP
from cache import cache_manager, news_cache
from auth import auth_manager, user_preferences, login_required
from search import advanced_search, SearchFilters
from security import rate_limit, security_headers
from export import data_exporter
import os
from dotenv import load_dotenv
import hashlib

load_dotenv()

app = Flask(__name__)
app.secret_key = os.getenv('SECRET_KEY', 'your-secret-key-here')
CORS(app)

# Register security headers middleware
@app.after_request
def apply_security_headers(response):
    return security_headers.add_security_headers(response)

bot = NewsBot()
analytics = NewsAnalytics()
nlp = AdvancedNLP()

# Register admin blueprint
from admin import admin_bp
app.register_blueprint(admin_bp)

@app.route('/')
def home():
    return render_template('index.html')

@app.route('/demo')
def demo():
    return render_template('demo.html')

@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        username = request.form.get('username')
        password = request.form.get('password')
        
        result = auth_manager.authenticate_user(username, password)
        if result['success']:
            # Create session
            session_token = auth_manager.create_session(
                result['user_id'], 
                request.remote_addr,
                request.headers.get('User-Agent')
            )
            session['user_token'] = session_token
            session['user_id'] = result['user_id']
            session['username'] = result['username']
            
            flash('Login successful!', 'success')
            return redirect(url_for('demo'))
        else:
            flash(result['error'], 'error')
    
    return render_template('login.html')

@app.route('/register', methods=['GET', 'POST'])
def register():
    if request.method == 'POST':
        username = request.form.get('username')
        email = request.form.get('email')
        password = request.form.get('password')
        confirm_password = request.form.get('confirm_password')
        
        if password != confirm_password:
            flash('Passwords do not match!', 'error')
            return render_template('register.html')
        
        result = auth_manager.register_user(username, email, password)
        if result['success']:
            flash('Registration successful! Please log in.', 'success')
            return redirect(url_for('login'))
        else:
            flash(result['error'], 'error')
    
    return render_template('register.html')

@app.route('/logout')
def logout():
    if 'user_token' in session:
        auth_manager.invalidate_session(session['user_token'])
        session.clear()
        flash('Logged out successfully!', 'info')
    return redirect(url_for('home'))

@app.route('/api/news', methods=['GET'])
@rate_limit(limit=100, window=3600)
def get_news():
    category = request.args.get('category', 'general')
    query = request.args.get('q')
    country = request.args.get('country', 'us')
    
    # Check cache first
    cache_key_data = news_cache.get_news(category, query, country)
    if cache_key_data:
        return jsonify(cache_key_data)
    
    # Fetch fresh data
    news_data = bot.fetch_news(category=category, query=query)
    
    # Cache the results
    if 'articles' in news_data:
        news_cache.set_news(category, query, country, news_data)
        
        # Index articles for search
        for article in news_data['articles'][:10]:  # Limit to avoid overloading
            article_data = {
                'article_id': str(article.get('publishedAt', '')) + str(hash(article.get('title', ''))),
                'title': article.get('title', ''),
                'content': article.get('description', ''),
                'category': category,
                'source': article.get('source', {}).get('name', ''),
                'author': article.get('author', ''),
                'published_at': article.get('publishedAt', ''),
                'word_count': len(article.get('description', '').split())
            }
            advanced_search.index_article(article_data)
    
    return jsonify(news_data)

@app.route('/api/summarize', methods=['POST'])
@rate_limit(limit=50, window=3600)
def summarize():
    data = request.get_json()
    content = data.get('content')
    language = data.get('language', 'English')
    
    if not content:
        return jsonify({"error": "No content provided for summarization"}), 400
    
    # Generate content hash for caching
    content_hash = hashlib.md5(content.encode()).hexdigest()
    
    # Check cache first
    cached_summary = news_cache.get_summary(content_hash, language)
    if cached_summary:
        return jsonify({"summary": cached_summary, "cached": True})
    
    # Generate summary
    summary = bot.summarize_article(content, language=language)
    
    # Cache the summary
    news_cache.set_summary(content_hash, language, summary)
    
    # Track user activity if logged in
    if 'user_id' in session:
        analytics.track_user_interaction(session['user_id'], 'summarize', content_hash=content_hash)
    
    return jsonify({"summary": summary, "cached": False})

@app.route('/api/analyze', methods=['POST'])
@rate_limit(limit=50, window=3600)
def analyze():
    data = request.get_json()
    content = data.get('content')
    
    if not content:
        return jsonify({"error": "No content provided for analysis"}), 400
    
    # Generate content hash for caching
    content_hash = hashlib.md5(content.encode()).hexdigest()
    
    # Check cache first
    cached_sentiment = news_cache.get_sentiment(content_hash)
    if cached_sentiment:
        return jsonify({"sentiment": cached_sentiment, "cached": True})
    
    # Analyze sentiment
    sentiment = bot.analyze_sentiment(content)
    
    # Cache the result
    news_cache.set_sentiment(content_hash, sentiment)
    
    # Store analytics
    if 'user_id' in session:
        analytics.track_user_interaction(session['user_id'], 'analyze', content_hash=content_hash)
    
    return jsonify({"sentiment": sentiment, "cached": False})

@app.route('/api/reliability', methods=['POST'])
@rate_limit(limit=50, window=3600)
def reliability():
    data = request.get_json()
    content = data.get('content')
    
    if not content:
        return jsonify({"error": "No content provided for reliability analysis"}), 400
    
    score = bot.analyze_reliability(content)
    return jsonify({"score": score})

@app.route('/api/search', methods=['GET'])
@rate_limit(limit=200, window=3600)
def search_articles():
    query = request.args.get('q', '')
    filters = {}
    
    # Parse filters from query parameters
    if request.args.get('category'):
        filters['category'] = request.args.get('category')
    if request.args.get('sentiment'):
        filters['sentiment'] = request.args.get('sentiment')
    if request.args.get('language'):
        filters['language'] = request.args.get('language')
    if request.args.get('min_reliability'):
        filters['min_reliability'] = request.args.get('min_reliability')
    if request.args.get('date_from'):
        filters['date_from'] = request.args.get('date_from')
    if request.args.get('date_to'):
        filters['date_to'] = request.args.get('date_to')
    
    # Validate filters
    filters = SearchFilters.validate_filters(filters)
    
    # Search parameters
    sort_by = request.args.get('sort', 'relevance')
    limit = int(request.args.get('limit', 20))
    offset = int(request.args.get('offset', 0))
    
    # Perform search
    results = advanced_search.search(query, filters, sort_by, limit, offset)
    
    # Track search
    user_id = session.get('user_id') if 'user_id' in session else None
    advanced_search.track_search(user_id, query, filters, results['total_count'])
    
    return jsonify(results)

@app.route('/api/search/suggestions')
def search_suggestions():
    query = request.args.get('q', '')
    limit = int(request.args.get('limit', 10))
    
    suggestions = advanced_search.get_suggestions(query, limit)
    return jsonify({"suggestions": suggestions})

@app.route('/api/search/popular')
def popular_searches():
    limit = int(request.args.get('limit', 10))
    popular = advanced_search.get_popular_searches(limit)
    return jsonify({"popular": popular})

@app.route('/api/user/preferences', methods=['GET', 'POST'])
@login_required
def user_preferences_api():
    user_id = session['user_id']
    
    if request.method == 'POST':
        data = request.get_json()
        for pref_type, pref_value in data.items():
            user_preferences.set_preference(user_id, pref_type, pref_value)
        return jsonify({"success": True})
    
    preferences = user_preferences.get_all_preferences(user_id)
    return jsonify(preferences)

@app.route('/api/user/history')
@login_required
def reading_history():
    user_id = session['user_id']
    limit = int(request.args.get('limit', 50))
    history = user_preferences.get_reading_history(user_id, limit)
    return jsonify(history)

@app.route('/api/user/stats')
@login_required
def user_stats():
    user_id = session['user_id']
    days = int(request.args.get('days', 30))
    stats = user_preferences.get_reading_stats(user_id, days)
    return jsonify(stats)

@app.route('/api/trending')
def trending_topics():
    days = int(request.args.get('days', 7))
    
    # Check cache first
    cached_trending = news_cache.get_trending_topics(days)
    if cached_trending:
        return jsonify({"trending": cached_trending, "cached": True})
    
    # Get fresh trending data
    trending = analytics.analyze_trending_topics(days)
    
    # Cache the results
    news_cache.set_trending_topics(days, trending)
    
    return jsonify({"trending": trending, "cached": False})

@app.route('/api/analytics/overview')
def analytics_overview():
    """Public analytics endpoint."""
    days = int(request.args.get('days', 30))
    
    # Get various analytics
    sentiment_trends = analytics.get_sentiment_trends(days)
    category_analytics = analytics.get_category_analytics()
    search_analytics = advanced_search.get_search_analytics(days)
    
    return jsonify({
        "sentiment_trends": sentiment_trends,
        "category_analytics": category_analytics,
        "search_analytics": search_analytics
    })

@app.route('/api/cache/stats')
def cache_stats():
    """Cache statistics endpoint."""
    stats = cache_manager.get_stats()
    return jsonify(stats)

@app.route('/api/cache/clear', methods=['POST'])
def clear_cache():
    """Clear cache endpoint."""
    # This should be protected in production
    cache_manager.clear()
    return jsonify({"success": True, "message": "Cache cleared successfully"})

@app.route('/api/export/user-data', methods=['GET'])
@login_required
@rate_limit(limit=10, window=3600)
def export_user_data():
    """Export user data in various formats."""
    user_id = session['user_id']
    format_type = request.args.get('format', 'json').lower()
    
    try:
        data = data_exporter.export_user_data(user_id, format_type)
        
        # Set appropriate content type and filename
        content_types = {
            'json': 'application/json',
            'csv': 'text/csv',
            'xml': 'application/xml'
        }
        
        filename = f'user_data_{user_id}_{datetime.now().strftime("%Y%m%d")}.{format_type}'
        
        return Response(
            data,
            mimetype=content_types.get(format_type, 'application/octet-stream'),
            headers={'Content-Disposition': f'attachment; filename={filename}'}
        )
    except Exception as e:
        return jsonify({"error": str(e)}), 500

@app.route('/api/export/analytics', methods=['GET'])
@rate_limit(limit=20, window=3600)
def export_analytics():
    """Export analytics data."""
    days = int(request.args.get('days', 30))
    format_type = request.args.get('format', 'json').lower()
    
    try:
        data = data_exporter.export_analytics_data(days, format_type)
        
        content_types = {
            'json': 'application/json',
            'csv': 'text/csv',
            'xml': 'application/xml'
        }
        
        filename = f'analytics_{days}days_{datetime.now().strftime("%Y%m%d")}.{format_type}'
        
        return Response(
            data,
            mimetype=content_types.get(format_type, 'application/octet-stream'),
            headers={'Content-Disposition': f'attachment; filename={filename}'}
        )
    except Exception as e:
        return jsonify({"error": str(e)}), 500

@app.route('/api/export/backup', methods=['GET'])
@rate_limit(limit=5, window=3600)
def export_backup():
    """Create full system backup."""
    include_user_data = request.args.get('include_users', 'false').lower() == 'true'
    
    try:
        data = data_exporter.create_full_backup(include_user_data=include_user_data)
        filename = f'backup_{datetime.now().strftime("%Y%m%d_%H%M%S")}.zip'
        
        return Response(
            data,
            mimetype='application/zip',
            headers={'Content-Disposition': f'attachment; filename={filename}'}
        )
    except Exception as e:
        return jsonify({"error": str(e)}), 500

if __name__ == '__main__':
    port = int(os.getenv("PORT", 5000))
    debug = os.getenv("DEBUG", "True") == "True"
    
    # Cleanup expired cache entries on startup
    cache_manager.cleanup_expired()
    
    app.run(host='0.0.0.0', port=port, debug=debug)

# Developed by Molla Samser | 2026 | RSK World
420 lines•14.1 KB
python
auth.py
Raw Download
Find: Go to:
"""
User Authentication and Personalization System
Developer: Molla Samser
Design & Testing: Rima Khatun
Company: RSK World
Year: 2026
Website: https://rskworld.in
"""

import sqlite3
import hashlib
import secrets
import json
from datetime import datetime, timedelta
from functools import wraps
from flask import session, request, redirect, url_for, flash, jsonify
import os

class AuthManager:
    def __init__(self, db_path='users.db'):
        self.db_path = db_path
        self.init_database()
    
    def init_database(self):
        """Initialize user database."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # Users table
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                username TEXT UNIQUE NOT NULL,
                email TEXT UNIQUE NOT NULL,
                password_hash TEXT NOT NULL,
                salt TEXT NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                last_login TIMESTAMP,
                is_active BOOLEAN DEFAULT 1,
                preferences TEXT DEFAULT '{}'
            )
        ''')
        
        # User sessions table
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS user_sessions (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id INTEGER,
                session_token TEXT UNIQUE NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                expires_at TIMESTAMP,
                ip_address TEXT,
                user_agent TEXT,
                is_active BOOLEAN DEFAULT 1,
                FOREIGN KEY (user_id) REFERENCES users (id)
            )
        ''')
        
        # User preferences and activity
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS user_preferences (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id INTEGER,
                category TEXT,
                preference_type TEXT,
                preference_value TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (user_id) REFERENCES users (id)
            )
        ''')
        
        # User reading history
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS reading_history (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id INTEGER,
                article_id TEXT,
                article_title TEXT,
                category TEXT,
                read_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                reading_time INTEGER DEFAULT 0,
                FOREIGN KEY (user_id) REFERENCES users (id)
            )
        ''')
        
        conn.commit()
        conn.close()
    
    def _generate_salt(self):
        """Generate a random salt for password hashing."""
        return secrets.token_hex(16)
    
    def _hash_password(self, password, salt):
        """Hash password with salt."""
        return hashlib.sha256(f"{password}{salt}".encode()).hexdigest()
    
    def register_user(self, username, email, password):
        """Register a new user."""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            # Check if user already exists
            cursor.execute('SELECT id FROM users WHERE username = ? OR email = ?', (username, email))
            if cursor.fetchone():
                return {'success': False, 'error': 'User already exists'}
            
            # Create new user
            salt = self._generate_salt()
            password_hash = self._hash_password(password, salt)
            
            cursor.execute('''
                INSERT INTO users (username, email, password_hash, salt)
                VALUES (?, ?, ?, ?)
            ''', (username, email, password_hash, salt))
            
            user_id = cursor.lastrowid
            conn.commit()
            conn.close()
            
            return {'success': True, 'user_id': user_id}
        
        except Exception as e:
            return {'success': False, 'error': str(e)}
    
    def authenticate_user(self, username, password):
        """Authenticate user credentials."""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            cursor.execute('''
                SELECT id, username, email, password_hash, salt, is_active
                FROM users WHERE username = ? OR email = ?
            ''', (username, username))
            
            user = cursor.fetchone()
            conn.close()
            
            if not user:
                return {'success': False, 'error': 'Invalid credentials'}
            
            user_id, username, email, stored_hash, salt, is_active = user
            
            if not is_active:
                return {'success': False, 'error': 'Account is disabled'}
            
            # Verify password
            password_hash = self._hash_password(password, salt)
            if password_hash != stored_hash:
                return {'success': False, 'error': 'Invalid credentials'}
            
            # Update last login
            self._update_last_login(user_id)
            
            return {
                'success': True,
                'user_id': user_id,
                'username': username,
                'email': email
            }
        
        except Exception as e:
            return {'success': False, 'error': str(e)}
    
    def _update_last_login(self, user_id):
        """Update user's last login timestamp."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            UPDATE users SET last_login = ? WHERE id = ?
        ''', (datetime.now(), user_id))
        
        conn.commit()
        conn.close()
    
    def create_session(self, user_id, ip_address=None, user_agent=None):
        """Create a new user session."""
        try:
            session_token = secrets.token_urlsafe(32)
            expires_at = datetime.now() + timedelta(days=7)  # 7 days expiry
            
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            cursor.execute('''
                INSERT INTO user_sessions (user_id, session_token, expires_at, ip_address, user_agent)
                VALUES (?, ?, ?, ?, ?)
            ''', (user_id, session_token, expires_at, ip_address, user_agent))
            
            conn.commit()
            conn.close()
            
            return session_token
        
        except Exception as e:
            return None
    
    def validate_session(self, session_token):
        """Validate session token and return user info."""
        conn = None
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            cursor.execute('''
                SELECT us.user_id, u.username, u.email, us.expires_at
                FROM user_sessions us
                JOIN users u ON us.user_id = u.id
                WHERE us.session_token = ? AND us.is_active = 1 AND u.is_active = 1
            ''', (session_token,))
            
            result = cursor.fetchone()
            
            if not result:
                return None
            
            user_id, username, email, expires_at = result
            
            # Check if session is expired
            if expires_at:
                if isinstance(expires_at, str):
                    expires_dt = datetime.fromisoformat(expires_at)
                else:
                    expires_dt = expires_at
                
                if datetime.now() > expires_dt:
                    self.invalidate_session(session_token)
                    return None
            
            return {
                'user_id': user_id,
                'username': username,
                'email': email
            }
        
        except Exception as e:
            return None
        finally:
            if conn:
                conn.close()
    
    def invalidate_session(self, session_token):
        """Invalidate a session."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            UPDATE user_sessions SET is_active = 0 WHERE session_token = ?
        ''', (session_token,))
        
        conn.commit()
        conn.close()
    
    def invalidate_all_sessions(self, user_id):
        """Invalidate all sessions for a user."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            UPDATE user_sessions SET is_active = 0 WHERE user_id = ?
        ''', (user_id,))
        
        conn.commit()
        conn.close()

class UserPreferences:
    """Manage user preferences and personalization."""
    
    def __init__(self, db_path='users.db'):
        self.db_path = db_path
    
    def set_preference(self, user_id, preference_type, preference_value, category=None):
        """Set a user preference."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT OR REPLACE INTO user_preferences 
            (user_id, category, preference_type, preference_value)
            VALUES (?, ?, ?, ?)
        ''', (user_id, category, preference_type, json.dumps(preference_value)))
        
        conn.commit()
        conn.close()
    
    def get_preference(self, user_id, preference_type, category=None):
        """Get a user preference."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            SELECT preference_value FROM user_preferences 
            WHERE user_id = ? AND preference_type = ? AND (category = ? OR category IS NULL)
        ''', (user_id, preference_type, category))
        
        result = cursor.fetchone()
        conn.close()
        
        if result:
            return json.loads(result[0])
        return None
    
    def get_all_preferences(self, user_id):
        """Get all user preferences."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            SELECT category, preference_type, preference_value 
            FROM user_preferences WHERE user_id = ?
        ''', (user_id,))
        
        results = cursor.fetchall()
        conn.close()
        
        preferences = {}
        for category, pref_type, pref_value in results:
            if category not in preferences:
                preferences[category] = {}
            preferences[category][pref_type] = json.loads(pref_value)
        
        return preferences
    
    def track_reading_history(self, user_id, article_id, article_title, category, reading_time=0):
        """Track user reading history."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT INTO reading_history 
            (user_id, article_id, article_title, category, reading_time)
            VALUES (?, ?, ?, ?, ?)
        ''', (user_id, article_id, article_title, category, reading_time))
        
        conn.commit()
        conn.close()
    
    def get_reading_history(self, user_id, limit=50):
        """Get user reading history."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            SELECT article_id, article_title, category, read_at, reading_time
            FROM reading_history 
            WHERE user_id = ? 
            ORDER BY read_at DESC 
            LIMIT ?
        ''', (user_id, limit))
        
        results = cursor.fetchall()
        conn.close()
        
        return [
            {
                'article_id': row[0],
                'article_title': row[1],
                'category': row[2],
                'read_at': row[3],
                'reading_time': row[4]
            }
            for row in results
        ]
    
    def get_reading_stats(self, user_id, days=30):
        """Get user reading statistics."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cutoff_date = datetime.now() - timedelta(days=days)
        
        # Total articles read
        cursor.execute('''
            SELECT COUNT(*) FROM reading_history 
            WHERE user_id = ? AND read_at >= ?
        ''', (user_id, cutoff_date))
        total_articles = cursor.fetchone()[0]
        
        # Articles by category
        cursor.execute('''
            SELECT category, COUNT(*) FROM reading_history 
            WHERE user_id = ? AND read_at >= ?
            GROUP BY category
        ''', (user_id, cutoff_date))
        category_stats = dict(cursor.fetchall())
        
        # Total reading time
        cursor.execute('''
            SELECT SUM(reading_time) FROM reading_history 
            WHERE user_id = ? AND read_at >= ?
        ''', (user_id, cutoff_date))
        total_time = cursor.fetchone()[0] or 0
        
        conn.close()
        
        return {
            'total_articles': total_articles,
            'category_stats': category_stats,
            'total_reading_time': total_time,
            'avg_reading_time': total_time / total_articles if total_articles > 0 else 0
        }

# Flask decorators
def login_required(f):
    """Decorator to require user login."""
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if 'user_token' not in session:
            flash('Please log in to access this page.', 'warning')
            return redirect(url_for('login'))
        
        # Validate session
        user_info = auth_manager.validate_session(session['user_token'])
        if not user_info:
            session.pop('user_token', None)
            flash('Session expired. Please log in again.', 'warning')
            return redirect(url_for('login'))
        
        return f(*args, **kwargs)
    return decorated_function

def admin_required(f):
    """Decorator to require admin privileges."""
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if 'user_token' not in session:
            flash('Admin access required.', 'error')
            return redirect(url_for('admin.login'))
        
        user_info = auth_manager.validate_session(session['user_token'])
        if not user_info or user_info.get('username') != 'admin':
            flash('Admin access required.', 'error')
            return redirect(url_for('admin.login'))
        
        return f(*args, **kwargs)
    return decorated_function

# Initialize global auth manager
auth_manager = AuthManager()
user_preferences = UserPreferences()

# Developer Details
# Created by Molla Samser (RSK World)
# 2026
447 lines•15 KB
python
🚀 Support RSK World

Subscribe to our YouTube channel for latest tutorials & updates!



Click subscribe & support our work ❤️

About RSK World

Founded by Molla Samser, with Designer & Tester Rima Khatun, RSK World is your one-stop destination for free programming resources, source code, and development tools.

Founder: Molla Samser
Designer & Tester: Rima Khatun

Development

  • Game Development
  • Web Development
  • Mobile Development
  • AI Development
  • Development Tools

Legal

  • Terms & Conditions
  • Privacy Policy
  • Disclaimer

Contact Info

Nutanhat, Mongolkote
Purba Burdwan, West Bengal
India, 713147

+91 93305 39277

hello@rskworld.in
support@rskworld.in

© 2026 RSK World. All rights reserved.

Content used for educational purposes only. View Disclaimer