src/Controllers/user_controller.py - master - wonkxe/Contact_and_task

src/Controllers/user_controller.py
206 строк8 KB
wonkxe
25 май 2026, 13:57
25 май 2026, 13:578620ad8
"""
Контроллер управления пользователями (дополнительные методы).
"""
import re
import json
from src.Models.user import User
from src.Connection.connect import connect_db, close_db


class UserController:

    @staticmethod
    def user_exists(login: str) -> dict:
        connect_db()
        try:
            user = User.get_or_none(User.login == login)
            if user:
                return {'exists': True, 'message': 'Пользователь существует', 'user_id': user.id}
            else:
                return {'exists': False, 'message': 'Пользователь не найден', 'user_id': None}
        finally:
            close_db()

    @staticmethod
    def get_all_users() -> dict:
        connect_db()
        try:
            users = list(User.select())
            data = [{'id': u.id, 'login': u.login} for u in users]
            return {'success': True, 'message': f'Найдено {len(data)} пользователей', 'users': data}
        finally:
            close_db()

    @staticmethod
    def search_users_by_login(search_term: str) -> dict:
        if len(search_term) < 2:
            return {'success': False, 'message': 'Поисковый запрос должен содержать минимум 2 символа', 'users': []}

        connect_db()
        try:
            users = list(User.select().where(User.login.contains(search_term)))
            data = [{'id': u.id, 'login': u.login} for u in users]
            return {'success': True, 'message': f'Найдено {len(data)} пользователей', 'users': data}
        finally:
            close_db()

    @staticmethod
    def validate_login(login: str) -> dict:
        if not login or len(login) < 3:
            return {'valid': False, 'message': 'Логин должен содержать минимум 3 символа'}
        if len(login) > 50:
            return {'valid': False, 'message': 'Логин не должен превышать 50 символов'}
        if not re.match(r'^[a-zA-Z0-9_.-]+$', login):
            return {'valid': False, 'message': 'Логин может содержать только буквы, цифры и символы _ . -'}
        return {'valid': True, 'message': 'Логин корректен'}

    @staticmethod
    def validate_password_strength(password: str) -> dict:
        suggestions = []
        criteria_met = 0

        if len(password) >= 8:
            criteria_met += 1
        else:
            suggestions.append('Используйте минимум 8 символов')

        if re.search(r'[A-Z]', password):
            criteria_met += 1
        else:
            suggestions.append('Добавьте заглавные буквы')

        if re.search(r'[a-z]', password):
            criteria_met += 1
        else:
            suggestions.append('Добавьте строчные буквы')

        if re.search(r'[0-9]', password):
            criteria_met += 1
        else:
            suggestions.append('Добавьте цифры')

        if re.search(r'[!@#$%^&*(),.?":{}|<>]', password):
            criteria_met += 1
        else:
            suggestions.append('Добавьте специальные символы')

        if criteria_met >= 4:
            return {'strong': True, 'message': 'Отличный пароль!', 'suggestions': []}
        elif criteria_met == 3:
            return {'strong': False, 'message': 'Хороший, но можно усилить', 'suggestions': suggestions}
        else:
            return {'strong': False, 'message': 'Слабый пароль', 'suggestions': suggestions}

    @staticmethod
    def bulk_create_users(users_data: list) -> dict:
        connect_db()
        created = 0
        failed = 0
        errors = []

        try:
            for index, data in enumerate(users_data):
                login = data.get('login')
                password = data.get('password')

                if not login or not password:
                    failed += 1
                    errors.append({'index': index, 'login': login, 'error': 'Отсутствует логин или пароль'})
                    continue

                if len(password) < 6:
                    failed += 1
                    errors.append({'index': index, 'login': login, 'error': 'Пароль слишком короткий'})
                    continue

                if User.get_or_none(User.login == login):
                    failed += 1
                    errors.append({'index': index, 'login': login, 'error': 'Логин уже существует'})
                    continue

                try:
                    user = User.create(login=login)
                    user.set_password(password)
                    user.save()
                    created += 1
                except Exception as e:
                    failed += 1
                    errors.append({'index': index, 'login': login, 'error': str(e)})

            return {'success': True, 'message': f'Создано: {created}, Ошибок: {failed}', 'created': created,
                    'failed': failed, 'errors': errors}
        finally:
            close_db()

    @staticmethod
    def delete_inactive_users(user_ids: list) -> dict:
        connect_db()
        deleted = 0
        failed = 0
        errors = []

        try:
            for uid in user_ids:
                try:
                    user = User.get_by_id(uid)
                    user.delete_instance()
                    deleted += 1
                except User.DoesNotExist:
                    failed += 1
                    errors.append({'user_id': uid, 'error': 'Пользователь не найден'})
                except Exception as e:
                    failed += 1
                    errors.append({'user_id': uid, 'error': str(e)})

            return {'success': True, 'message': f'Удалено: {deleted}, Ошибок: {failed}', 'deleted': deleted,
                    'failed': failed, 'errors': errors}
        finally:
            close_db()

    @staticmethod
    def export_users_to_json(filepath: str = None) -> dict:
        connect_db()
        try:
            users = list(User.select())
            data = [{'id': u.id, 'login': u.login} for u in users]

            if filepath:
                try:
                    with open(filepath, 'w', encoding='utf-8') as f:
                        json.dump(data, f, ensure_ascii=False, indent=2)
                    return {'success': True, 'message': f'Экспортировано {len(data)} пользователей в {filepath}',
                            'data': data, 'filepath': filepath}
                except Exception as e:
                    return {'success': False, 'message': f'Ошибка сохранения файла: {str(e)}', 'data': data,
                            'filepath': None}
            else:
                return {'success': True, 'message': f'Получено {len(data)} пользователей', 'data': data,
                        'filepath': None}
        finally:
            close_db()

    @staticmethod
    def get_user_statistics() -> dict:
        connect_db()
        try:
            count = User.select().count()
            return {'success': True, 'total_users': count, 'message': f'Всего пользователей в системе: {count}'}
        finally:
            close_db()

    @staticmethod
    def is_password_default(user_id: int) -> dict:
        connect_db()
        try:
            user = User.get_or_none(User.id == user_id)
            if not user:
                return {'is_default': False, 'message': 'Пользователь не найден'}

            default_passwords = ['123456', 'password', 'qwerty', 'admin123', '111111', '123456789']
            for pwd in default_passwords:
                if user.check_password(pwd):
                    return {'is_default': True, 'message': 'Используется стандартный пароль. Рекомендуется сменить его'}

            return {'is_default': False, 'message': 'Пароль не является стандартным'}
        finally:
            close_db()