pythonDev/lr12

# main.py
from flask import Flask
from views import api
from db import get_db_connection

def create_app():
    app = Flask(__name__)
    app.register_blueprint(api, url_prefix='/')
    get_db_connection()
    return app

if __name__ == '__main__':
    app = create_app()
    app.run(host='127.0.0.1', port=5000, debug=True)

#--------------------------------------------------------------------------------------------------
#db.py
import pymysql
import os
from werkzeug.security import generate_password_hash

DB_HOST = '127.0.0.1'
DB_USER = 'root'
DB_PASSWORD = ''
DB_NAME = 'taskdb'
SECRET_KEY = os.getenv('SECRET_KEY', 'super-secret-dev-key-change-in-prod')
TOKEN_EXPIRATION_HOURS = 1
_connection = None

def get_db_connection():
    global _connection
    if _connection is None or not _connection.open:
        _connection = pymysql.connect(
            host=DB_HOST,
            user=DB_USER,
            password=DB_PASSWORD,
            charset='utf8mb4',
            cursorclass=pymysql.cursors.DictCursor,
            autocommit=False
        )
        _init_database()
    return _connection


def _init_database():
    """Создаёт базу и таблицы, если не существуют."""
    conn = _connection
    try:
        with conn.cursor() as cursor:
            cursor.execute(f"CREATE DATABASE IF NOT EXISTS `{DB_NAME}` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci")
            print(f"База данных '{DB_NAME}' готова.")
            conn.select_db(DB_NAME)

            cursor.execute("""
                CREATE TABLE IF NOT EXISTS users (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    username VARCHAR(50) NOT NULL UNIQUE,
                    password_hash VARCHAR(255) NOT NULL,
                    is_admin TINYINT(1) NOT NULL DEFAULT 0,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
            """)
            print("Таблица 'users' готова.")

            cursor.execute("""
                CREATE TABLE IF NOT EXISTS tasks (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    user_id INT NOT NULL,
                    title VARCHAR(255) NOT NULL,
                    description TEXT,
                    status ENUM('pending', 'done') NOT NULL DEFAULT 'pending',
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
            """)
            print("Таблица 'tasks' готова.")

            conn.commit()
    except Exception as e:
        print(f"Ошибка инициализации БД: {e}")
        conn.rollback()
        raise

_blacklisted_tokens = set()

def blacklist_token(token):
    _blacklisted_tokens.add(token)

def is_token_blacklisted(token):
    return token in _blacklisted_tokens

def create_user(username, password, is_admin=False):
    conn = get_db_connection()
    with conn.cursor() as cursor:
        cursor.execute("SELECT id FROM users WHERE username = %s", (username,))
        if cursor.fetchone():
            return None
        password_hash = generate_password_hash(password)
        cursor.execute(
            "INSERT INTO users (username, password_hash, is_admin) VALUES (%s, %s, %s)",
            (username, password_hash, int(is_admin))
        )
        conn.commit()
        return cursor.lastrowid


def get_user_by_username(username):
    conn = get_db_connection()
    with conn.cursor() as cursor:
        cursor.execute("SELECT id, username, password_hash, is_admin FROM users WHERE username = %s", (username,))
        return cursor.fetchone()


def get_user_by_id(user_id):
    conn = get_db_connection()
    with conn.cursor() as cursor:
        cursor.execute("SELECT id, username, password_hash, is_admin FROM users WHERE id = %s", (user_id,))
        return cursor.fetchone()


def update_password(user_id, new_password):
    conn = get_db_connection()
    with conn.cursor() as cursor:
        password_hash = generate_password_hash(new_password)
        cursor.execute("UPDATE users SET password_hash = %s WHERE id = %s", (password_hash, user_id))
        conn.commit()
        return cursor.rowcount > 0


def create_task(user_id, title, description=None, status='pending'):
    conn = get_db_connection()
    with conn.cursor() as cursor:
        cursor.execute(
            "INSERT INTO tasks (user_id, title, description, status) VALUES (%s, %s, %s, %s)",
            (user_id, title.strip(), description.strip() if description else None, status)
        )
        conn.commit()
        return cursor.lastrowid


def get_tasks_for_user(user_id, is_admin=False):
    conn = get_db_connection()
    with conn.cursor() as cursor:
        if is_admin:
            cursor.execute("SELECT * FROM tasks")
        else:
            cursor.execute("SELECT * FROM tasks WHERE user_id = %s", (user_id,))
        return cursor.fetchall()


def get_task_by_id(task_id, user_id, is_admin=False):
    conn = get_db_connection()
    with conn.cursor() as cursor:
        if is_admin:
            cursor.execute("SELECT * FROM tasks WHERE id = %s", (task_id,))
        else:
            cursor.execute("SELECT * FROM tasks WHERE id = %s AND user_id = %s", (task_id, user_id))
        return cursor.fetchone()


def update_task(task_id, user_id, is_admin, title=None, description=None, status=None):
    conn = get_db_connection()
    with conn.cursor() as cursor:
        task = get_task_by_id(task_id, user_id, is_admin)
        if not task:
            return None

        updates = []
        params = []
        if title is not None:
            updates.append("title = %s")
            params.append(title.strip())
        if description is not None:
            updates.append("description = %s")
            params.append(description.strip() if description else None)
        if status is not None:
            if status not in ('pending', 'done'):
                return None
            updates.append("status = %s")
            params.append(status)

        if not updates:
            return task

        params.append(task_id)
        if is_admin:
            sql = f"UPDATE tasks SET {', '.join(updates)} WHERE id = %s"
        else:
            sql = f"UPDATE tasks SET {', '.join(updates)} WHERE id = %s AND user_id = %s"
            params.append(user_id)

        cursor.execute(sql, params)
        conn.commit()
        return get_task_by_id(task_id, user_id, is_admin)


def delete_task(task_id, user_id, is_admin=False):
    conn = get_db_connection()
    with conn.cursor() as cursor:
        if is_admin:
            cursor.execute("DELETE FROM tasks WHERE id = %s", (task_id,))
        else:
            cursor.execute("DELETE FROM tasks WHERE id = %s AND user_id = %s", (task_id, user_id))
        conn.commit()
        return cursor.rowcount > 0


def get_tasks_by_status(user_id, status, is_admin=False):
    conn = get_db_connection()
    with conn.cursor() as cursor:
        if is_admin:
            cursor.execute("SELECT * FROM tasks WHERE status = %s", (status,))
        else:
            cursor.execute("SELECT * FROM tasks WHERE user_id = %s AND status = %s", (user_id, status))
        return cursor.fetchall()

#--------------------------------------------------------------------------------------------------
#auth.py
from functools import wraps
from flask import request, jsonify
import jwt
from db import get_user_by_id, is_token_blacklisted, SECRET_KEY

def token_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token = None

        if 'Authorization' in request.headers:
            parts = request.headers['Authorization'].split()
            if len(parts) == 2 and parts[0] == "Bearer":
                token = parts[1]

        if not token:
            return jsonify({'error': 'Токен отсутствует'}), 401

        if is_token_blacklisted(token):
            return jsonify({'error': 'Токен отозван'}), 401

        try:
            data = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])
            user_id = data.get('user_id')
            if not user_id:
                raise jwt.InvalidTokenError("Нет user_id в токене")

            current_user = get_user_by_id(user_id)
            if not current_user:
                raise jwt.InvalidTokenError("Пользователь не найден")

        except jwt.ExpiredSignatureError:
            return jsonify({'error': 'Токен истёк'}), 401
        except jwt.InvalidTokenError as e:
            return jsonify({'error': f'Недействительный токен: {str(e)}'}), 401

        return f(current_user=current_user, *args, **kwargs)

    return decorated

#--------------------------------------------------------------------------------------------------
#views.py
from flask import Blueprint, request, jsonify
from werkzeug.security import check_password_hash
from db import (
    create_user, get_user_by_username, get_tasks_for_user, create_task,
    get_task_by_id, update_task, delete_task, get_tasks_by_status,
    update_password, SECRET_KEY, TOKEN_EXPIRATION_HOURS
)
from auth import token_required
import jwt
import re
from datetime import datetime, timedelta

api = Blueprint('api', __name__)


@api.route('/register', methods=['POST'])
def register():
    data = request.get_json()
    if not data or not data.get('username') or not data.get('password'):
        return jsonify({'error': 'Требуются username и password'}), 400

    username = data['username'].strip()
    password = data['password']
    is_admin = bool(data.get('is_admin', False))

    if not re.match(r"^[a-zA-Z0-9_]{3,20}$", username):
        return jsonify({'error': 'Имя: 3–20 символов, буквы, цифры, _'}), 400
    if get_user_by_username(username):
        return jsonify({'error': 'Пользователь уже существует'}), 409

    user_id = create_user(username, password, is_admin)
    if not user_id:
        return jsonify({'error': 'Ошибка создания'}), 500

    return jsonify({'message': 'Пользователь создан', 'user_id': user_id}), 201


@api.route('/login', methods=['POST'])
def login():
    data = request.get_json()
    if not data or not data.get('username') or not data.get('password'):
        return jsonify({'error': 'Требуются username и password'}), 400

    user = get_user_by_username(data['username'])
    if not user or not check_password_hash(user['password_hash'], data['password']):
        return jsonify({'error': 'Неверные данные'}), 401

    payload = {
        'user_id': user['id'],
        'exp': datetime.now() + timedelta(hours=TOKEN_EXPIRATION_HOURS)
    }
    token = jwt.encode(payload, SECRET_KEY, algorithm='HS256')

    return jsonify({
        'token': token,
        'user': {
            'id': user['id'],
            'username': user['username'],
            'is_admin': bool(user['is_admin'])
        }
    })


@api.route('/change-password', methods=['POST'])
@token_required
def change_password(current_user):
    data = request.get_json()
    old = data.get('old_password')
    new = data.get('new_password')
    if not old or not new:
        return jsonify({'error': 'Требуются old_password и new_password'}), 400
    if len(new) < 6:
        return jsonify({'error': 'Новый пароль ≥ 6 символов'}), 400
    if not check_password_hash(current_user['password_hash'], old):
        return jsonify({'error': 'Неверный текущий пароль'}), 401

    if update_password(current_user['id'], new):
        return jsonify({'message': 'Пароль изменён'})
    return jsonify({'error': 'Ошибка изменения пароля'}), 500


@api.route('/logout', methods=['POST'])
@token_required
def logout(current_user):
    auth = request.headers.get('Authorization')
    token = auth.split(' ')[1]
    from db import blacklist_token
    blacklist_token(token)
    return jsonify({'message': 'Вы успешно вышли из системы. Токен отозван.'})


@api.route('/tasks', methods=['GET'])
@token_required
def list_tasks(current_user):
    status = request.args.get('status')
    if status:
        if status not in ('pending', 'done'):
            return jsonify({'error': 'Недопустимый статус'}), 400
        tasks = get_tasks_by_status(current_user['id'], status, current_user['is_admin'])
    else:
        tasks = get_tasks_for_user(current_user['id'], current_user['is_admin'])
    return jsonify(tasks)


@api.route('/tasks', methods=['POST'])
@token_required
def create_task_route(current_user):
    data = request.get_json()
    if not data or 'title' not in data or not data['title'].strip():
        return jsonify({'error': 'Поле "title" обязательно'}), 400

    title = data['title']
    description = data.get('description')
    status = data.get('status', 'pending')
    if status not in ('pending', 'done'):
        return jsonify({'error': 'Недопустимый статус'}), 400

    task_id = create_task(current_user['id'], title, description, status)
    task = get_task_by_id(task_id, current_user['id'], current_user['is_admin'])
    return jsonify(task), 201


@api.route('/tasks/<int:task_id>', methods=['GET'])
@token_required
def get_task(current_user, task_id):
    task = get_task_by_id(task_id, current_user['id'], current_user['is_admin'])
    if not task:
        return jsonify({'error': 'Задача не найдена'}), 404
    return jsonify(task)


@api.route('/tasks/<int:task_id>', methods=['PUT'])
@token_required
def update_task_route(current_user, task_id):
    data = request.get_json()
    if not data:
        return jsonify({'error': 'Тело запроса пустое'}), 400

    title = data.get('title')
    description = data.get('description')
    status = data.get('status')

    updated = update_task(task_id, current_user['id'], current_user['is_admin'], title, description, status)
    if not updated:
        return jsonify({'error': 'Задача не найдена или недоступна'}), 404
    return jsonify(updated)


@api.route('/tasks/<int:task_id>', methods=['DELETE'])
@token_required
def delete_task_route(current_user, task_id):
    success = delete_task(task_id, current_user['id'], current_user['is_admin'])
    if not success:
        return jsonify({'error': 'Задача не найдена или недоступна'}), 404
    return jsonify({'message': 'Задача удалена'})