app.py - master - Maxdev/Wbbstatstest

app.py
324 строки13 KB
Maxdev
upload files
05 мар 2025, 02:58
05 мар 2025, 02:58f0225f5
import logging
from logging.handlers import RotatingFileHandler
import os
from flask import Flask, render_template, request, redirect, url_for, session, flash
from werkzeug.security import generate_password_hash, check_password_hash
import pymysql
from config import DB_HOST, DB_NAME, DB_USER, DB_PASSWORD, DB_PORT

# Создаём папку для логов, если её нет
if not os.path.exists('logs'):
    os.mkdir('logs')

# Настройка логирования
handler = RotatingFileHandler('logs/app.log', maxBytes=10240, backupCount=10)
handler.setFormatter(logging.Formatter(
    '%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'
))
handler.setLevel(logging.INFO)

app = Flask(__name__)
app.secret_key = 'your_secret_key_here'  # Замените на случайную строку

# Подключаем логирование к приложению
app.logger.addHandler(handler)
app.logger.setLevel(logging.INFO)
app.logger.info('Запуск приложения')

# Функция для получения данных из API Wildberries
def get_wildberries_data(api_key):
    url = "https://suppliers-api.wildberries.ru/api/v1/supplier/orders"
    headers = {
        "Authorization": api_key
    }
    try:
        response = requests.get(url, headers=headers)
        if response.status_code == 200:
            return response.json()
        elif response.status_code == 401:
            return {"error": "Ошибка авторизации: неверный API-ключ."}
        elif response.status_code == 404:
            return {"error": "Данные не найдены."}
        elif response.status_code == 500:
            return {"error": "Ошибка сервера Wildberries."}
        else:
            return {"error": f"Неизвестная ошибка: {response.status_code}"}
    except requests.exceptions.RequestException as e:
        app.logger.error(f'Ошибка подключения к API Wildberries: {str(e)}')
        return {"error": f"Ошибка подключения к API: {str(e)}"}

# Функция для получения отчёта о реализации
def get_wildberries_report(api_key, date_from, date_to):
    url = "https://suppliers-api.wildberries.ru/api/v5/supplier/reportDetailByPeriod"
    headers = {
        "Authorization": api_key
    }
    params = {
        "dateFrom": date_from,
        "dateTo": date_to
    }
    try:
        response = requests.get(url, headers=headers, params=params)
        if response.status_code == 200:
            return response.json()
        elif response.status_code == 401:
            return {"error": "Ошибка авторизации: неверный API-ключ."}
        elif response.status_code == 404:
            return {"error": "Данные не найдены."}
        elif response.status_code == 500:
            return {"error": "Ошибка сервера Wildberries."}
        else:
            return {"error": f"Неизвестная ошибка: {response.status_code}"}
    except requests.exceptions.RequestException as e:
        app.logger.error(f'Ошибка подключения к API Wildberries: {str(e)}')
        return {"error": f"Ошибка подключения к API: {str(e)}"}

# Функция для подключения к базе данных
def get_db_connection():
    try:
        conn = pymysql.connect(
            host=DB_HOST,
            database=DB_NAME,
            user=DB_USER,
            password=DB_PASSWORD,
            port=DB_PORT,
            cursorclass=pymysql.cursors.DictCursor
        )
        return conn
    except Exception as e:
        app.logger.error(f'Ошибка подключения к базе данных: {str(e)}')
        flash(f'Ошибка подключения к базе данных: {str(e)}', 'error')
        return None

# Главная страница
@app.route('/')
def home():
    return render_template('home.html')

# Регистрация
@app.route('/register', methods=['GET', 'POST'])
def register():
    if request.method == 'POST':
        username = request.form.get('username')
        password = request.form.get('password')
        confirm_password = request.form.get('confirm_password')

        if not username or not password or not confirm_password:
            flash('Пожалуйста, заполните все поля.', 'error')
            return redirect(url_for('register'))

        if password != confirm_password:
            flash('Пароли не совпадают!', 'error')
            return redirect(url_for('register'))

        hashed_password = generate_password_hash(password, method='pbkdf2:sha256')

        conn = get_db_connection()
        if not conn:
            return redirect(url_for('register'))

        cur = conn.cursor()
        try:
            cur.execute(
                'INSERT INTO users (username, password) VALUES (%s, %s)',
                (username, hashed_password)
            )
            conn.commit()
            flash('Регистрация успешна!', 'success')
            app.logger.info(f'Пользователь {username} успешно зарегистрирован.')
        except pymysql.IntegrityError:
            flash('Пользователь с таким именем уже существует!', 'error')
            app.logger.warning(f'Попытка регистрации существующего пользователя: {username}')
        except Exception as e:
            app.logger.error(f'Ошибка при регистрации пользователя: {str(e)}')
            flash('Ошибка при регистрации.', 'error')
        finally:
            cur.close()
            conn.close()

        return redirect(url_for('login'))

    return render_template('register.html')

# Авторизация
@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        username = request.form.get('username')
        password = request.form.get('password')

        if not username or not password:
            flash('Пожалуйста, заполните все поля.', 'error')
            return redirect(url_for('login'))

        conn = get_db_connection()
        if not conn:
            return redirect(url_for('login'))

        cur = conn.cursor()
        try:
            cur.execute('SELECT * FROM users WHERE username = %s', (username,))
            user = cur.fetchone()
        except Exception as e:
            app.logger.error(f'Ошибка при запросе к базе данных: {str(e)}')
            flash(f'Ошибка при запросе к базе данных: {str(e)}', 'error')
            return redirect(url_for('login'))
        finally:
            cur.close()
            conn.close()

        if user and check_password_hash(user['password'], password):
            session['username'] = user['username']
            session['user_id'] = user['id']
            app.logger.info(f'Пользователь {username} успешно авторизован.')
            return redirect(url_for('dashboard'))
        else:
            flash('Неверный логин или пароль!', 'error')
            app.logger.warning(f'Неудачная попытка входа для пользователя: {username}')

    return render_template('login.html')

# Дашборд
@app.route('/dashboard')
def dashboard():
    if 'username' not in session:
        return redirect(url_for('login'))

    conn = get_db_connection()
    if not conn:
        return redirect(url_for('login'))

    cur = conn.cursor()
    try:
        cur.execute('SELECT * FROM wildberries_cabinets WHERE user_id = %s', (session['user_id'],))
        cabinets = cur.fetchall()
    except Exception as e:
        app.logger.error(f'Ошибка при запросе к базе данных: {str(e)}')
        flash(f'Ошибка при запросе к базе данных: {str(e)}', 'error')
        return redirect(url_for('login'))
    finally:
        cur.close()
        conn.close()

    return render_template('dashboard.html', username=session['username'], cabinets=cabinets)

# Добавление кабинета Wildberries
@app.route('/add_wildberries_cabinet', methods=['POST'])
def add_wildberries_cabinet():
    if 'username' not in session:
        return redirect(url_for('login'))

    api_key = request.form.get('api_key')
    cabinet_name = request.form.get('cabinet_name')
    user_id = session['user_id']

    if not api_key or not cabinet_name:
        flash('Пожалуйста, заполните все поля.', 'error')
        return redirect(url_for('dashboard'))

    conn = get_db_connection()
    if not conn:
        return redirect(url_for('dashboard'))

    cur = conn.cursor()
    try:
        cur.execute(
            'INSERT INTO wildberries_cabinets (user_id, api_key, cabinet_name) VALUES (%s, %s, %s)',
            (user_id, api_key, cabinet_name)
        )
        conn.commit()
        flash('Кабинет успешно подключён!', 'success')
        app.logger.info(f'Кабинет {cabinet_name} успешно подключён пользователем {session["username"]}.')
    except pymysql.IntegrityError:
        flash('Ошибка при подключении кабинета.', 'error')
        app.logger.warning(f'Ошибка при подключении кабинета: {cabinet_name}')
    except Exception as e:
        app.logger.error(f'Ошибка при подключении кабинета: {str(e)}')
        flash('Ошибка при подключении кабинета.', 'error')
    finally:
        cur.close()
        conn.close()

    return redirect(url_for('dashboard'))

# Удаление кабинета Wildberries
@app.route('/delete_wildberries_cabinet/<int:cabinet_id>', methods=['POST'])
def delete_wildberries_cabinet(cabinet_id):
    if 'username' not in session:
        return redirect(url_for('login'))

    conn = get_db_connection()
    if not conn:
        return redirect(url_for('dashboard'))

    cur = conn.cursor()
    try:
        cur.execute('DELETE FROM wildberries_cabinets WHERE id = %s', (cabinet_id,))
        conn.commit()
        flash('Кабинет успешно удалён!', 'success')
        app.logger.info(f'Кабинет {cabinet_id} успешно удалён пользователем {session["username"]}.')
    except Exception as e:
        app.logger.error(f'Ошибка при удалении кабинета: {str(e)}')
        flash(f'Ошибка при удалении кабинета: {str(e)}', 'error')
    finally:
        cur.close()
        conn.close()

    return redirect(url_for('dashboard'))

# Получение отчёта о реализации
@app.route('/get_wildberries_report/<int:cabinet_id>', methods=['POST'])
def get_wildberries_report(cabinet_id):
    if 'username' not in session:
        return redirect(url_for('login'))

    date_from = request.form.get('date_from')
    date_to = request.form.get('date_to')
    user_id = session['user_id']

    if not date_from or not date_to:
        flash('Пожалуйста, заполните все поля.', 'error')
        return redirect(url_for('dashboard'))

    conn = get_db_connection()
    if not conn:
        return redirect(url_for('dashboard'))

    cur = conn.cursor()
    try:
        cur.execute('SELECT * FROM wildberries_cabinets WHERE id = %s AND user_id = %s', (cabinet_id, user_id))
        cabinet = cur.fetchone()
    except Exception as e:
        app.logger.error(f'Ошибка при запросе к базе данных: {str(e)}')
        flash(f'Ошибка при запросе к базе данных: {str(e)}', 'error')
        return redirect(url_for('dashboard'))
    finally:
        cur.close()
        conn.close()

    if not cabinet:
        flash('Кабинет не найден.', 'error')
        return redirect(url_for('dashboard'))

    # Получаем отчёт для кабинета
    report = get_wildberries_report(cabinet['api_key'], date_from, date_to)

    # Сохраняем отчёт в сессии для отображения
    if 'cabinet_reports' not in session:
        session['cabinet_reports'] = {}
    session['cabinet_reports'][cabinet_id] = report
    session.modified = True

    return redirect(url_for('dashboard'))

# Выход
@app.route('/logout')
def logout():
    username = session.get('username')
    session.pop('username', None)
    session.pop('user_id', None)
    app.logger.info(f'Пользователь {username} вышел из системы.')
    return redirect(url_for('home'))

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