# Proprietary License Agreement
# Copyright (c) 2024-29 CodWiz
# Permission is hereby granted to any person obtaining a copy of this software and associated documentation files (the "Software"), to use the Software for personal and non-commercial purposes, subject to the following conditions:
# 1. The Software may not be modified, altered, or otherwise changed in any way without the explicit written permission of the author.
# 2. Redistribution of the Software, in original or modified form, is strictly prohibited without the explicit written permission of the author.
# 3. The Software is provided "as is", without warranty of any kind, express or implied, including but not limited to the warranties of merchantability, fitness for a particular purpose, and non-infringement. In no event shall the author or copyright holder be liable for any claim, damages, or other liability, whether in an action of contract, tort, or otherwise, arising from, out of, or in connection with the Software or the use or other dealings in the Software.
# 4. Any use of the Software must include the above copyright notice and this permission notice in all copies or substantial portions of the Software.
# 5. By using the Software, you agree to be bound by the terms and conditions of this license.
# For any inquiries or requests for permissions, please contact codwiz@yandex.ru.
# ---------------------------------------------------------------------------------
# Name: TaskManager
# Description: Manages tasks with Telegram commands and inline keyboards.
# Author: @hikka_mods
# ---------------------------------------------------------------------------------
# meta developer: @hikka_mods
# scope: TaskManager
# scope: TaskManager 0.0.1
# ---------------------------------------------------------------------------------
import asyncio
import datetime
import json
import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional
from .. import loader, utils
logger = logging.getLogger(__name__)
@dataclass
class Task:
"""Represents a task."""
description: str
due_date: Optional[datetime.datetime] = None
completed: bool = False
created_at: datetime.datetime = field(default_factory=datetime.datetime.now)
id: str = field(default_factory=lambda: f"{datetime.datetime.now().timestamp()}")
def to_dict(self) -> dict:
"""Convert task to dictionary for JSON serialization."""
return {
"id": self.id,
"description": self.description,
"due_date": self.due_date.isoformat() if self.due_date else None,
"completed": self.completed,
"created_at": self.created_at.isoformat(),
}
@classmethod
def from_dict(cls, data: dict) -> "Task":
"""Create task from dictionary."""
return cls(
id=data.get("id", f"{datetime.datetime.now().timestamp()}"),
description=data["description"],
due_date=datetime.datetime.fromisoformat(data["due_date"])
if data.get("due_date")
else None,
completed=data["completed"],
created_at=datetime.datetime.fromisoformat(data["created_at"]),
)
class TaskManager:
"""Manages tasks, storing them in a JSON file."""
def __init__(self, data_file: str):
self.data_file = Path(data_file)
self.tasks: Dict[int, List[Task]] = {}
self._lock = asyncio.Lock()
self.load_data()
def load_data(self):
"""Loads task data from the JSON file."""
if not self.data_file.exists():
self.tasks = {}
logger.info("Task data file not found. Starting empty.")
return
try:
with open(self.data_file, "r", encoding="utf-8") as f:
data = json.load(f)
self.tasks = {
int(user_id): [Task.from_dict(task) for task in task_list]
for user_id, task_list in data.items()
}
except (json.JSONDecodeError, KeyError, ValueError) as e:
logger.warning(f"Failed to load task data: {e}. Starting empty.")
self.tasks = {}
except Exception as e:
logger.error(f"Unexpected error loading task data: {e}")
self.tasks = {}
async def save_data(self):
"""Saves task data to the JSON file."""
async with self._lock:
try:
self.data_file.parent.mkdir(parents=True, exist_ok=True)
data = {
str(user_id): [task.to_dict() for task in task_list]
for user_id, task_list in self.tasks.items()
}
with open(self.data_file, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
except IOError as e:
logger.error(f"Failed to save task data: {e}")
except Exception as e:
logger.error(f"Unexpected error saving task data: {e}")
async def add_task(self, user_id: int, task: Task):
self.tasks.setdefault(user_id, []).append(task)
await self.save_data()
async def remove_task(self, user_id: int, index: int) -> bool:
if user_id in self.tasks and 0 <= index < len(self.tasks[user_id]):
del self.tasks[user_id][index]
await self.save_data()
return True
logger.warning(f"Invalid index for removal: {index}, user: {user_id}")
return False
async def complete_task(self, user_id: int, index: int) -> bool:
if user_id in self.tasks and 0 <= index < len(self.tasks[user_id]):
self.tasks[user_id][index].completed = True
await self.save_data()
return True
logger.warning(f"Invalid index for completion: {index}, user: {user_id}")
return False
def get_tasks(self, user_id: int, include_completed: bool = True) -> List[Task]:
tasks = self.tasks.get(user_id, [])
if not include_completed:
tasks = [task for task in tasks if not task.completed]
return tasks
async def clear_tasks(self, user_id: int) -> bool:
if user_id in self.tasks:
self.tasks[user_id] = []
await self.save_data()
return True
logger.info(f"No tasks to clear for user: {user_id}")
return False
def get_task(self, user_id: int, index: int) -> Optional[Task]:
if user_id in self.tasks and 0 <= index < len(self.tasks[user_id]):
return self.tasks[user_id][index]
return None
def get_overdue_tasks(self, user_id: int) -> List[Task]:
"""Get overdue tasks for a user."""
now = datetime.datetime.now()
return [
task
for task in self.get_tasks(user_id)
if task.due_date and task.due_date < now and not task.completed
]
async def update_task(self, user_id: int, index: int, **kwargs) -> bool:
"""Update task properties."""
task = self.get_task(user_id, index)
if not task:
return False
for key, value in kwargs.items():
if hasattr(task, key):
setattr(task, key, value)
await self.save_data()
return True
@loader.tds
class TaskManagerModule(loader.Module):
"""Manages tasks with Telegram commands and inline keyboards."""
strings = {
"name": "TaskManager",
"task_added": "✅ Task added.",
"task_removed": "✅ Task removed.",
"task_completed": "✅ Task completed.",
"task_not_found": "❌ Task not found.",
"no_tasks": "📄 No active tasks.",
"task_list": "📄 Your tasks:\n{}",
"invalid_index": "❌ Invalid index. Provide valid integer.",
"description_required": "❗️ Provide task description.",
"clear_confirmation": "⚠️ Delete all tasks?",
"tasks_cleared": "✅ All tasks deleted.",
"due_date_format": "❌ Invalid date. Use YYYY-MM-DD HH:MM.",
"task_info": "ℹ Task: {description}\n📅 Due: {due_date}\n✔️ Completed: {completed}\n🎛 Created: {created_at}",
"confirm_clear": "Confirm",
"cancel_clear": "Cancel",
"clear_cancelled": "❌ Deletion cancelled.",
"index_required": "⚠️ Provide task index.",
"clear_confirmation_text": "Are you sure you want to clear all tasks?",
"confirm": "Confirm",
"cancel": "Cancel",
}
strings_ru = {
"task_added": "✅ Задача добавлена.",
"task_removed": "✅ Задача удалена.",
"task_completed": "✅ Задача выполнена.",
"task_not_found": "❌ Задача не найдена.",
"no_tasks": "📄 Нет активных задач.",
"task_list": "📄 Ваши задачи:\n{}",
"invalid_index": "❌ Неверный индекс. Укажите целое число.",
"description_required": "❗️ Укажите описание задачи.",
"clear_confirmation": "⚠️ Удалить все задачи?",
"tasks_cleared": "✅ Все задачи удалены.",
"due_date_format": "❌ Неверный формат даты. Используйте ГГГГ-ММ-ДД ЧЧ:ММ.",
"task_info": "ℹ Задача: {description}\n📅 Срок: {due_date}\n✔️ Выполнена: {completed}\n🎛 Создана: {created_at}",
"confirm_clear": "Подтвердить",
"cancel_clear": "Отменить",
"clear_cancelled": "❌ Удаление отменено.",
"index_required": "⚠️ Укажите индекс задачи.",
"clear_confirmation_text": "Вы уверены, что хотите удалить все задачи?",
"confirm": "Подтвердить",
"cancel": "Отменить",
}
def __init__(self):
self.task_manager: Optional[TaskManager] = None
async def client_ready(self, client, db):
data_dir = Path.cwd() / "data"
data_dir.mkdir(exist_ok=True)
self.task_manager = TaskManager(str(data_dir / "tasks.json"))
@loader.command(
ru_doc="Добавить задачу:\n.taskadd <описание> | <дата (необязательно)>",
en_doc="Add task:\n.taskadd | ",
)
async def taskadd(self, message):
args = utils.get_args_raw(message)
if not args:
await utils.answer(message, self.strings("description_required"))
return
try:
description, due_date_str = (
args.split("|", 1) if "|" in args else (args, None)
)
description = description.strip()
due_date_str = due_date_str.strip() if due_date_str else None
due_date = (
datetime.datetime.fromisoformat(due_date_str) if due_date_str else None
)
except ValueError:
await utils.answer(message, self.strings("due_date_format"))
return
except Exception as e:
logger.error(f"Error adding task: {e}")
await utils.answer(
message, f"❌ Error: {e}"
)
return
task = Task(description=description, due_date=due_date)
await self.task_manager.add_task(message.sender_id, task)
await utils.answer(message, self.strings("task_added"))
@loader.command(ru_doc="[index] - удалить задачу", en_doc="[index] - remove task")
async def taskremove(self, message):
args = utils.get_args_raw(message)
if not args:
await utils.answer(message, self.strings("index_required"))
return
try:
index = int(args) - 1
except ValueError:
await utils.answer(message, self.strings("invalid_index"))
return
if await self.task_manager.remove_task(message.sender_id, index):
await utils.answer(message, self.strings("task_removed"))
else:
await utils.answer(message, self.strings("task_not_found"))
@loader.command(
ru_doc="[index] - Завершите задачу", en_doc="[index] - Complete task"
)
async def taskcomplete(self, message):
args = utils.get_args_raw(message)
if not args:
await utils.answer(message, self.strings("index_required"))
return
try:
index = int(args) - 1
except ValueError:
await utils.answer(message, self.strings("invalid_index"))
return
if await self.task_manager.complete_task(message.sender_id, index):
await utils.answer(message, self.strings("task_completed"))
else:
await utils.answer(message, self.strings("task_not_found"))
@loader.command(ru_doc="Список задач", en_doc="List tasks")
async def tasklist(self, message):
tasks = self.task_manager.get_tasks(message.sender_id)
if not tasks:
await utils.answer(message, self.strings("no_tasks"))
return
task_list_str = "\n".join(
[
f" {i + 1}. {'✅' if task.completed else '❌'} {task.description} (Due: {task.due_date.strftime('%Y-%m-%d %H:%M') if task.due_date else 'None'})"
for i, task in enumerate(tasks)
]
)
await utils.answer(message, self.strings("task_list").format(task_list_str))
@loader.command(
ru_doc="[index] - Посмотреть информацию о задаче",
en_doc="[index] - Show task info",
)
async def taskinfo(self, message):
args = utils.get_args_raw(message)
if not args:
await utils.answer(message, self.strings("index_required"))
return
try:
index = int(args) - 1
except ValueError:
await utils.answer(message, self.strings("invalid_index"))
return
task = self.task_manager.get_task(message.sender_id, index)
if not task:
await utils.answer(message, self.strings("task_not_found"))
return
due_date_str = (
task.due_date.strftime("%Y-%m-%d %H:%M") if task.due_date else "None"
)
created_at_str = task.created_at.strftime("%Y-%m-%d %H:%M")
await utils.answer(
message,
self.strings("task_info").format(
description=task.description,
due_date=due_date_str,
completed="Yes" if task.completed else "No",
created_at=created_at_str,
),
)
@loader.command(ru_doc="Удалить все задачи", en_doc="Clear all tasks")
async def taskclear(self, message):
await self.inline.form(
text=self.strings("clear_confirmation_text"),
message=message,
reply_markup=[
[
{"text": self.strings("confirm"), "callback": self.clear_confirm},
{"text": self.strings("cancel"), "callback": self.clear_cancel},
]
],
silent=True,
)
async def clear_confirm(self, call):
"""Callback for confirming task clearing."""
if await self.task_manager.clear_tasks(call.from_user.id):
await call.edit(self.strings("tasks_cleared"))
else:
await call.edit(self.strings("no_tasks"))
async def clear_cancel(self, call):
"""Callback for canceling task clearing."""
await call.edit(self.strings("clear_cancelled"))