add-todo-all/scripts/service/todo_service.py

"""待办事项领域服务（业务内核）。

按模板铁律：「单业务内核、多入口」——CLI / Action / 数据管理按钮 / 定时任务
都调用这里的同一个函数，禁止按入口分叉业务行为。

直接用 sqlite3 做持久化（demo 简化，未抽 repository）。
"""
from __future__ import annotations

import time
from dataclasses import dataclass
from typing import Optional
import logging

from db.connection import get_conn
from util.logging_config import get_skill_logger, setup_skill_logging

logger = logging.getLogger(__name__)

# 业务常量（不暴露给宿主，技能自定）
STATUS_PENDING = "pending"
STATUS_DONE = "done"
VALID_STATUS = {STATUS_PENDING, STATUS_DONE}

PRIORITY_HIGH = "high"
PRIORITY_NORMAL = "normal"
PRIORITY_LOW = "low"
VALID_PRIORITY = {PRIORITY_HIGH, PRIORITY_NORMAL, PRIORITY_LOW}


@dataclass
class Todo:
    id: int
    title: str
    status: str
    priority: str
    due_at: Optional[int]
    created_at: int
    updated_at: int

    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "title": self.title,
            "status": self.status,
            "priority": self.priority,
            "due_at": self.due_at,
            "created_at": self.created_at,
            "updated_at": self.updated_at,
        }


def _now() -> int:
    return int(time.time())


def add_todo(
    *,
    title: str,
    priority: str = PRIORITY_NORMAL,
    due_at: Optional[int] = None,
) -> Todo:
    """新增一条待办。领域校验：标题非空、优先级合法。"""
    title = (title or "").strip()
    if not title:
        logger.warning("标题不能为空")
        raise ValueError("标题不能为空")
    if priority not in VALID_PRIORITY:
        logger.warning(f"优先级不合法，可选：{sorted(VALID_PRIORITY)}")
        raise ValueError(f"优先级不合法，可选：{sorted(VALID_PRIORITY)}")
    now = _now()
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute(
            "INSERT INTO todos (title, status, priority, due_at, created_at, updated_at) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            (title, STATUS_PENDING, priority, due_at, now, now),
        )
        new_id = cur.lastrowid
        conn.commit()
        return Todo(new_id, title, STATUS_PENDING, priority, due_at, now, now)
    finally:
        conn.close()


def list_todos(*, status: Optional[str] = None, limit: int = 100) -> list[Todo]:
    """列出待办；可按状态过滤。默认按创建时间倒序。"""
    sql = "SELECT id, title, status, priority, due_at, created_at, updated_at FROM todos"
    params: list = []
    if status:
        if status not in VALID_STATUS:
            logger.warning(f"状态不合法，可选：{sorted(VALID_STATUS)}")
            raise ValueError(f"状态不合法，可选：{sorted(VALID_STATUS)}")
        sql += " WHERE status = ?"
        params.append(status)
    sql += " ORDER BY created_at DESC LIMIT ?"
    params.append(limit)
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute(sql, params)
        rows = cur.fetchall()
        return [Todo(*row) for row in rows]
    finally:
        conn.close()


def get_todo(todo_id: int) -> Optional[Todo]:
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute(
            "SELECT id, title, status, priority, due_at, created_at, updated_at "
            "FROM todos WHERE id = ?",
            (todo_id,),
        )
        row = cur.fetchone()
        return Todo(*row) if row else None
    finally:
        conn.close()


def mark_done(todo_id: int) -> Todo:
    """标记一条待办为完成。"""
    todo = get_todo(todo_id)
    if not todo:
        logger.warning(f"待办 {todo_id} 不存在")
        raise ValueError(f"待办 {todo_id} 不存在")
    if todo.status == STATUS_DONE:
        logger.warning(f"待办 {todo_id} 已是完成状态")
        raise ValueError(f"待办 {todo_id} 已是完成状态")
    now = _now()
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute(
            "UPDATE todos SET status = ?, updated_at = ? WHERE id = ?",
            (STATUS_DONE, now, todo_id),
        )
        conn.commit()
        todo.status = STATUS_DONE
        todo.updated_at = now
        return todo
    finally:
        conn.close()


def delete_todo(todo_id: int) -> bool:
    """删除一条待办；返回是否真的删了一行。"""
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute("DELETE FROM todos WHERE id = ?", (todo_id,))
        conn.commit()
        return cur.rowcount > 0
    finally:
        conn.close()


def stats() -> dict:
    """简单统计：用于 ``todo stats`` 命令与 demo 报告。"""
    conn = get_conn()
    try:
        cur = conn.cursor()
        cur.execute("SELECT status, COUNT(*) FROM todos GROUP BY status")
        by_status = {row[0]: int(row[1]) for row in cur.fetchall()}
        cur.execute("SELECT COUNT(*) FROM todos")
        total = int(cur.fetchone()[0])
        return {
            "total": total,
            "pending": by_status.get(STATUS_PENDING, 0),
            "done": by_status.get(STATUS_DONE, 0),
        }
    finally:
        conn.close()