# -*- coding: utf-8 -*-
"""Base Crawler
==================
.. module:: pcapkit.vendor.default
:mod:`pcapkit.vendor.default` exports :class:`~pcapkit.vendor.default.Vendor`
only, which is the base class for all vendor crawlers.
"""
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 Callable, Optional
__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'
#: Default constant template of enumeration registry from IANA CSV.
LINE = lambda NAME, DOCS, FLAG, ENUM, MISS, MODL: 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
from pcapkit.utilities.exceptions import EnumValueError
if TYPE_CHECKING:
from typing import Any
__all__ = ['{NAME}']
class {NAME}(EnumRegistry, IntEnum):
"""[{NAME}] {DOCS}"""
{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, str, str, str, str, str], 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|_.
.. |requests| replace:: ``requests``
.. _requests: https://requests.readthedocs.io
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__}'
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]
###############
# 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 did.
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__)
[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)
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/``.
``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, ``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
``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: nothing
in the body ever needed instance state -- it only ever consulted
``type(self)``, so ``cls`` is exactly that, 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 --
:func:`pcapkit.vendor.__main__.run`'s pre-run snapshot (GitHub issue
:issue:`872`) is exactly such a caller. Every crawler defined in this
codebase inherits this implementation unchanged; none 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 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 :)
Would proxies failed 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 below send this. Wikimedia rejects ``requests``'
# default agent with HTTP 403, so a crawler without it fetches 126 bytes
# of robot-policy text and retries MAX_RETRY times against a refusal
# that no amount of retrying will lift; see #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: manual intervention needs an operator, and a process
# with no terminal has none -- nobody to read the instructions
# below, nobody to save the page, so the wait it ends in could
# only ever time out the whole run. Fail exactly as CI_MODE
# does rather than prompting into the void; see #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: whatever stopped the prompt, ``input()`` is no
# longer waiting -- and this loop only waits because
# ``input()`` does, so the file it waits for can never
# arrive. Discarding the exception, as this used to,
# left a ``while True`` that printed three lines per
# iteration at full speed: 26.7 million lines and 2.6 GB
# measured in #522. The terminal this started on can
# still go away mid-wait -- a closed ``stdin`` raises
# :exc:`ValueError`, a dropped one :exc:`EOFError`, a
# vanished :data:`sys.stdin` :exc:`RuntimeError` -- so the
# check before the prompt does not make this redundant.
# ``KeyboardInterrupt`` is not an :exc:`Exception` and so
# still aborts as itself.
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)