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

src/Controllers/contact_controller.py
374 строки15 KB
wonkxe
25 май 2026, 13:57
25 май 2026, 13:578620ad8
"""
Контроллер контактов.

Отвечает за CRUD-операции с контактами пользователя.
Включает базовые методы и расширенные функции управления.
"""

import csv
from src.Models.contact import Contact
from src.Models.user import User
from src.Connection.connect import connect_db, close_db
from peewee import IntegrityError


# --- Базовые CRUD методы (для совместимости с тестами Этапа 3) ---

def add_contact(user_id: int, first_name: str, last_name: str, email: str, phone: str = None) -> tuple:
    """
    Добавление нового контакта для пользователя
    """
    connect_db()
    try:
        if not User.select().where(User.id == user_id).exists():
            return (False, "Пользователь не найден", None)

        contact = Contact.create(
            first_name=first_name,
            last_name=last_name,
            email=email,
            phone=phone,
            user=user_id
        )
        return (True, "Контакт добавлен", contact.id)

    except IntegrityError:
        return (False, "Контакт с таким email уже существует", None)
    except Exception as e:
        return (False, f"Ошибка добавления контакта: {str(e)}", None)
    finally:
        close_db()


def get_all_contacts(user_id: int) -> list:
    """
    Получение всех контактов пользователя
    """
    connect_db()
    try:
        return list(Contact.select().where(Contact.user == user_id))
    finally:
        close_db()


def find_contacts(user_id: int, search_term: str) -> list:
    """
    Поиск контактов по имени, фамилии или email
    """
    connect_db()
    try:
        term = f"%{search_term}%"
        return list(Contact.select().where(
            (Contact.user == user_id) &
            (
                    (Contact.first_name ** term) |
                    (Contact.last_name ** term) |
                    (Contact.email ** term)
            )
        ))
    finally:
        close_db()


def update_contact(contact_id: int, user_id: int, **kwargs) -> tuple:
    """
    Обновление данных контакта.
    """
    connect_db()
    try:
        contact = Contact.get_or_none((Contact.id == contact_id) & (Contact.user == user_id))

        if not contact:
            return (False, "Контакт не найден или доступ запрещен")

        allowed_fields = ['first_name', 'last_name', 'email', 'phone']

        for field, value in kwargs.items():
            if field in allowed_fields:
                setattr(contact, field, value)

        contact.save()
        return (True, "Контакт обновлен")

    except IntegrityError:
        return (False, "Ошибка обновления: возможно, email уже занят")
    except Exception as e:
        return (False, f"Ошибка обновления: {str(e)}")
    finally:
        close_db()


def delete_contact(contact_id: int, user_id: int) -> tuple:
    """
    Удаление контакта (каскадно удаляет задачи)
    """
    connect_db()
    try:
        contact = Contact.get_or_none((Contact.id == contact_id) & (Contact.user == user_id))

        if not contact:
            return (False, "Контакт не найден или доступ запрещен")

        contact.delete_instance()
        return (True, "Контакт удален")

    except Exception as e:
        return (False, f"Ошибка удаления: {str(e)}")
    finally:
        close_db()


# --- Дополнительные методы из tasks_contact.md ---
# Для удобства использования в тестах и коде, оформим их как статические методы класса
# или как отдельные функции. Здесь используем класс для группировки,
# но если нужно, чтобы они импортировались как функции, можно вынести их наружу.
# Оставим их в классе ContactController, как требовалось в задании tasks_contact.md

class ContactController:

    @staticmethod
    def get_contact_by_id(contact_id: int, user_id: int) -> dict:
        connect_db()
        try:
            contact = Contact.get_or_none((Contact.id == contact_id) & (Contact.user == user_id))
            if not contact:
                return {'success': False, 'message': 'Контакт не найден', 'contact_data': None}

            data = {
                'id': contact.id,
                'first_name': contact.first_name,
                'last_name': contact.last_name,
                'email': contact.email,
                'phone': contact.phone
            }
            return {'success': True, 'message': 'Контакт найден', 'contact_data': data}
        finally:
            close_db()

    @staticmethod
    def get_contacts_count(user_id: int) -> dict:
        connect_db()
        try:
            count = Contact.select().where(Contact.user == user_id).count()
            return {'success': True, 'count': count, 'message': f'У пользователя {count} контактов'}
        finally:
            close_db()

    @staticmethod
    def delete_all_contacts(user_id: int) -> dict:
        connect_db()
        try:
            count = Contact.select().where(Contact.user == user_id).count()
            Contact.delete().where(Contact.user == user_id).execute()
            return {'success': True, 'message': f'Удалено {count} контактов', 'deleted_count': count}
        finally:
            close_db()

    @staticmethod
    def search_contacts_by_phone(user_id: int, phone_pattern: str) -> dict:
        if len(phone_pattern) < 3:
            return {'success': False, 'message': 'Шаблон телефона должен содержать минимум 3 символа', 'contacts': []}

        connect_db()
        try:
            contacts = list(Contact.select().where(
                (Contact.user == user_id) & (Contact.phone.contains(phone_pattern))
            ))
            data = [
                {'id': c.id, 'first_name': c.first_name, 'last_name': c.last_name, 'email': c.email, 'phone': c.phone}
                for c in contacts]
            return {'success': True, 'message': f'Найдено {len(data)} контакта', 'contacts': data}
        finally:
            close_db()

    @staticmethod
    def get_contacts_sorted(user_id: int, sort_by: str = 'last_name', ascending: bool = True) -> dict:
        valid_fields = ['first_name', 'last_name', 'email', 'id']
        if sort_by not in valid_fields:
            return {'success': False, 'message': f'Недопустимое поле сортировки: {sort_by}', 'contacts': []}

        connect_db()
        try:
            query = Contact.select().where(Contact.user == user_id)
            field = getattr(Contact, sort_by)
            if ascending:
                query = query.order_by(field.asc())
            else:
                query = query.order_by(field.desc())

            contacts = list(query)
            data = [
                {'id': c.id, 'first_name': c.first_name, 'last_name': c.last_name, 'email': c.email, 'phone': c.phone}
                for c in contacts]
            direction = "возрастание" if ascending else "убывание"
            return {'success': True, 'message': f'Найдено {len(data)} контактов, сортировка по {sort_by} ({direction})',
                    'contacts': data}
        finally:
            close_db()

    @staticmethod
    def check_contact_exists(user_id: int, email: str = None, phone: str = None) -> dict:
        connect_db()
        try:
            conditions = [Contact.user == user_id]
            if email:
                conditions.append(Contact.email == email)
            if phone:
                conditions.append(Contact.phone == phone)

            if not email and not phone:
                return {'exists': False, 'message': 'Не указан email или телефон', 'contact_id': None}

            contact = Contact.get_or_none(*conditions)
            if contact:
                msg = 'Контакт с таким email существует' if email else 'Контакт с таким телефоном существует'
                return {'exists': True, 'message': msg, 'contact_id': contact.id}
            else:
                return {'exists': False, 'message': 'Контакт не найден', 'contact_id': None}
        finally:
            close_db()

    @staticmethod
    def get_contacts_by_email_domain(user_id: int, domain: str) -> dict:
        if not domain.startswith('@'):
            domain = '@' + domain

        connect_db()
        try:
            contacts = list(Contact.select().where(
                (Contact.user == user_id) & (Contact.email.contains(domain))
            ))
            data = [
                {'id': c.id, 'first_name': c.first_name, 'last_name': c.last_name, 'email': c.email, 'phone': c.phone}
                for c in contacts]
            return {'success': True, 'message': f'Найдено {len(data)} контактов с email доменом {domain}',
                    'contacts': data}
        finally:
            close_db()

    @staticmethod
    def batch_add_contacts(user_id: int, contacts_data: list) -> dict:
        connect_db()
        created = 0
        failed = 0
        errors = []

        try:
            for index, data in enumerate(contacts_data):
                first_name = data.get('first_name')
                last_name = data.get('last_name')
                email = data.get('email')
                phone = data.get('phone')

                if not first_name or not last_name or not email:
                    failed += 1
                    errors.append({'index': index, 'contact': f"{first_name} {last_name}",
                                   'error': 'Отсутствуют обязательные поля'})
                    continue

                # Проверка на существование
                if Contact.get_or_none((Contact.user == user_id) & (Contact.email == email)):
                    failed += 1
                    errors.append(
                        {'index': index, 'contact': f"{first_name} {last_name}", 'error': 'Контакт уже существует'})
                    continue

                try:
                    Contact.create(first_name=first_name, last_name=last_name, email=email, phone=phone, user=user_id)
                    created += 1
                except Exception as e:
                    failed += 1
                    errors.append({'index': index, 'contact': f"{first_name} {last_name}", 'error': str(e)})

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

    @staticmethod
    def export_contacts_to_csv(user_id: int, filepath: str = None) -> dict:
        connect_db()
        try:
            contacts = list(Contact.select().where(Contact.user == user_id))
            data = [
                {'id': c.id, 'first_name': c.first_name, 'last_name': c.last_name, 'email': c.email, 'phone': c.phone}
                for c in contacts]

            if filepath:
                try:
                    with open(filepath, 'w', newline='', encoding='utf-8') as f:
                        writer = csv.DictWriter(f, fieldnames=['id', 'first_name', 'last_name', 'email', 'phone'])
                        writer.writeheader()
                        writer.writerows(data)
                    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_duplicate_contacts(user_id: int) -> dict:
        connect_db()
        try:
            contacts = list(Contact.select().where(Contact.user == user_id))
            email_counts = {}
            phone_counts = {}

            for c in contacts:
                if c.email:
                    email_counts.setdefault(c.email, []).append(c.id)
                if c.phone:
                    phone_counts.setdefault(c.phone, []).append(c.id)

            duplicates = []
            for email, ids in email_counts.items():
                if len(ids) > 1:
                    duplicates.append({'email': email, 'contacts': ids})
            for phone, ids in phone_counts.items():
                if len(ids) > 1:
                    duplicates.append({'phone': phone, 'contacts': ids})

            return {'success': True, 'message': f'Найдено {len(duplicates)} дублирующихся контакта',
                    'duplicates': duplicates}
        finally:
            close_db()

    @staticmethod
    def transfer_contacts(from_user_id: int, to_user_id: int) -> dict:
        connect_db()
        try:
            if not User.get_or_none(User.id == from_user_id) or not User.get_or_none(User.id == to_user_id):
                return {'success': False, 'message': 'Один из пользователей не найден', 'transferred_count': 0}

            count = Contact.select().where(Contact.user == from_user_id).count()
            Contact.update(user=to_user_id).where(Contact.user == from_user_id).execute()

            return {'success': True,
                    'message': f'Передано {count} контактов от пользователя {from_user_id} пользователю {to_user_id}',
                    'transferred_count': count}
        finally:
            close_db()

    @staticmethod
    def get_contacts_paginated(user_id: int, page: int = 1, per_page: int = 10) -> dict:
        connect_db()
        try:
            total = Contact.select().where(Contact.user == user_id).count()
            offset = (page - 1) * per_page
            contacts = list(Contact.select().where(Contact.user == user_id).limit(per_page).offset(offset))

            data = [
                {'id': c.id, 'first_name': c.first_name, 'last_name': c.last_name, 'email': c.email, 'phone': c.phone}
                for c in contacts]
            start = offset + 1 if total > 0 else 0
            end = min(offset + per_page, total)
            message = f'Показаны контакты {start}-{end} из {total}' if total > 0 else 'Контакты не найдены'

            return {'success': True, 'message': message, 'contacts': data, 'total': total, 'page': page,
                    'per_page': per_page}
        finally:
            close_db()