Coverage for product_risk_suite/scraper/management/commands/sync_evidence_scrapers.py: 0%

1import logging 

2import time 

3 

4from django.conf import settings 

5from django.core.management.base import BaseCommand 

6 

7from risk_assessment.models import Evidence 

8from scraper.services import run_scraper_for_evidence 

9 

10logger = logging.getLogger("scraper") 

11 

12 

13class Command(BaseCommand): 

14 help = "Runs every Evidence's assigned scraper once, or forever on a configurable cycle with --loop." 

15 

16 def add_arguments(self, parser): 

17 parser.add_argument( 

18 "--loop", 

19 action="store_true", 

20 help="Keep running, sleeping EVIDENCE_SCRAPER_INTERVAL_SECONDS between passes.", 

21 ) 

22 

23 def handle(self, *args, **options): 

24 if options["loop"]: 

25 while True: 

26 self._run_once() 

27 time.sleep(settings.EVIDENCE_SCRAPER_INTERVAL_SECONDS) 

28 else: 

29 self._run_once() 

30 

31 def _run_once(self): 

32 succeeded = failed = 0 

33 for evidence in Evidence.objects.filter(scraper__isnull=False): 

34 result = run_scraper_for_evidence(evidence) 

35 if result: 

36 succeeded += 1 

37 elif result is False: 

38 failed += 1 

39 summary = f"Evidence scraper sync: {succeeded} succeeded, {failed} failed." 

40 logger.info(summary) 

41 self.stdout.write(summary)