| 1 | import asyncio |
| 2 | import json |
| 3 | from datetime import datetime |
| 4 | from typing import Any, Dict, List, Optional |
| 5 | |
| 6 | import aiosqlite |
| 7 | |
| 8 | |
| 9 | class Database: |
| 10 | def __init__(self, db_path: str = "dy_downloader.db"): |
| 11 | self.db_path = db_path |
| 12 | self._initialized = False |
| 13 | self._conn: Optional[aiosqlite.Connection] = None |
| 14 | # 延迟到首次 _get_conn 调用时在当前 event loop 上创建 Lock， |
| 15 | # 避免在 __init__ 阶段抢到错误的 loop。 |
| 16 | self._conn_lock: Optional[asyncio.Lock] = None |
| 17 | |
| 18 | async def _get_conn(self) -> aiosqlite.Connection: |
| 19 | if self._conn is not None: |
| 20 | return self._conn |
| 21 | if self._conn_lock is None: |
| 22 | self._conn_lock = asyncio.Lock() |
| 23 | async with self._conn_lock: |
| 24 | if self._conn is None: |
| 25 | self._conn = await aiosqlite.connect(self.db_path) |
| 26 | return self._conn |
| 27 | |
| 28 | async def initialize(self): |
| 29 | if self._initialized: |
| 30 | return |
| 31 | |
| 32 | db = await self._get_conn() |
| 33 | |
| 34 | # WAL gives concurrent reader/writer; NORMAL avoids fsync on every commit |
| 35 | # (loses at most last few txns on power loss — acceptable for download history). |
| 36 | await db.execute("PRAGMA journal_mode=WAL") |
| 37 | await db.execute("PRAGMA synchronous=NORMAL") |
| 38 | |
| 39 | await db.execute(""" |
| 40 | CREATE TABLE IF NOT EXISTS aweme ( |
| 41 | id INTEGER PRIMARY KEY AUTOINCREMENT, |
| 42 | aweme_id TEXT UNIQUE NOT NULL, |
| 43 | aweme_type TEXT NOT NULL, |
| 44 | title TEXT, |
| 45 | author_id TEXT, |
| 46 | author_name TEXT, |
| 47 | create_time INTEGER, |
| 48 | download_time INTEGER, |
| 49 | file_path TEXT, |
| 50 | metadata TEXT |
| 51 | ) |
| 52 | """) |
| 53 | |
| 54 | await db.execute(""" |
| 55 | CREATE TABLE IF NOT EXISTS download_history ( |
| 56 | id INTEGER PRIMARY KEY AUTOINCREMENT, |
| 57 | url TEXT NOT NULL, |
| 58 | url_type TEXT NOT NULL, |
| 59 | download_time INTEGER, |
| 60 | total_count INTEGER, |
| 61 | success_count INTEGER, |
| 62 | config TEXT |
| 63 | ) |
| 64 | """) |
| 65 | |
| 66 | await db.execute(""" |
| 67 | CREATE TABLE IF NOT EXISTS transcript_job ( |
| 68 | id INTEGER PRIMARY KEY AUTOINCREMENT, |
| 69 | aweme_id TEXT NOT NULL, |
| 70 | video_path TEXT NOT NULL, |
| 71 | transcript_dir TEXT, |
| 72 | text_path TEXT, |
| 73 | json_path TEXT, |
| 74 | model TEXT NOT NULL, |
| 75 | status TEXT NOT NULL, |
| 76 | skip_reason TEXT, |
| 77 | error_message TEXT, |
| 78 | created_at INTEGER, |
| 79 | updated_at INTEGER, |
| 80 | UNIQUE(aweme_id, video_path, model) |
| 81 | ) |
| 82 | """) |
| 83 | |
| 84 | # `job` persists the task-center JobManager records so they survive |
| 85 | # a sidecar restart. Only terminal jobs (success / failed / cancelled) |
| 86 | # are ever written here — see server/jobs.py. `last_retry_summary` |
| 87 | # and `overrides` are stored as JSON text. |
| 88 | await db.execute(""" |
| 89 | CREATE TABLE IF NOT EXISTS job ( |
| 90 | job_id TEXT PRIMARY KEY, |
| 91 | url TEXT NOT NULL, |
| 92 | status TEXT NOT NULL, |
| 93 | created_at TEXT NOT NULL, |
| 94 | started_at TEXT, |
| 95 | finished_at TEXT, |
| 96 | total INTEGER NOT NULL DEFAULT 0, |
| 97 | success INTEGER NOT NULL DEFAULT 0, |
| 98 | failed INTEGER NOT NULL DEFAULT 0, |
| 99 | skipped INTEGER NOT NULL DEFAULT 0, |
| 100 | error TEXT, |
| 101 | author_nickname TEXT, |
| 102 | author_sec_uid TEXT, |
| 103 | retry_count INTEGER NOT NULL DEFAULT 0, |
| 104 | last_retry_at TEXT, |
| 105 | last_retry_summary TEXT, |
| 106 | retry_history TEXT, |
| 107 | overrides TEXT |
| 108 | ) |
| 109 | """) |
| 110 | |
| 111 | await db.execute("CREATE INDEX IF NOT EXISTS idx_aweme_id ON aweme(aweme_id)") |
| 112 | await db.execute("CREATE INDEX IF NOT EXISTS idx_author_id ON aweme(author_id)") |
| 113 | await db.execute("CREATE INDEX IF NOT EXISTS idx_download_time ON aweme(download_time)") |
| 114 | await db.execute( |
| 115 | "CREATE INDEX IF NOT EXISTS idx_transcript_aweme_id ON transcript_job(aweme_id)" |
| 116 | ) |
| 117 | await db.execute( |
| 118 | "CREATE INDEX IF NOT EXISTS idx_transcript_status ON transcript_job(status)" |
| 119 | ) |
| 120 | await db.execute("CREATE INDEX IF NOT EXISTS idx_job_created_at ON job(created_at)") |
| 121 | await db.execute("CREATE INDEX IF NOT EXISTS idx_job_status ON job(status)") |
| 122 | |
| 123 | # Incremental migration: add author_sec_uid column to legacy aweme tables. |
| 124 | # Running initialize() twice must be a no-op. |
| 125 | cursor = await db.execute("PRAGMA table_info(aweme)") |
| 126 | existing_columns = {row[1] for row in await cursor.fetchall()} |
| 127 | if "author_sec_uid" not in existing_columns: |
| 128 | await db.execute("ALTER TABLE aweme ADD COLUMN author_sec_uid TEXT") |
| 129 | |
| 130 | # Incremental migration: add retry_history column to legacy job |
| 131 | # tables so pre-existing DB files (created before retry-history |
| 132 | # persistence landed) continue to work. NULL for old rows; the |
| 133 | # restore path maps NULL -> [] so the renderer gracefully shows |
| 134 | # no history for those jobs. |
| 135 | cursor = await db.execute("PRAGMA table_info(job)") |
| 136 | existing_job_columns = {row[1] for row in await cursor.fetchall()} |
| 137 | if "retry_history" not in existing_job_columns: |
| 138 | await db.execute("ALTER TABLE job ADD COLUMN retry_history TEXT") |
| 139 | |
| 140 | await db.commit() |
| 141 | self._initialized = True |
| 142 | |
| 143 | async def is_downloaded(self, aweme_id: str) -> bool: |
| 144 | db = await self._get_conn() |
| 145 | cursor = await db.execute("SELECT id FROM aweme WHERE aweme_id = ?", (aweme_id,)) |
| 146 | result = await cursor.fetchone() |
| 147 | return result is not None |
| 148 | |
| 149 | async def add_aweme( |
| 150 | self, |
| 151 | aweme_data: Dict[str, Any], |
| 152 | *, |
| 153 | author_sec_uid: Optional[str] = None, |
| 154 | ): |
| 155 | db = await self._get_conn() |
| 156 | # Prefer the explicit kwarg; fall back to a key on the payload so existing |
| 157 | # callers (tests, legacy downloaders) keep working. |
| 158 | sec_uid = author_sec_uid if author_sec_uid is not None else aweme_data.get("author_sec_uid") |
| 159 | await db.execute( |
| 160 | """ |
| 161 | INSERT OR REPLACE INTO aweme |
| 162 | (aweme_id, aweme_type, title, author_id, author_name, author_sec_uid, |
| 163 | create_time, download_time, file_path, metadata) |
| 164 | VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) |
| 165 | """, |
| 166 | ( |
| 167 | aweme_data.get("aweme_id"), |
| 168 | aweme_data.get("aweme_type"), |
| 169 | aweme_data.get("title"), |
| 170 | aweme_data.get("author_id"), |
| 171 | aweme_data.get("author_name"), |
| 172 | sec_uid, |
| 173 | aweme_data.get("create_time"), |
| 174 | int(datetime.now().timestamp()), |
| 175 | aweme_data.get("file_path"), |
| 176 | aweme_data.get("metadata"), |
| 177 | ), |
| 178 | ) |
| 179 | await db.commit() |
| 180 | |
| 181 | async def add_aweme_batch(self, items: List[Dict[str, Any]]) -> None: |
| 182 | """Insert N awemes in a single transaction. Replaces existing rows by aweme_id.""" |
| 183 | if not items: |
| 184 | return |
| 185 | db = await self._get_conn() |
| 186 | now_ts = int(datetime.now().timestamp()) |
| 187 | rows = [ |
| 188 | ( |
| 189 | item.get("aweme_id"), |
| 190 | item.get("aweme_type"), |
| 191 | item.get("title"), |
| 192 | item.get("author_id"), |
| 193 | item.get("author_name"), |
| 194 | item.get("author_sec_uid"), |
| 195 | item.get("create_time"), |
| 196 | now_ts, |
| 197 | item.get("file_path"), |
| 198 | item.get("metadata"), |
| 199 | ) |
| 200 | for item in items |
| 201 | ] |
| 202 | await db.executemany( |
| 203 | """ |
| 204 | INSERT OR REPLACE INTO aweme |
| 205 | (aweme_id, aweme_type, title, author_id, author_name, author_sec_uid, |
| 206 | create_time, download_time, file_path, metadata) |
| 207 | VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) |
| 208 | """, |
| 209 | rows, |
| 210 | ) |
| 211 | await db.commit() |
| 212 | |
| 213 | async def get_latest_aweme_time(self, author_id: str) -> Optional[int]: |
| 214 | db = await self._get_conn() |
| 215 | cursor = await db.execute( |
| 216 | "SELECT MAX(create_time) FROM aweme WHERE author_id = ?", (author_id,) |
| 217 | ) |
| 218 | result = await cursor.fetchone() |
| 219 | return result[0] if result and result[0] else None |
| 220 | |
| 221 | async def add_history(self, history_data: Dict[str, Any]): |
| 222 | db = await self._get_conn() |
| 223 | await db.execute( |
| 224 | """ |
| 225 | INSERT INTO download_history |
| 226 | (url, url_type, download_time, total_count, success_count, config) |
| 227 | VALUES (?, ?, ?, ?, ?, ?) |
| 228 | """, |
| 229 | ( |
| 230 | history_data.get("url"), |
| 231 | history_data.get("url_type"), |
| 232 | int(datetime.now().timestamp()), |
| 233 | history_data.get("total_count"), |
| 234 | history_data.get("success_count"), |
| 235 | history_data.get("config"), |
| 236 | ), |
| 237 | ) |
| 238 | await db.commit() |
| 239 | |
| 240 | async def get_aweme_history( |
| 241 | self, |
| 242 | *, |
| 243 | page: int = 1, |
| 244 | size: int = 50, |
| 245 | author: Optional[str] = None, |
| 246 | date_from: Optional[int] = None, |
| 247 | date_to: Optional[int] = None, |
| 248 | aweme_type: Optional[str] = None, |
| 249 | title: Optional[str] = None, |
| 250 | ) -> Dict[str, Any]: |
| 251 | """Paginated aweme history, newest download first. |
| 252 | |
| 253 | `date_from` / `date_to` are unix-seconds (filter against `create_time`). |
| 254 | `aweme_type` matches the `aweme_type` column (e.g. 'video', 'gallery'). |
| 255 | `title` is a case-insensitive substring match on the title column. |
| 256 | """ |
| 257 | db = await self._get_conn() |
| 258 | where: list = [] |
| 259 | params: list = [] |
| 260 | if author: |
| 261 | where.append("author_name = ?") |
| 262 | params.append(author) |
| 263 | if date_from is not None: |
| 264 | where.append("create_time >= ?") |
| 265 | params.append(int(date_from)) |
| 266 | if date_to is not None: |
| 267 | where.append("create_time <= ?") |
| 268 | params.append(int(date_to)) |
| 269 | if aweme_type: |
| 270 | where.append("aweme_type = ?") |
| 271 | params.append(aweme_type) |
| 272 | if title: |
| 273 | where.append("LOWER(COALESCE(title, '')) LIKE ?") |
| 274 | params.append(f"%{title.lower()}%") |
| 275 | where_sql = ("WHERE " + " AND ".join(where)) if where else "" |
| 276 | |
| 277 | cursor = await db.execute(f"SELECT COUNT(*) FROM aweme {where_sql}", params) |
| 278 | row = await cursor.fetchone() |
| 279 | total = int(row[0]) if row else 0 |
| 280 | |
| 281 | offset = max(0, (page - 1) * size) |
| 282 | cursor = await db.execute( |
| 283 | f"SELECT aweme_id, aweme_type, title, author_id, author_name, " |
| 284 | f"author_sec_uid, create_time, download_time, file_path FROM aweme " |
| 285 | f"{where_sql} ORDER BY download_time DESC, id DESC LIMIT ? OFFSET ?", |
| 286 | params + [int(size), int(offset)], |
| 287 | ) |
| 288 | rows = await cursor.fetchall() |
| 289 | items = [ |
| 290 | { |
| 291 | "aweme_id": r[0], |
| 292 | "aweme_type": r[1], |
| 293 | "title": r[2], |
| 294 | "author_id": r[3], |
| 295 | "author_name": r[4], |
| 296 | "author_sec_uid": r[5], |
| 297 | "create_time": r[6], |
| 298 | "download_time": r[7], |
| 299 | "file_path": r[8], |
| 300 | } |
| 301 | for r in rows |
| 302 | ] |
| 303 | return {"total": total, "page": int(page), "size": int(size), "items": items} |
| 304 | |
| 305 | async def get_aweme_count_by_author(self, author_id: str) -> int: |
| 306 | db = await self._get_conn() |
| 307 | cursor = await db.execute("SELECT COUNT(*) FROM aweme WHERE author_id = ?", (author_id,)) |
| 308 | result = await cursor.fetchone() |
| 309 | return result[0] if result else 0 |
| 310 | |
| 311 | async def get_top_authors(self, *, days: int, limit: int) -> List[Dict[str, Any]]: |
| 312 | """Return the most-downloaded authors in the last ``days`` days. |
| 313 | |
| 314 | Aggregates rows in `aweme` with ``create_time >= now - days*86400`` and |
| 315 | non-empty / non-null ``author_sec_uid``. Groups by ``author_sec_uid`` |
| 316 | and orders by ``COUNT(*) DESC, author_sec_uid ASC`` (stable tie-break |
| 317 | so property tests are deterministic). Truncates to ``limit`` rows. |
| 318 | |
| 319 | ``author_name`` for each result row is the latest non-empty |
| 320 | ``author_name`` for that ``sec_uid`` (ordered by ``download_time`` |
| 321 | descending). If all rows for that sec_uid have empty/null names, |
| 322 | falls back to the Chinese placeholder ``"未知作者"``. |
| 323 | |
| 324 | Each returned dict contains ``sec_uid`` / ``author_name`` / |
| 325 | ``download_count``. |
| 326 | """ |
| 327 | cutoff = int(datetime.now().timestamp()) - int(days) * 86400 |
| 328 | db = await self._get_conn() |
| 329 | cursor = await db.execute( |
| 330 | """ |
| 331 | SELECT a.author_sec_uid, |
| 332 | (SELECT a2.author_name FROM aweme a2 |
| 333 | WHERE a2.author_sec_uid = a.author_sec_uid |
| 334 | AND a2.author_name IS NOT NULL |
| 335 | AND a2.author_name != '' |
| 336 | ORDER BY a2.download_time DESC |
| 337 | LIMIT 1) AS author_name, |
| 338 | COUNT(*) AS download_count |
| 339 | FROM aweme a |
| 340 | WHERE a.create_time >= ? |
| 341 | AND a.author_sec_uid IS NOT NULL |
| 342 | AND a.author_sec_uid != '' |
| 343 | GROUP BY a.author_sec_uid |
| 344 | ORDER BY download_count DESC, a.author_sec_uid ASC |
| 345 | LIMIT ? |
| 346 | """, |
| 347 | (cutoff, int(limit)), |
| 348 | ) |
| 349 | rows = await cursor.fetchall() |
| 350 | return [ |
| 351 | { |
| 352 | "sec_uid": row[0], |
| 353 | "author_name": row[1] if row[1] else "未知作者", |
| 354 | "download_count": int(row[2]), |
| 355 | } |
| 356 | for row in rows |
| 357 | ] |
| 358 | |
| 359 | async def upsert_transcript_job(self, job_data: Dict[str, Any]): |
| 360 | now_ts = int(datetime.now().timestamp()) |
| 361 | db = await self._get_conn() |
| 362 | await db.execute( |
| 363 | """ |
| 364 | INSERT INTO transcript_job ( |
| 365 | aweme_id, |
| 366 | video_path, |
| 367 | transcript_dir, |
| 368 | text_path, |
| 369 | json_path, |
| 370 | model, |
| 371 | status, |
| 372 | skip_reason, |
| 373 | error_message, |
| 374 | created_at, |
| 375 | updated_at |
| 376 | ) |
| 377 | VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) |
| 378 | ON CONFLICT(aweme_id, video_path, model) DO UPDATE SET |
| 379 | transcript_dir = excluded.transcript_dir, |
| 380 | text_path = excluded.text_path, |
| 381 | json_path = excluded.json_path, |
| 382 | status = excluded.status, |
| 383 | skip_reason = excluded.skip_reason, |
| 384 | error_message = excluded.error_message, |
| 385 | updated_at = excluded.updated_at |
| 386 | """, |
| 387 | ( |
| 388 | job_data.get("aweme_id"), |
| 389 | job_data.get("video_path"), |
| 390 | job_data.get("transcript_dir"), |
| 391 | job_data.get("text_path"), |
| 392 | job_data.get("json_path"), |
| 393 | job_data.get("model") or "gpt-4o-mini-transcribe", |
| 394 | job_data.get("status"), |
| 395 | job_data.get("skip_reason"), |
| 396 | job_data.get("error_message"), |
| 397 | now_ts, |
| 398 | now_ts, |
| 399 | ), |
| 400 | ) |
| 401 | await db.commit() |
| 402 | |
| 403 | async def get_transcript_job(self, aweme_id: str) -> Optional[Dict[str, Any]]: |
| 404 | db = await self._get_conn() |
| 405 | cursor = await db.execute( |
| 406 | """ |
| 407 | SELECT aweme_id, video_path, transcript_dir, text_path, json_path, |
| 408 | model, status, skip_reason, error_message, created_at, updated_at |
| 409 | FROM transcript_job |
| 410 | WHERE aweme_id = ? |
| 411 | ORDER BY updated_at DESC, id DESC |
| 412 | LIMIT 1 |
| 413 | """, |
| 414 | (aweme_id,), |
| 415 | ) |
| 416 | row = await cursor.fetchone() |
| 417 | if not row: |
| 418 | return None |
| 419 | return { |
| 420 | "aweme_id": row[0], |
| 421 | "video_path": row[1], |
| 422 | "transcript_dir": row[2], |
| 423 | "text_path": row[3], |
| 424 | "json_path": row[4], |
| 425 | "model": row[5], |
| 426 | "status": row[6], |
| 427 | "skip_reason": row[7], |
| 428 | "error_message": row[8], |
| 429 | "created_at": row[9], |
| 430 | "updated_at": row[10], |
| 431 | } |
| 432 | |
| 433 | async def delete_aweme_by_ids(self, aweme_ids: List[str]) -> int: |
| 434 | """Delete aweme rows by their string id. Returns the number of rows removed. |
| 435 | |
| 436 | Empty input is a no-op that returns 0 without issuing any SQL. |
| 437 | |
| 438 | Uses a parameterized ``DELETE ... WHERE aweme_id IN (?,?,...)`` statement |
| 439 | because ``aiosqlite.Cursor.rowcount`` is not reliably populated after |
| 440 | ``executemany`` across all versions. Chunked at 500 ids per statement to |
| 441 | stay well below SQLite's host-parameter limit (historically 999). |
| 442 | """ |
| 443 | if not aweme_ids: |
| 444 | return 0 |
| 445 | # De-duplicate input while preserving a stable order. Duplicate ids would |
| 446 | # otherwise match the same row twice in different chunks and inflate the |
| 447 | # returned count beyond the rows actually affected. |
| 448 | seen: Dict[str, None] = {} |
| 449 | for aid in aweme_ids: |
| 450 | if aid not in seen: |
| 451 | seen[aid] = None |
| 452 | unique_ids = list(seen.keys()) |
| 453 | |
| 454 | db = await self._get_conn() |
| 455 | if self._conn_lock is None: |
| 456 | self._conn_lock = asyncio.Lock() |
| 457 | deleted = 0 |
| 458 | chunk_size = 500 |
| 459 | async with self._conn_lock: |
| 460 | for start in range(0, len(unique_ids), chunk_size): |
| 461 | chunk = unique_ids[start : start + chunk_size] |
| 462 | placeholders = ",".join("?" for _ in chunk) |
| 463 | cursor = await db.execute( |
| 464 | f"DELETE FROM aweme WHERE aweme_id IN ({placeholders})", |
| 465 | chunk, |
| 466 | ) |
| 467 | if cursor.rowcount is not None and cursor.rowcount > 0: |
| 468 | deleted += cursor.rowcount |
| 469 | await db.commit() |
| 470 | return deleted |
| 471 | |
| 472 | async def truncate_history(self) -> None: |
| 473 | """Delete every row from `aweme` and `download_history`. |
| 474 | |
| 475 | Does not touch disk files or any other table (e.g. transcript_job). |
| 476 | """ |
| 477 | db = await self._get_conn() |
| 478 | if self._conn_lock is None: |
| 479 | self._conn_lock = asyncio.Lock() |
| 480 | async with self._conn_lock: |
| 481 | await db.execute("DELETE FROM aweme") |
| 482 | await db.execute("DELETE FROM download_history") |
| 483 | await db.commit() |
| 484 | |
| 485 | # ------------------------------------------------------------------ |
| 486 | # Task-center job persistence (see server/jobs.py) |
| 487 | # ------------------------------------------------------------------ |
| 488 | |
| 489 | async def upsert_job(self, job_dict: Dict[str, Any]) -> None: |
| 490 | """Insert or replace a task-center job record. |
| 491 | |
| 492 | Accepts the dict produced by :py:meth:`server.jobs.DownloadJob.to_dict` |
| 493 | plus an optional ``overrides`` key (the JobManager stores overrides |
| 494 | separately on the in-memory job but we persist them too so future |
| 495 | retries/re-runs can inherit them). Unknown keys are ignored — any |
| 496 | renderer-only computed fields (``url_type``, ``duration_ms`` etc.) |
| 497 | are recomputed from raw columns on read. |
| 498 | """ |
| 499 | db = await self._get_conn() |
| 500 | if self._conn_lock is None: |
| 501 | self._conn_lock = asyncio.Lock() |
| 502 | |
| 503 | last_retry_summary = job_dict.get("last_retry_summary") |
| 504 | retry_history = job_dict.get("retry_history") |
| 505 | overrides = job_dict.get("overrides") |
| 506 | params = ( |
| 507 | job_dict.get("job_id"), |
| 508 | job_dict.get("url") or "", |
| 509 | job_dict.get("status") or "", |
| 510 | job_dict.get("created_at") or "", |
| 511 | job_dict.get("started_at"), |
| 512 | job_dict.get("finished_at"), |
| 513 | int(job_dict.get("total") or 0), |
| 514 | int(job_dict.get("success") or 0), |
| 515 | int(job_dict.get("failed") or 0), |
| 516 | int(job_dict.get("skipped") or 0), |
| 517 | job_dict.get("error"), |
| 518 | job_dict.get("author_nickname"), |
| 519 | job_dict.get("author_sec_uid"), |
| 520 | int(job_dict.get("retry_count") or 0), |
| 521 | job_dict.get("last_retry_at"), |
| 522 | json.dumps(last_retry_summary) if last_retry_summary else None, |
| 523 | json.dumps(retry_history) if retry_history else None, |
| 524 | json.dumps(overrides) if overrides else None, |
| 525 | ) |
| 526 | async with self._conn_lock: |
| 527 | await db.execute( |
| 528 | """ |
| 529 | INSERT OR REPLACE INTO job ( |
| 530 | job_id, url, status, created_at, started_at, finished_at, |
| 531 | total, success, failed, skipped, error, |
| 532 | author_nickname, author_sec_uid, |
| 533 | retry_count, last_retry_at, last_retry_summary, |
| 534 | retry_history, overrides |
| 535 | ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) |
| 536 | """, |
| 537 | params, |
| 538 | ) |
| 539 | await db.commit() |
| 540 | |
| 541 | async def delete_jobs(self, job_ids: List[str]) -> int: |
| 542 | """Delete job rows by id. Returns the number of rows deleted.""" |
| 543 | if not job_ids: |
| 544 | return 0 |
| 545 | seen: Dict[str, None] = {} |
| 546 | for jid in job_ids: |
| 547 | if jid and jid not in seen: |
| 548 | seen[jid] = None |
| 549 | unique_ids = list(seen.keys()) |
| 550 | if not unique_ids: |
| 551 | return 0 |
| 552 | |
| 553 | db = await self._get_conn() |
| 554 | if self._conn_lock is None: |
| 555 | self._conn_lock = asyncio.Lock() |
| 556 | deleted = 0 |
| 557 | chunk_size = 500 |
| 558 | async with self._conn_lock: |
| 559 | for start in range(0, len(unique_ids), chunk_size): |
| 560 | chunk = unique_ids[start : start + chunk_size] |
| 561 | placeholders = ",".join("?" for _ in chunk) |
| 562 | cursor = await db.execute( |
| 563 | f"DELETE FROM job WHERE job_id IN ({placeholders})", |
| 564 | chunk, |
| 565 | ) |
| 566 | if cursor.rowcount is not None and cursor.rowcount > 0: |
| 567 | deleted += cursor.rowcount |
| 568 | await db.commit() |
| 569 | return deleted |
| 570 | |
| 571 | async def load_terminal_jobs(self, limit: Optional[int] = None) -> List[Dict[str, Any]]: |
| 572 | """Load persisted terminal jobs ordered by created_at DESC. |
| 573 | |
| 574 | Only rows whose ``status`` is a terminal value (success / failed / |
| 575 | cancelled) are returned. Running/pending rows shouldn't exist on |
| 576 | disk — see server/jobs.py — but we filter defensively in case an |
| 577 | older build left stale rows. |
| 578 | """ |
| 579 | db = await self._get_conn() |
| 580 | if self._conn_lock is None: |
| 581 | self._conn_lock = asyncio.Lock() |
| 582 | |
| 583 | sql = ( |
| 584 | "SELECT job_id, url, status, created_at, started_at, finished_at, " |
| 585 | "total, success, failed, skipped, error, author_nickname, " |
| 586 | "author_sec_uid, retry_count, last_retry_at, last_retry_summary, " |
| 587 | "retry_history, overrides FROM job " |
| 588 | "WHERE status IN ('success', 'failed', 'cancelled') " |
| 589 | "ORDER BY created_at DESC" |
| 590 | ) |
| 591 | if limit is not None and limit > 0: |
| 592 | sql += f" LIMIT {int(limit)}" |
| 593 | |
| 594 | async with self._conn_lock: |
| 595 | cursor = await db.execute(sql) |
| 596 | rows = await cursor.fetchall() |
| 597 | |
| 598 | result: List[Dict[str, Any]] = [] |
| 599 | for row in rows: |
| 600 | summary_raw = row[15] |
| 601 | history_raw = row[16] |
| 602 | overrides_raw = row[17] |
| 603 | try: |
| 604 | summary = json.loads(summary_raw) if summary_raw else None |
| 605 | except (TypeError, ValueError): |
| 606 | summary = None |
| 607 | try: |
| 608 | history = json.loads(history_raw) if history_raw else [] |
| 609 | if not isinstance(history, list): |
| 610 | history = [] |
| 611 | except (TypeError, ValueError): |
| 612 | history = [] |
| 613 | try: |
| 614 | overrides = json.loads(overrides_raw) if overrides_raw else None |
| 615 | except (TypeError, ValueError): |
| 616 | overrides = None |
| 617 | result.append( |
| 618 | { |
| 619 | "job_id": row[0], |
| 620 | "url": row[1], |
| 621 | "status": row[2], |
| 622 | "created_at": row[3], |
| 623 | "started_at": row[4], |
| 624 | "finished_at": row[5], |
| 625 | "total": row[6] or 0, |
| 626 | "success": row[7] or 0, |
| 627 | "failed": row[8] or 0, |
| 628 | "skipped": row[9] or 0, |
| 629 | "error": row[10], |
| 630 | "author_nickname": row[11], |
| 631 | "author_sec_uid": row[12], |
| 632 | "retry_count": row[13] or 0, |
| 633 | "last_retry_at": row[14], |
| 634 | "last_retry_summary": summary, |
| 635 | "retry_history": history, |
| 636 | "overrides": overrides, |
| 637 | } |
| 638 | ) |
| 639 | return result |
| 640 | |
| 641 | async def close(self): |
| 642 | if self._conn is not None: |
| 643 | await self._conn.close() |
| 644 | self._conn = None |
| 645 |