100 lines
4.5 KiB
Python
100 lines
4.5 KiB
Python
import aiosqlite
|
||
import datetime
|
||
import uuid
|
||
from typing import List, Optional, Dict, Any
|
||
|
||
class Database:
|
||
def __init__(self, db_path: str = "tasks.db"):
|
||
self.db_path = db_path
|
||
|
||
async def initialize(self):
|
||
async with aiosqlite.connect(self.db_path) as db:
|
||
await db.execute("""
|
||
CREATE TABLE IF NOT EXISTS tasks (
|
||
id TEXT PRIMARY KEY,
|
||
user_id INTEGER NOT NULL,
|
||
user_name TEXT,
|
||
guild_id INTEGER,
|
||
channel_id INTEGER,
|
||
title TEXT NOT NULL,
|
||
description TEXT,
|
||
priority TEXT NOT NULL,
|
||
deadline TEXT NOT NULL,
|
||
is_completed INTEGER DEFAULT 0,
|
||
notion_page_id TEXT,
|
||
created_at TEXT DEFAULT (DATETIME('now', 'localtime')),
|
||
last_reminded_type TEXT DEFAULT ''
|
||
)
|
||
""")
|
||
await db.commit()
|
||
|
||
async def add_task(self, user_id: int, user_name: str, guild_id: Optional[int],
|
||
channel_id: int, title: str, description: str,
|
||
priority: str, deadline: str) -> str:
|
||
task_id = str(uuid.uuid4())[:8] # 短めのUUID（8文字）を使用
|
||
async with aiosqlite.connect(self.db_path) as db:
|
||
await db.execute("""
|
||
INSERT INTO tasks (id, user_id, user_name, guild_id, channel_id, title, description, priority, deadline)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""", (task_id, user_id, user_name, guild_id, channel_id, title, description, priority, deadline))
|
||
await db.commit()
|
||
return task_id
|
||
|
||
async def get_user_tasks(self, user_id: int, channel_id: Optional[int] = None) -> List[Dict[str, Any]]:
|
||
async with aiosqlite.connect(self.db_path) as db:
|
||
db.row_factory = aiosqlite.Row
|
||
if channel_id:
|
||
query = "SELECT * FROM tasks WHERE user_id = ? AND channel_id = ? AND is_completed = 0 ORDER BY deadline ASC"
|
||
params = (user_id, channel_id)
|
||
else:
|
||
query = "SELECT * FROM tasks WHERE user_id = ? AND is_completed = 0 ORDER BY deadline ASC"
|
||
params = (user_id,)
|
||
|
||
async with db.execute(query, params) as cursor:
|
||
rows = await cursor.fetchall()
|
||
return [dict(row) for row in rows]
|
||
|
||
async def get_all_active_tasks(self) -> List[Dict[str, Any]]:
|
||
async with aiosqlite.connect(self.db_path) as db:
|
||
db.row_factory = aiosqlite.Row
|
||
async with db.execute("SELECT * FROM tasks WHERE is_completed = 0") as cursor:
|
||
rows = await cursor.fetchall()
|
||
return [dict(row) for row in rows]
|
||
|
||
async def get_task_by_id(self, task_id: str) -> Optional[Dict[str, Any]]:
|
||
async with aiosqlite.connect(self.db_path) as db:
|
||
db.row_factory = aiosqlite.Row
|
||
async with db.execute("SELECT * FROM tasks WHERE id = ?", (task_id,)) as cursor:
|
||
row = await cursor.fetchone()
|
||
return dict(row) if row else None
|
||
|
||
async def update_priority(self, task_id: str, new_priority: str):
|
||
async with aiosqlite.connect(self.db_path) as db:
|
||
await db.execute("UPDATE tasks SET priority = ? WHERE id = ?", (new_priority, task_id))
|
||
await db.commit()
|
||
|
||
async def update_task(self, task_id: str, title: str, description: str,
|
||
priority: str, deadline: str):
|
||
async with aiosqlite.connect(self.db_path) as db:
|
||
await db.execute("""
|
||
UPDATE tasks
|
||
SET title = ?, description = ?, priority = ?, deadline = ?
|
||
WHERE id = ?
|
||
""", (title, description, priority, deadline, task_id))
|
||
await db.commit()
|
||
|
||
async def update_reminded_type(self, task_id: str, reminded_type: str):
|
||
async with aiosqlite.connect(self.db_path) as db:
|
||
await db.execute("UPDATE tasks SET last_reminded_type = ? WHERE id = ?", (reminded_type, task_id))
|
||
await db.commit()
|
||
|
||
async def complete_task(self, task_id: str):
|
||
async with aiosqlite.connect(self.db_path) as db:
|
||
await db.execute("UPDATE tasks SET is_completed = 1 WHERE id = ?", (task_id,))
|
||
await db.commit()
|
||
|
||
async def delete_task(self, task_id: str):
|
||
async with aiosqlite.connect(self.db_path) as db:
|
||
await db.execute("DELETE FROM tasks WHERE id = ?", (task_id,))
|
||
await db.commit()
|