|
|
import sqlite3, os, datetime, csv |
|
|
import matplotlib.pyplot as plt |
|
|
|
|
|
DB_PATH = os.path.join(os.path.dirname(__file__), "tasks.db") |
|
|
|
|
|
def init_db(): |
|
|
conn = sqlite3.connect(DB_PATH) |
|
|
c = conn.cursor() |
|
|
c.execute("""CREATE TABLE IF NOT EXISTS tasks ( |
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT, |
|
|
user TEXT, |
|
|
title TEXT, |
|
|
description TEXT, |
|
|
status TEXT, |
|
|
tags TEXT, |
|
|
created_at TEXT |
|
|
)""") |
|
|
conn.commit() |
|
|
conn.close() |
|
|
|
|
|
def add_task(user, payload): |
|
|
title = payload.get("title") or payload.get("name") or "Untitled" |
|
|
desc = payload.get("description", "") |
|
|
status = payload.get("status", "todo") |
|
|
tags = ",".join(payload.get("tags", [])) if isinstance(payload.get("tags"), list) else (payload.get("tags") or "") |
|
|
created_at = datetime.datetime.utcnow().isoformat() |
|
|
conn = sqlite3.connect(DB_PATH) |
|
|
c = conn.cursor() |
|
|
c.execute("INSERT INTO tasks (user, title, description, status, tags, created_at) VALUES (?,?,?,?,?,?)", |
|
|
(user, title, desc, status, tags, created_at)) |
|
|
conn.commit() |
|
|
conn.close() |
|
|
|
|
|
def list_tasks(user, limit=20): |
|
|
conn = sqlite3.connect(DB_PATH) |
|
|
c = conn.cursor() |
|
|
c.execute("SELECT id, title, status, created_at FROM tasks WHERE user=? ORDER BY id DESC LIMIT ?", (user, limit)) |
|
|
rows = c.fetchall() |
|
|
conn.close() |
|
|
return rows |
|
|
|
|
|
def get_dashboard_fig(user): |
|
|
conn = sqlite3.connect(DB_PATH) |
|
|
c = conn.cursor() |
|
|
c.execute("SELECT status, COUNT(*) FROM tasks WHERE user=? GROUP BY status", (user,)) |
|
|
rows = c.fetchall() |
|
|
conn.close() |
|
|
statuses = [r[0] for r in rows] if rows else [] |
|
|
counts = [r[1] for r in rows] if rows else [] |
|
|
if not rows: |
|
|
|
|
|
fig, ax = plt.subplots() |
|
|
ax.text(0.5, 0.5, 'No tasks yet', horizontalalignment='center', verticalalignment='center', fontsize=12) |
|
|
ax.axis('off') |
|
|
return fig |
|
|
fig, ax = plt.subplots() |
|
|
ax.pie(counts, labels=statuses, autopct='%1.1f%%') |
|
|
ax.set_title('Tasks by Status') |
|
|
return fig |
|
|
|
|
|
def export_tasks_csv(user): |
|
|
path = os.path.join(os.path.dirname(__file__), f"tasks_export_{user}.csv") |
|
|
conn = sqlite3.connect(DB_PATH) |
|
|
c = conn.cursor() |
|
|
c.execute("SELECT id, title, description, status, tags, created_at FROM tasks WHERE user=? ORDER BY id DESC", (user,)) |
|
|
rows = c.fetchall() |
|
|
conn.close() |
|
|
with open(path, 'w', newline='', encoding='utf-8') as f: |
|
|
writer = csv.writer(f) |
|
|
writer.writerow(["id","title","description","status","tags","created_at"]) |
|
|
for r in rows: |
|
|
writer.writerow(r) |
|
|
return path |
|
|
|
