diff --git a/cl/scrapers/management/commands/back_scrape_fl_dockets.py b/cl/scrapers/management/commands/back_scrape_fl_dockets.py index 8d62d6fcde..838eb1c81d 100644 --- a/cl/scrapers/management/commands/back_scrape_fl_dockets.py +++ b/cl/scrapers/management/commands/back_scrape_fl_dockets.py @@ -6,12 +6,14 @@ from datetime import date from functools import lru_cache from pathlib import Path +from typing import ClassVar +from uuid import UUID import botocore.exceptions from asgiref.sync import async_to_sync from django.core.management import CommandParser from httpx import Response -from juriscraper.state.florida import FloridaScraper +from juriscraper.state.florida import FloridaCase, FloridaScraper from juriscraper.state.florida.courts import FloridaCourtID from juriscraper.state.RequestManager import ( ExponentialBackoff, @@ -35,6 +37,10 @@ S3_BASE = Path("responses/dockets/florida") +def _make_case_key(court_id: FloridaCourtID, docket_number: str) -> str: + return f"{S3_BASE}/parsed/{court_id.value}/{_make_case_number_key(docket_number)}.html" + + @dataclass class S3Cache(RequestHandler): """Handler to use S3 as a (successful) response cache @@ -157,6 +163,60 @@ def __hash__(self) -> int: } +async def _get_full_case( # type: ignore[return] + court_id: str, case_uuid: str, scraper: FloridaScraper +) -> FloridaCase | None: + """Attempt to fetch the full case data for a given case, returning `None` if any error occurred.""" + match await scraper.fetch_case_data(case_uuid, court_id): + case Exception() as e: + logger.error( + "Failed to fetch full case data for %s: %s", + case_uuid, + e, + ) + return None + case (_, errors) if errors: + logger.error( + "Failed to fetch full case data for %s: %r", + case_uuid, + errors, + ) + return None + case (full_case, _): + return full_case + + +async def _backfill_targeted( + cases: list[tuple[FloridaCourtID, UUID]], + throttle: CeleryThrottle, + queue_name: str, + scraper: FloridaScraper, +): + logger.info("Starting targeted backfill of %d cases...", len(cases)) + for i, (court_id, case_uuid) in enumerate(cases): + logger.info("Fetching case %s for %s", case_uuid, court_id) + case = await _get_full_case(court_id.value, str(case_uuid), scraper) + if case is None: + continue + + content = case.model_dump_json(ensure_ascii=True).encode("utf-8") + + key = _make_case_key(court_id, case.docket_number) + + throttle.maybe_wait() + save_response_to_s3.si(key, content).set( + queue=queue_name + ).apply_async() + + if i % 10 == 0: + logger.info( + "Completed scrape of %d/%d cases (%.2f%%)", + i + 1, + len(cases), + (i + 1) / len(cases) * 100, + ) + + async def _backfill( date_ranges: dict[FloridaCourtID, tuple[date, date]], full_scrape: bool, @@ -164,9 +224,11 @@ async def _backfill( scraper: FloridaScraper, throttle: CeleryThrottle, queue_name: str, + skip_parsed: bool, + cache: S3Cache, ): logger.info("Starting Florida backfill...") - loop = asyncio.get_running_loop() + full_scrape_loop = full_scrape and not skip_parsed try: for court_id, (start_date, end_date) in date_ranges.items(): i = 0 @@ -174,9 +236,19 @@ async def _backfill( start_date, end_date, court_ids=[court_id], - full_scrape=full_scrape, + full_scrape=full_scrape_loop, ): - key = f"{S3_BASE}/parsed/{court_id.value}/{_make_case_number_key(case.docket_number)}.json" + key = _make_case_key(court_id, case.docket_number) + if skip_parsed and cache.s3_key_exists(key): + continue + if full_scrape and skip_parsed: + full_case = await _get_full_case( + court_id.value, str(case.case_uuid), scraper + ) + if not full_case: + continue + case = full_case + content = case.model_dump_json(ensure_ascii=True).encode( "utf-8" ) @@ -212,6 +284,7 @@ class Command(StateBackScrapeCommand): FloridaCourtID.FIFTH_COA: ScraperCheckpointTracker("acisfl5ac"), FloridaCourtID.SIXTH_COA: ScraperCheckpointTracker("acisfl6ac"), } + date_range_required: ClassVar[bool] = False def add_arguments(self, parser: CommandParser): super().add_arguments(parser) @@ -273,6 +346,19 @@ def add_arguments(self, parser: CommandParser): help="CeleryThrottle min queue depth; the throttle keeps the " "queue between this and 2x this value.", ) + parser.add_argument( + "--skip-parsed", + dest="skip_parsed", + action="store_true", + help="Whether to skip items which already have parsed files saved in S3.", + ) + parser.add_argument( + "--scrape-uuids", + dest="scrape_uuids", + type=str, + default="", + help="Path to a CSV of Florida court IDs and case UUIDs for targeted rescraping. Overrides the --backscrape-start, --backscrape-end, and --courts options.", + ) def handle( self, @@ -287,25 +373,28 @@ def handle( archive_responses: bool, queue: str, throttle_min_items: int, - backscrape_start: str, - backscrape_end: str, + backscrape_start: str = "", + backscrape_end: str = "", courts: str, + skip_parsed: bool, + scrape_uuids: str, **options, ): - logger.info("Setting up Florida back-scrape...") - start = _parse_date(backscrape_start) - end = _parse_date(backscrape_end) - court_ids = [ - _COURT_ID_MAP[c.strip()] for c in courts.split(",") if c.strip() - ] - if not court_ids: - logger.warning("No courts specified. Defaulting to all courts.") - court_ids = list(_COURT_ID_MAP.values()) - throttle = CeleryThrottle( queue_name=queue, min_items=throttle_min_items ) + storage = S3GlacierInstantRetrievalStorage() + + cache = S3Cache( + base=S3_BASE, + save=archive_responses, + load=use_cache, + throttle=throttle, + queue_name=queue, + storage=storage, + ) + scraper = FloridaScraper( rps=rps, retry=ExponentialBackoff( @@ -313,17 +402,42 @@ def handle( backoff=backoff, backoff_growth=backoff_growth, ), - handlers=[ - S3Cache( - base=S3_BASE, - save=archive_responses, - load=use_cache, - throttle=throttle, - queue_name=queue, - ) - ], + handlers=[cache], ) + logger.info("Setting up Florida back-scrape...") + if scrape_uuids: + p = Path(scrape_uuids) + with p.open() as f: + cases = [ + ( + _COURT_ID_MAP[c.strip()], + UUID(case_uuid.strip()), + ) + for c, case_uuid in (line.split(",") for line in f) + ] + async_to_sync(_backfill_targeted)( + cases=cases, + throttle=throttle, + queue_name=queue, + scraper=scraper, + ) + return + + if not backscrape_start or not backscrape_end: + logger.error( + "Both --backscrape-start and --backscrape-end are required." + ) + return + start = _parse_date(backscrape_start) + end = _parse_date(backscrape_end) + court_ids = [ + _COURT_ID_MAP[c.strip()] for c in courts.split(",") if c.strip() + ] + if not court_ids: + logger.warning("No courts specified. Defaulting to all courts.") + court_ids = list(_COURT_ID_MAP.values()) + if auto_resume: logger.info("Auto resume enabled. Getting checkpoints...") checkpoints = { @@ -353,4 +467,6 @@ def handle( scraper=scraper, throttle=throttle, queue_name=queue, + skip_parsed=skip_parsed, + cache=cache, ) diff --git a/cl/scrapers/management/utils.py b/cl/scrapers/management/utils.py index fc4492d075..b061e50f44 100644 --- a/cl/scrapers/management/utils.py +++ b/cl/scrapers/management/utils.py @@ -75,6 +75,7 @@ def set(self, checkpoint: date) -> None: class StateBackScrapeCommand(ABC, BaseCommand): checkpoint_tracker: ClassVar[ScraperCheckpointTracker] + date_range_required: ClassVar[bool] = True def add_arguments(self, parser: CommandParser): parser.add_argument( @@ -83,13 +84,13 @@ def add_arguments(self, parser: CommandParser): help="Starting value for backscraper iterable creation. " "Each scraper handles the parsing of the argument," "since the value may represent a year, a string, a date, etc.", - required=True, + required=self.date_range_required, ) parser.add_argument( "--backscrape-end", dest="backscrape_end", help="End value for backscraper iterable creation.", - required=True, + required=self.date_range_required, ) parser.add_argument( "--courts",