|
11 | 11 |
|
12 | 12 | import asyncio |
13 | 13 | import os |
| 14 | +import time |
14 | 15 | import concurrent.futures |
15 | 16 | from abc import ABC |
| 17 | +from datetime import datetime, timezone |
16 | 18 | from typing import List, Dict, Any, Optional, Union |
17 | 19 |
|
18 | 20 | from ..core.engine import AsyncEngine |
@@ -356,6 +358,137 @@ def _fetch_results(self, snapshot_id: str, format: str = "json") -> Any: |
356 | 358 | """Fetch scrape job results (internal sync wrapper).""" |
357 | 359 | return _run_blocking(self._fetch_results_async(snapshot_id, format=format)) |
358 | 360 |
|
| 361 | + # ============================================================================ |
| 362 | + # PUBLIC SERVICE VERBS (drive a triggered job by snapshot_id) |
| 363 | + # ============================================================================ |
| 364 | + # These let any triggered scrape be polled/fetched from the service, given |
| 365 | + # only its snapshot_id — the "colorless job" pattern the crawler already uses |
| 366 | + # (CrawlJob is data; CrawlerService.status/download own the verbs). They are |
| 367 | + # purely additive: the same logic that ScrapeJob.status/wait/fetch/to_result |
| 368 | + # run, exposed on the service so a plain snapshot_id (or a future colorless |
| 369 | + # snapshot) is all a caller needs. |
| 370 | + |
| 371 | + async def status(self, snapshot_id: str) -> str: |
| 372 | + """ |
| 373 | + Check a triggered job's status by snapshot_id. |
| 374 | +
|
| 375 | + Returns: |
| 376 | + Status string: "ready", "in_progress", "error", etc. |
| 377 | +
|
| 378 | + Example: |
| 379 | + >>> sid = (await scraper.products_trigger(url)).snapshot_id |
| 380 | + >>> s = await scraper.status(sid) |
| 381 | + """ |
| 382 | + return await self._check_status_async(snapshot_id) |
| 383 | + |
| 384 | + async def wait( |
| 385 | + self, |
| 386 | + snapshot_id: str, |
| 387 | + timeout: Optional[int] = None, |
| 388 | + poll_interval: int = DEFAULT_POLL_INTERVAL, |
| 389 | + verbose: bool = False, |
| 390 | + ) -> str: |
| 391 | + """ |
| 392 | + Poll a triggered job until it is ready (by snapshot_id). |
| 393 | +
|
| 394 | + Args: |
| 395 | + snapshot_id: Snapshot identifier from a trigger operation. |
| 396 | + timeout: Maximum seconds to wait (uses MIN_POLL_TIMEOUT if None). |
| 397 | + poll_interval: Seconds between status checks. |
| 398 | + verbose: Print status updates. |
| 399 | +
|
| 400 | + Returns: |
| 401 | + Final status ("ready"). |
| 402 | +
|
| 403 | + Raises: |
| 404 | + TimeoutError: If timeout is reached. |
| 405 | + APIError: If the job fails. |
| 406 | + """ |
| 407 | + wait_timeout = timeout or self.MIN_POLL_TIMEOUT |
| 408 | + start_time = time.time() |
| 409 | + |
| 410 | + while True: |
| 411 | + elapsed = time.time() - start_time |
| 412 | + if elapsed > wait_timeout: |
| 413 | + raise TimeoutError(f"Job {snapshot_id} timed out after {wait_timeout}s") |
| 414 | + |
| 415 | + current = await self.status(snapshot_id) |
| 416 | + |
| 417 | + if verbose: |
| 418 | + print(f" [{elapsed:.1f}s] Job status: {current}") |
| 419 | + |
| 420 | + if current == "ready": |
| 421 | + return current |
| 422 | + elif current == "error" or current == "failed": |
| 423 | + raise APIError(f"Job {snapshot_id} failed with status: {current}") |
| 424 | + |
| 425 | + await asyncio.sleep(poll_interval) |
| 426 | + |
| 427 | + async def fetch(self, snapshot_id: str, format: str = "json") -> Any: |
| 428 | + """ |
| 429 | + Fetch a triggered job's results by snapshot_id. |
| 430 | +
|
| 431 | + Note: Does not check readiness. Use wait() first or check status(). |
| 432 | +
|
| 433 | + Args: |
| 434 | + snapshot_id: Snapshot identifier from a trigger operation. |
| 435 | + format: Result format ("json" or "raw"). |
| 436 | +
|
| 437 | + Returns: |
| 438 | + The job results. |
| 439 | + """ |
| 440 | + return await self._fetch_results_async(snapshot_id, format=format) |
| 441 | + |
| 442 | + async def to_result( |
| 443 | + self, |
| 444 | + snapshot_id: str, |
| 445 | + timeout: Optional[int] = None, |
| 446 | + poll_interval: int = DEFAULT_POLL_INTERVAL, |
| 447 | + ) -> ScrapeResult: |
| 448 | + """ |
| 449 | + Wait for a triggered job to complete, fetch it, and wrap in a ScrapeResult. |
| 450 | +
|
| 451 | + Convenience that combines wait() + fetch() + result construction, by |
| 452 | + snapshot_id. Cost/timing are derived from the service's own |
| 453 | + PLATFORM_NAME / COST_PER_RECORD, so nothing is needed beyond the id. |
| 454 | +
|
| 455 | + Args: |
| 456 | + snapshot_id: Snapshot identifier from a trigger operation. |
| 457 | + timeout: Maximum seconds to wait (uses MIN_POLL_TIMEOUT if None). |
| 458 | + poll_interval: Seconds between status checks. |
| 459 | +
|
| 460 | + Returns: |
| 461 | + ScrapeResult (success or, on failure, success=False with error set). |
| 462 | + """ |
| 463 | + start_time = datetime.now(timezone.utc) |
| 464 | + |
| 465 | + try: |
| 466 | + await self.wait(snapshot_id, timeout=timeout, poll_interval=poll_interval) |
| 467 | + data = await self.fetch(snapshot_id) |
| 468 | + end_time = datetime.now(timezone.utc) |
| 469 | + |
| 470 | + record_count = len(data) if isinstance(data, list) else 1 |
| 471 | + estimated_cost = record_count * self.COST_PER_RECORD |
| 472 | + |
| 473 | + return ScrapeResult( |
| 474 | + success=True, |
| 475 | + data=data, |
| 476 | + platform=self.PLATFORM_NAME or None, |
| 477 | + cost=estimated_cost, |
| 478 | + snapshot_id=snapshot_id, |
| 479 | + trigger_sent_at=start_time, |
| 480 | + data_fetched_at=end_time, |
| 481 | + ) |
| 482 | + except Exception as e: |
| 483 | + return ScrapeResult( |
| 484 | + success=False, |
| 485 | + error=str(e), |
| 486 | + platform=self.PLATFORM_NAME or None, |
| 487 | + snapshot_id=snapshot_id, |
| 488 | + trigger_sent_at=start_time, |
| 489 | + data_fetched_at=datetime.now(timezone.utc), |
| 490 | + ) |
| 491 | + |
359 | 492 | # ============================================================================ |
360 | 493 | # CONTEXT MANAGER SUPPORT (for standalone usage) |
361 | 494 | # ============================================================================ |
|
0 commit comments