Создать ИИ помощника для автоматизации задач

.env.example

@@ -0,0 +1,4 @@
+# Copy this to .env and fill in your values
+TELEGRAM_BOT_TOKEN=
+# IANA timezone, e.g. Europe/Warsaw, Europe/Moscow, America/New_York
+TIMEZONE=

bot/db.py

@@ -0,0 +1,174 @@
+import sqlite3
+from dataclasses import dataclass
+from typing import List, Optional, Tuple, Iterable
+from datetime import date
+
+
+@dataclass
+class Plan:
+    id: int
+    chat_id: int
+    title: str
+    time_hhmm: str  # 'HH:MM'
+    recurrence: str  # 'daily' | 'weekly' | 'monthly' | 'once'
+    dow: Optional[int]  # 0..6 (Mon..Sun)
+    dom: Optional[int]  # 1..31
+    once_date: Optional[str]  # 'YYYY-MM-DD'
+    active: int
+
+
+class Database:
+    def __init__(self, path: str) -> None:
+        self.path = path
+        self._ensure_schema()
+
+    def _conn(self) -> sqlite3.Connection:
+        conn = sqlite3.connect(self.path)
+        conn.row_factory = sqlite3.Row
+        return conn
+
+    def _ensure_schema(self) -> None:
+        with self._conn() as conn:
+            cur = conn.cursor()
+            cur.execute(
+                """
+                CREATE TABLE IF NOT EXISTS plans (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    chat_id INTEGER NOT NULL,
+                    title TEXT NOT NULL,
+                    time_hhmm TEXT NOT NULL,
+                    recurrence TEXT NOT NULL,
+                    dow INTEGER,
+                    dom INTEGER,
+                    once_date TEXT,
+                    active INTEGER NOT NULL DEFAULT 1
+                )
+                """
+            )
+            cur.execute(
+                """
+                CREATE TABLE IF NOT EXISTS completions (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    plan_id INTEGER NOT NULL,
+                    date TEXT NOT NULL, -- 'YYYY-MM-DD'
+                    done INTEGER NOT NULL,
+                    UNIQUE(plan_id, date),
+                    FOREIGN KEY(plan_id) REFERENCES plans(id) ON DELETE CASCADE
+                )
+                """
+            )
+            conn.commit()
+
+    # Plans
+    def add_plan(
+        self,
+        chat_id: int,
+        title: str,
+        time_hhmm: str,
+        recurrence: str,
+        dow: Optional[int] = None,
+        dom: Optional[int] = None,
+        once_date: Optional[str] = None,
+    ) -> int:
+        with self._conn() as conn:
+            cur = conn.cursor()
+            cur.execute(
+                """
+                INSERT INTO plans (chat_id, title, time_hhmm, recurrence, dow, dom, once_date)
+                VALUES (?, ?, ?, ?, ?, ?, ?)
+                """,
+                (chat_id, title, time_hhmm, recurrence, dow, dom, once_date),
+            )
+            conn.commit()
+            return int(cur.lastrowid)
+
+    def get_plan(self, plan_id: int) -> Optional[Plan]:
+        with self._conn() as conn:
+            row = conn.execute("SELECT * FROM plans WHERE id = ?", (plan_id,)).fetchone()
+            return self._row_to_plan(row) if row else None
+
+    def list_plans_for_chat(self, chat_id: int) -> List[Plan]:
+        with self._conn() as conn:
+            rows = conn.execute(
+                "SELECT * FROM plans WHERE chat_id = ? AND active = 1 ORDER BY time_hhmm",
+                (chat_id,),
+            ).fetchall()
+            return [self._row_to_plan(r) for r in rows]
+
+    def get_all_plans(self) -> List[Plan]:
+        with self._conn() as conn:
+            rows = conn.execute("SELECT * FROM plans WHERE active = 1").fetchall()
+            return [self._row_to_plan(r) for r in rows]
+
+    def _row_to_plan(self, row: sqlite3.Row) -> Plan:
+        return Plan(
+            id=row["id"],
+            chat_id=row["chat_id"],
+            title=row["title"],
+            time_hhmm=row["time_hhmm"],
+            recurrence=row["recurrence"],
+            dow=row["dow"],
+            dom=row["dom"],
+            once_date=row["once_date"],
+            active=row["active"],
+        )
+
+    # Completions
+    def get_completion(self, plan_id: int, day: date) -> bool:
+        day_str = day.isoformat()
+        with self._conn() as conn:
+            row = conn.execute(
+                "SELECT done FROM completions WHERE plan_id = ? AND date = ?",
+                (plan_id, day_str),
+            ).fetchone()
+            return bool(row["done"]) if row else False
+
+    def set_completion(self, plan_id: int, day: date, done: bool) -> None:
+        day_str = day.isoformat()
+        with self._conn() as conn:
+            conn.execute(
+                """
+                INSERT INTO completions (plan_id, date, done)
+                VALUES (?, ?, ?)
+                ON CONFLICT(plan_id, date) DO UPDATE SET done = excluded.done
+                """,
+                (plan_id, day_str, int(done)),
+            )
+            conn.commit()
+
+    def toggle_completion(self, plan_id: int, day: date) -> bool:
+        new_value = not self.get_completion(plan_id, day)
+        self.set_completion(plan_id, day, new_value)
+        return new_value
+
+    # Queries for today
+    def plans_for_date(self, chat_id: int, day: date) -> List[Tuple[Plan, bool]]:
+        plans = self.list_plans_for_chat(chat_id)
+        result: List[Tuple[Plan, bool]] = []
+        for p in plans:
+            if self._occurs_on(p, day):
+                result.append((p, self.get_completion(p.id, day)))
+        # sort by time
+        result.sort(key=lambda t: t[0].time_hhmm)
+        return result
+
+    def _occurs_on(self, plan: Plan, day: date) -> bool:
+        if plan.recurrence == "daily":
+            return True
+        if plan.recurrence == "weekly":
+            return plan.dow == day.weekday()
+        if plan.recurrence == "monthly":
+            # If day-of-month larger than actual month days, schedule on last day
+            dom = plan.dom or 1
+            return day.day == dom or (
+                day.day == self._last_day_of_month(day.year, day.month) and dom > day.day
+            )
+        if plan.recurrence == "once":
+            return (plan.once_date == day.isoformat())
+        return False
+
+    @staticmethod
+    def _last_day_of_month(year: int, month: int) -> int:
+        # February leap year logic minimal
+        from calendar import monthrange
+        return monthrange(year, month)[1]