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

src/Controllers/task_controller.py
366 строк15 KB
wonkxe
25 май 2026, 13:57
25 май 2026, 13:578620ad8
"""
Контроллер задач.

Отвечает за CRUD-операции с задачами.
"""

import json
import datetime
from src.Models.task import Task
from src.Models.contact import Contact
from src.Connection.connect import connect_db, close_db


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

def add_task(contact_id: int, user_id: int, title: str, due_date, description: str = None, status: str = 'new', priority: str = 'medium') -> tuple:
    connect_db()
    try:
        contact = Contact.get_or_none((Contact.id == contact_id) & (Contact.user == user_id))
        if not contact:
            return (False, "Контакт не найден или доступ запрещен", None)

        task = Task.create(
            title=title,
            description=description,
            status=status,
            priority=priority,
            due_date=due_date,
            contact=contact
        )
        return (True, "Задача добавлена", task.id)

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


def get_tasks_by_contact(contact_id: int, user_id: int) -> list:
    connect_db()
    try:
        if not Contact.select().where((Contact.id == contact_id) & (Contact.user == user_id)).exists():
            return []
        return list(Task.select().where(Task.contact == contact_id))
    finally:
        close_db()


def update_task_status(task_id: int, user_id: int, new_status: str) -> tuple:
    connect_db()
    try:
        task = Task.select().join(Contact).where(
            (Task.id == task_id) & (Contact.user == user_id)
        ).get_or_none()

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

        valid_statuses = ['new', 'progress', 'done']
        if new_status not in valid_statuses:
            return (False, "Недопустимый статус")

        task.status = new_status
        task.save()
        return (True, "Статус обновлен")

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


def filter_tasks_by_priority(user_id: int, priority: str) -> list:
    connect_db()
    try:
        return list(Task.select().join(Contact).where(
            (Contact.user == user_id) & (Task.priority == priority)
        ))
    finally:
        close_db()


def delete_task(task_id: int, user_id: int) -> tuple:
    connect_db()
    try:
        task = Task.select().join(Contact).where(
            (Task.id == task_id) & (Contact.user == user_id)
        ).get_or_none()

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

        task.delete_instance()
        return (True, "Задача удалена")

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


# --- Дополнительные методы из tasks_task.md ---

class TaskController:

    @staticmethod
    def get_task_by_id(task_id: int, user_id: int) -> dict:
        connect_db()
        try:
            task = Task.select().join(Contact).where((Task.id == task_id) & (Contact.user == user_id)).get_or_none()
            if not task:
                return {'success': False, 'message': 'Задача не найдена', 'task_data': None}

            data = {
                'id': task.id,
                'title': task.title,
                'description': task.description,
                'due_date': task.due_date.strftime('%Y-%m-%d') if task.due_date else None,
                'status': task.status,
                'priority': task.priority,
                'contact_id': task.contact.id
            }
            return {'success': True, 'message': 'Задача найдена', 'task_data': data}
        finally:
            close_db()

    @staticmethod
    def get_all_user_tasks(user_id: int) -> dict:
        connect_db()
        try:
            tasks = list(Task.select().join(Contact).where(Contact.user == user_id))
            data = []
            for t in tasks:
                data.append({
                    'id': t.id,
                    'title': t.title,
                    'status': t.status,
                    'priority': t.priority,
                    'due_date': t.due_date.strftime('%Y-%m-%d') if t.due_date else None,
                    'contact': {'id': t.contact.id, 'first_name': t.contact.first_name, 'last_name': t.contact.last_name}
                })
            return {'success': True, 'message': f'Найдено {len(data)} задач', 'total': len(data), 'tasks': data}
        finally:
            close_db()

    @staticmethod
    def get_tasks_by_status(user_id: int, status: str) -> dict:
        valid_statuses = ['new', 'progress', 'done']
        if status not in valid_statuses:
            return {'success': False, 'message': 'Недопустимый статус', 'tasks': [], 'count': 0}

        connect_db()
        try:
            tasks = list(Task.select().join(Contact).where((Contact.user == user_id) & (Task.status == status)))
            data = [{'id': t.id, 'title': t.title, 'priority': t.priority, 'due_date': t.due_date.strftime('%Y-%m-%d') if t.due_date else None} for t in tasks]
            return {'success': True, 'message': f'Найдено {len(data)} задач со статусом {status}', 'count': len(data), 'tasks': data}
        finally:
            close_db()

    @staticmethod
    def get_overdue_tasks(user_id: int) -> dict:
        connect_db()
        try:
            today = datetime.date.today()
            tasks = list(Task.select().join(Contact).where(
                (Contact.user == user_id) &
                (Task.due_date < today) &
                (Task.status != 'done')
            ).order_by(Task.due_date.asc()))

            data = [{'id': t.id, 'title': t.title, 'due_date': t.due_date.strftime('%Y-%m-%d'), 'priority': t.priority} for t in tasks]
            return {'success': True, 'message': f'Найдено {len(data)} просроченные задачи', 'count': len(data), 'tasks': data}
        finally:
            close_db()

    @staticmethod
    def get_tasks_by_priority(user_id: int, priority: str) -> dict:
        valid_priorities = ['low', 'medium', 'high']
        if priority not in valid_priorities:
            return {'success': False, 'message': 'Недопустимый приоритет', 'tasks': [], 'count': 0}

        connect_db()
        try:
            tasks = list(Task.select().join(Contact).where((Contact.user == user_id) & (Task.priority == priority)).order_by(Task.due_date.asc()))
            data = [{'id': t.id, 'title': t.title, 'status': t.status, 'due_date': t.due_date.strftime('%Y-%m-%d') if t.due_date else None} for t in tasks]
            return {'success': True, 'message': f'Найдено {len(data)} задач с приоритетом {priority}', 'count': len(data), 'tasks': data}
        finally:
            close_db()

    @staticmethod
    def get_tasks_by_date_range(user_id: int, start_date: str, end_date: str) -> dict:
        connect_db()
        try:
            start = datetime.datetime.strptime(start_date, '%Y-%m-%d').date()
            end = datetime.datetime.strptime(end_date, '%Y-%m-%d').date()

            tasks = list(Task.select().join(Contact).where(
                (Contact.user == user_id) &
                (Task.due_date.between(start, end))
            ).order_by(Task.due_date.asc()))

            data = [{'id': t.id, 'title': t.title, 'due_date': t.due_date.strftime('%Y-%m-%d')} for t in tasks]
            return {'success': True, 'message': f'Найдено {len(data)} задач за период {start_date} - {end_date}', 'count': len(data), 'tasks': data}
        except ValueError:
            return {'success': False, 'message': 'Неверный формат даты. Используйте YYYY-MM-DD', 'count': 0, 'tasks': []}
        finally:
            close_db()

    @staticmethod
    def get_task_statistics(user_id: int) -> dict:
        connect_db()
        try:
            total = Task.select().join(Contact).where(Contact.user == user_id).count()

            by_status = {}
            for status in ['new', 'progress', 'done']:
                by_status[status] = Task.select().join(Contact).where((Contact.user == user_id) & (Task.status == status)).count()

            by_priority = {}
            for priority in ['low', 'medium', 'high']:
                by_priority[priority] = Task.select().join(Contact).where((Contact.user == user_id) & (Task.priority == priority)).count()

            today = datetime.date.today()
            overdue = Task.select().join(Contact).where(
                (Contact.user == user_id) &
                (Task.due_date < today) &
                (Task.status != 'done')
            ).count()

            completion_rate = (by_status['done'] / total * 100) if total > 0 else 0.0

            stats = {
                'total': total,
                'by_status': by_status,
                'by_priority': by_priority,
                'overdue': overdue,
                'completion_rate': round(completion_rate, 2)
            }
            return {'success': True, 'stats': stats, 'message': 'Статистика задач пользователя'}
        finally:
            close_db()

    @staticmethod
    def bulk_update_status(user_id: int, task_ids: list, new_status: str) -> dict:
        valid_statuses = ['new', 'progress', 'done']
        if new_status not in valid_statuses:
            return {'success': False, 'message': 'Недопустимый статус', 'updated': 0, 'failed': len(task_ids), 'errors': [{'task_id': tid, 'error': 'Недопустимый статус'} for tid in task_ids]}

        connect_db()
        updated = 0
        failed = 0
        errors = []

        try:
            for tid in task_ids:
                task = Task.select().join(Contact).where((Task.id == tid) & (Contact.user == user_id)).get_or_none()
                if task:
                    task.status = new_status
                    task.save()
                    updated += 1
                else:
                    failed += 1
                    errors.append({'task_id': tid, 'error': 'Задача не найдена или не принадлежит пользователю'})

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

    @staticmethod
    def delete_completed_tasks(user_id: int) -> dict:
        connect_db()
        try:
            count = Task.select().join(Contact).where((Contact.user == user_id) & (Task.status == 'done')).count()
            if count > 0:
                Task.delete().where(Task.id.in_(
                    Task.select().join(Contact).where((Contact.user == user_id) & (Task.status == 'done'))
                )).execute()
            return {'success': True, 'message': f'Удалено {count} выполненных задач', 'deleted_count': count}
        finally:
            close_db()

    @staticmethod
    def get_upcoming_tasks(user_id: int, days: int = 7) -> dict:
        connect_db()
        try:
            today = datetime.date.today()
            end_date = today + datetime.timedelta(days=days)

            tasks = list(Task.select().join(Contact).where(
                (Contact.user == user_id) &
                (Task.due_date >= today) &
                (Task.due_date <= end_date) &
                (Task.status != 'done')
            ).order_by(Task.due_date.asc()))

            data = [{'id': t.id, 'title': t.title, 'due_date': t.due_date.strftime('%Y-%m-%d'), 'priority': t.priority} for t in tasks]
            return {'success': True, 'message': f'Найдено {len(data)} задач на ближайшие {days} дней', 'count': len(data), 'tasks': data}
        finally:
            close_db()

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

        connect_db()
        try:
            term = f"%{search_term}%"
            tasks = list(Task.select().join(Contact).where(
                (Contact.user == user_id) &
                ((Task.title ** term) | (Task.description ** term))
            ))

            data = [{'id': t.id, 'title': t.title, 'status': t.status, 'priority': t.priority} for t in tasks]
            return {'success': True, 'message': f'Найдено {len(data)} задачи по запросу "{search_term}"', 'count': len(data), 'tasks': data}
        finally:
            close_db()

    @staticmethod
    def export_tasks_to_json(user_id: int, filepath: str = None) -> dict:
        connect_db()
        try:
            tasks = list(Task.select().join(Contact).where(Contact.user == user_id))
            data = []
            for t in tasks:
                data.append({
                    'id': t.id,
                    'title': t.title,
                    'description': t.description,
                    'status': t.status,
                    'priority': t.priority,
                    'due_date': t.due_date.strftime('%Y-%m-%d') if t.due_date else None,
                    'contact': {'id': t.contact.id, 'name': f"{t.contact.first_name} {t.contact.last_name}"}
                })

            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_tasks_grouped_by_priority(user_id: int) -> dict:
        connect_db()
        try:
            tasks = list(Task.select().join(Contact).where(Contact.user == user_id))
            groups = {
                'high': {'count': 0, 'tasks': []},
                'medium': {'count': 0, 'tasks': []},
                'low': {'count': 0, 'tasks': []}
            }

            for t in tasks:
                if t.priority in groups:
                    groups[t.priority]['count'] += 1
                    groups[t.priority]['tasks'].append({'id': t.id, 'title': t.title, 'status': t.status})

            return {'success': True, 'message': 'Задачи сгруппированы по приоритету', 'groups': groups}
        finally:
            close_db()