Run download jobs and deliver books to CWA
Jobs download new chapters, recheck chapters, replace the book, update metadata or rebuild from stored chapters. Locked metadata fields keep their values. A scheduler queues updates. An update finds the library book of a story through CWA, and fails with an error if it cannot, so that it never makes a duplicate book.
This commit is contained in:
@@ -0,0 +1,40 @@
|
||||
"""Daily copies of the database and personal.ini.
|
||||
|
||||
The database holds chapters that the sites removed, so it is the only copy of
|
||||
them. A backup a day, kept for a week, limits a bad change or a broken disk to
|
||||
the loss of one day.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import shutil
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
|
||||
from bindery.store import Store
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def daily_backup(store: Store, folder: Path, personal_ini: Path, keep: int, today: date | None = None) -> Path | None:
|
||||
"""Make the backup of today if there is none yet, then delete all but the newest `keep` backups.
|
||||
|
||||
Return the path of the new database copy, or None if today had one already.
|
||||
"""
|
||||
today = today or date.today()
|
||||
folder.mkdir(parents=True, exist_ok=True)
|
||||
target = folder / f"bindery-{today.isoformat()}.db"
|
||||
made = None
|
||||
if not target.exists():
|
||||
# A copy under another name first, so that a stop never leaves half a backup.
|
||||
partial = folder / f"{target.name}.part"
|
||||
partial.unlink(missing_ok=True)
|
||||
store.backup(partial)
|
||||
partial.replace(target)
|
||||
if personal_ini.exists():
|
||||
shutil.copy2(personal_ini, folder / f"personal-{today.isoformat()}.ini")
|
||||
logger.info(f"Backed up the database to {target}")
|
||||
made = target
|
||||
for pattern in ("bindery-*.db", "personal-*.ini"):
|
||||
for old in sorted(folder.glob(pattern))[:-keep]:
|
||||
old.unlink()
|
||||
return made
|
||||
+239
@@ -0,0 +1,239 @@
|
||||
"""Download modes, jobs and the job queue."""
|
||||
|
||||
import json
|
||||
import sqlite3
|
||||
import uuid
|
||||
from collections import deque
|
||||
from dataclasses import asdict, dataclass, field
|
||||
from datetime import datetime, timedelta
|
||||
from enum import StrEnum
|
||||
from pathlib import Path
|
||||
from threading import Lock
|
||||
from typing import Any
|
||||
|
||||
from bindery import config
|
||||
|
||||
# Calibre fields that a story can lock. A locked field keeps its current
|
||||
# value until a download explicitly forces an update of that field.
|
||||
METADATA_FIELDS: dict[str, str] = {
|
||||
"title": "Title",
|
||||
"authors": "Authors",
|
||||
"comments": "Description",
|
||||
"tags": "Tags",
|
||||
"series": "Series",
|
||||
"pubdate": "Published date",
|
||||
"languages": "Language",
|
||||
}
|
||||
|
||||
|
||||
class QueueStatus(StrEnum):
|
||||
"""Enum for possible job statuses."""
|
||||
|
||||
QUEUED = "queued"
|
||||
RUNNING = "running"
|
||||
DONE = "done"
|
||||
ERROR = "error"
|
||||
CANCELLED = "cancelled"
|
||||
|
||||
|
||||
class DownloadMode(StrEnum):
|
||||
"""What a job fetches from the site and how it changes the book."""
|
||||
|
||||
NEW = "new" # new chapters only, appended after the chapters we have
|
||||
RECHECK = "recheck" # download selected chapters again and keep changes
|
||||
RECHECK_ALL = "recheck_all" # download every chapter still on the site again
|
||||
FULL = "full" # replace the whole book with what is on the site now
|
||||
METADATA = "metadata" # update metadata only, no chapter changes
|
||||
REBUILD = "rebuild" # no site access: rebuild the epub from stored chapters
|
||||
IMPORT = "import" # start tracking a book that is already in the library
|
||||
IMPORT_FILE = "import_file" # add the chapters of an uploaded epub to a story
|
||||
|
||||
|
||||
MODE_LABELS: dict[str, str] = {
|
||||
DownloadMode.NEW.value: "Download new chapters (default)",
|
||||
DownloadMode.RECHECK.value: "Recheck selected chapters",
|
||||
DownloadMode.RECHECK_ALL.value: "Recheck all chapters",
|
||||
DownloadMode.FULL.value: "Download and replace entire book",
|
||||
DownloadMode.METADATA.value: "Update metadata from site",
|
||||
DownloadMode.REBUILD.value: "Rebuild book from stored chapters",
|
||||
DownloadMode.IMPORT.value: "Import from library",
|
||||
DownloadMode.IMPORT_FILE.value: "Import chapters from an epub file",
|
||||
}
|
||||
|
||||
# A job that was running when Bindery stopped this many times is not started again.
|
||||
MAX_RESTARTS = 3
|
||||
|
||||
|
||||
@dataclass
|
||||
class Job:
|
||||
"""One unit of work for the download thread."""
|
||||
|
||||
mode: DownloadMode
|
||||
story_id: int | None = None
|
||||
url: str | None = None
|
||||
calibre_id: int | None = None
|
||||
chapter_ids: list[int] = field(default_factory=list)
|
||||
force_fields: list[str] = field(default_factory=list)
|
||||
options: dict[str, Any] = field(default_factory=dict)
|
||||
id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
|
||||
status: QueueStatus = QueueStatus.QUEUED
|
||||
title: str = ""
|
||||
message: str = ""
|
||||
progress: float = 0.0
|
||||
created: datetime = field(default_factory=datetime.now)
|
||||
updated: datetime = field(default_factory=datetime.now)
|
||||
result: dict[str, Any] = field(default_factory=dict)
|
||||
restarts: int = 0
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
d = asdict(self)
|
||||
d["mode"] = self.mode.value
|
||||
d["mode_label"] = MODE_LABELS[self.mode.value]
|
||||
d["status"] = self.status.value
|
||||
d["created"] = self.created.isoformat(timespec="seconds")
|
||||
d["updated"] = self.updated.isoformat(timespec="seconds")
|
||||
return d
|
||||
|
||||
def to_record(self) -> str:
|
||||
d = asdict(self)
|
||||
d["created"] = self.created.isoformat()
|
||||
d["updated"] = self.updated.isoformat()
|
||||
return json.dumps(d)
|
||||
|
||||
@classmethod
|
||||
def from_record(cls, record: str) -> "Job":
|
||||
d = json.loads(record)
|
||||
d["mode"] = DownloadMode(d["mode"])
|
||||
d["status"] = QueueStatus(d["status"])
|
||||
d["created"] = datetime.fromisoformat(d["created"])
|
||||
d["updated"] = datetime.fromisoformat(d["updated"])
|
||||
return cls(**d)
|
||||
|
||||
|
||||
class JobQueue:
|
||||
"""Thread-safe FIFO job queue that also keeps recent job history.
|
||||
|
||||
With a database path, the queue keeps each job in SQLite, so the jobs outlive
|
||||
a restart. A job that was running goes back to the front of the queue.
|
||||
"""
|
||||
|
||||
def __init__(self, path: Path | None = None) -> None:
|
||||
self._lock = Lock()
|
||||
self._pending: deque[str] = deque()
|
||||
self._jobs: dict[str, Job] = {}
|
||||
self._history = timedelta(seconds=config.JOB_HISTORY)
|
||||
self._db: sqlite3.Connection | None = None
|
||||
if path:
|
||||
self._load(Path(path))
|
||||
|
||||
def _load(self, path: Path) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
# One connection, used only while the lock is held. Each statement commits.
|
||||
self._db = sqlite3.connect(path, timeout=30, check_same_thread=False, isolation_level=None)
|
||||
self._db.execute("PRAGMA journal_mode=WAL")
|
||||
self._db.execute("CREATE TABLE IF NOT EXISTS jobs (id TEXT PRIMARY KEY, record TEXT NOT NULL)")
|
||||
jobs = [Job.from_record(r[0]) for r in self._db.execute("SELECT record FROM jobs ORDER BY rowid")]
|
||||
interrupted = [j for j in jobs if j.status == QueueStatus.RUNNING]
|
||||
for job in interrupted:
|
||||
job.restarts += 1
|
||||
if job.restarts >= MAX_RESTARTS:
|
||||
job.status = QueueStatus.ERROR
|
||||
job.message = f"Stopped {job.restarts} times, giving up"
|
||||
else:
|
||||
job.status = QueueStatus.QUEUED
|
||||
job.message = "Resumed"
|
||||
job.updated = datetime.now()
|
||||
self._save(job)
|
||||
self._jobs = {j.id: j for j in jobs}
|
||||
waiting = [
|
||||
j.id for j in interrupted + [j for j in jobs if j not in interrupted] if j.status == QueueStatus.QUEUED
|
||||
]
|
||||
self._pending.extend(waiting)
|
||||
self._expire()
|
||||
|
||||
def _save(self, job: Job) -> None:
|
||||
if self._db:
|
||||
self._db.execute(
|
||||
"INSERT INTO jobs (id, record) VALUES (?, ?) ON CONFLICT (id) DO UPDATE SET record = excluded.record",
|
||||
(job.id, job.to_record()),
|
||||
)
|
||||
|
||||
def _expire(self) -> None:
|
||||
"""Forget finished jobs after JOB_HISTORY seconds."""
|
||||
now = datetime.now()
|
||||
expired = [
|
||||
k
|
||||
for k, j in self._jobs.items()
|
||||
if j.status not in (QueueStatus.QUEUED, QueueStatus.RUNNING) and now - j.updated > self._history
|
||||
]
|
||||
for job_id in expired:
|
||||
del self._jobs[job_id]
|
||||
if self._db and expired:
|
||||
self._db.executemany("DELETE FROM jobs WHERE id = ?", [(k,) for k in expired])
|
||||
|
||||
def add(self, job: Job) -> Job:
|
||||
"""Queue a job. If the same story already waits for the same mode, return that job."""
|
||||
with self._lock:
|
||||
for job_id in self._pending:
|
||||
other = self._jobs[job_id]
|
||||
if (
|
||||
other.mode == job.mode
|
||||
and other.chapter_ids == job.chapter_ids
|
||||
and other.force_fields == job.force_fields
|
||||
and other.options == job.options
|
||||
and (
|
||||
(job.story_id and other.story_id == job.story_id)
|
||||
or (job.url and other.url == job.url)
|
||||
or (job.calibre_id and other.calibre_id == job.calibre_id)
|
||||
)
|
||||
):
|
||||
return other
|
||||
self._expire()
|
||||
self._jobs[job.id] = job
|
||||
self._pending.append(job.id)
|
||||
self._save(job)
|
||||
return job
|
||||
|
||||
def get_next(self) -> Job | None:
|
||||
with self._lock:
|
||||
while self._pending:
|
||||
job = self._jobs[self._pending.popleft()]
|
||||
if job.status == QueueStatus.QUEUED:
|
||||
return job
|
||||
return None
|
||||
|
||||
def update(self, job: Job, **changes: Any) -> None:
|
||||
with self._lock:
|
||||
for key, value in changes.items():
|
||||
setattr(job, key, value)
|
||||
job.updated = datetime.now()
|
||||
self._save(job)
|
||||
|
||||
def cancel(self, job_id: str) -> bool:
|
||||
with self._lock:
|
||||
job = self._jobs.get(job_id)
|
||||
if not job or job.status != QueueStatus.QUEUED:
|
||||
return False
|
||||
job.status = QueueStatus.CANCELLED
|
||||
job.updated = datetime.now()
|
||||
# A cancelled job must not be shared, and it can expire before the download thread reaches it.
|
||||
self._pending.remove(job_id)
|
||||
self._save(job)
|
||||
return True
|
||||
|
||||
def has_active(self, story_id: int) -> bool:
|
||||
with self._lock:
|
||||
return any(
|
||||
j.story_id == story_id and j.status in (QueueStatus.QUEUED, QueueStatus.RUNNING)
|
||||
for j in self._jobs.values()
|
||||
)
|
||||
|
||||
def get_status(self, story_id: int | None = None) -> list[dict[str, Any]]:
|
||||
"""Return recent jobs, newest first."""
|
||||
with self._lock:
|
||||
self._expire()
|
||||
jobs = [j for j in self._jobs.values() if story_id is None or j.story_id == story_id]
|
||||
return [j.to_dict() for j in sorted(jobs, key=lambda j: j.created, reverse=True)]
|
||||
|
||||
|
||||
job_queue = JobQueue(config.DB_PATH)
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user