# -*- coding: utf-8 -*-
"""Base Crawler
==================
.. module:: pcapkit.vendor.default
:mod:`pcapkit.vendor.default` exports :class:`~pcapkit.vendor.default.Vendor`,
the base class of every vendor crawler.
"""
import abc
import collections
import contextlib
import csv
import functools
import importlib.metadata
import inspect
import os
import re
import sys
import tempfile
import textwrap
import webbrowser
from typing import TYPE_CHECKING
import requests
from pcapkit import __version__
from pcapkit.utilities import config
from pcapkit.utilities.exceptions import VendorNotImplemented, VendorPathNotFound
from pcapkit.utilities.warnings import VendorRequestWarning, warn
if TYPE_CHECKING:
from collections import Counter
from typing import Any, Callable, Optional
from pcapkit.corekit.module import ModuleDescriptor
__all__ = ['Vendor']
#: int: How many times :meth:`Vendor._request` asks for a crawler's page, both
#: directly and then through the proxies of :func:`get_proxies`, before giving
#: up, :envvar:`PCAPKIT_VENDOR_RETRY`.
#:
#: .. versionchanged:: 1.5.0
#:
#: Read with :func:`~pcapkit.utilities.config.read_limit` (:issue:`1726`). A
#: value that is not an integer of at least ``1`` warns with
#: :exc:`~pcapkit.utilities.warnings.ConfigWarning` and falls back to ``5``,
#: where one that was not an integer crashed the import, and ``0`` was read
#: as ``1``.
MAX_RETRY = config.read_limit('VENDOR_RETRY', 5, minimum=1)
#: bool: Whether the crawlers run in CI, where a failed fetch fails the run
#: instead of prompting, :envvar:`PCAPKIT_CI_MODE`.
#:
#: .. versionchanged:: 1.5.0
#:
#: Read with :func:`~pcapkit.utilities.config.read_flag` (:issue:`1726`). A
#: value it does not accept warns with
#: :exc:`~pcapkit.utilities.warnings.ConfigWarning` and falls back to off,
#: where it was taken as off silently.
CI_MODE = config.read_flag('CI_MODE')
#: Distribution name of this package, i.e. the name its metadata is registered
#: under, which is not the import name (:mod:`pcapkit`).
DISTRIBUTION = 'pypcapkit'
#: Project URL used as the contact address in :func:`get_user_agent` when the
#: distribution metadata cannot be read, i.e. when running straight from a
#: source checkout that was never installed. Kept in step with ``repository``
#: under ``[project.urls]`` in :file:`pyproject.toml`.
PROJECT_URL = 'https://github.com/JarryShaw/PyPCAPKit'
#: Import line the generated module needs for a non-empty ``__proto_dest__``.
#: :data:`LINE` renders it only when the crawler declares
#: :attr:`Vendor.PROTO_DEST`, so an enumeration naming no destination keeps the
#: import block it had. The trailing newline lives here rather than in the
#: f-string because an f-string replacement field may not contain a backslash
#: before Python 3.12, and this package still builds on 3.8.
PROTO_DEST_IMPORT = 'from pcapkit.corekit.module import ModuleDescriptor\n'
[docs]
def render_proto_dest(descriptors: 'tuple[ModuleDescriptor[Any], ...]') -> 'str':
"""Render the generated class's ``__proto_dest__`` declaration.
:data:`LINE` calls this with the crawler's :attr:`Vendor.PROTO_DEST`. It is
a function rather than an expression inside the f-string for the reason
:data:`PROTO_DEST_IMPORT` gives: the block it returns contains newlines,
which an f-string replacement field cannot spell before Python 3.12.
Args:
descriptors: The crawler's :attr:`Vendor.PROTO_DEST`.
Returns:
The declaration, indented as a class body and followed by a blank line,
or the empty string when ``descriptors`` is empty -- which is what keeps
the template's output byte-identical for every enumeration that names no
destination.
"""
if not descriptors:
return ''
entries = ''.join(f" ModuleDescriptor({module!r}, {name!r}),\n"
for module, name in descriptors)
return (' #: Protocol class(es) owning the ``__proto__`` dispatch registry keyed by\n'
' #: this enumeration, read by\n'
' #: :func:`~pcapkit.foundation.registry.protocols.protocol.register_protocol_code`\n'
' #: to resolve a bare member of it into its destination(s). The classes are\n'
' #: named as lazy :class:`~pcapkit.corekit.module.ModuleDescriptor`\n'
' #: references rather than imported, because :mod:`pcapkit.const` sits below\n'
' #: :mod:`pcapkit.protocols` and importing one from here would be circular.\n'
' __proto_dest__ = (\n'
f'{entries}'
' )\n'
'\n')
#: Default constant template of enumeration registry from IANA CSV.
#:
#: ``DEST`` is the crawler's :attr:`Vendor.PROTO_DEST` and defaults to empty, so
#: a caller that renders an enumeration keying no ``__proto__`` leaves it out.
#:
#: NOTE: the type comment says ``Callable[..., str]`` rather than spelling the
#: parameters out, because mypy cannot infer a lambda's type against an explicit
#: ``Callable[[...], ...]`` once the lambda carries a default (measured: one
#: extra ``Cannot infer type of lambda`` error). The parameters are documented in
#: ``docs/source/pcapkit/vendor/default.rst`` instead.
LINE = lambda NAME, DOCS, FLAG, ENUM, MISS, MODL, DEST=(): f'''\
# -*- coding: utf-8 -*-
# pylint: disable=line-too-long,consider-using-f-string
"""{(name := DOCS.split(' [', maxsplit=1)[0])}
{'=' * (len(name) + 6)}
.. module:: {MODL.replace('vendor', 'const')}
This module contains the constant enumeration for **{name.replace('``', '')}**,
which is automatically generated from :class:`{MODL}.{NAME}`.
"""
from typing import TYPE_CHECKING
from aenum import IntEnum{', extend_enum' if 'extend_enum(' in MISS else ''}
from pcapkit.corekit.enum import EnumRegistry
{PROTO_DEST_IMPORT if DEST else ''}from pcapkit.utilities.exceptions import EnumValueError
if TYPE_CHECKING:
from typing import Any
__all__ = ['{NAME}']
class {NAME}(EnumRegistry, IntEnum):
"""[{NAME}] {DOCS}"""
{render_proto_dest(DEST)} {ENUM}
@classmethod
def _validate_value(cls, value: 'Any') -> 'None':
"""Reject ``value`` unless it passes the range check :meth:`_missing_` makes.
Overrides :meth:`~pcapkit.corekit.enum.EnumLookup._validate_value`
with that check and the message it raises, so
:meth:`~pcapkit.corekit.enum.EnumRegistry.register` refuses exactly the
values :meth:`~pcapkit.corekit.enum.EnumLookup.get` refuses, in the
same words (:issue:`1816`).
Args:
value: Value about to be looked up or minted.
Raises:
EnumValueError: If ``value`` fails that check. A
:exc:`ValueError`, so an ``except ValueError`` caller is
unaffected.
"""
if not ({FLAG}):
raise EnumValueError('%r is not a valid %s' % (value, cls.__name__))
@classmethod
def _missing_(cls, value: 'int') -> '{NAME}':
"""Lookup function used when value is not found.
Args:
value: Value to get enum item.
"""
if not ({FLAG}):
raise ValueError('%r is not a valid %s' % (value, cls.__name__))
{MISS}
{'' if (test := ''.join(MISS.splitlines()[-1:])).startswith('return') or test[8:].startswith('return') else 'return super()._missing_(value)'}
'''.strip() # type: Callable[..., str]
[docs]
def get_proxies() -> 'dict[str, str]':
"""Get proxy for blocked sites.
The function will read :envvar:`PCAPKIT_HTTP_PROXY` and
:envvar:`PCAPKIT_HTTPS_PROXY`, if any, on every call, for the proxy
settings of |requests|_.
Returns:
Proxy settings for |requests|_.
"""
HTTP_PROXY = os.getenv('PCAPKIT_HTTP_PROXY')
HTTPS_PROXY = os.getenv('PCAPKIT_HTTPS_PROXY')
PROXIES = {} # type: dict[str, str]
if HTTP_PROXY is not None:
PROXIES['http'] = HTTP_PROXY
if HTTPS_PROXY is not None:
PROXIES['https'] = HTTPS_PROXY
return PROXIES
@functools.lru_cache(maxsize=1)
def get_user_agent() -> 'str':
"""Get the ``User-Agent`` header the crawlers identify themselves with.
Many :attr:`~Vendor.LINK` registries are Wikipedia articles, and the
Wikimedia Foundation's User-Agent policy refuses |requests|_' default
``python-requests/<version>`` agent outright -- HTTP 403 with a body reading
*"Please set a user-agent and respect our robot policy"*. What the policy
asks for is an agent that names the tool and gives a contact address, so that
a misbehaving client can be reached instead of simply blocked. It does
**not** ask for a browser agent, and sending one would misrepresent what is
making the request, so this deliberately identifies the crawler as itself.
The string is composed from the package's own metadata -- distribution name,
:data:`pcapkit.__version__` and the ``repository`` project URL -- rather than
written out as a literal, so that it follows the package instead of going
stale. Where the distribution metadata cannot be read, i.e. when running from
a source checkout that was never installed, :data:`DISTRIBUTION` and
:data:`PROJECT_URL` stand in for it.
Returns:
Value for the ``User-Agent`` request header.
See Also:
`Wikimedia Foundation User-Agent policy
<https://foundation.wikimedia.org/wiki/Policy:Wikimedia_Foundation_User-Agent_Policy>`__
"""
name = DISTRIBUTION
url = PROJECT_URL
with contextlib.suppress(importlib.metadata.PackageNotFoundError):
metadata = importlib.metadata.metadata(DISTRIBUTION)
name = metadata.get('Name') or name
for entry in metadata.get_all('Project-URL') or []:
label, _, value = entry.partition(',')
if label.strip().casefold() == 'repository' and value.strip():
url = value.strip()
break
return f'{name}/{__version__} (+{url}) python-requests/{requests.__version__}'
[docs]
def stdin_is_interactive() -> 'bool':
"""Whether there is somebody at a keyboard to be prompted.
:meth:`Vendor._request`'s last resort is to ask an operator to fetch the page
by hand, which needs a terminal: a process whose ``stdin`` is a pipe, a file,
or :file:`/dev/null` has nobody to open a browser for and nobody who could
ever save the file that path then waits for.
``stdin`` can be absent rather than merely redirected, and it can stop being
usable while the process runs, so neither is allowed to propagate out of a
predicate: :data:`sys.stdin` is :obj:`None` under a GUI launcher such as
:program:`pythonw`, and one that has been closed raises :exc:`ValueError` from
:meth:`~io.IOBase.isatty` instead of answering. Every such case counts as
*not* interactive, which is the conservative answer -- it costs a crawler the
manual-intervention path it could not have used anyway.
Returns:
Whether :data:`sys.stdin` is attached to a terminal.
"""
stdin = getattr(sys, 'stdin', None)
if stdin is None:
return False
try:
return bool(stdin.isatty())
except (AttributeError, ValueError, OSError):
return False
[docs]
class Vendor(metaclass=VendorMeta):
"""Default vendor generator.
Inherit this class with :attr:`~Vendor.FLAG` &
:attr:`~Vendor.LINK` attributes, etc., to implement
a new vendor generator.
"""
###############
# Macros
###############
#: Name of constant enumeration.
NAME: 'str'
#: Docstring of constant enumeration.
DOCS: 'str'
#: Value limit checker.
FLAG: 'str' = None # type: ignore[assignment]
#: Link to registry.
LINK: 'str' = None # type: ignore[assignment]
#: Protocol class(es) owning the ``__proto__`` dispatch registry keyed by
#: the generated enumeration, rendered into its class body as
#: ``__proto_dest__`` -- the declaration
#: :func:`~pcapkit.foundation.registry.protocols.protocol.register_protocol_code`
#: reads to resolve a bare member of that enumeration into its
#: destination(s). Asked for in review of :pr:`2392`; no issue holds the
#: ruling.
#:
#: Only the registries that key a ``__proto__`` declare one, and the
#: template renders nothing for a crawler that leaves this empty, so every
#: other generated module is unaffected. A class is named as a lazy
#: :class:`~pcapkit.corekit.module.ModuleDescriptor` rather than imported,
#: because :mod:`pcapkit.const` sits below :mod:`pcapkit.protocols`.
#:
#: .. versionadded:: 1.5.0
PROTO_DEST: 'tuple[ModuleDescriptor[Any], ...]' = ()
###############
# Processors
###############
@staticmethod
def split_references(text: 'str') -> 'list[str]':
"""Split an IANA reference cell into its citations.
IANA's CSV exports have spelt a cell citing two documents in two ways
(:issue:`1664`). The older form is ``[RFC1072][RFC6247]``, or one
citation per line of a quoted cell. The port number registry began
writing ``RFC 3501`` by 2026-09-29, and every registry re-exported on
2026-10-13 (17:26Z to 17:54Z) reads ``[RFC 1072]; [RFC 6247]``. Both give
``['RFC1072', 'RFC6247']`` here, so every crawler renders them
identically:
* a segment of nothing but whitespace and ``;`` separates two
citations rather than being one, and is dropped;
* ``RFC`` followed by whitespace and a number loses the whitespace,
which is the spelling the crawlers render as an :rst:role:`rfc` role.
``RFC Errata 5748`` and draft names such as
``RFC-ietf-tcpm-accurate-ecn-34`` are left as they are.
Any other segment, including text outside the brackets, is returned
verbatim and in order, as :func:`re.split` on the brackets alone gives it.
Args:
text: Raw reference cell.
Returns:
The cell's citations, without their brackets.
"""
refs = [] # type: list[str]
for ref in re.split(r'\[|\]', text):
if re.fullmatch(r'[\s;]*', ref) is None:
refs.append(re.sub(r'\A(\s*)RFC\s+(?=\d)', r'\1RFC', ref))
return refs
[docs]
def safe_name(self, name: 'str') -> 'str':
"""Convert enumeration name to :class:`enum.Enum` friendly.
Args:
name: original enumeration name
Returns:
Converted enumeration name.
"""
temp = '_'.join(
filter(
None,
re.sub(
r'\W',
'_',
'_'.join(
re.sub(
r'\(.*\)',
'',
name
).split(),
),
).split('_')
)
)
if temp.isidentifier():
return temp
return f'{self.NAME}_{temp}'
[docs]
def rename(self, name: 'str', code: 'str', *, original: 'Optional[str]' = None) -> 'str': # pylint: disable=redefined-outer-name
"""Rename duplicated fields.
Args:
name: Field name.
code: Field code.
original: Original field name (extracted from CSV records).
Returns:
Revised field name.
Example:
If ``name`` has multiple occurrences in the source registry,
the field name will be sanitised as ``${name}_${code}``.
Otherwise, the plain ``name`` will be returned.
"""
index = original or name
if self.record[self.safe_name(index)] > 1 or self.safe_name(index).upper() in ['RESERVED', 'UNASSIGNED']:
name = f'{name}_{code}'
return self.safe_name(name)
[docs]
def process(self, data: 'list[str]') -> 'tuple[list[str], list[str]]':
"""Process CSV data.
Args:
data: CSV data.
Returns:
Enumeration fields and missing fields.
"""
reader = csv.reader(data)
next(reader)
enum = [] # type: list[str]
miss = [] # type: list[str]
for item in reader:
name = item[1]
rfcs = item[2]
temp = [] # type: list[str]
for rfc in self.split_references(rfcs):
if 'RFC' in rfc and re.match(r'\d+', rfc[3:]):
#temp.append(f'[{rfc[:3]} {rfc[3:]}]')
temp_split = rfc[3:].split(', ', maxsplit=1)
if len(temp_split) > 1:
temp.append(f'[:rfc:`{temp_split[0]}#{temp_split[1].lower()}`]'.replace(' ', '-'))
else:
temp.append(f'[:rfc:`{temp_split[0]}`]')
else:
temp.append(f'[{rfc}]'.replace('_', ' '))
desc = self.wrap_comment(re.sub(r'\r*\n', ' ', '%s %s' % ( # pylint: disable=consider-using-f-string
name, ''.join(temp) if rfcs else '',
), flags=re.MULTILINE))
try:
code, _ = item[0], int(item[0])
renm = self.rename(name, code)
pres = f'{renm} = {code}'
sufs = f'#: {desc}'
#if len(pres) > 74:
# sufs = f"\n{' '*80}{sufs}"
#enum.append(f'{pres.ljust(76)}{sufs}')
enum.append(f'{sufs}\n {pres}')
except ValueError:
start, stop = item[0].split('-')
miss.append(f'if {start} <= value <= {stop}:')
miss.append(f' #: {desc}')
miss.append(f" return cls._unregistered_member(value, '{self.safe_name(name)}')")
return enum, miss
[docs]
def count(self, data: 'list[str]') -> 'Counter[str]':
"""Count field records.
Args:
data: CSV data.
Returns:
Field recordings.
"""
reader = csv.reader(data)
next(reader) # header
return collections.Counter(map(lambda item: self.safe_name(item[1]),
filter(lambda item: len(item[0].split('-')) != 2, reader)))
[docs]
def context(self, data: 'list[str]') -> 'str':
"""Generate constant context.
Args:
data: CSV data.
Returns:
Constant context.
"""
enum, miss = self.process(data)
ENUM = '\n\n '.join(map(lambda s: s.rstrip(), enum)).strip()
MISS = '\n '.join(map(lambda s: s.rstrip(), miss)).strip()
return LINE(self.NAME, self.DOCS, self.FLAG, ENUM, MISS, self.__module__, self.PROTO_DEST)
[docs]
def request(self, text: 'Optional[str]' = None) -> 'list[str]':
"""Fetch CSV file.
Args:
text: Context from :attr:`~Vendor.LINK`.
Returns:
CSV data.
"""
if text is None:
return []
return text.strip().split('\r\n')
###############
# Defaults
###############
def __new__(cls) -> 'Vendor':
"""Subclassing checkpoint.
Raises:
VendorNotImplemented: If ``cls`` is not a subclass of
:class:`~pcapkit.vendor.default.Vendor`.
"""
if cls is Vendor:
raise VendorNotImplemented('cannot initiate Vendor instance')
return super().__new__(cls)
[docs]
def __init__(self) -> 'None':
"""Generate new constant files."""
#: Name of constant enumeration.
self.NAME = type(self).__name__
#: Docstring of constant enumeration.
self.DOCS = type(self).__doc__ # type: ignore[assignment]
data = self._request()
self.record = self.count(data)
temp_ctx = [] # type: list[str]
orig_ctx = self.context(data)
for line in orig_ctx.splitlines():
if line:
if line.strip():
temp_ctx.append(line.rstrip())
else:
temp_ctx.append(line)
context = '\n'.join(temp_ctx)
const_file = self._dest_path()
os.makedirs(os.path.dirname(const_file), exist_ok=True)
with open(const_file, 'w') as file: # pylint: disable=unspecified-encoding
print(context, file=file)
[docs]
@classmethod
def _dest_path(cls) -> 'str':
"""Resolve the ``const/`` file this crawler's module mirrors.
:attr:`~Vendor.__module__` sits somewhere under the :mod:`pcapkit.vendor`
package, at any depth -- e.g. :mod:`pcapkit.vendor.reg.linktype` one
level down, or :mod:`pcapkit.vendor.reg.apptype.tcp` (:issue:`732`)
two. The output file mirrors that same position under
:mod:`pcapkit.const`.
The position is taken relative to the ``vendor`` package's own
``__file__`` rather than by splitting the module's path a fixed number
of times, so any nesting works: whatever path a module sits at
*relative to* ``vendor/``, the same relative path is mirrored under
``const/``. A fixed two-level split would send
``vendor/reg/apptype/tcp.py`` to ``vendor/const/apptype/tcp.py``,
*inside* ``vendor/`` itself rather than under ``pcapkit/const/``.
:mod:`pcapkit.vendor` is imported here, inside the method, rather than at
module scope: it imports *this* module while it is still initialising
(see ``pcapkit/vendor/__init__.py``), so a module-level import would have
to reason about that partial state. By the time any concrete
:class:`Vendor` subclass is instantiated, :mod:`pcapkit.vendor` has always
finished importing.
That relative-path mirroring is only safe once the module is confirmed
to actually sit *under* ``vendor_root`` -- ``const/`` and ``vendor/`` are
siblings at equal depth, so for a module ``m`` that is *not* under
``vendor_root``, ``os.path.relpath(m, vendor_root)`` starts with enough
``..`` segments that re-joining them under ``const_root`` cancels back
out to ``m`` itself. That is reachable: every crawler ends
``sys.exit(SomeCrawler())``, so running one from a second checkout, or
from a module that never sits under ``vendor/`` at all, resolves
:mod:`pcapkit.vendor` from wherever it is installed while
``inspect.getfile(cls)`` names a file elsewhere entirely -- and
:meth:`Vendor.__init__ <pcapkit.vendor.default.Vendor.__init__>` then opens that path with ``'w'``, silently
truncating whatever :attr:`~Vendor.__module__` actually names instead of
raising. So the escape is rejected outright, rather than trusted to
produce a harmless-looking wrong path.
This is a classmethod, not an instance method, deliberately: the body
needs no instance state, only the crawler's class, which ``cls`` is
with no instance required to obtain it. That lets a caller resolve a
crawler's destination *without* instantiating it, and so without triggering
:meth:`Vendor.__init__`'s network fetch, render and write --
the pre-run snapshot that ``run()`` in :mod:`pcapkit.vendor.__main__`
takes (:issue:`872`) is such a caller. No crawler overrides
``_dest_path``, so the classmethod is authoritative for all of them.
Returns:
Absolute path of the constant module this crawler should write.
Raises:
VendorPathNotFound: If this crawler's own module is not located
under the :mod:`pcapkit.vendor` package root, so mirroring it
under :mod:`pcapkit.const` cannot be done safely.
"""
import pcapkit.vendor # pylint: disable=import-outside-toplevel
vendor_root = os.path.dirname(os.path.abspath(pcapkit.vendor.__file__))
module_file = os.path.abspath(inspect.getfile(cls))
rel_path = os.path.relpath(module_file, vendor_root)
if rel_path.split(os.sep, 1)[0] == os.pardir:
raise VendorPathNotFound(
f'{module_file!r} is not inside the vendor package root {vendor_root!r}; '
f'refusing to derive a const/ path for it')
const_root = os.path.join(os.path.dirname(vendor_root), 'const')
return os.path.join(const_root, rel_path)
[docs]
def _request(self) -> 'list[str]':
"""Fetch CSV data from :attr:`~Vendor.LINK`.
This is the low-level call of :meth:`~Vendor.request`.
If :attr:`~Vendor.LINK` is ``None``, it will directly
call the upper method :meth:`~Vendor.request` with **NO**
arguments.
The method will first try to *GET* the content of :attr:`~Vendor.LINK`.
Should any exception be raised, it will first try with proxy settings from
:func:`~pcapkit.vendor.default.get_proxies`.
.. note::
Since some :attr:`~Vendor.LINK` links are from Wikipedia, etc., they
might not be available in certain areas, e.g. the amazing PRC :)
Should the proxies fail again, it will prompt for user intervention, i.e.
it will use :func:`webbrowser.open` to open the page in browser for you, and
you can manually load that page and save the HTML source at the location
it provides.
That last resort is only taken when somebody can actually act on it, i.e.
when :envvar:`PCAPKIT_CI_MODE` is unset *and*
:func:`stdin_is_interactive` finds a terminal. A non-interactive run --
under a pipe, in a container, from a scheduled job -- is failed with the
fetch error instead, the same way :envvar:`PCAPKIT_CI_MODE` fails it,
rather than printing instructions nobody will read. See :issue:`522`.
Returns:
CSV data.
Raises:
requests.RequestException: If the registry could not be fetched and
manual intervention is unavailable, i.e. under
:envvar:`PCAPKIT_CI_MODE`, with no interactive ``stdin``, or where
the prompt itself fails because ``stdin`` went away mid-wait.
Warns:
VendorRequestWarning: If connection failed with and/or without proxies.
See Also:
:meth:`~Vendor.request`
"""
if self.LINK is None:
return self.request() # type: ignore[unreachable]
# NOTE: both branches send this, as Wikimedia rejects ``requests``' default
# agent with HTTP 403 that no retry lifts (#518).
headers = {'User-Agent': get_user_agent()}
try:
counter = 1
while True:
if counter > MAX_RETRY:
raise requests.exceptions.RequestException
page = requests.get(self.LINK, headers=headers) # nosec: B113
if not page.ok or not page.text:
warn(f'Connection failed; retry for {counter}/{MAX_RETRY}...',
VendorRequestWarning, stacklevel=2)
counter += 1
continue
break
except requests.RequestException:
warn('Connection failed; retry with proxies (if any)...',
VendorRequestWarning, stacklevel=2)
try:
proxies = get_proxies() or None
if proxies is None:
raise
counter = 1
while True:
if counter > MAX_RETRY:
raise
page = requests.get(self.LINK, headers=headers, proxies=proxies) # nosec: B113
if not page.ok or not page.text:
warn(f'Connection failed; retry with proxy for {counter}/{MAX_RETRY}...',
VendorRequestWarning, stacklevel=2)
counter += 1
continue
break
except requests.RequestException as error:
if CI_MODE:
warn('Connection failed; exit on CI mode...',
VendorRequestWarning, stacklevel=2)
raise
if not stdin_is_interactive():
# NOTE: with no terminal, nobody can save the page, so fail as
# CI_MODE does rather than wait (#522).
warn('Connection failed; exit as stdin is not interactive...',
VendorRequestWarning, stacklevel=2)
raise
warn('Connection failed; retry with manual intervene...',
VendorRequestWarning, stacklevel=2)
with tempfile.TemporaryDirectory(suffix='-tempdir',
prefix='pcapkit-',
dir=os.path.abspath(os.curdir)) as tempdir:
temp_file = os.path.join(tempdir, f'{self.NAME}.html')
flag = False
with contextlib.suppress(Exception):
flag = webbrowser.open(self.LINK)
if flag:
print('Please save the page source at')
print(f' {temp_file}')
else:
print('Please navigate to the following address')
print(f' {self.LINK}')
print('and save the page source at')
print(f' {temp_file}')
while True:
try:
input('Press ENTER to continue...') # nosec
except Exception as exc:
# NOTE: give up once ``input()`` fails on a closed, dropped or
# vanished stdin, or this spins at full speed (2.6 GB in #522).
warn(f'Connection failed; cannot prompt for manual intervene ({exc})...',
VendorRequestWarning, stacklevel=2)
raise error from exc
if os.path.isfile(temp_file):
break
print('File not found; please save the page source at')
print(f' {temp_file}')
with open(temp_file) as file: # pylint: disable=unspecified-encoding
text = file.read()
else:
text = page.text
else:
text = page.text
return self.request(text)