237 lines
8.4 KiB
Python
237 lines
8.4 KiB
Python
"""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 .db import initialize_database, open_database
|
||
from .task_models import (
|
||
NewClaimedTask,
|
||
TaskDetail,
|
||
TaskFilters,
|
||
TaskStatus,
|
||
TaskSummary,
|
||
TaskType,
|
||
)
|
||
|
||
|
||
PathValue = Union[str, Path]
|
||
MAX_PAGE_SIZE = 500
|
||
|
||
|
||
class DuplicateTaskError(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
|
||
|
||
@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
|