24920572770

Committed 25 Apr 2026 02:35AM UTC coverage: 72.339% (-0.6%) from 72.893%

Build # 24920572770

Build Type

push

github

Committed by

mborsetti

Commit Message

Version 3.35.0rc0

Coverage Stats

1454 of 2420 branches covered (60.08%)

Branch coverage included in aggregate %.

129 of 209 new or added lines in 7 files covered. (61.72%)

62 existing lines in 1 file now uncovered.

5126 of 6676 relevant lines covered (76.78%)

11.01 hits per line

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

79.66

/webchanges/filters/_shell.py

"""Shell execution filters."""

# 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 json
import logging
import os
import re
import shlex
import subprocess
import sys
from typing import Any

from webchanges import __project_name__
from webchanges.filters._base import FilterBase

logger = logging.getLogger(__name__)


def _pipe_filter(f_cls: FilterBase, data: str | bytes, subfilter: dict[str, Any]) -> str:
    if 'command' not in subfilter:
        raise ValueError(f"The '{f_cls.__kind__}' filter needs a command. ({f_cls.job.get_indexed_location()})")

    # Work on a copy of the environment as not to modify the outside environment
    env = os.environ.copy()
    env.update(
        {
            f'{__project_name__.upper()}_JOB_JSON': json.dumps(f_cls.job.to_dict()),
            f'{__project_name__.upper()}_JOB_NAME': f_cls.job.pretty_name(),
            f'{__project_name__.upper()}_JOB_LOCATION': f_cls.job.get_location(),
            f'{__project_name__.upper()}_JOB_INDEX_NUMBER': str(f_cls.job.index_number),
            'URLWATCH_JOB_NAME': f_cls.job.pretty_name(),  # urlwatch 2 compatibility
            'URLWATCH_JOB_LOCATION': f_cls.job.get_location(),  # urlwatch 2 compatibility
        }
    )

    if subfilter.get('escape_characters') and sys.platform == 'win32':
        escaped_command = re.sub(r'([()!^"<>&|])', r'^\1', subfilter['command']).replace('%', '%%')
        # escaped_command = _windows_escape_cmd(subfilter['command'])
    else:
        escaped_command = subfilter['command']

    if f_cls.__kind__ == 'execute':
        command = shlex.split(escaped_command)
        shell = False
    else:  # 'shellpipe'
        command = escaped_command
        shell = True

    try:
        return subprocess.run(  # noqa: S603 Check for untrusted input
            command,
            input=data,
            capture_output=True,
            shell=shell,
            check=True,
            text=True,
            env=env,
        ).stdout
    except subprocess.CalledProcessError as e:
        logger.error(
            f"The '{f_cls.__kind__}' filter returned error code {e.returncode} ({f_cls.job.get_indexed_location()}):\n"
            f'{e.stderr}\n---\n{e.stdout}'
        )
        raise e
    except FileNotFoundError as e:
        logger.error(f"The '{f_cls.__kind__}' filter returned error ({f_cls.job.get_indexed_location()}):\n{e}")
        raise FileNotFoundError(e, f'with command {command}') from None


class ExecuteFilter(FilterBase):
    """Filter using a command."""

    __kind__ = 'execute'

    __supported_subfilters__: dict[str, str] = {
        'command': 'Command to execute for filtering (required)',
        'escape_characters': 'Whether to escape characters when running in Windows',
    }

    __default_subfilter__ = 'command'

    def filter(self, data: str | bytes, mime_type: str, subfilter: dict[str, Any]) -> tuple[str | bytes, str]:
        if not mime_type.startswith('text'):
            mime_type = 'text/plain'
        return _pipe_filter(self, data, subfilter), mime_type


class ShellPipeFilter(FilterBase):
    """Filter using a shell command."""

    __kind__ = 'shellpipe'

    __supported_subfilters__: dict[str, str] = {
        'command': 'Shell command to execute for filtering (required)',
        'escape_characters': 'Whether to escape characters when running in Windows',
    }

    __default_subfilter__ = 'command'

    def filter(self, data: str | bytes, mime_type: str, subfilter: dict[str, Any]) -> tuple[str | bytes, str]:
        if not mime_type.startswith('text'):
            mime_type = 'text/plain'
        return _pipe_filter(self, data, subfilter), mime_type

1	"""Shell execution filters."""
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 json	15✔
8	import logging	15✔
9	import os	15✔
10	import re	15✔
11	import shlex	15✔
12	import subprocess	15✔
13	import sys	15✔
14	from typing import Any	15✔
15
16	from webchanges import __project_name__	15✔
17	from webchanges.filters._base import FilterBase	15✔
18
19	logger = logging.getLogger(__name__)	15✔
20
21
22	def _pipe_filter(f_cls: FilterBase, data: str \| bytes, subfilter: dict[str, Any]) -> str:	15✔
23	if 'command' not in subfilter:	15✔
24	raise ValueError(f"The '{f_cls.__kind__}' filter needs a command. ({f_cls.job.get_indexed_location()})")	15✔
25
26	# Work on a copy of the environment as not to modify the outside environment
27	env = os.environ.copy()	15✔
28	env.update(	15✔
29	{
30	f'{__project_name__.upper()}_JOB_JSON': json.dumps(f_cls.job.to_dict()),
31	f'{__project_name__.upper()}_JOB_NAME': f_cls.job.pretty_name(),
32	f'{__project_name__.upper()}_JOB_LOCATION': f_cls.job.get_location(),
33	f'{__project_name__.upper()}_JOB_INDEX_NUMBER': str(f_cls.job.index_number),
34	'URLWATCH_JOB_NAME': f_cls.job.pretty_name(), # urlwatch 2 compatibility
35	'URLWATCH_JOB_LOCATION': f_cls.job.get_location(), # urlwatch 2 compatibility
36	}
37	)
38
39	if subfilter.get('escape_characters') and sys.platform == 'win32':	15!
40	escaped_command = re.sub(r'([()!^"<>&\|])', r'^\1', subfilter['command']).replace('%', '%%')	×
41	# escaped_command = _windows_escape_cmd(subfilter['command'])
42	else:
43	escaped_command = subfilter['command']	15✔
44
45	if f_cls.__kind__ == 'execute':	15✔
46	command = shlex.split(escaped_command)	15✔
47	shell = False	15✔
48	else: # 'shellpipe'
49	command = escaped_command	15✔
50	shell = True	15✔
51
52	try:	15✔
53	return subprocess.run( # noqa: S603 Check for untrusted input	15✔
54	command,
55	input=data,
56	capture_output=True,
57	shell=shell,
58	check=True,
59	text=True,
60	env=env,
61	).stdout
62	except subprocess.CalledProcessError as e:	×
NEW 63	logger.error(	×
64	f"The '{f_cls.__kind__}' filter returned error code {e.returncode} ({f_cls.job.get_indexed_location()}):\n"
65	f'{e.stderr}\n---\n{e.stdout}'
66	)
67	raise e	×
68	except FileNotFoundError as e:	×
69	logger.error(f"The '{f_cls.__kind__}' filter returned error ({f_cls.job.get_indexed_location()}):\n{e}")	×
70	raise FileNotFoundError(e, f'with command {command}') from None	×
71
72
73	class ExecuteFilter(FilterBase):	15✔
74	"""Filter using a command."""
75
76	__kind__ = 'execute'	15✔
77
78	__supported_subfilters__: dict[str, str] = {	15✔
79	'command': 'Command to execute for filtering (required)',
80	'escape_characters': 'Whether to escape characters when running in Windows',
81	}
82
83	__default_subfilter__ = 'command'	15✔
84
85	def filter(self, data: str \| bytes, mime_type: str, subfilter: dict[str, Any]) -> tuple[str \| bytes, str]:	15✔
86	if not mime_type.startswith('text'):	15!
87	mime_type = 'text/plain'	×
88	return _pipe_filter(self, data, subfilter), mime_type	15✔
89
90
91	class ShellPipeFilter(FilterBase):	15✔
92	"""Filter using a shell command."""
93
94	__kind__ = 'shellpipe'	15✔
95
96	__supported_subfilters__: dict[str, str] = {	15✔
97	'command': 'Shell command to execute for filtering (required)',
98	'escape_characters': 'Whether to escape characters when running in Windows',
99	}
100
101	__default_subfilter__ = 'command'	15✔
102
103	def filter(self, data: str \| bytes, mime_type: str, subfilter: dict[str, Any]) -> tuple[str \| bytes, str]:	15✔
104	if not mime_type.startswith('text'):	15!
105	mime_type = 'text/plain'	×
106	return _pipe_filter(self, data, subfilter), mime_type	15✔

mborsetti / webchanges / 24920572770

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