24977678469

Committed 27 Apr 2026 05:10AM UTC coverage: 73.171% (-0.09%) from 73.256%

Build # 24977678469

Build Type

push

github

Committed by

mborsetti

Commit Message

Version 3.36.1

Coverage Stats

1534 of 2498 branches covered (61.41%)

Branch coverage included in aggregate %.

1 of 1 new or added line in 1 file covered. (100.0%)

4 existing lines in 2 files now uncovered.

5336 of 6891 relevant lines covered (77.43%)

11.12 hits per line

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

86.15

/webchanges/handler.py

"""Handles the running of jobs and, afterward, of the reports."""

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

from __future__ import annotations

import logging
import os
import subprocess
import sys
import time
import traceback
from typing import TYPE_CHECKING, Any, ContextManager, Iterator, Literal, NamedTuple, Self, TypedDict
from zoneinfo import ZoneInfo

from webchanges.differs import DifferBase, ReportKind
from webchanges.filters import FilterBase
from webchanges.jobs import NotModifiedError
from webchanges.reporters import ReporterBase

# https://stackoverflow.com/questions/39740632
if TYPE_CHECKING:
    from concurrent.futures import Future
    from pathlib import Path
    from types import TracebackType

    from webchanges.jobs import JobBase
    from webchanges.main import Urlwatch
    from webchanges.storage import SsdbStorage, _Config, _ConfigDifferDefaults

logger = logging.getLogger(__name__)


class Snapshot(NamedTuple):
    """Type for Snapshot named tuple.

    * 0: data: str | bytes
    * 1: timestamp: float
    * 2: tries: int
    * 3: etag: str
    * 4: mime_type: mime_type
    * 5: error: ErrorData
    """

    data: str | bytes
    timestamp: float
    tries: int
    etag: str
    mime_type: str
    error_data: ErrorData


Verb = Literal[
    'new',  # new job
    'changed',  # valid data received, and it has changed
    'changed,no_report',  # valid data received, and it has changed, but filtered diff yields no report
    'unchanged',  # valid data received, no changes
    'unchanged,error_ended',  # valid data received, no changes from the last data received before an error
    'error',  # error, prior state was different (either data or different error)
    'error,repeated',  # error, same as before
]
ErrorData = TypedDict('ErrorData', {'type': str, 'message': str}, total=False)


class JobState(ContextManager):
    """The JobState class, which contains run information about a job."""

    _http_client_used: Literal['httpx', 'requests', 'curl_cffi', 'playwright'] | None = None
    error_ignored: bool
    exception: Exception | None = None
    generated_diff: dict[ReportKind, str]
    history_dic_snapshots: dict[str | bytes, Snapshot]
    new_data: str | bytes = ''
    new_error_data: ErrorData = {}
    new_etag: str = ''
    new_mime_type: str = ''
    new_timestamp: float
    old_snapshot = Snapshot(
        data='',
        timestamp=1605147837.511478,  # initialized to the first release of webchanges!
        tries=0,
        etag='',
        mime_type='text/plain',
        error_data={},
    )
    old_data: str | bytes = ''
    old_error_data: ErrorData = {}
    old_etag: str = ''
    old_mime_type: str = 'text/plain'
    old_timestamp: float = 1605147837.511478  # initialized to the first release of webchanges!
    traceback: str
    tries: int = 0  # if >1, an error; value is the consecutive number of runs leading to an error
    unfiltered_diff: dict[ReportKind, str]
    verb: Verb

    def __init__(self, snapshots_db: SsdbStorage, job: JobBase) -> None:
        """Initializes the class

        :param snapshots_db: The SsdbStorage object with the snapshot database methods.
        :param job: A JobBase object with the job information.
        """
        self.snapshots_db = snapshots_db
        self.job: JobBase = job

        self.generated_diff = {}
        self.unfiltered_diff = {}
        self.history_dic_snapshots = {}

    def __enter__(self) -> Self:
        """Context manager invoked on entry to the body of a with statement to make it possible to factor out standard
        uses of try/finally statements. Calls the main_thread_enter method of the Job.

        :returns: Class object.
        """
        # Below is legacy code that now does nothing, so it's being skipped
        # try:
        #     self.job.main_thread_enter()
        # except Exception as e:
        #     logger.info(f'Job {self.job.index_number}: Exception while creating resources for job', exc_info=True)
        #     self.exception = e
        #     self.traceback = self.job.format_error(e, traceback.format_exc())

        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        tb: TracebackType | None,
    ) -> bool | None:
        """Context manager invoked on exit from the body of a with statement to make it possible to factor out standard
        uses of try/finally statements. Calls the main_thread_exit() method of the Job.

        :returns: None.
        """
        # Below is legacy code that now does nothing, so it's being skipped
        # try:
        #     self.job.main_thread_exit()
        # except Exception:
        #     # We don't want exceptions from releasing resources to override job run results
        #     logger.warning(f'Job {self.index_number}: Exception while releasing resources for job', exc_info=True)
        # if isinstance(exc_value, subprocess.CalledProcessError):
        #     raise subprocess.SubprocessError(exc_value.stderr)
        # elif isinstance(exc_value, FileNotFoundError):
        #     raise OSError(exc_value)
        return None

    @staticmethod
    def debugging_session() -> bool:
        """Checks if the code is currently running within an external debugger (e.g. IDE) and *NOT* in a testing
        environment.

        :returns: True if an external debugger is attached and it's not a pytest session, False otherwise.
        """
        return sys.breakpointhook.__module__ != 'sys' and 'PYTEST_CURRENT_TEST' not in os.environ

    def added_data(self) -> dict[str, bool | str | Exception | float | None]:
        """Returns a dict with the data added in the processing of the job."""
        attrs = ('error_ignored', 'exception', 'new_data', 'new_etag', 'new_timestamp')
        return {attr: getattr(self, attr) for attr in attrs if hasattr(self, attr)}

    def load(self) -> None:
        """Loads form the database the last snapshot(s) for the job."""
        guid = self.job.guid
        self.old_snapshot = self.snapshots_db.load(guid)
        # TODO: Remove these
        (
            self.old_data,
            self.old_timestamp,
            self.tries,
            self.old_etag,
            self.old_mime_type,
            self.old_error_data,
        ) = self.old_snapshot
        if self.job.compared_versions and self.job.compared_versions > 1:
            self.history_dic_snapshots = {
                s.data: s for s in self.snapshots_db.get_history_snapshots(guid, self.job.compared_versions)
            }

    def save(self) -> None:
        """Saves new data retrieved by the job into the snapshot database."""
        if self.new_error_data:  # have encountered an exception, so save the old data
            new_snapshot = Snapshot(
                data=self.old_data,
                timestamp=self.new_timestamp,
                tries=self.tries,
                etag=self.old_etag,
                mime_type=self.old_mime_type,
                error_data=self.new_error_data,
            )
        else:
            new_snapshot = Snapshot(
                data=self.new_data,
                timestamp=self.new_timestamp,
                tries=self.tries,
                etag=self.new_etag,
                mime_type=self.new_mime_type,
                error_data=self.new_error_data,
            )
        self.snapshots_db.save(guid=self.job.guid, snapshot=new_snapshot)
        logger.info(f'Job {self.job.index_number}: Saved new data to database')

    def delete_latest(self, temporary: bool = True) -> None:
        """Removes the last instance in the snapshot database."""
        self.snapshots_db.delete_latest(guid=self.job.guid, temporary=temporary)

    def process(self, headless: bool = True) -> JobState:
        """Processes the job: loads it (i.e. runs it) and handles Exceptions (errors).

        :returns: a JobState object containing information of the job run.
        """
        logger.info(f'{self.job.get_indexed_location()} started processing ({type(self.job).__name__})')
        logger.debug(f'Job {self.job.index_number}: {self.job}')

        if self.exception and not isinstance(self.exception, NotModifiedError):
            self.new_timestamp = time.time()
            self.new_error_data = {'type': type(self.exception).__name__, 'message': str(self.exception)}
            logger.info(f'{self.job.get_indexed_location()} ended processing due to exception: {self.exception}')
            return self

        try:
            self.load()

            self.new_timestamp = time.time()
            data, self.new_etag, mime_type = self.job.retrieve(self, headless)
            logger.info(f'Job {self.job.index_number}: Retrieved {len(data)} bytes of {mime_type} content.')

            logger.debug(
                f'Job {self.job.index_number}: Retrieved data={data!r} | etag={self.new_etag} | mime_type={mime_type}'
            )

            # Apply automatic filters first
            filtered_data, mime_type = FilterBase.auto_process(self, data, mime_type)

            # Apply any specified filters
            for filter_kind, subfilter in FilterBase.normalize_filter_list(self.job.filters, self.job.index_number):  # ty:ignore[invalid-argument-type]
                filtered_data, mime_type = FilterBase.process(filter_kind, subfilter, self, filtered_data, mime_type)

            self.new_data = filtered_data
            self.new_mime_type = mime_type

        except NotModifiedError as e:
            # HTTP 304 response has been received
            self.exception = e
            self.error_ignored = False
        except Exception as e:
            # Processing error of job failed its chance to handle error
            # Job has a chance to format and ignore its error
            if self.debugging_session():
                logger.warning('Running in a debugging session: raising the exception instead of processing it')
                raise
            self.exception = e
            self.error_ignored = self.job.ignore_error(e)
            if not self.error_ignored:
                self.new_timestamp = time.time()
                # Check for specific exception types to provide more detailed tracebacks
                if self.job.__class__.__module__ == 'hooks':
                    logger.info('Job is from hooks.py: including full traceback in error message')
                    self.traceback = ''.join(traceback.format_exception(e)).rstrip()
                elif isinstance(e, subprocess.CalledProcessError):
                    self.traceback = (
                        f'subprocess.CalledProcessError: Command returned non-zero exit status {e.returncode}.\n\n'
                        + '\n'.join(filter(None, (e.stderr, e.stdout)))
                    )
                else:
                    # Generic traceback for other exceptions
                    self.traceback = self.job.format_error(e, traceback.format_exc())

                self.tries += 1
                self.new_error_data = {
                    'type': '.'.join(filter(None, [getattr(e, '__module__', None), e.__class__.__name__])),
                    'message': str(e),
                }
                logger.info(
                    f'Job {self.job.index_number}: Job ended with an error; incrementing cumulative error runs to '
                    f'{self.tries}'
                )

        logger.debug(f'Job {self.job.index_number}: Processed as {self.added_data()}')
        logger.info(f'{self.job.get_indexed_location()} ended processing')
        return self

    def get_diff(
        self,
        report_kind: ReportKind = 'plain',
        differ: dict[str, Any] | None = None,
        differ_defaults: _ConfigDifferDefaults | None = None,
        tz: ZoneInfo | None = None,
    ) -> str:
        """Generates the job's diff and applies diff_filters to it (if any). Memoized.

        :parameter report_kind: the kind of report that needs the differ.
        :parameter differ: the name of the differ to override self.job.differ.
        :parameter tz: The IANA tz_info name of the timezone to use for diff in the job's report (e.g. 'Etc/UTC').
        :returns: The job's diff.
        """
        if report_kind in self.generated_diff:
            return self.generated_diff[report_kind]

        if report_kind not in self.unfiltered_diff:
            differ_kind, subdiffer = DifferBase.normalize_differ(
                differ or self.job.differ,
                self.job.index_number,
                differ_defaults,
            )
            unfiltered_diff = DifferBase.process(differ_kind, subdiffer, self, report_kind, tz, self.unfiltered_diff)
            self.unfiltered_diff.update(unfiltered_diff)
        _generated_diff = self.unfiltered_diff[report_kind]
        if _generated_diff:
            # Apply any specified diff_filters
            _mime_type = 'text/plain'
            for filter_kind, subfilter in FilterBase.normalize_filter_list(
                self.job.diff_filters, self.job.index_number
            ):
                _generated_diff, _mime_type = FilterBase.process(
                    filter_kind, subfilter, self, _generated_diff, _mime_type
                )
        self.generated_diff[report_kind] = str(_generated_diff)

        return self.generated_diff[report_kind]

    def is_markdown(self) -> bool:
        """Returns whether the new data is in markdown."""
        return self.new_mime_type == 'text/markdown' or bool(self.job.is_markdown)


class Report:
    """The base class for reporting."""

    job_states: list[JobState] = []
    new_release_future: Future[str | bool] | None = None
    start: float = time.perf_counter()

    def __init__(self, urlwatch: Urlwatch) -> None:
        """:param urlwatch: The Urlwatch object with the program configuration information."""
        self.config: _Config = urlwatch.config_storage.config
        self.tz = (
            ZoneInfo(self.config['report']['tz'])
            if 'report' in self.config and self.config['report']['tz'] is not None
            else None
        )

    def _result(
        self,
        verb: Verb,
        job_state: JobState,
    ) -> None:
        """Logs error and appends the verb to the job_state.

        :param verb: Description of the result of the job run. Can be one of
          • 'new': new job;
          • 'changed': valid data received, and it has changed;
          • 'changed,no_report': valid data received, and it has changed, but no report;
          • 'unchanged': valid data received, no changes;
          • 'unchanged,error_ended': valid data received, no changes from the last data received before an error;
          • 'error': error, prior state was different (either data or different error);
          • 'error,repeated': error, same as before;
        or a custom message such as  'test'.  Ultimately called by job_runner.

        :param job_state: The JobState object with the information of the job run.
        """
        if job_state.exception is not None:
            logger.info(
                f'Job {job_state.job.index_number}: Got exception while processing job {job_state.job}',
                exc_info=job_state.exception,
            )

        job_state.verb = verb
        self.job_states.append(job_state)

    def new(self, job_state: JobState) -> None:
        """Sets the verb of the job in job_state to 'new'. Called by :py:func:`run_jobs` and tests.

        :param job_state: The JobState object with the information of the job run.
        """
        self._result('new', job_state)

    def changed(self, job_state: JobState) -> None:
        """Sets the verb of the job in job_state to 'changed'. Called by :py:func:`run_jobs` and tests.

        :param job_state: The JobState object with the information of the job run.
        """
        self._result('changed', job_state)

    def changed_no_report(self, job_state: JobState) -> None:
        """Sets the verb of the job in job_state to 'changed,no_report'. Called by :py:func:`run_jobs` and tests.

        :param job_state: The JobState object with the information of the job run.
        """
        self._result('changed,no_report', job_state)

    def unchanged(self, job_state: JobState) -> None:
        """Sets the verb of the job in job_state to 'unchanged'. Called by :py:func:`run_jobs` and tests.

        :param job_state: The JobState object with the information of the job run.
        """
        self._result('unchanged', job_state)

    def unchanged_from_error(self, job_state: JobState) -> None:
        """Sets the verb of the job in job_state to 'unchanged'. Called by :py:func:`run_jobs` and tests.

        :param job_state: The JobState object with the information of the job run.
        """
        self._result('unchanged,error_ended', job_state)

    def error(self, job_state: JobState) -> None:
        """Sets the verb of the job in job_state to 'error'. Called by :py:func:`run_jobs` and tests.

        :param job_state: The JobState object with the information of the job run.
        """
        self._result('error', job_state)

    def error_same_error(self, job_state: JobState) -> None:
        """Sets the verb of the job in job_state to 'error'. Called by :py:func:`run_jobs` and tests.

        :param job_state: The JobState object with the information of the job run.
        """
        self._result('error,repeated', job_state)

    def custom(
        self,
        job_state: JobState,
        label: Verb,
    ) -> None:
        """Sets the verb of the job in job_state to a custom label. Called by
        :py:func:`UrlwatchCommand.check_test_reporter`.

        :param job_state: The JobState object with the information of the job run.
        :param label: The label to set the information of the job run to.
        """
        self._result(label, job_state)

    def get_filtered_job_states(self, job_states: list[JobState]) -> Iterator[JobState]:
        """Returns JobStates that have reportable changes per config['display'].  Called from :py:Class:`ReporterBase`.

        :param job_states: The list of JobState objects with the information of the job runs.
        :returns: An iterable of JobState objects that have reportable changes per config['display'].
        """

        def job_is_reportable(self: Report, job_state: JobState) -> bool:
            """Identify whether the job should generate a report based on its verb, relevant sub-directives, and the
            reporters configuration."""
            logger.debug(f'Job {job_state.job.index_number}: Checking if job is reportable ({job_state.verb})')
            display_cfg = self.config['display']
            match job_state.verb:
                case 'unchanged':
                    return display_cfg['unchanged']
                case 'changed':
                    if not display_cfg['empty-diff']:
                        diff = job_state.get_diff(tz=self.tz, differ_defaults=self.config['differ_defaults'])
                        if diff == '':
                            return False
                    return True
                case 'changed,no_report':
                    # set by differ
                    return False
                case e if e.startswith('error'):
                    if job_state.job.suppress_errors is not None:
                        return not job_state.job.suppress_errors
                    return display_cfg['error']
                case 'error,repeated':
                    if job_state.job.suppress_repeated_errors is not None:
                        return job_state.job.suppress_repeated_errors
                    return display_cfg['error']
                case 'unchanged,error_ended':
                    if job_state.job.suppress_errors is not None:
                        return job_state.job.suppress_errors
                    return display_cfg['error']
                case 'new':
                    return display_cfg['new']
                case _:
                    return True

        for job_state in job_states:
            if job_is_reportable(self, job_state):
                yield job_state

    def finish(self, jobs_file: list[Path]) -> None:
        """Finish job run: determine its duration and generate reports by submitting job_states to
        :py:Class:`ReporterBase` :py:func:`submit_all`.

        :param jobs_file: The path to the file containing the list of jobs (optional, used in footers).
        """
        end = time.perf_counter()
        duration = end - self.start

        ReporterBase.submit_all(self, self.job_states, duration, jobs_file)

    def finish_one(self, name: str, jobs_file: list[Path] | None = None, check_enabled: bool | None = True) -> None:
        """Finish job run of one: determine its duration and generate reports by submitting job_states to
        :py:Class:`ReporterBase` :py:func:`submit_one`.  Used in testing.

        :param name: The name of the reporter to run.
        :param jobs_file: The path to the file containing the list of jobs (optional, used in footers).
        :param check_enabled: If True (default), run reports only if they are enabled in the configuration.
        """
        end = time.perf_counter()
        duration = end - self.start
        if jobs_file is None:
            jobs_file = []

        ReporterBase.submit_one(name, self, self.job_states, duration, jobs_file, check_enabled)

1	"""Handles the running of jobs and, afterward, of the reports."""
2
3	# The code below is subject to the license contained in the LICENSE.md file, which is part of the source code.
4
5	from __future__ import annotations	15✔
6
7	import logging	15✔
8	import os	15✔
9	import subprocess	15✔
10	import sys	15✔
11	import time	15✔
12	import traceback	15✔
13	from typing import TYPE_CHECKING, Any, ContextManager, Iterator, Literal, NamedTuple, Self, TypedDict	15✔
14	from zoneinfo import ZoneInfo	15✔
15
16	from webchanges.differs import DifferBase, ReportKind	15✔
17	from webchanges.filters import FilterBase	15✔
18	from webchanges.jobs import NotModifiedError	15✔
19	from webchanges.reporters import ReporterBase	15✔
20
21	# https://stackoverflow.com/questions/39740632
22	if TYPE_CHECKING:
23	from concurrent.futures import Future
24	from pathlib import Path
25	from types import TracebackType
26
27	from webchanges.jobs import JobBase
28	from webchanges.main import Urlwatch
29	from webchanges.storage import SsdbStorage, _Config, _ConfigDifferDefaults
30
31	logger = logging.getLogger(__name__)	15✔
32
33
34	class Snapshot(NamedTuple):	15✔
35	"""Type for Snapshot named tuple.
36
37	* 0: data: str \| bytes
38	* 1: timestamp: float
39	* 2: tries: int
40	* 3: etag: str
41	* 4: mime_type: mime_type
42	* 5: error: ErrorData
43	"""
44
45	data: str \| bytes	15✔
46	timestamp: float	15✔
47	tries: int	15✔
48	etag: str	15✔
49	mime_type: str	15✔
50	error_data: ErrorData	15✔
51
52
53	Verb = Literal[	15✔
54	'new', # new job
55	'changed', # valid data received, and it has changed
56	'changed,no_report', # valid data received, and it has changed, but filtered diff yields no report
57	'unchanged', # valid data received, no changes
58	'unchanged,error_ended', # valid data received, no changes from the last data received before an error
59	'error', # error, prior state was different (either data or different error)
60	'error,repeated', # error, same as before
61	]
62	ErrorData = TypedDict('ErrorData', {'type': str, 'message': str}, total=False)	15✔
63
64
65	class JobState(ContextManager):	15✔
66	"""The JobState class, which contains run information about a job."""
67
68	_http_client_used: Literal['httpx', 'requests', 'curl_cffi', 'playwright'] \| None = None	15✔
69	error_ignored: bool	15✔
70	exception: Exception \| None = None	15✔
71	generated_diff: dict[ReportKind, str]	15✔
72	history_dic_snapshots: dict[str \| bytes, Snapshot]	15✔
73	new_data: str \| bytes = ''	15✔
74	new_error_data: ErrorData = {}	15✔
75	new_etag: str = ''	15✔
76	new_mime_type: str = ''	15✔
77	new_timestamp: float	15✔
78	old_snapshot = Snapshot(	15✔
79	data='',
80	timestamp=1605147837.511478, # initialized to the first release of webchanges!
81	tries=0,
82	etag='',
83	mime_type='text/plain',
84	error_data={},
85	)
86	old_data: str \| bytes = ''	15✔
87	old_error_data: ErrorData = {}	15✔
88	old_etag: str = ''	15✔
89	old_mime_type: str = 'text/plain'	15✔
90	old_timestamp: float = 1605147837.511478 # initialized to the first release of webchanges!	15✔
91	traceback: str	15✔
92	tries: int = 0 # if >1, an error; value is the consecutive number of runs leading to an error	15✔
93	unfiltered_diff: dict[ReportKind, str]	15✔
94	verb: Verb	15✔
95
96	def __init__(self, snapshots_db: SsdbStorage, job: JobBase) -> None:	15✔
97	"""Initializes the class
98
99	:param snapshots_db: The SsdbStorage object with the snapshot database methods.
100	:param job: A JobBase object with the job information.
101	"""
102	self.snapshots_db = snapshots_db	15✔
103	self.job: JobBase = job	15✔
104
105	self.generated_diff = {}	15✔
106	self.unfiltered_diff = {}	15✔
107	self.history_dic_snapshots = {}	15✔
108
109	def __enter__(self) -> Self:	15✔
110	"""Context manager invoked on entry to the body of a with statement to make it possible to factor out standard
111	uses of try/finally statements. Calls the main_thread_enter method of the Job.
112
113	:returns: Class object.
114	"""
115	# Below is legacy code that now does nothing, so it's being skipped
116	# try:
117	# self.job.main_thread_enter()
118	# except Exception as e:
119	# logger.info(f'Job {self.job.index_number}: Exception while creating resources for job', exc_info=True)
120	# self.exception = e
121	# self.traceback = self.job.format_error(e, traceback.format_exc())
122
123	return self	15✔
124
125	def __exit__(	15✔
126	self,
127	exc_type: type[BaseException] \| None,
128	exc_value: BaseException \| None,
129	tb: TracebackType \| None,
130	) -> bool \| None:
131	"""Context manager invoked on exit from the body of a with statement to make it possible to factor out standard
132	uses of try/finally statements. Calls the main_thread_exit() method of the Job.
133
134	:returns: None.
135	"""
136	# Below is legacy code that now does nothing, so it's being skipped
137	# try:
138	# self.job.main_thread_exit()
139	# except Exception:
140	# # We don't want exceptions from releasing resources to override job run results
141	# logger.warning(f'Job {self.index_number}: Exception while releasing resources for job', exc_info=True)
142	# if isinstance(exc_value, subprocess.CalledProcessError):
143	# raise subprocess.SubprocessError(exc_value.stderr)
144	# elif isinstance(exc_value, FileNotFoundError):
145	# raise OSError(exc_value)
146	return None	15✔
147
148	@staticmethod	15✔
149	def debugging_session() -> bool:	15✔
150	"""Checks if the code is currently running within an external debugger (e.g. IDE) and NOT in a testing
151	environment.
152
153	:returns: True if an external debugger is attached and it's not a pytest session, False otherwise.
154	"""
155	return sys.breakpointhook.__module__ != 'sys' and 'PYTEST_CURRENT_TEST' not in os.environ	15✔
156
157	def added_data(self) -> dict[str, bool \| str \| Exception \| float \| None]:	15✔
158	"""Returns a dict with the data added in the processing of the job."""
159	attrs = ('error_ignored', 'exception', 'new_data', 'new_etag', 'new_timestamp')	15✔
160	return {attr: getattr(self, attr) for attr in attrs if hasattr(self, attr)}	15✔
161
162	def load(self) -> None:	15✔
163	"""Loads form the database the last snapshot(s) for the job."""
164	guid = self.job.guid	15✔
165	self.old_snapshot = self.snapshots_db.load(guid)	15✔
166	# TODO: Remove these
167	(	15✔
168	self.old_data,
169	self.old_timestamp,
170	self.tries,
171	self.old_etag,
172	self.old_mime_type,
173	self.old_error_data,
174	) = self.old_snapshot
175	if self.job.compared_versions and self.job.compared_versions > 1:	15✔
176	self.history_dic_snapshots = {	6✔
177	s.data: s for s in self.snapshots_db.get_history_snapshots(guid, self.job.compared_versions)
178	}
179
180	def save(self) -> None:	15✔
181	"""Saves new data retrieved by the job into the snapshot database."""
182	if self.new_error_data: # have encountered an exception, so save the old data	15✔
183	new_snapshot = Snapshot(	15✔
184	data=self.old_data,
185	timestamp=self.new_timestamp,
186	tries=self.tries,
187	etag=self.old_etag,
188	mime_type=self.old_mime_type,
189	error_data=self.new_error_data,
190	)
191	else:
192	new_snapshot = Snapshot(	15✔
193	data=self.new_data,
194	timestamp=self.new_timestamp,
195	tries=self.tries,
196	etag=self.new_etag,
197	mime_type=self.new_mime_type,
198	error_data=self.new_error_data,
199	)
200	self.snapshots_db.save(guid=self.job.guid, snapshot=new_snapshot)	15✔
201	logger.info(f'Job {self.job.index_number}: Saved new data to database')	15✔
202
203	def delete_latest(self, temporary: bool = True) -> None:	15✔
204	"""Removes the last instance in the snapshot database."""
205	self.snapshots_db.delete_latest(guid=self.job.guid, temporary=temporary)	15✔
206
207	def process(self, headless: bool = True) -> JobState:	15✔
208	"""Processes the job: loads it (i.e. runs it) and handles Exceptions (errors).
209
210	:returns: a JobState object containing information of the job run.
211	"""
212	logger.info(f'{self.job.get_indexed_location()} started processing ({type(self.job).__name__})')	15✔
213	logger.debug(f'Job {self.job.index_number}: {self.job}')	15✔
214
215	if self.exception and not isinstance(self.exception, NotModifiedError):	15!
216	self.new_timestamp = time.time()	×
217	self.new_error_data = {'type': type(self.exception).__name__, 'message': str(self.exception)}	×
218	logger.info(f'{self.job.get_indexed_location()} ended processing due to exception: {self.exception}')	×
219	return self	×
220
221	try:	15✔
222	self.load()	15✔
223
224	self.new_timestamp = time.time()	15✔
225	data, self.new_etag, mime_type = self.job.retrieve(self, headless)	15✔
226	logger.info(f'Job {self.job.index_number}: Retrieved {len(data)} bytes of {mime_type} content.')	15✔
227
228	logger.debug(	15✔
229	f'Job {self.job.index_number}: Retrieved data={data!r} \| etag={self.new_etag} \| mime_type={mime_type}'
230	)
231
232	# Apply automatic filters first
233	filtered_data, mime_type = FilterBase.auto_process(self, data, mime_type)	15✔
234
235	# Apply any specified filters
236	for filter_kind, subfilter in FilterBase.normalize_filter_list(self.job.filters, self.job.index_number): # ty:ignore[invalid-argument-type]	15!
UNCOV 237	filtered_data, mime_type = FilterBase.process(filter_kind, subfilter, self, filtered_data, mime_type)	×
238
239	self.new_data = filtered_data	15✔
240	self.new_mime_type = mime_type	15✔
241
242	except NotModifiedError as e:	15✔
243	# HTTP 304 response has been received
244	self.exception = e	×
245	self.error_ignored = False	×
246	except Exception as e:	15✔
247	# Processing error of job failed its chance to handle error
248	# Job has a chance to format and ignore its error
249	if self.debugging_session():	15!
250	logger.warning('Running in a debugging session: raising the exception instead of processing it')	×
251	raise	×
252	self.exception = e	15✔
253	self.error_ignored = self.job.ignore_error(e)	15✔
254	if not self.error_ignored:	15✔
255	self.new_timestamp = time.time()	15✔
256	# Check for specific exception types to provide more detailed tracebacks
257	if self.job.__class__.__module__ == 'hooks':	15!
258	logger.info('Job is from hooks.py: including full traceback in error message')	×
259	self.traceback = ''.join(traceback.format_exception(e)).rstrip()	×
260	elif isinstance(e, subprocess.CalledProcessError):	15✔
261	self.traceback = (	15✔
262	f'subprocess.CalledProcessError: Command returned non-zero exit status {e.returncode}.\n\n'
263	+ '\n'.join(filter(None, (e.stderr, e.stdout)))
264	)
265	else:
266	# Generic traceback for other exceptions
267	self.traceback = self.job.format_error(e, traceback.format_exc())	15✔
268
269	self.tries += 1	15✔
270	self.new_error_data = {	15✔
271	'type': '.'.join(filter(None, [getattr(e, '__module__', None), e.__class__.__name__])),
272	'message': str(e),
273	}
274	logger.info(	15✔
275	f'Job {self.job.index_number}: Job ended with an error; incrementing cumulative error runs to '
276	f'{self.tries}'
277	)
278
279	logger.debug(f'Job {self.job.index_number}: Processed as {self.added_data()}')	15✔
280	logger.info(f'{self.job.get_indexed_location()} ended processing')	15✔
281	return self	15✔
282
283	def get_diff(	15✔
284	self,
285	report_kind: ReportKind = 'plain',
286	differ: dict[str, Any] \| None = None,
287	differ_defaults: _ConfigDifferDefaults \| None = None,
288	tz: ZoneInfo \| None = None,
289	) -> str:
290	"""Generates the job's diff and applies diff_filters to it (if any). Memoized.
291
292	:parameter report_kind: the kind of report that needs the differ.
293	:parameter differ: the name of the differ to override self.job.differ.
294	:parameter tz: The IANA tz_info name of the timezone to use for diff in the job's report (e.g. 'Etc/UTC').
295	:returns: The job's diff.
296	"""
297	if report_kind in self.generated_diff:	15✔
298	return self.generated_diff[report_kind]	15✔
299
300	if report_kind not in self.unfiltered_diff:	15✔
301	differ_kind, subdiffer = DifferBase.normalize_differ(	15✔
302	differ or self.job.differ,
303	self.job.index_number,
304	differ_defaults,
305	)
306	unfiltered_diff = DifferBase.process(differ_kind, subdiffer, self, report_kind, tz, self.unfiltered_diff)	15✔
307	self.unfiltered_diff.update(unfiltered_diff)	15✔
308	_generated_diff = self.unfiltered_diff[report_kind]	15✔
309	if _generated_diff:	15✔
310	# Apply any specified diff_filters
311	_mime_type = 'text/plain'	15✔
312	for filter_kind, subfilter in FilterBase.normalize_filter_list(	15✔
313	self.job.diff_filters, self.job.index_number
314	):
315	_generated_diff, _mime_type = FilterBase.process(	15✔
316	filter_kind, subfilter, self, _generated_diff, _mime_type
317	)
318	self.generated_diff[report_kind] = str(_generated_diff)	15✔
319
320	return self.generated_diff[report_kind]	15✔
321
322	def is_markdown(self) -> bool:	15✔
323	"""Returns whether the new data is in markdown."""
324	return self.new_mime_type == 'text/markdown' or bool(self.job.is_markdown)	15✔
325
326
327	class Report:	15✔
328	"""The base class for reporting."""
329
330	job_states: list[JobState] = []	15✔
331	new_release_future: Future[str \| bool] \| None = None	15✔
332	start: float = time.perf_counter()	15✔
333
334	def __init__(self, urlwatch: Urlwatch) -> None:	15✔
335	""":param urlwatch: The Urlwatch object with the program configuration information."""
336	self.config: _Config = urlwatch.config_storage.config	15✔
337	self.tz = (	15✔
338	ZoneInfo(self.config['report']['tz'])
339	if 'report' in self.config and self.config['report']['tz'] is not None
340	else None
341	)
342
343	def _result(	15✔
344	self,
345	verb: Verb,
346	job_state: JobState,
347	) -> None:
348	"""Logs error and appends the verb to the job_state.
349
350	:param verb: Description of the result of the job run. Can be one of
351	• 'new': new job;
352	• 'changed': valid data received, and it has changed;
353	• 'changed,no_report': valid data received, and it has changed, but no report;
354	• 'unchanged': valid data received, no changes;
355	• 'unchanged,error_ended': valid data received, no changes from the last data received before an error;
356	• 'error': error, prior state was different (either data or different error);
357	• 'error,repeated': error, same as before;
358	or a custom message such as 'test'. Ultimately called by job_runner.
359
360	:param job_state: The JobState object with the information of the job run.
361	"""
362	if job_state.exception is not None:	15✔
363	logger.info(	15✔
364	f'Job {job_state.job.index_number}: Got exception while processing job {job_state.job}',
365	exc_info=job_state.exception,
366	)
367
368	job_state.verb = verb	15✔
369	self.job_states.append(job_state)	15✔
370
371	def new(self, job_state: JobState) -> None:	15✔
372	"""Sets the verb of the job in job_state to 'new'. Called by :py:func:`run_jobs` and tests.
373
374	:param job_state: The JobState object with the information of the job run.
375	"""
376	self._result('new', job_state)	15✔
377
378	def changed(self, job_state: JobState) -> None:	15✔
379	"""Sets the verb of the job in job_state to 'changed'. Called by :py:func:`run_jobs` and tests.
380
381	:param job_state: The JobState object with the information of the job run.
382	"""
383	self._result('changed', job_state)	15✔
384
385	def changed_no_report(self, job_state: JobState) -> None:	15✔
386	"""Sets the verb of the job in job_state to 'changed,no_report'. Called by :py:func:`run_jobs` and tests.
387
388	:param job_state: The JobState object with the information of the job run.
389	"""
390	self._result('changed,no_report', job_state)	15✔
391
392	def unchanged(self, job_state: JobState) -> None:	15✔
393	"""Sets the verb of the job in job_state to 'unchanged'. Called by :py:func:`run_jobs` and tests.
394
395	:param job_state: The JobState object with the information of the job run.
396	"""
397	self._result('unchanged', job_state)	15✔
398
399	def unchanged_from_error(self, job_state: JobState) -> None:	15✔
400	"""Sets the verb of the job in job_state to 'unchanged'. Called by :py:func:`run_jobs` and tests.
401
402	:param job_state: The JobState object with the information of the job run.
403	"""
404	self._result('unchanged,error_ended', job_state)	×
405
406	def error(self, job_state: JobState) -> None:	15✔
407	"""Sets the verb of the job in job_state to 'error'. Called by :py:func:`run_jobs` and tests.
408
409	:param job_state: The JobState object with the information of the job run.
410	"""
411	self._result('error', job_state)	15✔
412
413	def error_same_error(self, job_state: JobState) -> None:	15✔
414	"""Sets the verb of the job in job_state to 'error'. Called by :py:func:`run_jobs` and tests.
415
416	:param job_state: The JobState object with the information of the job run.
417	"""
418	self._result('error,repeated', job_state)	15✔
419
420	def custom(	15✔
421	self,
422	job_state: JobState,
423	label: Verb,
424	) -> None:
425	"""Sets the verb of the job in job_state to a custom label. Called by
426	:py:func:`UrlwatchCommand.check_test_reporter`.
427
428	:param job_state: The JobState object with the information of the job run.
429	:param label: The label to set the information of the job run to.
430	"""
431	self._result(label, job_state)	15✔
432
433	def get_filtered_job_states(self, job_states: list[JobState]) -> Iterator[JobState]:	15✔
434	"""Returns JobStates that have reportable changes per config['display']. Called from :py:Class:`ReporterBase`.
435
436	:param job_states: The list of JobState objects with the information of the job runs.
437	:returns: An iterable of JobState objects that have reportable changes per config['display'].
438	"""
439
440	def job_is_reportable(self: Report, job_state: JobState) -> bool:	15✔
441	"""Identify whether the job should generate a report based on its verb, relevant sub-directives, and the
442	reporters configuration."""
443	logger.debug(f'Job {job_state.job.index_number}: Checking if job is reportable ({job_state.verb})')	15✔
444	display_cfg = self.config['display']	15✔
445	match job_state.verb:	15✔
446	case 'unchanged':	15✔
447	return display_cfg['unchanged']	15✔
448	case 'changed':	15✔
449	if not display_cfg['empty-diff']:	15!
450	diff = job_state.get_diff(tz=self.tz, differ_defaults=self.config['differ_defaults'])	15✔
451	if diff == '':	15!
452	return False	×
453	return True	15✔
454	case 'changed,no_report':	15✔
455	# set by differ
456	return False	15✔
457	case e if e.startswith('error'):	15✔
458	if job_state.job.suppress_errors is not None:	15!
459	return not job_state.job.suppress_errors	×
460	return display_cfg['error']	15✔
461	case 'error,repeated':	15!
462	if job_state.job.suppress_repeated_errors is not None:	×
463	return job_state.job.suppress_repeated_errors	×
464	return display_cfg['error']	×
465	case 'unchanged,error_ended':	15!
466	if job_state.job.suppress_errors is not None:	×
467	return job_state.job.suppress_errors	×
468	return display_cfg['error']	×
469	case 'new':	15✔
470	return display_cfg['new']	15✔
471	case _:	15✔
472	return True	15✔
473
474	for job_state in job_states:	15✔
475	if job_is_reportable(self, job_state):	15✔
476	yield job_state	15✔
477
478	def finish(self, jobs_file: list[Path]) -> None:	15✔
479	"""Finish job run: determine its duration and generate reports by submitting job_states to
480	:py:Class:`ReporterBase` :py:func:`submit_all`.
481
482	:param jobs_file: The path to the file containing the list of jobs (optional, used in footers).
483	"""
484	end = time.perf_counter()	×
485	duration = end - self.start	×
486
487	ReporterBase.submit_all(self, self.job_states, duration, jobs_file)	×
488
489	def finish_one(self, name: str, jobs_file: list[Path] \| None = None, check_enabled: bool \| None = True) -> None:	15✔
490	"""Finish job run of one: determine its duration and generate reports by submitting job_states to
491	:py:Class:`ReporterBase` :py:func:`submit_one`. Used in testing.
492
493	:param name: The name of the reporter to run.
494	:param jobs_file: The path to the file containing the list of jobs (optional, used in footers).
495	:param check_enabled: If True (default), run reports only if they are enabled in the configuration.
496	"""
497	end = time.perf_counter()	15✔
498	duration = end - self.start	15✔
499	if jobs_file is None:	15✔
500	jobs_file = []	15✔
501
502	ReporterBase.submit_one(name, self, self.job_states, duration, jobs_file, check_enabled)	15✔

mborsetti / webchanges / 24977678469

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