cmautobuy/client/src/task_repository.py

"""PDD 任务的 SQLite Repository。

Repository 是数据库访问入口。界面和自动化代码不应自行拼接任务 SQL。
"""

import json
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, List, Optional, Tuple, Union
from uuid import uuid4

from .db import initialize_database, open_database
from .task_models import (
    NewClaimedTask,
    OutboxEventRecord,
    OutboxEventType,
    OutboxStatus,
    RunStatus,
    StartedTaskRun,
    TaskDetail,
    TaskFilters,
    TaskStatus,
    TaskSummary,
    TaskType,
)


PathValue = Union[str, Path]
MAX_PAGE_SIZE = 500


class DuplicateTaskError(ValueError):
    """相同远程任务编号已经存在，不能覆盖。"""


class CollectRerunError(ValueError):
    """当前任务不满足重新采集条件。"""


def utc_now_iso() -> str:
    """返回精确到秒的 UTC ISO 8601 时间。"""

    return datetime.now(timezone.utc).isoformat(timespec="seconds").replace(
        "+00:00", "Z"
    )


class TaskRepository:
    """保存和查询本机已经领取的 PDD 任务。"""

    def __init__(self, db_path: Optional[PathValue] = None):
        self._db_path = initialize_database(db_path)

    def add_claimed_task(
        self, task: NewClaimedTask, received_at: Optional[str] = None
    ) -> int:
        """写入一条新领取任务，返回本地自增编号。

        相同 ``remote_task_id`` 已存在时抛出 ``DuplicateTaskError``，
        不覆盖已经保存的本地状态。
        """

        now = received_at or utc_now_iso()
        payload = json.dumps(task.admin_payload, ensure_ascii=False)
        connection = open_database(self._db_path)
        try:
            try:
                with connection:
                    cursor = connection.execute(
                        "INSERT INTO pdd_tasks ("
                        " remote_task_id, task_type, goods_id, goods_url, title,"
                        " target_color, target_size, price_cent, quantity, status,"
                        " priority, version, admin_payload, received_at, created_at,"
                        " updated_at"
                        ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
                        (
                            task.remote_task_id.strip(),
                            task.task_type.value,
                            task.goods_id,
                            task.goods_url.strip(),
                            task.title,
                            task.target_color,
                            task.target_size,
                            task.price_cent,
                            task.quantity,
                            TaskStatus.CLAIMED.value,
                            task.priority,
                            task.version,
                            payload,
                            now,
                            now,
                            now,
                        ),
                    )
                return int(cursor.lastrowid)
            except sqlite3.IntegrityError as exc:
                if "pdd_tasks.remote_task_id" in str(exc):
                    raise DuplicateTaskError(
                        f"任务 {task.remote_task_id} 已经存在"
                    ) from exc
                raise
        finally:
            connection.close()

    def list_tasks(
        self,
        filters: Optional[TaskFilters] = None,
        limit: int = 50,
        offset: int = 0,
    ) -> List[TaskSummary]:
        """分页查询任务摘要，默认按更新时间和本地编号倒序。"""

        self._validate_page(limit, offset)
        where_sql, parameters = self._build_where(filters or TaskFilters())
        parameters.extend((limit, offset))
        connection = open_database(self._db_path)
        try:
            rows = connection.execute(
                "SELECT id, remote_task_id, task_type, goods_id, title,"
                " target_color, target_size, price_cent, quantity, status, updated_at"
                " FROM pdd_tasks"
                f"{where_sql}"
                " ORDER BY updated_at DESC, id DESC"
                " LIMIT ? OFFSET ?",
                parameters,
            ).fetchall()
        finally:
            connection.close()
        return [self._to_summary(row) for row in rows]

    def count_tasks(self, filters: Optional[TaskFilters] = None) -> int:
        """返回符合筛选条件的任务总数。"""

        where_sql, parameters = self._build_where(filters or TaskFilters())
        connection = open_database(self._db_path)
        try:
            row = connection.execute(
                f"SELECT COUNT(*) FROM pdd_tasks{where_sql}", parameters
            ).fetchone()
        finally:
            connection.close()
        return int(row[0])

    def get_task(self, remote_task_id: str) -> Optional[TaskDetail]:
        """按稳定远程编号读取完整任务；不存在时返回 None。"""

        connection = open_database(self._db_path)
        try:
            row = connection.execute(
                "SELECT * FROM pdd_tasks WHERE remote_task_id = ?",
                (remote_task_id,),
            ).fetchone()
        finally:
            connection.close()
        return self._to_detail(row) if row is not None else None

    def recover_interrupted_work(self) -> None:
        """恢复上次异常退出留下的可重试状态。"""

        now = utc_now_iso()
        connection = open_database(self._db_path)
        try:
            with connection:
                connection.execute(
                    "UPDATE outbox_events SET status = 'pending', updated_at = ?"
                    " WHERE status = 'sending'",
                    (now,),
                )
                # #35 之前，Admin 的失败响应缺少 result_id。Admin 实际已经
                # 接收，但旧 Client 把这类 2xx 响应误标为永久失败。恢复为
                # pending 后使用原幂等键重试，不会重复写入业务结果。
                connection.execute(
                    "UPDATE outbox_events SET status = 'pending', updated_at = ?"
                    " WHERE status = 'failed'"
                    " AND last_error = 'Admin 提交响应字段不完整'",
                    (now,),
                )
                connection.execute(
                    "UPDATE pdd_tasks SET status = 'retry_wait',"
                    " current_step = 'interrupted', retry_count = retry_count + 1,"
                    " last_error_code = 'CLIENT_INTERRUPTED',"
                    " last_error_message = '客户端上次执行期间退出', updated_at = ?"
                    " WHERE status = 'running' AND task_type = 'collect'",
                    (now,),
                )
                connection.execute(
                    "UPDATE task_runs SET run_status = 'failed',"
                    " error_code = 'CLIENT_INTERRUPTED',"
                    " error_message = '客户端上次执行期间退出',"
                    " finished_at = ?, updated_at = ?"
                    " WHERE run_status = 'running' AND irreversible_action_at IS NULL"
                    " AND task_id IN (SELECT id FROM pdd_tasks WHERE task_type = 'collect')",
                    (now, now),
                )
        finally:
            connection.close()

    def next_collect_task(self) -> Optional[TaskDetail]:
        """返回最早的本地待执行采集任务。"""

        connection = open_database(self._db_path)
        try:
            row = connection.execute(
                "SELECT * FROM pdd_tasks"
                " WHERE task_type = 'collect' AND status IN ('claimed', 'retry_wait')"
                " ORDER BY received_at ASC, id ASC LIMIT 1"
            ).fetchone()
        finally:
            connection.close()
        return self._to_detail(row) if row is not None else None

    def validate_collect_rerun(self, remote_task_id: str) -> TaskDetail:
        """校验任务能否重新采集，成功时返回任务详情。"""

        connection = open_database(self._db_path)
        try:
            row = connection.execute(
                "SELECT * FROM pdd_tasks WHERE remote_task_id = ?",
                (remote_task_id,),
            ).fetchone()
            if row is None:
                raise CollectRerunError(f"任务 {remote_task_id} 不存在")
            self._check_collect_rerun(connection, row)
        finally:
            connection.close()
        return self._to_detail(row)

    def prepare_collect_rerun(self, remote_task_id: str) -> TaskDetail:
        """事务内把已结束的采集任务恢复为待执行，保留旧结果。"""

        now = utc_now_iso()
        connection = open_database(self._db_path)
        try:
            with connection:
                row = connection.execute(
                    "SELECT * FROM pdd_tasks WHERE remote_task_id = ?",
                    (remote_task_id,),
                ).fetchone()
                if row is None:
                    raise CollectRerunError(f"任务 {remote_task_id} 不存在")
                self._check_collect_rerun(connection, row)
                connection.execute(
                    "UPDATE pdd_tasks SET status = 'claimed',"
                    " current_step = 'rerun_requested', finished_at = NULL,"
                    " last_error_code = NULL, last_error_message = NULL,"
                    " updated_at = ? WHERE id = ?",
                    (now, row["id"]),
                )
        finally:
            connection.close()
        task = self.get_task(remote_task_id)
        assert task is not None
        return task

    @staticmethod
    def _check_collect_rerun(
        connection: sqlite3.Connection, row: sqlite3.Row
    ) -> None:
        """检查重新采集的类型、终态和 Outbox 约束。"""

        if row["task_type"] != TaskType.COLLECT.value:
            raise CollectRerunError("采购任务不能重新执行，以免重复下单")
        allowed_statuses = {
            TaskStatus.SUCCEEDED.value,
            TaskStatus.FAILED.value,
            TaskStatus.CANCELLED.value,
            TaskStatus.MANUAL_REVIEW.value,
        }
        if row["status"] not in allowed_statuses:
            status_name = {
                TaskStatus.CLAIMED.value: "待执行",
                TaskStatus.RUNNING.value: "执行中",
                TaskStatus.RESULT_PENDING.value: "结果待提交",
                TaskStatus.RETRY_WAIT.value: "等待重试",
            }.get(row["status"], row["status"])
            raise CollectRerunError(f"任务当前为“{status_name}”，不能重新采集")
        unsent_count = int(
            connection.execute(
                "SELECT COUNT(*) FROM outbox_events"
                " WHERE task_id = ? AND status != 'sent'",
                (row["id"],),
            ).fetchone()[0]
        )
        if unsent_count:
            raise CollectRerunError("任务仍有未发送的结果，请先完成提交")

    def start_collect_run(
        self, remote_task_id: str, device_address: str
    ) -> StartedTaskRun:
        """原子地把待执行任务改为执行中，并创建一次运行记录。"""

        now = utc_now_iso()
        attempt_id = str(uuid4())
        connection = open_database(self._db_path)
        try:
            with connection:
                row = connection.execute(
                    "SELECT * FROM pdd_tasks WHERE remote_task_id = ?",
                    (remote_task_id,),
                ).fetchone()
                if row is None:
                    raise ValueError(f"任务 {remote_task_id} 不存在")
                if row["task_type"] != TaskType.COLLECT.value:
                    raise ValueError("当前只能执行采集任务")
                if row["status"] not in {
                    TaskStatus.CLAIMED.value,
                    TaskStatus.RETRY_WAIT.value,
                }:
                    raise ValueError(f"任务状态 {row['status']} 不能开始采集")
                attempt_no = int(
                    connection.execute(
                        "SELECT COALESCE(MAX(attempt_no), 0) + 1"
                        " FROM task_runs WHERE task_id = ?",
                        (row["id"],),
                    ).fetchone()[0]
                )
                connection.execute(
                    "UPDATE pdd_tasks SET status = 'running',"
                    " current_step = 'collecting', started_at = COALESCE(started_at, ?),"
                    " last_error_code = NULL, last_error_message = NULL, updated_at = ?"
                    " WHERE id = ?",
                    (now, now, row["id"]),
                )
                connection.execute(
                    "INSERT INTO task_runs (task_id, attempt_id, attempt_no,"
                    " device_address, run_status, current_step, started_at,"
                    " created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
                    (
                        row["id"], attempt_id, attempt_no, device_address,
                        RunStatus.RUNNING.value, "collecting", now, now, now,
                    ),
                )
            task = self.get_task(remote_task_id)
            assert task is not None
            return StartedTaskRun(task, attempt_id, attempt_no)
        finally:
            connection.close()

    def save_collect_result(
        self,
        remote_task_id: str,
        attempt_id: str,
        pdd_data: Dict[str, object],
    ) -> OutboxEventRecord:
        """在一个事务中保存采集结果并创建待提交事件。"""

        now = utc_now_iso()
        connection = open_database(self._db_path)
        try:
            with connection:
                task = connection.execute(
                    "SELECT id, version FROM pdd_tasks WHERE remote_task_id = ?",
                    (remote_task_id,),
                ).fetchone()
                if task is None:
                    raise ValueError(f"任务 {remote_task_id} 不存在")
                payload = {
                    "task_version": task["version"],
                    "attempt_id": attempt_id,
                    "result_type": "collect",
                    "completed_at": now,
                    "pdd_data": pdd_data,
                }
                idempotency_key = f"{remote_task_id}:{attempt_id}:result-v1"
                result_json = json.dumps(pdd_data, ensure_ascii=False)
                connection.execute(
                    "UPDATE pdd_tasks SET status = 'result_pending',"
                    " current_step = 'submit_result', pdd_data = ?, goods_id = ?,"
                    " title = ?, price_cent = ?, finished_at = ?, updated_at = ?"
                    " WHERE id = ?",
                    (
                        result_json,
                        pdd_data.get("goods_id"), pdd_data.get("title"),
                        self._summary_price(pdd_data), now, now, task["id"],
                    ),
                )
                connection.execute(
                    "UPDATE task_runs SET run_status = 'succeeded',"
                    " current_step = 'submit_result', result_data = ?,"
                    " finished_at = ?, updated_at = ?"
                    " WHERE attempt_id = ?",
                    (result_json, now, now, attempt_id),
                )
                cursor = connection.execute(
                    "INSERT INTO outbox_events (task_id, event_type, idempotency_key,"
                    " payload_json, status, created_at, updated_at)"
                    " VALUES (?, 'collect_result', ?, ?, 'pending', ?, ?)",
                    (
                        task["id"], idempotency_key,
                        json.dumps(payload, ensure_ascii=False), now, now,
                    ),
                )
                event_id = int(cursor.lastrowid)
            event = self.get_outbox_event(event_id)
            assert event is not None
            return event
        finally:
            connection.close()

    def save_collect_failure(
        self,
        remote_task_id: str,
        attempt_id: str,
        status: TaskStatus,
        error_code: str,
        error_message: str,
        retryable: bool,
        diagnostics: Optional[Dict[str, object]] = None,
    ) -> OutboxEventRecord:
        """保存结构化失败，并可靠排队提交 Admin。"""

        if status not in {
            TaskStatus.RETRY_WAIT, TaskStatus.MANUAL_REVIEW,
            TaskStatus.FAILED, TaskStatus.CANCELLED,
        }:
            raise ValueError("失败状态无效")
        now = utc_now_iso()
        connection = open_database(self._db_path)
        try:
            with connection:
                task = connection.execute(
                    "SELECT id, version FROM pdd_tasks WHERE remote_task_id = ?",
                    (remote_task_id,),
                ).fetchone()
                if task is None:
                    raise ValueError(f"任务 {remote_task_id} 不存在")
                payload = {
                    "task_version": task["version"],
                    "attempt_id": attempt_id,
                    "status": status.value,
                    "error": {
                        "code": error_code,
                        "message": error_message,
                        "retryable": retryable,
                        "step": "collecting",
                    },
                    "diagnostics": diagnostics or {"artifacts": []},
                    "reported_at": now,
                }
                idempotency_key = f"{remote_task_id}:{attempt_id}:failure-v1"
                connection.execute(
                    "UPDATE pdd_tasks SET status = ?, current_step = 'failed',"
                    " retry_count = retry_count + ?, last_error_code = ?,"
                    " last_error_message = ?, finished_at = ?, updated_at = ?"
                    " WHERE id = ?",
                    (
                        status.value, 1 if status is TaskStatus.RETRY_WAIT else 0,
                        error_code, error_message, now, now, task["id"],
                    ),
                )
                run_status = {
                    TaskStatus.CANCELLED: RunStatus.CANCELLED,
                    TaskStatus.MANUAL_REVIEW: RunStatus.MANUAL_REVIEW,
                }.get(status, RunStatus.FAILED)
                connection.execute(
                    "UPDATE task_runs SET run_status = ?, current_step = 'failed',"
                    " error_code = ?, error_message = ?, finished_at = ?, updated_at = ?"
                    " WHERE attempt_id = ?",
                    (run_status.value, error_code, error_message, now, now, attempt_id),
                )
                cursor = connection.execute(
                    "INSERT INTO outbox_events (task_id, event_type, idempotency_key,"
                    " payload_json, status, created_at, updated_at)"
                    " VALUES (?, 'task_failure', ?, ?, 'pending', ?, ?)",
                    (
                        task["id"], idempotency_key,
                        json.dumps(payload, ensure_ascii=False), now, now,
                    ),
                )
                event_id = int(cursor.lastrowid)
            event = self.get_outbox_event(event_id)
            assert event is not None
            return event
        finally:
            connection.close()

    def next_pending_outbox(self) -> Optional[OutboxEventRecord]:
        connection = open_database(self._db_path)
        try:
            row = connection.execute(
                "SELECT o.* FROM outbox_events o JOIN pdd_tasks t ON t.id = o.task_id"
                " WHERE o.status = 'pending'"
                " AND (o.next_retry_at IS NULL OR o.next_retry_at <= ?)"
                " ORDER BY o.id ASC LIMIT 1",
                (utc_now_iso(),),
            ).fetchone()
        finally:
            connection.close()
        return self._to_outbox(row) if row is not None else None

    def get_outbox_event(self, event_id: int) -> Optional[OutboxEventRecord]:
        connection = open_database(self._db_path)
        try:
            row = connection.execute(
                "SELECT * FROM outbox_events WHERE id = ?", (event_id,)
            ).fetchone()
        finally:
            connection.close()
        return self._to_outbox(row) if row is not None else None

    def outbox_task_id(self, event_id: int) -> str:
        connection = open_database(self._db_path)
        try:
            row = connection.execute(
                "SELECT t.remote_task_id FROM outbox_events o"
                " JOIN pdd_tasks t ON t.id = o.task_id WHERE o.id = ?",
                (event_id,),
            ).fetchone()
        finally:
            connection.close()
        if row is None:
            raise ValueError(f"Outbox {event_id} 不存在")
        return str(row[0])

    def mark_outbox_sending(self, event_id: int) -> None:
        self._update_outbox(event_id, "sending", None)

    def mark_outbox_retry(self, event_id: int, message: str) -> None:
        self._update_outbox(event_id, "pending", message, increment=True)

    def mark_outbox_failed(self, event_id: int, message: str) -> None:
        self._update_outbox(event_id, "failed", message, increment=True)

    def mark_outbox_sent(self, event_id: int) -> None:
        now = utc_now_iso()
        connection = open_database(self._db_path)
        try:
            with connection:
                row = connection.execute(
                    "SELECT task_id, event_type FROM outbox_events WHERE id = ?",
                    (event_id,),
                ).fetchone()
                if row is None:
                    raise ValueError(f"Outbox {event_id} 不存在")
                connection.execute(
                    "UPDATE outbox_events SET status = 'sent', sent_at = ?,"
                    " updated_at = ?, attempt_count = attempt_count + 1"
                    " WHERE id = ?",
                    (now, now, event_id),
                )
                if row["event_type"] == OutboxEventType.COLLECT_RESULT.value:
                    connection.execute(
                        "UPDATE pdd_tasks SET status = 'succeeded',"
                        " current_step = 'completed', updated_at = ? WHERE id = ?",
                        (now, row["task_id"]),
                    )
        finally:
            connection.close()

    def _update_outbox(
        self, event_id: int, status: str, message: Optional[str], increment: bool = False
    ) -> None:
        connection = open_database(self._db_path)
        try:
            with connection:
                cursor = connection.execute(
                    "UPDATE outbox_events SET status = ?, last_error = ?, updated_at = ?,"
                    f" attempt_count = attempt_count + {1 if increment else 0} WHERE id = ?",
                    (status, message, utc_now_iso(), event_id),
                )
                if cursor.rowcount != 1:
                    raise ValueError(f"Outbox {event_id} 不存在")
        finally:
            connection.close()

    @staticmethod
    def _summary_price(pdd_data: Dict[str, object]) -> Optional[int]:
        skus = pdd_data.get("skus")
        if not isinstance(skus, list):
            return None
        prices = [
            item.get("price_cent") for item in skus
            if isinstance(item, dict) and isinstance(item.get("price_cent"), int)
        ]
        return min(prices) if prices else None

    @staticmethod
    def _to_outbox(row: sqlite3.Row) -> OutboxEventRecord:
        return OutboxEventRecord(
            id=row["id"], task_id=row["task_id"],
            event_type=OutboxEventType(row["event_type"]),
            idempotency_key=row["idempotency_key"],
            payload_json=TaskRepository._load_json_object(row["payload_json"]),
            status=OutboxStatus(row["status"]), attempt_count=row["attempt_count"],
            next_retry_at=row["next_retry_at"], last_error=row["last_error"],
            created_at=row["created_at"], updated_at=row["updated_at"],
            sent_at=row["sent_at"],
        )

    @staticmethod
    def _validate_page(limit: int, offset: int) -> None:
        if not 1 <= limit <= MAX_PAGE_SIZE:
            raise ValueError(f"limit 必须在 1 到 {MAX_PAGE_SIZE} 之间")
        if offset < 0:
            raise ValueError("offset 不能小于 0")

    @staticmethod
    def _build_where(filters: TaskFilters) -> Tuple[str, List[object]]:
        clauses = []
        parameters: List[object] = []
        if filters.task_type is not None:
            clauses.append("task_type = ?")
            parameters.append(filters.task_type.value)
        if filters.status is not None:
            clauses.append("status = ?")
            parameters.append(filters.status.value)
        if filters.keyword.strip():
            keyword = TaskRepository._escape_like(filters.keyword.strip())
            pattern = f"%{keyword}%"
            clauses.append(
                "(remote_task_id LIKE ? ESCAPE '\\'"
                " OR COALESCE(goods_id, '') LIKE ? ESCAPE '\\'"
                " OR COALESCE(title, '') LIKE ? ESCAPE '\\')"
            )
            parameters.extend((pattern, pattern, pattern))
        return (" WHERE " + " AND ".join(clauses) if clauses else "", parameters)

    @staticmethod
    def _escape_like(value: str) -> str:
        return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")

    @staticmethod
    def _to_summary(row: sqlite3.Row) -> TaskSummary:
        return TaskSummary(
            id=row["id"],
            remote_task_id=row["remote_task_id"],
            task_type=TaskType(row["task_type"]),
            goods_id=row["goods_id"],
            title=row["title"],
            target_color=row["target_color"],
            target_size=row["target_size"],
            price_cent=row["price_cent"],
            quantity=row["quantity"],
            status=TaskStatus(row["status"]),
            updated_at=row["updated_at"],
        )

    @staticmethod
    def _to_detail(row: sqlite3.Row) -> TaskDetail:
        admin_payload = TaskRepository._load_json_object(row["admin_payload"])
        pdd_data = (
            TaskRepository._load_json_object(row["pdd_data"])
            if row["pdd_data"] is not None
            else None
        )
        return TaskDetail(
            id=row["id"],
            remote_task_id=row["remote_task_id"],
            task_type=TaskType(row["task_type"]),
            goods_id=row["goods_id"],
            goods_url=row["goods_url"],
            title=row["title"],
            target_color=row["target_color"],
            target_size=row["target_size"],
            price_cent=row["price_cent"],
            quantity=row["quantity"],
            status=TaskStatus(row["status"]),
            current_step=row["current_step"],
            priority=row["priority"],
            version=row["version"],
            admin_payload=admin_payload,
            pdd_data=pdd_data,
            retry_count=row["retry_count"],
            last_error_code=row["last_error_code"],
            last_error_message=row["last_error_message"],
            received_at=row["received_at"],
            started_at=row["started_at"],
            finished_at=row["finished_at"],
            created_at=row["created_at"],
            updated_at=row["updated_at"],
        )

    @staticmethod
    def _load_json_object(text: str) -> Dict[str, object]:
        value = json.loads(text)
        if not isinstance(value, dict):
            raise ValueError("数据库 JSON 字段必须是对象")
        return value