server.py - master - tracie/Fullstack_website_1

server.py
625 строк23 KB
tracie
done
13 окт 2025, 07:23
13 окт 2025, 07:231e0c427
from flask import Flask, render_template, request, redirect, url_for, flash, g
from werkzeug.security import generate_password_hash, check_password_hash
import sqlite3
from datetime import datetime
import json
import hashlib
import os
from web3 import Web3
from flask_login import LoginManager, UserMixin, login_user, logout_user, current_user, login_required

app = Flask(__name__, template_folder='pages', static_folder='static')
app.secret_key = 'your_very_secure_secret_key_here'
app.config['DATABASE'] = 'poll_analytics.db'
app.config['PERMANENT_SESSION_LIFETIME'] = 3600
app.config['BLOCKCHAIN_PROVIDER'] = 'https://sepolia.infura.io/v3/YOUR_INFURA_KEY'  # Используем testnet
app.config['CONTRACT_ADDRESS'] = '0x0000000000000000000000000000000000000000'  # Можете заменить на свой контракт
app.config['WALLET_PRIVATE_KEY'] = 'your_private_key'  # Для реального использования храните безопасно

def get_db():
    if 'db' not in g:
        g.db = sqlite3.connect(app.config['DATABASE'])
        g.db.row_factory = sqlite3.Row
    return g.db

def close_db(e=None):
    db = g.pop('db', None)
    if db is not None:
        db.close()

def init_db():
    with app.app_context():
        db = get_db()
        cursor = db.cursor()

        # Users table with roles
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                username TEXT UNIQUE NOT NULL,
                email TEXT UNIQUE NOT NULL,
                password TEXT NOT NULL,
                secret_salt TEXT NOT NULL,
                role TEXT DEFAULT 'user',
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                last_login TIMESTAMP
            )
        ''')

        # Polls table
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS polls (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id INTEGER NOT NULL,
                title TEXT NOT NULL,
                description TEXT,
                questions JSON NOT NULL,
                is_public BOOLEAN DEFAULT 1,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (user_id) REFERENCES users(id)
            )
        ''')

        # Responses table
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS responses (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                poll_id INTEGER NOT NULL,
                user_id INTEGER,
                answers JSON NOT NULL,
                tx_hash TEXT,
                submitted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (poll_id) REFERENCES polls(id),
                FOREIGN KEY (user_id) REFERENCES users(id)
            )
        ''')

        # Analytics cache table
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS analytics_cache (
                poll_id INTEGER PRIMARY KEY,
                stats JSON NOT NULL,
                last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (poll_id) REFERENCES polls(id)
            )
        ''')

        # Anonymous participations table
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS anonymous_participations (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id INTEGER NOT NULL,
                poll_hash TEXT NOT NULL,
                tx_hash TEXT NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (user_id) REFERENCES users(id)
            )
        ''')

        # Проверяем, существует ли администратор
        admin = cursor.execute('SELECT id FROM users WHERE username = ?', ('admin',)).fetchone()
        if not admin:
            # Создаем администратора
            cursor.execute(
                'INSERT INTO users (username, email, password, secret_salt, role) VALUES (?, ?, ?, ?, ?)',
                ('admin', 'admin@example.com', generate_password_hash('admin'),
                 hashlib.sha256(os.urandom(32)).hexdigest(), 'admin')
            )
            db.commit()
            print("[INFO] Admin user created successfully")

        db.commit()
        print("[INFO] Database initialized successfully")

class User(UserMixin):
    def __init__(self, user_dict):
        self.id = user_dict['id']
        self.username = user_dict['username']
        self.email = user_dict['email']
        # Используем доступ по индексу вместо .get()
        self.role = user_dict['role'] if 'role' in user_dict.keys() else 'user'

    def is_admin(self):
        return self.role == 'admin'

login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'login'

@login_manager.user_loader
def load_user(user_id):
    db = get_db()
    user = db.execute('SELECT * FROM users WHERE id = ?', (user_id,)).fetchone()
    if user:
        return User(user)
    return None

# Blockchain functions
def init_blockchain():
    return Web3(Web3.HTTPProvider(app.config['BLOCKCHAIN_PROVIDER']))

def send_to_blockchain(data):
    w3 = init_blockchain()
    account = w3.eth.account.from_key(app.config['WALLET_PRIVATE_KEY'])

    tx = {
        'to': app.config['CONTRACT_ADDRESS'],
        'value': 0,
        'gas': 200000,
        'gasPrice': w3.toWei('10', 'gwei'),
        'nonce': w3.eth.getTransactionCount(account.address),
        'data': w3.toHex(text=data)
    }

    signed_tx = account.signTransaction(tx)
    tx_hash = w3.eth.sendRawTransaction(signed_tx.rawTransaction)
    return tx_hash.hex()

# Initialize the database
init_db()

# Template filters
@app.template_filter('format_datetime')
def format_datetime(value):
    if value is None:
        return ""
    return datetime.strptime(value, "%Y-%m-%d %H:%M:%S").strftime("%d.%m.%Y %H:%M")

@app.template_filter('short_hash')
def short_hash(value):
    if not value:
        return ""
    return f"{value[:6]}...{value[-4:]}"

# Routes
@app.route('/')
def index():
    db = get_db()
    recent_polls = []
    if not current_user.is_authenticated:
        cursor = db.execute('''
            SELECT polls.*, users.username 
            FROM polls 
            JOIN users ON polls.user_id = users.id 
            WHERE is_public = 1 
            ORDER BY created_at DESC 
            LIMIT 3
        ''')
        recent_polls = cursor.fetchall()
    return render_template('index.html', recent_polls=recent_polls)

@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        username = request.form.get('username')
        password = request.form.get('password')
        remember = True if request.form.get('remember') else False

        db = get_db()
        user = db.execute('SELECT * FROM users WHERE username = ?', (username,)).fetchone()

        if user and check_password_hash(user['password'], password):
            user_obj = User(user)
            login_user(user_obj, remember=remember)

            db.execute(
                'UPDATE users SET last_login = CURRENT_TIMESTAMP WHERE id = ?',
                (user['id'],))
            db.commit()

            flash('Вы успешно вошли в систему!', 'success')
            next_page = request.args.get('next')
            return redirect(next_page or url_for('index'))
        else:
            flash('Неверное имя пользователя или пароль', 'danger')

    return render_template('auth/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')

        db = get_db()
        error = None

        if not username:
            error = 'Требуется имя пользователя'
        elif not email:
            error = 'Требуется email'
        elif not password:
            error = 'Требуется пароль'
        elif db.execute('SELECT id FROM users WHERE username = ?', (username,)).fetchone() is not None:
            error = f'Пользователь {username} уже зарегистрирован'
        elif db.execute('SELECT id FROM users WHERE email = ?', (email,)).fetchone() is not None:
            error = f'Email {email} уже зарегистрирован'

        if error is None:
            secret_salt = hashlib.sha256(os.urandom(32)).hexdigest()
            db.execute(
                'INSERT INTO users (username, email, password, secret_salt) VALUES (?, ?, ?, ?)',
                (username, email, generate_password_hash(password), secret_salt)
            )
            db.commit()
            flash('Регистрация прошла успешно! Пожалуйста, войдите в систему.', 'success')
            return redirect(url_for('login'))

        flash(error, 'danger')

    return render_template('auth/register.html')

@app.route('/logout')
@login_required
def logout():
    logout_user()
    flash('Вы успешно вышли из системы', 'info')
    return redirect(url_for('index'))

@app.route('/dashboard')
@login_required
def dashboard():
    db = get_db()
    user_polls = db.execute('''
        SELECT * FROM polls 
        WHERE user_id = ? 
        ORDER BY created_at DESC 
        LIMIT 5
    ''', (current_user.id,)).fetchall()

    recent_responses = db.execute('''
        SELECT r.*, p.title as poll_title 
        FROM responses r
        JOIN polls p ON r.poll_id = p.id
        WHERE p.user_id = ?
        ORDER BY r.submitted_at DESC
        LIMIT 5
    ''', (current_user.id,)).fetchall()

    return render_template('dashboard.html',
                           user_polls=user_polls,
                           recent_responses=recent_responses)

@app.route('/polls/create', methods=['GET', 'POST'])
@login_required
def create_poll():
    if request.method == 'POST':
        try:
            # Basic poll info
            title = request.form.get('title')
            description = request.form.get('description')
            is_public = 1 if request.form.get('is_public') else 0

            if not title:
                flash('Название опроса обязательно', 'danger')
                return redirect(url_for('create_poll'))

            # Process questions
            questions = []
            form_data = request.form.to_dict()

            # Debug: выводим все полученные данные формы
            app.logger.debug(f"Полученные данные формы: {form_data}")

            # Find all question texts
            question_texts = {k: v for k, v in form_data.items()
                              if k.startswith('question_') and k.endswith('_text')}

            app.logger.debug(f"Найдены тексты вопросов: {question_texts}")

            for q_key, q_text in question_texts.items():
                if not q_text:
                    app.logger.debug(f"Пропуск пустого вопроса: {q_key}")
                    continue

                q_num = q_key.split('_')[1]
                q_type = form_data.get(f'question_{q_num}_type', 'radio')

                app.logger.debug(f"Обработка вопроса {q_num}: текст='{q_text}', тип='{q_type}'")

                question = {
                    'text': q_text,
                    'type': q_type,
                    'options': []
                }

                if q_type in ['radio', 'checkbox']:
                    # Получаем ВСЕ варианты для этого вопроса (без сортировки)
                    question_options = [v for k, v in form_data.items()
                                        if k.startswith(f'question_{q_num}_option_') and v.strip()]
                    question['options'] = question_options


                questions.append(question)
                app.logger.debug(f"Добавлен вопрос: {question}")

            if not questions:
                flash('Опрос должен содержать хотя бы один вопрос', 'danger')
                return redirect(url_for('create_poll'))

            # Debug: выводим итоговую структуру вопросов перед сохранением
            app.logger.debug(f"Итоговая структура вопросов для сохранения: {questions}")

            # Save to database
            db = get_db()
            questions_json = json.dumps(questions, ensure_ascii=False)
            app.logger.debug(f"JSON для сохранения в БД: {questions_json}")

            db.execute(
                'INSERT INTO polls (user_id, title, description, questions, is_public) VALUES (?, ?, ?, ?, ?)',
                (current_user.id, title, description, questions_json, is_public))
            db.commit()

            # Debug: проверяем, что сохранилось в базе
            saved_poll = db.execute(
                'SELECT questions FROM polls WHERE user_id = ? ORDER BY id DESC LIMIT 1',
                (current_user.id,)
            ).fetchone()

            if saved_poll:
                app.logger.debug(f"Фактически сохраненные вопросы в БД: {saved_poll['questions']}")
            else:
                app.logger.debug("Не удалось получить сохраненный опрос для проверки")

            flash('Опрос успешно создан!', 'success')
            return redirect(url_for('my_polls'))

        except Exception as e:
            db.rollback()
            error_msg = f'Ошибка при сохранении опроса: {str(e)}'
            app.logger.error(error_msg, exc_info=True)
            flash(error_msg, 'danger')
            return redirect(url_for('create_poll'))

    return render_template('polls/create.html')

@app.route('/poll_view')
@login_required
def poll_view():
    print('DEBUG: poll_view called')
    return render_template('polls/poll_view.html')

@app.route('/polls')
@login_required
def my_polls():
    db = get_db()

    # Получаем опросы текущего пользователя и публичные опросы
    polls = db.execute('''
        SELECT polls.*, 
               COUNT(responses.id) as response_count,
               users.username as author
        FROM polls
        LEFT JOIN responses ON polls.id = responses.poll_id
        LEFT JOIN users ON polls.user_id = users.id
        WHERE polls.user_id = ? OR polls.is_public = 1
        GROUP BY polls.id
        ORDER BY polls.created_at DESC
    ''', (current_user.id,)).fetchall()

    return render_template('polls/list.html',
                           polls=polls,
                           now=datetime.now())

@app.route('/polls/<int:poll_id>')
def view_poll(poll_id):
    db = get_db()
    poll = db.execute('''
        SELECT polls.*, users.username 
        FROM polls 
        JOIN users ON polls.user_id = users.id 
        WHERE polls.id = ?
    ''', (poll_id,)).fetchone()

    if poll is None:
        flash('Опрос не найден', 'danger')
        return redirect(url_for('index'))

    if not poll['is_public'] and (not current_user.is_authenticated or current_user.id != poll['user_id']):
        flash('Это приватный опрос', 'warning')
        return redirect(url_for('index'))

    # Преобразуем JSON вопросы в Python объект
    try:
        questions = json.loads(poll['questions'])
        # Добавляем отладочный вывод
        print("Questions data:", questions)
    except json.JSONDecodeError as e:
        flash('Ошибка при загрузке вопросов опроса', 'danger')
        app.logger.error(f'Error decoding questions JSON: {str(e)}')
        questions = []

    return render_template('polls/poll_view.html',
                           poll=poll,
                           questions=questions)

@app.route('/polls/<int:poll_id>/delete', methods=['POST'])
@login_required
def delete_poll(poll_id):
    db = get_db()
    poll = db.execute('SELECT * FROM polls WHERE id = ?', (poll_id,)).fetchone()

    if poll is None:
        flash('Опрос не найден', 'danger')
        return redirect(url_for('my_polls'))

    if current_user.id != poll['user_id'] and not current_user.is_admin():
        flash('У вас нет прав для удаления этого опроса', 'danger')
        return redirect(url_for('my_polls'))

    try:
        # Удаляем все связанные ответы и аналитику
        db.execute('DELETE FROM responses WHERE poll_id = ?', (poll_id,))
        db.execute('DELETE FROM analytics_cache WHERE poll_id = ?', (poll_id,))
        db.execute('DELETE FROM polls WHERE id = ?', (poll_id,))
        db.commit()
        flash('Опрос успешно удален', 'success')
    except Exception as e:
        db.rollback()
        flash(f'Ошибка при удалении опроса: {str(e)}', 'danger')

    return redirect(url_for('my_polls'))

@app.route('/polls/<int:poll_id>/submit', methods=['POST'])
def submit_poll(poll_id):
    print("DEBUG: Начало обработки submit_poll")

    try:
        db = get_db()
        poll = db.execute('SELECT * FROM polls WHERE id = ?', (poll_id,)).fetchone()

        if not poll:
            flash('Опрос не найден', 'danger')
            return redirect(url_for('index'))

        # Получаем вопросы из опроса
        questions = json.loads(poll['questions'])
        answers = {}

        # Собираем все ответы из формы
        for i, question in enumerate(questions, 1):
            q_key = f'q_{i}'

            if question['type'] == 'checkbox':
                # Для чекбоксов получаем список значений
                answers[str(i)] = request.form.getlist(q_key + '[]')  # Обратите внимание на '[]' для чекбоксов
            else:
                # Для остальных типов получаем одно значение
                answers[str(i)] = request.form.get(q_key, '')

        # Сохраняем ответы в базу данных
        user_id = current_user.id if current_user.is_authenticated else None

        # Для демонстрации просто сохраняем в базу без блокчейна
        db.execute(
            'INSERT INTO responses (poll_id, user_id, answers) VALUES (?, ?, ?)',
            (poll_id, user_id, json.dumps(answers)))
        db.commit()

        flash('Ваши ответы успешно сохранены!', 'success')
        return redirect(url_for('view_poll', poll_id=poll_id))

    except Exception as e:
        db.rollback()
        print(f"Ошибка при обработке ответов: {str(e)}")
        flash(f'Ошибка при обработке ответов: {str(e)}', 'danger')
        return redirect(url_for('view_poll', poll_id=poll_id))

@app.route('/polls/<int:poll_id>/analytics')
@login_required
def poll_analytics(poll_id):
    db = get_db()
    poll = db.execute('SELECT * FROM polls WHERE id = ? AND user_id = ?',
                      (poll_id, current_user.id)).fetchone()

    if poll is None:
        flash('Опрос не найден или у вас нет прав доступа', 'danger')
        return redirect(url_for('my_polls'))

    analytics = db.execute(
        'SELECT stats FROM analytics_cache WHERE poll_id = ?', (poll_id,)
    ).fetchone()

    if analytics:
        stats = json.loads(analytics['stats'])
    else:
        stats = compute_poll_analytics(poll_id)
        db.execute(
            'INSERT INTO analytics_cache (poll_id, stats) VALUES (?, ?)',
            (poll_id, json.dumps(stats))
        )
        db.commit()

    # Получаем blockchain proof
    blockchain_proofs = db.execute('''
        SELECT tx_hash FROM responses WHERE poll_id = ? AND tx_hash IS NOT NULL
    ''', (poll_id,)).fetchall()

    questions = json.loads(poll['questions'])
    return render_template('polls/analytics.html',
                           poll=poll,
                           questions=questions,
                           stats=stats,
                           blockchain_proofs=blockchain_proofs)

@app.route('/polls/proof/<tx_hash>')
@login_required
def poll_proof(tx_hash):
    w3 = init_blockchain()
    try:
        tx = w3.eth.getTransaction(tx_hash)
        return render_template('polls/blockchain_proof.html',
                               tx_hash=tx_hash,
                               tx_data=tx.input)
    except Exception as e:
        flash(f'Ошибка при получении данных из блокчейна: {str(e)}', 'danger')
        return redirect(url_for('index'))

def compute_poll_analytics(poll_id):
    db = get_db()
    responses = db.execute(
        'SELECT answers FROM responses WHERE poll_id = ?', (poll_id,)
    ).fetchall()

    if not responses:
        return {}

    poll = db.execute('SELECT questions FROM polls WHERE id = ?', (poll_id,)).fetchone()
    questions = json.loads(poll['questions'])
    stats = {
        'total_responses': len(responses),
        'blockchain_verified': db.execute(
            'SELECT COUNT(*) FROM responses WHERE poll_id = ? AND tx_hash IS NOT NULL',
            (poll_id,)
        ).fetchone()[0],
        'questions': {}
    }

    for i, question in enumerate(questions, 1):
        q_key = str(i)
        stats['questions'][q_key] = {
            'text': question['text'],
            'type': question['type'],
            'answers': {}
        }

        if question['type'] in ['radio', 'select']:
            for option in question.get('options', []):
                stats['questions'][q_key]['answers'][option] = 0

            for response in responses:
                answer = json.loads(response['answers']).get(q_key, '')
                if answer and answer in stats['questions'][q_key]['answers']:
                    stats['questions'][q_key]['answers'][answer] += 1

        elif question['type'] == 'checkbox':
            for option in question.get('options', []):
                stats['questions'][q_key]['answers'][option] = 0

            for response in responses:
                answers = json.loads(response['answers']).get(q_key, [])
                for answer in answers:
                    if answer in stats['questions'][q_key]['answers']:
                        stats['questions'][q_key]['answers'][answer] += 1

        elif question['type'] == 'text':
            text_answers = []
            for response in responses:
                answer = json.loads(response['answers']).get(q_key, '')
                if answer:
                    text_answers.append(answer)
            stats['questions'][q_key]['answers'] = text_answers

    return stats

@app.errorhandler(404)
def page_not_found(e):
    return render_template('errors/404.html'), 404

@app.errorhandler(500)
def internal_server_error(e):
    return render_template('errors/500.html'), 500

if __name__ == '__main__':
    app.run(debug=True)