Coverage for src/lilbee/crawler/task.py: 100%
125 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""Background crawl task management: start, track, and query crawl operations."""
3import asyncio
4import logging
5import threading
6import uuid
7from dataclasses import dataclass, field
8from datetime import UTC, datetime
9from enum import StrEnum
11from lilbee.core.config.enums import CrawlRenderMode
12from lilbee.crawler import crawl_and_save
13from lilbee.crawler.models import NO_LOGIN, CrawlLogin
14from lilbee.crawler.url_filter import without_userinfo
15from lilbee.runtime.progress import (
16 CrawlPageEvent,
17 CrawlPageFailedEvent,
18 DetailedProgressCallback,
19 EventType,
20 ProgressEvent,
21)
23log = logging.getLogger(__name__)
25# Maximum completed tasks to retain in memory before evicting oldest.
26_MAX_COMPLETED_TASKS = 100
28# Maximum per-page failure reasons stored on a task; pages_failed keeps the
29# true count so a large all-failing crawl doesn't grow task memory unboundedly.
30_MAX_FAILURE_REASONS = 20
33class TaskStatus(StrEnum):
34 """Lifecycle states for a crawl task."""
36 PENDING = "pending"
37 RUNNING = "running"
38 DONE = "done"
39 FAILED = "failed"
40 CANCELLED = "cancelled"
43@dataclass
44class CrawlTask:
45 """Tracks a single crawl operation.
47 depth / max_pages follow the crawl_and_save three-state convention: None =
48 unbounded, 0 (depth only) = single URL, positive int = explicit cap.
49 """
51 task_id: str
52 url: str
53 depth: int | None
54 max_pages: int | None
55 render_mode: CrawlRenderMode | None = None
56 include_subdomains: bool = False
57 # Dropped when the crawl ends, so a finished task keeps no secret.
58 login: CrawlLogin = field(default=NO_LOGIN, repr=False)
59 status: TaskStatus = TaskStatus.PENDING
60 pages_crawled: int = 0
61 pages_total: int | None = None
62 pages_failed: int = 0
63 failure_reasons: list[str] = field(default_factory=list)
64 error: str | None = None
65 started_at: str = ""
66 finished_at: str = ""
67 # Polled by crawl_and_save between pages and by the follow-on sync between
68 # files, so a stop lands on a boundary with the pages already fetched saved.
69 # Cancelling _async_task instead would abort mid-page and lose them.
70 cancel: threading.Event = field(default_factory=threading.Event, repr=False)
71 _async_task: asyncio.Task[None] | None = field(default=None, repr=False, init=False)
73 @property
74 def shown_url(self) -> str:
75 """The crawled address without its username and password."""
76 return without_userinfo(self.url)
79class TaskRegistry:
80 """In-memory registry of active and completed crawl tasks.
81 A single module-level instance (_registry) is used because task tracking
82 is inherently per-process state (asyncio.Task references, etc.).
83 """
85 def __init__(self) -> None:
86 self.tasks: dict[str, CrawlTask] = {}
88 def clear(self) -> None:
89 """Remove all tasks from the registry."""
90 self.tasks.clear()
93_registry = TaskRegistry()
96def now_iso() -> str:
97 """Current UTC time as ISO 8601 string."""
98 return datetime.now(UTC).isoformat()
101def make_progress_updater(task: CrawlTask) -> DetailedProgressCallback:
102 """Return a progress callback that updates task fields from crawl events."""
104 def _on_progress(event_type: EventType, data: ProgressEvent) -> None:
105 if event_type == EventType.CRAWL_PAGE:
106 if not isinstance(data, CrawlPageEvent):
107 raise TypeError(f"Expected CrawlPageEvent, got {type(data).__name__}")
108 task.pages_crawled = data.current
109 task.pages_total = data.total
110 elif event_type == EventType.CRAWL_PAGE_FAILED:
111 if not isinstance(data, CrawlPageFailedEvent):
112 raise TypeError(f"Expected CrawlPageFailedEvent, got {type(data).__name__}")
113 task.pages_failed += 1
114 if len(task.failure_reasons) < _MAX_FAILURE_REASONS:
115 task.failure_reasons.append(f"{data.url}: {data.reason}")
117 return _on_progress
120async def run_crawl(task: CrawlTask) -> None:
121 """Execute crawl, save results, and trigger sync."""
122 task.status = TaskStatus.RUNNING
123 task.started_at = now_iso()
124 progress = make_progress_updater(task)
126 try:
127 paths = await crawl_and_save(
128 task.url,
129 depth=task.depth,
130 max_pages=task.max_pages,
131 on_progress=progress,
132 cancel=task.cancel,
133 include_subdomains=task.include_subdomains,
134 render_mode=task.render_mode,
135 login=task.login,
136 )
137 if task.cancel.is_set():
138 task.status = TaskStatus.CANCELLED
139 task.pages_crawled = task.pages_crawled or len(paths)
140 task.finished_at = now_iso()
141 log.info("Crawl cancelled: %s after %d files", task.shown_url, len(paths))
142 return
143 task.status = TaskStatus.DONE
144 task.pages_crawled = task.pages_crawled or len(paths)
145 task.finished_at = now_iso()
146 log.info("Crawl complete: %s → %d files", task.shown_url, len(paths))
147 from lilbee.app.ingest import LIBRARY_REFUSALS
149 try:
150 from lilbee.data.ingest import sync
152 await sync(quiet=True, cancel=task.cancel, started_by="a crawl")
153 except LIBRARY_REFUSALS as exc:
154 # A refusal says what to do; a traceback adds nothing to it.
155 log.warning("Post-crawl sync failed for %s: %s", task.shown_url, exc)
156 except Exception:
157 log.warning("Post-crawl sync failed for %s", task.shown_url, exc_info=True)
158 except Exception as exc:
159 task.status = TaskStatus.FAILED
160 task.error = str(exc)
161 task.finished_at = now_iso()
162 log.warning("Crawl failed: %s: %s", task.shown_url, exc)
163 finally:
164 task.login = NO_LOGIN
165 task._async_task = None
168def _evict_completed() -> None:
169 """Remove completed tasks with the earliest ``finished_at`` when over cap.
171 Finish order diverges from start order whenever short tasks complete
172 while a long one is still running, so sort on ``finished_at``
173 rather than dict insertion order.
174 """
175 done_statuses = (TaskStatus.DONE, TaskStatus.FAILED, TaskStatus.CANCELLED)
176 tasks = _registry.tasks
177 completed = [(tid, t) for tid, t in tasks.items() if t.status in done_statuses]
178 excess = len(completed) - _MAX_COMPLETED_TASKS
179 if excess <= 0:
180 return
181 completed.sort(key=lambda pair: pair[1].finished_at)
182 for tid, _ in completed[:excess]:
183 del tasks[tid]
186def start_crawl(
187 url: str,
188 depth: int | None = None,
189 max_pages: int | None = None,
190 render_mode: CrawlRenderMode | None = None,
191 *,
192 include_subdomains: bool = False,
193 login: CrawlLogin = NO_LOGIN,
194) -> str:
195 """Create a crawl task and launch it as an asyncio background task.
197 Defaults to whole-site unbounded recursion. Pass depth=0 for single URL.
198 ``render_mode`` of ``None`` defers to ``cfg.crawl_render_mode``.
199 ``include_subdomains`` widens whole-site scope to the host's subdomains.
200 ``login`` goes to the seed host and is dropped when the crawl ends.
201 Returns the task_id for status polling.
202 """
203 _evict_completed()
204 task_id = uuid.uuid4().hex[:12]
205 task = CrawlTask(
206 task_id=task_id,
207 url=url,
208 depth=depth,
209 max_pages=max_pages,
210 render_mode=render_mode,
211 include_subdomains=include_subdomains,
212 login=login,
213 )
214 _registry.tasks[task_id] = task
215 task._async_task = asyncio.create_task(run_crawl(task))
216 return task_id
219def get_task(task_id: str) -> CrawlTask | None:
220 """Look up a crawl task by ID."""
221 return _registry.tasks.get(task_id)
224def cancel_crawl(task_id: str) -> bool:
225 """Ask a running crawl to stop, returning whether it was still running.
227 Cooperative: the crawl stops at the next page boundary and keeps what it
228 already saved, so the pages fetched so far still reach the corpus. A task
229 that already finished is left alone.
230 """
231 task = _registry.tasks.get(task_id)
232 if task is None or task.status not in (TaskStatus.PENDING, TaskStatus.RUNNING):
233 return False
234 task.cancel.set()
235 return True
238def list_tasks() -> list[CrawlTask]:
239 """Return all tracked crawl tasks (active and completed)."""
240 return list(_registry.tasks.values())
243def clear_tasks() -> None:
244 """Remove all tasks from the registry (for testing)."""
245 _registry.clear()