2465821143

Build Type

push

github

Committed by mborsetti

Commit Message

Version 3.10.2rc5

Run Details

1368 of 1917 branches covered (71.36%)

Branch coverage included in aggregate %.

14 of 14 new or added lines in 4 files covered. (100.0%)

2 existing lines in 1 file now uncovered.

3244 of 3975 relevant lines covered (81.61%)

6.41 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.72

/webchanges/worker.py

"""The worker that runs jobs in parallel.  Called from main module."""

# The code below is subject to the license contained in the LICENSE file, which is part of the source code.

from __future__ import annotations

import difflib
import logging
import os
import random
import urllib.parse
from concurrent.futures import ThreadPoolExecutor
from contextlib import ExitStack
from typing import Iterable, Optional, TYPE_CHECKING

from .handler import JobState
from .jobs import BrowserJob, NotModifiedError, UrlJobBase

# https://stackoverflow.com/questions/39740632
if TYPE_CHECKING:
    from typing import List

    from .jobs import JobBase
    from .main import Urlwatch

logger = logging.getLogger(__name__)


def run_jobs(urlwatcher: Urlwatch) -> None:
    """Process (run) jobs in parallel.

    :raises IndexError: If any index(es) is/are out of range.
    """

    def insert_delay(jobs: List[JobBase]) -> List[JobBase]:  # pragma: no cover
        """
        TODO: Evaluate whether this is necessary; currently not being called.  Remove pragma no cover.

        Sets a _delay value for URL jobs hitting the network location already hit. Used to prevent multiple jobs
        hitting the same network location at the exact same time and being blocked as a result.

        CHANGELOG:
        * When multiple URL jobs have the same network location (www.example.com), a random delay between 0.1 and 1.0
          seconds is added to all jobs to that network location after the first one. This prevents being blocked by
          the site as a result of being flooded by **webchanges**'s parallelism sending multiple requests from the same
          source at the same exact time.

        :param jobs: The list of jobs.
        :return: The list of jobs with the _delay value set.
        """
        previous_netlocs = set()
        for job in jobs:
            if isinstance(job, UrlJobBase):
                netloc = urllib.parse.urlparse(job.url).netloc
                if netloc in previous_netlocs:
                    job._delay = random.uniform(0.1, 1)  # nosec [B311:blacklist] Standard pseudo-random generator
                else:
                    previous_netlocs.add(netloc)
        return jobs

    def job_runner(
        stack: ExitStack,
        jobs: Iterable[JobBase],
        max_workers: Optional[int] = None,
    ) -> None:
        """
        Runs the jobs in parallel.

        :param stack: The context manager.
        :param jobs: The jobs to run.
        :param max_workers: The number of maximum workers for ThreadPoolExecutor.
        :return: None
        """
        executor = ThreadPoolExecutor(max_workers=max_workers)

        # launch future to retrieve if new version is available
        if urlwatcher.report.new_release_future is None:
            urlwatcher.report.new_release_future = executor.submit(urlwatcher.get_new_release_version)

        for job_state in executor.map(
            lambda jobstate: jobstate.process(headless=not urlwatcher.urlwatch_config.no_headless),
            (stack.enter_context(JobState(urlwatcher.cache_storage, job)) for job in jobs),
        ):

            max_tries = 0 if not job_state.job.max_tries else job_state.job.max_tries
            # tries is incremented by JobState.process when an exception (including 304) is encountered.

            if job_state.exception is not None:
                # Oops, we have captured an error (which could also be 304)
                if job_state.error_ignored:
                    # We captured an error to ignore
                    logger.info(
                        f'Job {job_state.job.index_number}: Error while executing job was ignored due to job config'
                    )
                elif isinstance(job_state.exception, NotModifiedError):
                    # We captured a 304 Not Modified
                    logger.info(
                        f'Job {job_state.job.index_number}: Job has not changed (HTTP 304 response or same strong '
                        f'ETag)'
                    )
                    if job_state.tries > 0:
                        job_state.tries = 0
                        job_state.save(use_old_data=True)  # data is not returned by 304 therefore reuse old data
                    urlwatcher.report.unchanged(job_state)
                elif job_state.tries < max_tries:
                    # We're not reporting the error yet because we haven't yet hit 'max_tries'
                    logger.debug(
                        f'Job {job_state.job.index_number}: Error suppressed as cumulative number of '
                        f'failures ({job_state.tries}) does not exceed max_tries={max_tries}'
                    )
                    job_state.save(use_old_data=True)  # do not save error data but reuse old data
                else:
                    # Reporting the error
                    logger.debug(
                        f'Job {job_state.job.index_number}: Flagged as error as max_tries={max_tries} has been '
                        f'met or exceeded ({job_state.tries}'
                    )
                    job_state.save(use_old_data=True)  # do not save error data but reuse old data
                    urlwatcher.report.error(job_state)

            elif len(job_state.old_data) or job_state.old_timestamp != 0:
                # This is not the first time running this job (we have snapshots)
                if job_state.new_data in (job_state.old_data, job_state.history_data):
                    # exactly matches one of the previous snapshots
                    if job_state.tries > 0:
                        job_state.tries = 0
                        job_state.save()
                    urlwatcher.report.unchanged(job_state)
                else:
                    # no match
                    if len(job_state.history_data) > 1:
                        # replace old with best "good enough" previous snapshot
                        close_matches = difflib.get_close_matches(job_state.new_data, job_state.history_data, n=1)
                        if close_matches:
                            job_state.old_data = close_matches[0]
                            job_state.old_timestamp = job_state.history_data[close_matches[0]]
                    job_state.tries = 0
                    job_state.save()
                    urlwatcher.report.changed(job_state)
            else:
                # We have never run this job before (there are no snapshots)
                job_state.tries = 0
                job_state.save()
                urlwatcher.report.new(job_state)

    def get_virt_mem() -> int:
        try:
            import psutil
        except ImportError:
            raise ImportError(
                "Python package psutil is not installed; cannot use 'use_browser: true'. Please install "
                "dependencies with 'pip install webchanges[use_browser]'."
            )
        try:
            virt_mem = psutil.virtual_memory().available
            logger.debug(
                f'Found {virt_mem / 1e6:,.0f} MB of available physical memory (plus '
                f'{psutil.swap_memory().free / 1e6:,.0f} MB of swap).'
            )
        except psutil.Error as e:  # pragma: no cover
            virt_mem = 0
            logger.debug(f'Could not read memory: {e}')

        return virt_mem

    # extract subset of jobs to run if joblist CLI was set
    if urlwatcher.urlwatch_config.joblist:
        for idx in urlwatcher.urlwatch_config.joblist:
            if not (-len(urlwatcher.jobs) <= idx <= -1 or 1 <= idx <= len(urlwatcher.jobs)):
                raise IndexError(f'Job index {idx} out of range (found {len(urlwatcher.jobs)} jobs)')
        urlwatcher.urlwatch_config.joblist = [
            jn if jn > 0 else len(urlwatcher.jobs) + jn + 1 for jn in urlwatcher.urlwatch_config.joblist
        ]
        jobs = [
            job.with_defaults(urlwatcher.config_storage.config)
            for job in urlwatcher.jobs
            if job.index_number in urlwatcher.urlwatch_config.joblist
        ]
        logger.debug(f'Processing {len(jobs)} job as specified (# {urlwatcher.urlwatch_config.joblist})')
    else:
        jobs = [job.with_defaults(urlwatcher.config_storage.config) for job in urlwatcher.jobs]
        logger.debug(f'Processing {len(jobs)} jobs')

    #    jobs = insert_delay(jobs)

    with ExitStack() as stack:
        # run non-BrowserJob jobs first
        jobs_to_run = [job for job in jobs if type(job) != BrowserJob]
        if jobs_to_run:
            logger.debug(
                "Running jobs that do not require Chrome (without 'use_browser: true') in parallel with Python's "
                'default max_workers.'
            )
            job_runner(stack, jobs_to_run)
        else:
            logger.debug("Found no jobs that do not require Chrome (i.e. without 'use_browser: true').")

        # run BrowserJob jobs after
        jobs_to_run = [job for job in jobs if type(job) == BrowserJob]
        if jobs_to_run:
            virt_mem = get_virt_mem()
            max_workers = max(int(virt_mem / 120e6), 1)
            max_workers = min(max_workers, os.cpu_count() or 1)
            logger.debug(
                f"Running jobs that require Chrome (i.e. with 'use_browser: true') in parallel with {max_workers} "
                f'max_workers.'
            )
            job_runner(stack, jobs_to_run, max_workers)
        else:
            logger.debug("Found no jobs that require Chrome (i.e. with 'use_browser: true').")

1	"""The worker that runs jobs in parallel. Called from main module."""
2
3	# The code below is subject to the license contained in the LICENSE file, which is part of the source code.
4
5	from __future__ import annotations	8✔
6
7	import difflib	8✔
8	import logging	8✔
9	import os	8✔
10	import random	8✔
11	import urllib.parse	8✔
12	from concurrent.futures import ThreadPoolExecutor	8✔
13	from contextlib import ExitStack	8✔
14	from typing import Iterable, Optional, TYPE_CHECKING	8✔
15
16	from .handler import JobState	8✔
17	from .jobs import BrowserJob, NotModifiedError, UrlJobBase	8✔
18
19	# https://stackoverflow.com/questions/39740632
20	if TYPE_CHECKING:
21	from typing import List
22
23	from .jobs import JobBase
24	from .main import Urlwatch
25
26	logger = logging.getLogger(__name__)	8✔
27
28
29	def run_jobs(urlwatcher: Urlwatch) -> None:	8✔
30	"""Process (run) jobs in parallel.
31
32	:raises IndexError: If any index(es) is/are out of range.
33	"""
34
35	def insert_delay(jobs: List[JobBase]) -> List[JobBase]: # pragma: no cover
36	"""
37	TODO: Evaluate whether this is necessary; currently not being called. Remove pragma no cover.
38
39	Sets a _delay value for URL jobs hitting the network location already hit. Used to prevent multiple jobs
40	hitting the same network location at the exact same time and being blocked as a result.
41
42	CHANGELOG:
43	* When multiple URL jobs have the same network location (www.example.com), a random delay between 0.1 and 1.0
44	seconds is added to all jobs to that network location after the first one. This prevents being blocked by
45	the site as a result of being flooded by webchanges's parallelism sending multiple requests from the same
46	source at the same exact time.
47
48	:param jobs: The list of jobs.
49	:return: The list of jobs with the _delay value set.
50	"""
51	previous_netlocs = set()
52	for job in jobs:
53	if isinstance(job, UrlJobBase):
54	netloc = urllib.parse.urlparse(job.url).netloc
55	if netloc in previous_netlocs:
56	job._delay = random.uniform(0.1, 1) # nosec [B311:blacklist] Standard pseudo-random generator
57	else:
58	previous_netlocs.add(netloc)
59	return jobs
60
61	def job_runner(	8✔
62	stack: ExitStack,
63	jobs: Iterable[JobBase],
64	max_workers: Optional[int] = None,
65	) -> None:
66	"""
67	Runs the jobs in parallel.
68
69	:param stack: The context manager.
70	:param jobs: The jobs to run.
71	:param max_workers: The number of maximum workers for ThreadPoolExecutor.
72	:return: None
73	"""
74	executor = ThreadPoolExecutor(max_workers=max_workers)	8✔
75
76	# launch future to retrieve if new version is available
77	if urlwatcher.report.new_release_future is None:	8✔
78	urlwatcher.report.new_release_future = executor.submit(urlwatcher.get_new_release_version)	8✔
79
80	for job_state in executor.map(	8✔
81	lambda jobstate: jobstate.process(headless=not urlwatcher.urlwatch_config.no_headless),
82	(stack.enter_context(JobState(urlwatcher.cache_storage, job)) for job in jobs),
83	):
84
85	max_tries = 0 if not job_state.job.max_tries else job_state.job.max_tries	8✔
86	# tries is incremented by JobState.process when an exception (including 304) is encountered.
87
88	if job_state.exception is not None:	8✔
89	# Oops, we have captured an error (which could also be 304)
90	if job_state.error_ignored:	8!
91	# We captured an error to ignore
UNCOV 92	logger.info(	×
93	f'Job {job_state.job.index_number}: Error while executing job was ignored due to job config'
94	)
95	elif isinstance(job_state.exception, NotModifiedError):	8!
96	# We captured a 304 Not Modified
UNCOV 97	logger.info(	×
98	f'Job {job_state.job.index_number}: Job has not changed (HTTP 304 response or same strong '
99	f'ETag)'
100	)
101	if job_state.tries > 0:	×
102	job_state.tries = 0	×
103	job_state.save(use_old_data=True) # data is not returned by 304 therefore reuse old data	×
104	urlwatcher.report.unchanged(job_state)	×
105	elif job_state.tries < max_tries:	8✔
106	# We're not reporting the error yet because we haven't yet hit 'max_tries'
107	logger.debug(	8✔
108	f'Job {job_state.job.index_number}: Error suppressed as cumulative number of '
109	f'failures ({job_state.tries}) does not exceed max_tries={max_tries}'
110	)
111	job_state.save(use_old_data=True) # do not save error data but reuse old data	8✔
112	else:
113	# Reporting the error
114	logger.debug(	8✔
115	f'Job {job_state.job.index_number}: Flagged as error as max_tries={max_tries} has been '
116	f'met or exceeded ({job_state.tries}'
117	)
118	job_state.save(use_old_data=True) # do not save error data but reuse old data	8✔
119	urlwatcher.report.error(job_state)	8✔
120
121	elif len(job_state.old_data) or job_state.old_timestamp != 0:	8✔
122	# This is not the first time running this job (we have snapshots)
123	if job_state.new_data in (job_state.old_data, job_state.history_data):	8✔
124	# exactly matches one of the previous snapshots
125	if job_state.tries > 0:	8✔
126	job_state.tries = 0	8✔
127	job_state.save()	8✔
128	urlwatcher.report.unchanged(job_state)	8✔
129	else:
130	# no match
131	if len(job_state.history_data) > 1:	8!
132	# replace old with best "good enough" previous snapshot
133	close_matches = difflib.get_close_matches(job_state.new_data, job_state.history_data, n=1)	×
134	if close_matches:	×
135	job_state.old_data = close_matches[0]	×
136	job_state.old_timestamp = job_state.history_data[close_matches[0]]	×
137	job_state.tries = 0	8✔
138	job_state.save()	8✔
139	urlwatcher.report.changed(job_state)	8✔
140	else:
141	# We have never run this job before (there are no snapshots)
142	job_state.tries = 0	8✔
143	job_state.save()	8✔
144	urlwatcher.report.new(job_state)	8✔
145
146	def get_virt_mem() -> int:	8✔
147	try:	6✔
148	import psutil	6✔
149	except ImportError:
150	raise ImportError(
151	"Python package psutil is not installed; cannot use 'use_browser: true'. Please install "
152	"dependencies with 'pip install webchanges[use_browser]'."
153	)
154	try:	6✔
155	virt_mem = psutil.virtual_memory().available	6✔
156	logger.debug(	6✔
157	f'Found {virt_mem / 1e6:,.0f} MB of available physical memory (plus '
158	f'{psutil.swap_memory().free / 1e6:,.0f} MB of swap).'
159	)
160	except psutil.Error as e: # pragma: no cover
161	virt_mem = 0
162	logger.debug(f'Could not read memory: {e}')
163
164	return virt_mem	6✔
165
166	# extract subset of jobs to run if joblist CLI was set
167	if urlwatcher.urlwatch_config.joblist:	8✔
168	for idx in urlwatcher.urlwatch_config.joblist:	8✔
169	if not (-len(urlwatcher.jobs) <= idx <= -1 or 1 <= idx <= len(urlwatcher.jobs)):	8✔
170	raise IndexError(f'Job index {idx} out of range (found {len(urlwatcher.jobs)} jobs)')	8✔
171	urlwatcher.urlwatch_config.joblist = [	8✔
172	jn if jn > 0 else len(urlwatcher.jobs) + jn + 1 for jn in urlwatcher.urlwatch_config.joblist
173	]
174	jobs = [	8✔
175	job.with_defaults(urlwatcher.config_storage.config)
176	for job in urlwatcher.jobs
177	if job.index_number in urlwatcher.urlwatch_config.joblist
178	]
179	logger.debug(f'Processing {len(jobs)} job as specified (# {urlwatcher.urlwatch_config.joblist})')	8✔
180	else:
181	jobs = [job.with_defaults(urlwatcher.config_storage.config) for job in urlwatcher.jobs]	8✔
182	logger.debug(f'Processing {len(jobs)} jobs')	8✔
183
184	# jobs = insert_delay(jobs)
185
186	with ExitStack() as stack:	8✔
187	# run non-BrowserJob jobs first
188	jobs_to_run = [job for job in jobs if type(job) != BrowserJob]	8✔
189	if jobs_to_run:	8✔
190	logger.debug(	8✔
191	"Running jobs that do not require Chrome (without 'use_browser: true') in parallel with Python's "
192	'default max_workers.'
193	)
194	job_runner(stack, jobs_to_run)	8✔
195	else:
196	logger.debug("Found no jobs that do not require Chrome (i.e. without 'use_browser: true').")	6✔
197
198	# run BrowserJob jobs after
199	jobs_to_run = [job for job in jobs if type(job) == BrowserJob]	8✔
200	if jobs_to_run:	8✔
201	virt_mem = get_virt_mem()	6✔
202	max_workers = max(int(virt_mem / 120e6), 1)	6✔
203	max_workers = min(max_workers, os.cpu_count() or 1)	6✔
204	logger.debug(	6✔
205	f"Running jobs that require Chrome (i.e. with 'use_browser: true') in parallel with {max_workers} "
206	f'max_workers.'
207	)
208	job_runner(stack, jobs_to_run, max_workers)	6✔
209	else:
210	logger.debug("Found no jobs that require Chrome (i.e. with 'use_browser: true').")	8✔

mborsetti / webchanges / 2465821143

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous