shopdb-flask/scripts/gen_openapi.py

"""Generate docs/openapi.json from docs/api-inventory.json, and refresh that
inventory from the routes the app actually serves.

The inventory is a list of {surface, endpoints:[{method,path,auth,params,purpose,
example}]} objects (one per API surface). Two halves, maintained differently:

  STRUCTURE (surface, method, path, auth) is DERIVED. It is walked out of the
  live url_map and the decorators on each view, so it cannot drift from the
  code. It had: this branch carried no inventory at all, so the generator had no
  input, the parity test skipped, and 24 routes served traffic documented
  nowhere.

  PROSE (purpose, params, example) is HUMAN. What an endpoint is FOR cannot be
  derived, so a refresh never overwrites text that is already there - it fills
  in a new route from its docstring and leaves every existing entry's words
  alone.

So: add a route, run --refresh, and replace the seeded prose with the reason the
route exists. Coverage is enforced on top of that by
tests/test_api_inventory_parity.py, which fails when a served route has no entry
AND when a committed auth string disagrees with what this file re-derives.
Served interactively at /api/docs (see shopdb/core/api/docs.py).

    venv/bin/python scripts/gen_openapi.py             # spec from the inventory
    venv/bin/python scripts/gen_openapi.py --refresh   # inventory first, then spec

THE AUTH STRING IS A GRAMMAR, not free text. Its first word names the mechanism
and everything after it is prose; '-optional' on that word means the gate can be
satisfied without a credential. See AUTH_SCHEME and _security. Derived strings
come from DECORATOR_AUTH, HANDLER_AUTH and INLINE_AUTH below, so a route whose
gate this cannot see says so rather than guessing.

QUERY PARAMETERS ARE DERIVED TOO, and on the same terms as auth: only what the
code says. The sweep reads each view's AST for request.args reads and for the
two helpers this repository reads arguments through, and emits ONLY the ones
whose name resolves to a literal. A key built at runtime produces no parameter
rather than a guessed one - see the section above build(). They land in the
inventory as `queryparams`, which is DERIVED like auth, not prose: hand-editing
it does not survive the next --refresh.

EXAMPLES ARE HARVESTED, NOT WRITTEN. --harvest runs the test suite with this
file loaded as a pytest plugin, records what the test client actually sent and
actually got back, redacts it and folds it into the inventory. Nothing is
composed: an invented example body is a lie in a contract that a generated
client sends verbatim. Harvesting is optional and re-runnable - a checkout with
no tests, and a checkout that has never harvested, both generate a spec.

    venv/bin/python scripts/gen_openapi.py --harvest            # whole suite
    venv/bin/python scripts/gen_openapi.py --harvest tests/test_core
"""

import json
import os
import re
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
REPO = os.path.dirname(HERE)
INVENTORY = os.path.join(REPO, 'docs', 'api-inventory.json')
OUT = os.path.join(REPO, 'docs', 'openapi.json')

VERBS = ('get', 'post', 'put', 'patch', 'delete', 'head', 'options')


def _servers():
    """Servers block for the spec: the relative mount, plus this site's own URL
    if one was supplied.

    SHOPDB_PUBLIC_URL is read from the environment rather than stored, because
    the generated spec is published to a public wiki - a site URL baked into the
    generator ends up in everyone's documentation, including sites it is wrong
    for.
    """
    servers = []
    siteurl = (os.environ.get('SHOPDB_PUBLIC_URL') or '').strip().rstrip('/')
    if siteurl:
        servers.append({'url': siteurl, 'description': 'this site'})
    servers.append({'url': '/', 'description': 'relative to the deployed mount'})
    return servers


def _product_version():
    """Read __version__ out of shopdb/__init__.py without importing the app.

    Importing shopdb here would pull in Flask, the plugin loader and a database
    configuration, none of which this generator needs.
    """
    here = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    init = os.path.join(here, 'shopdb', '__init__.py')
    with open(init, 'r', encoding='utf-8') as handle:
        match = re.search(r"^__version__ = '([^']+)'", handle.read(), re.M)
    if not match:
        raise SystemExit('could not read __version__ from ' + init)
    return match.group(1)


def _norm_path(p):
    # Flask <int:id> / <id> -> OpenAPI {id}
    normalised = re.sub(r'<(?:[^:>]+:)?([^>]+)>', r'{\1}', p or '')
    # A trailing slash in a published path is a distinct resource to a strict
    # proxy (IIS), which answers the unslashed URL with a redirect a client
    # behind a subpath mount cannot follow. Emit the canonical, slashless form.
    if len(normalised) > 1:
        normalised = normalised.rstrip('/')
    return normalised


# The machine-readable half of an auth string is its FIRST word. Everything
# after it is prose for a human reader. An optional gate is spelled by suffixing
# that word with '-optional'.
#
# Substring matching is what this replaces, and it is why nine operations
# published `security: []` - "no credentials needed" - while their handler
# required one. 'none (single-use bootstrap token...)' starts with 'none', so
# the old sweep published the two bootstrap routes as open, and one of them
# hands out the installable agent package and spends a use. A leading token
# cannot be fooled by the prose that explains it.
AUTH_SCHEME = {
    'jwt': 'bearerAuth',
    'permission': 'bearerAuth',
    'api-key': 'apiKeyAuth',
    'setup-code': 'setupCodeAuth',
}

OPTIONAL_SUFFIX = '-optional'


def _auth_head(auth):
    """The mechanism word at the front of an auth string, lowercased."""
    match = re.match(r'[a-z-]+', (auth or '').strip().lower())
    return match.group(0) if match else ''


def _security(auth):
    """OpenAPI security for one endpoint's documented auth string.

    Optional auth is the interesting case. It means "works without a
    credential, does more with one", which OpenAPI expresses as a list carrying
    BOTH the empty requirement and the scheme. Two shapes reach it:

      jwt-optional       works signed out; the handler decides what shows.
      api-key-optional   token OR network. A kiosk admits either the per-device
                         token or a source address inside the site's CIDR
                         allowlist, so an allowlisted caller sends no
                         credential. Publishing that as `[]` is not a lie, but
                         it is not the truth either: it tells a reader there is
                         no token to send, and the allowlist is invisible.
                         Optional says both halves exist.

    An unrecognised word publishes as bearer-required, which is the safe
    direction to be wrong in: a reader who brings a token they did not need
    loses nothing, a reader told none was needed is locked out on the day.
    """
    head = _auth_head(auth)
    optional = head.endswith(OPTIONAL_SUFFIX)
    if optional:
        head = head[:-len(OPTIONAL_SUFFIX)]
    if not head or head in ('none', 'public'):
        return []
    scheme = AUTH_SCHEME.get(head, 'bearerAuth')
    if optional:
        return [{scheme: []}, {}]
    return [{scheme: []}]


def _path_parameters(rawpath):
    """Path parameters, typed from the Flask converter that declared them.

    The generator emitted no `parameters` at all, which makes the document
    invalid OpenAPI 3.1 - and breaks its only machine consumer outright, because
    a tool built from an operation with no parameters has nowhere to put the id
    the caller supplied, so it is dropped in silence.
    """
    parameters = []
    for converter, name in re.findall(r'<(?:([^:>]+):)?([^>]+)>', rawpath or ''):
        schema = {'type': 'integer'} if converter in ('int',) else {'type': 'string'}
        if converter == 'float':
            schema = {'type': 'number'}
        if converter == 'path':
            schema = {'type': 'string', 'description': 'may contain slashes'}
        parameters.append({
            'name': name,
            'in': 'path',
            'required': True,
            'schema': schema,
        })
    return parameters


def _request_body(verb, entry):
    """A JSON body for the verbs that take one.

    The inventory describes bodies in prose, not as schemas, so this does not
    invent field names it cannot verify - it declares an object and carries the
    prose. That is honest, and it is the difference between a machine client
    that can send a body and one that cannot send anything at all.
    """
    harvested = entry.get('requestexample')
    if verb not in ('post', 'put', 'patch'):
        # A DELETE that reads a body. Two here do, and one of them refuses
        # without {"confirm": "OK"} - so the spec described a call that cannot
        # succeed, and a client built from it fails on the day. The evidence is
        # a transcript of a request the suite really sent; with no transcript
        # this stays silent rather than offering every DELETE a body it ignores.
        # A transcript proves the CLIENT sent a body, not that the ROUTE reads
        # one. The suite posts {} to routes that ignore it, and publishing that
        # as "this route reads a JSON body" is a contract claim the code does
        # not support: DELETE /api/endpoints/agentkey never calls get_json.
        # So the handler has to show a read before the body is declared.
        if harvested is None or not entry.get('readsbody'):
            return None
        if harvested == {} or harvested is None:
            # An empty body teaches nothing and reads as "send {}", which for
            # a route with required fields is wrong. Declare the body, omit the
            # example.
            return {
                'required': False,
                'content': {'application/json': {
                    'schema': {'type': 'object', 'additionalProperties': True,
                               'description': 'This route reads a JSON body.'},
                }},
            }
        return {
            'required': False,
            'content': {'application/json': {
                'schema': {'type': 'object', 'additionalProperties': True,
                           'description': 'This route reads a JSON body. The '
                                          'example is a request that worked.'},
                'example': harvested,
            }},
        }
    described = entry.get('params') or ''
    media = {
        'schema': {
            'type': 'object',
            'additionalProperties': True,
            'description': described or 'See the endpoint description.',
        }
    }
    # The one thing the schema cannot say: which fields. It is an open object
    # because the inventory describes bodies in prose, so a harvested example is
    # the only field list a reader gets - and it is a transcript of a body the
    # suite actually sent, not a guess at one.
    if harvested is not None:
        media['example'] = harvested
    return {
        'required': verb in ('post', 'put'),
        'content': {'application/json': media},
    }


def _is_envelope(example):
    """Whether a harvested response is the success_response envelope.

    The envelope is `{status: 'success', data, meta}`. Anything else - a bare
    list, a plain object, a manifest - is a route that answers in its own shape.
    """
    return isinstance(example, dict) and example.get('status') == 'success'


def _responses(security, parameters, example=None):
    """The envelope, plus the failures a caller has to handle.

    One hardcoded 200 was the whole response section, so a generated client had
    no idea any call could fail, and 19 operations that do not return the
    envelope at all were documented as if they did.

    The 200 media type carried a bare `$ref` where OpenAPI expects a Media Type
    Object, i.e. `{"schema": {"$ref": ...}}`. Redoc tolerated it; a strict
    parser does not, and there is nowhere to hang an example on the wrong shape.
    """
    envelope = 'Success. Body is the success_response envelope: {status, data, meta}.'
    success = {'schema': {'$ref': '#/components/schemas/SuccessEnvelope'}}
    description = envelope
    if example is not None:
        success['example'] = example
        # A harvested example is EVIDENCE. Nineteen routes here do not return
        # the envelope, and every one of them was documented as if it did - a
        # $ref that a generated client validates against and a reader writes
        # code to. When the transcript disagrees with the envelope, the
        # transcript is what the route actually returned, so the claim comes
        # off rather than the example.
        if not _is_envelope(example):
            success = {'example': example}
            description = ('Success. This route does NOT return the standard '
                           'envelope; the example is a transcript of a real '
                           'response.')
    responses = {
        '200': {
            'description': description,
            'content': {'application/json': success},
        },
        'default': {
            'description': 'Error. Body is the error envelope; the code and '
                           'message are nested under data.error.',
            'content': {'application/json': {
                '$ref': '#/components/schemas/ErrorEnvelope'}},
        },
    }
    # An endpoint that can be called without a token cannot answer 401.
    if security and security != [{}]:
        requires = not any(item == {} for item in security)
        if requires:
            responses['401'] = {'description': 'Missing or invalid credentials.'}
            responses['403'] = {'description': 'Authenticated, but not permitted.'}
    if parameters:
        responses['404'] = {'description': 'No such record.'}
    return responses


def _summary(text, fallback):
    """A summary that ends on a word.

    Cutting at a fixed 120 characters truncated 95 of them mid-word, which is
    what a tool picker shows a user as the whole description of the call.
    """
    source = (text or fallback or '').strip()
    if len(source) <= 120:
        return source
    head = source[:120]
    cut = head.rfind(' ')
    # Trailing connectives read worse than a clean cut: "servicelevel +..."
    # promises a continuation the reader will never see.
    return (head[:cut] if cut > 40 else head).rstrip(' ,;:.+-/&|(') + '...'


# ---------------------------------------------------------------------------
# Derived query parameters
# ---------------------------------------------------------------------------
#
# A Flask rule carries its path arguments and nothing else, so a route whose
# real interface is '?code=' or '?search=' published an operation with no query
# parameters at all - 0 of 573. A machine client built from that document could
# call those endpoints but could not filter, search or page them, and
# /api/usb/kiosk/setup, where '?code=' is the ONLY way to call it successfully,
# said nothing.
#
# So the handler is read instead. This walks the view's AST for the reads that
# name their argument as a literal - request.args.get('x'), request.args['x'],
# request.args.getlist('x') - and for the two helpers this repository reads
# arguments through, shopdb.utils.pagination.get_pagination_params and
# shopdb.utils.queryargs.intarg/intargs.
#
# THE RULE IS: A NAME THAT DOES NOT RESOLVE TO A LITERAL IS NOT EMITTED. A key
# built at runtime, or read through a variable this cannot see, produces no
# parameter rather than a guessed one. The published spec is a contract, and a
# parameter that is not really there is worse than a missing one: a reader
# believes it, sends it, and is silently ignored. Silence is recoverable.
#
# Required is derived the same way. request.args['x'] raises without the
# argument, so it is required; a .get() call cannot be - it has a fallback, even
# when that fallback is None. Nothing here promotes a parameter to required
# because it looks important.

# The argument-reading helpers, by the tail of the call. Each maps to the
# parameters that helper reads, which is a property of the helper and not of the
# route - so it is written once here rather than re-derived per view.
#
# get_pagination_params reads all three names and int()s them, which is why they
# publish as integers while a bare request.args.get() publishes as a string: the
# query string is text, and only a handler that converts makes it anything else.
PAGINATION_PARAMS = (
    {'name': 'page', 'type': 'integer', 'required': False, 'default': 1},
    {'name': 'perpage', 'type': 'integer', 'required': False},
    # The legacy spelling, still read as a fallback by the same helper. A caller
    # reading only the spec would never find it, and a caller migrating from it
    # would not know it still works.
    {'name': 'per_page', 'type': 'integer', 'required': False},
)


def _constant_string(node, constants):
    """The literal string a call argument names, or None.

    Two forms resolve: a string literal, and a module-level Name bound to one
    (plugins/endpoints/api/bootstrap.py reads its query token through
    QUERY_TOKEN_ARG). Anything else - an f-string, a variable from an argument,
    a key built at runtime - returns None and is dropped, which is the whole
    honesty rule of this section.
    """
    import ast
    if isinstance(node, ast.Constant) and isinstance(node.value, str):
        return node.value
    if isinstance(node, ast.Name):
        return constants.get(node.id)
    return None


def _wire_type(call):
    """The OpenAPI type for a request.args read, from its `type=` converter.

    A query string is text. Werkzeug returns a string unless the caller passed
    type=int (or float, or bool), so that keyword is the only evidence here that
    the value is anything else, and without it the parameter publishes as a
    string.
    """
    import ast
    for keyword in call.keywords:
        if keyword.arg != 'type':
            continue
        name = getattr(keyword.value, 'id', '')
        return {'int': 'integer', 'float': 'number', 'bool': 'boolean'}.get(name, 'string')
    return 'string'


def _literal_default(node, wiretype):
    """A default worth publishing, or None.

    Only literals. `request.args.get('sort', DEFAULT_SORT)` has a default this
    cannot see, and publishing 'DEFAULT_SORT' as the value would be a lie in the
    contract; publishing no default is merely incomplete.

    A numeric default under a string type is stringified rather than dropped,
    because that is what the caller actually puts on the wire: `?page=1`.
    """
    import ast
    if not isinstance(node, ast.Constant) or node.value is None:
        return None
    value = node.value
    if isinstance(value, bool):
        return value if wiretype == 'boolean' else str(value).lower()
    if isinstance(value, (int, float)):
        return value if wiretype in ('integer', 'number') else str(value)
    if isinstance(value, str):
        return value
    return None


def _args_read(call, constants):
    """One parameter from a request.args.get/getlist call, or None."""
    import ast
    attribute = call.func
    if not isinstance(attribute, ast.Attribute):
        return None
    if attribute.attr not in ('get', 'getlist'):
        return None
    # `<anything>.args.get(...)`: the receiver is what matters, not whether it
    # is spelled `request` or a handler's `request_obj` parameter.
    owner = attribute.value
    if not isinstance(owner, ast.Attribute) or owner.attr != 'args':
        return None
    if not call.args:
        return None
    name = _constant_string(call.args[0], constants)
    if not name:
        return None
    if attribute.attr == 'getlist':
        # Repeatable: '?id=1&id=2'. Published as an array of strings, which is
        # what getlist returns without a converter.
        return {'name': name, 'type': 'array', 'required': False}
    wiretype = _wire_type(call)
    parameter = {'name': name, 'type': wiretype, 'required': False}
    fallback = None
    if len(call.args) > 1:
        fallback = call.args[1]
    for keyword in call.keywords:
        if keyword.arg == 'default':
            fallback = keyword.value
    if fallback is not None:
        default = _literal_default(fallback, wiretype)
        if default is not None:
            parameter['default'] = default
    return parameter


def _helper_read(call, constants):
    """Parameters from one call to an argument-reading helper."""
    import ast
    tail = call.func.attr if isinstance(call.func, ast.Attribute) else getattr(call.func, 'id', '')
    if tail == 'get_pagination_params':
        return [dict(item) for item in PAGINATION_PARAMS]
    if tail == 'intarg':
        if not call.args:
            return []
        name = _constant_string(call.args[0], constants)
        if not name:
            return []
        parameter = {'name': name, 'type': 'integer', 'required': False}
        fallback = call.args[1] if len(call.args) > 1 else None
        for keyword in call.keywords:
            if keyword.arg == 'default':
                fallback = keyword.value
        if fallback is not None:
            default = _literal_default(fallback, 'integer')
            if default is not None:
                parameter['default'] = default
        return [parameter]
    if tail == 'intargs':
        found = []
        for node in call.args:
            name = _constant_string(node, constants)
            if name:
                found.append({'name': name, 'type': 'integer', 'required': False})
        return found
    return []


def _args_subscript(node, constants):
    """A parameter from `request.args['x']`, which is REQUIRED.

    Werkzeug raises a 400 BadRequestKeyError when the argument is absent, so
    this is the one read in the language that proves the caller must send it.
    """
    import ast
    owner = node.value
    if not isinstance(owner, ast.Attribute) or owner.attr != 'args':
        return None
    name = _constant_string(node.slice, constants)
    if not name:
        return None
    return {'name': name, 'type': 'string', 'required': True}


def _merge_parameter(found, parameter):
    """Fold one read into the set, keeping the strongest evidence.

    The same argument is often read twice in one handler - once to test it, once
    to use it - and the two reads do not always carry the same evidence. Take
    required from either (one required read is proof), the more specific type,
    and the first default seen.
    """
    existing = found.get(parameter['name'])
    if existing is None:
        found[parameter['name']] = dict(parameter)
        return
    if parameter.get('required'):
        existing['required'] = True
    if existing.get('type') == 'string' and parameter.get('type') != 'string':
        existing['type'] = parameter['type']
    if 'default' not in existing and 'default' in parameter:
        existing['default'] = parameter['default']


def _module_constants(tree):
    """Module-level names bound to a string literal, for name resolution."""
    import ast
    constants = {}
    for node in tree.body:
        if not isinstance(node, ast.Assign):
            continue
        if not isinstance(node.value, ast.Constant) or not isinstance(node.value.value, str):
            continue
        for target in node.targets:
            if isinstance(target, ast.Name):
                constants[target.id] = node.value.value
    return constants


def _query_index(filename, _cache={}):
    """{function name: [parameter, ...]} of derived query arguments for one file.

    Read from the source for the same reason the auth sweep is: by the time a
    view is in url_map, functools.wraps has left nothing of its body to inspect.

    ast.walk over the function covers a nested def as well, which is deliberate
    - a handler that reads its arguments inside an inner function still reads
    them.
    """
    import ast
    if filename in _cache:
        return _cache[filename]
    index = {}
    try:
        with open(filename, 'r', encoding='utf-8') as handle:
            tree = ast.parse(handle.read())
    except (OSError, SyntaxError):
        _cache[filename] = index
        return index
    constants = _module_constants(tree)
    for node in ast.walk(tree):
        if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            continue
        found = {}
        for item in ast.walk(node):
            if isinstance(item, ast.Call):
                parameter = _args_read(item, constants)
                if parameter:
                    _merge_parameter(found, parameter)
                for parameter in _helper_read(item, constants):
                    _merge_parameter(found, parameter)
            elif isinstance(item, ast.Subscript):
                parameter = _args_subscript(item, constants)
                if parameter:
                    _merge_parameter(found, parameter)
        index[node.name] = [found[name] for name in sorted(found)]
    _cache[filename] = index
    return index


def _reads_a_json_body(view):
    """True when the handler really reads a request body.

    Evidence has to come from the HANDLER, not from a transcript. The suite
    sends {} to routes that ignore it, and a transcript-driven guess published
    "this route reads a JSON body" on DELETE /api/endpoints/agentkey, which
    never calls get_json. A client built from that would send a body to a route
    that discards it, and believe the body mattered.

    Matches request.get_json(...), request.json, request.data, request.form and
    request.files, on any receiver named request, anywhere in the function.
    """
    import ast
    import inspect
    import textwrap
    try:
        original = inspect.unwrap(view)
        source = textwrap.dedent(inspect.getsource(original))
        tree = ast.parse(source)
    except (OSError, TypeError, SyntaxError, IndentationError):
        # A view defined dynamically has no readable source. Silence, not a
        # guess: this decides whether the published contract claims a body.
        return False
    for node in ast.walk(tree):
        if not isinstance(node, ast.Attribute):
            continue
        if node.attr not in ('get_json', 'json', 'data', 'form', 'files'):
            continue
        base = node.value
        if isinstance(base, ast.Name) and base.id == 'request':
            return True
    return False


def _query_parameters(view):
    """Derived query parameters for one url_map view, sorted by name."""
    import inspect
    original = inspect.unwrap(view)
    filename = inspect.getsourcefile(original) or ''
    return _query_index(filename).get(original.__name__, [])


def _query_parameter_objects(entry):
    """The inventory's derived `queryparams` as OpenAPI parameter objects.

    Defensive about the key being absent: an inventory written before this
    existed, or one hand-trimmed, still generates a spec.
    """
    parameters = []
    for parameter in entry.get('queryparams') or []:
        name = parameter.get('name')
        if not name:
            continue
        wiretype = parameter.get('type') or 'string'
        if wiretype == 'array':
            schema = {'type': 'array', 'items': {'type': 'string'}}
        else:
            schema = {'type': wiretype}
        if 'default' in parameter:
            schema['default'] = parameter['default']
        parameters.append({
            'name': name,
            'in': 'query',
            'required': bool(parameter.get('required')),
            'schema': schema,
        })
    return parameters


def build(surfaces):
    paths = {}
    tags = []
    for s in surfaces:
        tags.append({'name': s['surface']})
        for e in s.get('endpoints', []):
            # 'PUT|PATCH' is how the inventory writes a multi-verb route.
            # Without the pipe here, neither verb matched VERBS and the
            # operation vanished from the spec without a word.
            for verb in re.split(r'[\/,|]', (e.get('method') or 'GET')):
                verb = verb.strip().lower()
                if verb not in VERBS:
                    continue
                path = _norm_path(e.get('path'))
                if not path:
                    continue
                desc = []
                if e.get('purpose'):
                    desc.append(e['purpose'])
                if e.get('auth'):
                    desc.append('\n\n**Auth:** ' + e['auth'])
                if e.get('params'):
                    desc.append('\n\n**Params:** ' + e['params'])
                if e.get('example'):
                    desc.append('\n\n**Example:**\n```\n' + e['example'] + '\n```')
                security = _security(e.get('auth'))
                pathparameters = _path_parameters(e.get('path'))
                # Path first, then query, so a reader sees the URL in the order
                # they type it.
                parameters = pathparameters + _query_parameter_objects(e)
                operation = {
                    'tags': [s['surface']],
                    'summary': _summary(e.get('purpose'), path),
                    'description': ''.join(desc),
                    'security': security,
                    # Only PATH parameters decide whether 404 is documented. A
                    # query filter that matches nothing is an empty list and a
                    # 200, not a missing record.
                    'responses': _responses(security, pathparameters,
                                            e.get('responseexample')),
                }
                if parameters:
                    operation['parameters'] = parameters
                body = _request_body(verb, e)
                if body:
                    operation['requestBody'] = body
                paths.setdefault(path, {})[verb] = operation
    return {
        'openapi': '3.1.0',
        'info': {
            'title': 'ShopDB Flask API',
            # Read from the code, not restated. A hardcoded copy here had
            # already drifted a release behind, and the same mistake in the
            # installer script shipped an exe stamped with the wrong version.
            'version': _product_version(),
            'description': (
                'Asset-management API (core + plugins). Responses use a '
                '`success_response` envelope: `{status, data, meta}`. Auth: Bearer '
                'JWT (login or a managed PAT) for `jwt`/`admin`/`permission:*`; '
                '`X-API-Key` for collector/managed-token endpoints; public endpoints '
                'need neither. A permission-gated route is published as '
                'bearer-required, which is what a default install does - a site '
                'that grants the permission to the `guest` role makes that same '
                'route answer a caller with no token.'),
        },
        # One site's production hostname was hardcoded here, so every generated
        # spec published it to the public wiki and offered a second site a
        # server it cannot reach. The relative mount is the only server this
        # generator can honestly name; a site that wants its own in the spec
        # sets SHOPDB_PUBLIC_URL when regenerating.
        'servers': _servers(),
        'components': {
            'schemas': {
                # The envelope every JSON endpoint returns. Worth spelling out
                # because the error shape nests one level deeper than most
                # people assume, and code written against the assumption reads
                # undefined on every error it tries to report.
                'SuccessEnvelope': {
                    'type': 'object',
                    'properties': {
                        'status': {'type': 'string', 'enum': ['success']},
                        'data': {'description': 'The payload. Shape is per endpoint.'},
                        'meta': {
                            'type': 'object',
                            'properties': {
                                'timestamp': {'type': 'string', 'format': 'date-time'},
                                'requestid': {'type': 'string'},
                                'pagination': {
                                    'type': 'object',
                                    'properties': {
                                        'page': {'type': 'integer'},
                                        'perpage': {'type': 'integer'},
                                        'total': {'type': 'integer'},
                                        'pages': {'type': 'integer'},
                                    },
                                },
                            },
                        },
                    },
                    'required': ['status'],
                },
                'ErrorEnvelope': {
                    'type': 'object',
                    'properties': {
                        'status': {'type': 'string', 'enum': ['error']},
                        'data': {
                            'type': 'object',
                            'properties': {
                                'error': {
                                    'type': 'object',
                                    'properties': {
                                        'code': {'type': 'string'},
                                        'message': {'type': 'string'},
                                        'details': {'type': 'object'},
                                    },
                                    'required': ['code', 'message'],
                                },
                            },
                            'required': ['error'],
                        },
                        'meta': {'type': 'object'},
                    },
                    'required': ['status', 'data'],
                },
            },
            'securitySchemes': {
            'bearerAuth': {'type': 'http', 'scheme': 'bearer', 'bearerFormat': 'JWT',
                           'description': 'Login token or a managed Personal Access Token (scoped).'},
            # One scheme covers every non-JWT header credential in this API,
            # because the header NAME varies per surface and each operation's
            # **Auth:** line says which one it wants. Publishing a scheme per
            # header would be tidier and no more informative.
            'apiKeyAuth': {'type': 'apiKey', 'in': 'header', 'name': 'X-API-Key',
                           'description': 'A managed service or device token in a header. '
                                          'X-API-Key for collector and managed tokens '
                                          '(collector.ingest, geenforce.fetch); '
                                          'X-Endpoint-Bootstrap-Token for agent bootstrap; '
                                          'X-Kiosk-Token for a kiosk screen. The operation '
                                          'description names the header and, where the gate '
                                          'is token-OR-network, the allowlist that is the '
                                          'other half.'},
            'setupCodeAuth': {'type': 'apiKey', 'in': 'query', 'name': 'code',
                              'description': 'A single-use provisioning code, redeemed in the '
                                             'query string because the caller is a browser being '
                                             'pointed at a URL. Spent on any attempt, right or '
                                             'wrong, and short-lived.'},
        }},
        'tags': tags,
        'paths': paths,
    }


# ---------------------------------------------------------------------------
# The derived half: refresh the inventory from the live url_map (--refresh)
# ---------------------------------------------------------------------------

# Only the API. The frontend shell and its assets are served by this app but are
# not the contract this document describes, and tests/test_api_inventory_parity
# walks the same /api-only surface - an entry for anything else reports there as
# a documented route that is not served.
API_PREFIX = '/api'

# The auth string for a route whose gate this cannot see. Deliberately not
# 'none': an unclassified route is the one case where a guess writes a lie into
# the published contract, and a site reads that contract to decide what its
# guest role and its firewall may reach. This greps, and it renders at /api/docs.
# Hand-editing the inventory does NOT clear this: auth is derived, so the next
# --refresh writes it straight back. Teaching the sweep below is the fix, which
# is also the only way the classification survives the route moving.
NEEDS_REVIEW = ('review: no gate found on the view - teach the sweep in '
                'scripts/gen_openapi.py, or gate the view')

# The jwt-optional wording, in one place because three branches reach it.
JWT_OPTIONAL = 'jwt-optional (works signed out; the handler decides what shows)'

# Decorators that ARE the gate. require_permission and require_role are not
# here: they stamp _requires_permission onto the wrapper, which is read first
# and carries the permission name itself.
DECORATOR_AUTH = {
    'require_api_key': 'api-key (per-plugin collector token)',
    'require_api_key_legacy': 'api-key (collector token, legacy shared key)',
    'require_fetch_token': 'api-key (geenforce fetch token, or the IP allowlist)',
    'require_report_token': 'api-key (geenforce report token)',
}

# Gates that live INSIDE the handler, found by the call the handler makes. An
# agent, a kiosk screen and a bootstrapping PC each authenticate on something
# that is not a JWT, so none of them can be gated by a decorator - and from
# outside the view, an inline gate and no gate at all look exactly alike. That
# is how 165 GET routes came to answer anybody.
HANDLER_AUTH = (
    ('_agent_auth_or_error', 'api-key (endpoint agent credential, or the fleet '
                             'key while sharedkeyallowed is on)'),
    ('authenticate_agent', 'api-key (endpoint agent credential, or the fleet '
                           'key while sharedkeyallowed is on)'),
    ('collector_ingest', 'api-key (endpoints collector token)'),
    ('_check_collector_auth', 'api-key (per-plugin collector token)'),
    ('collector_authorized', 'api-key (per-plugin collector token)'),
    # Token OR network, either sufficient, so neither 'none' nor 'api-key' is
    # the truth on its own: an allowlisted PC sends no credential at all, and a
    # screen by the cabinet sends the device token from anywhere. 'none' was
    # published here, which told a reader there is no token to send and left
    # the allowlist invisible. Optional says both halves exist.
    ('_kiosk_open', 'api-key-optional (kiosk: the per-device token in the '
                    'X-Kiosk-Token header, OR a caller address inside the '
                    "site's kiosk CIDR allowlist - an allowlisted caller sends "
                    "no credential; 404 while the site's kiosk flag is off)"),
    # REQUIRED, not 'none'. The handler reads the token and 404s without it,
    # and /bootstrap/package hands out the installable agent package and spends
    # a use doing it.
    ('_presented_token', 'api-key (single-use bootstrap token in the '
                         'X-Endpoint-Bootstrap-Token header, or ?t= while that '
                         'is enabled; per-IP rate limited, 404 when not usable)'),
    # The code IS the credential: unguessable, spent on any attempt, and gone in
    # minutes. No header and no session, because the caller is a browser being
    # pointed at a URL by an administrator.
    ('_redeem_kiosk_setup_code', 'setup-code (single-use kiosk setup code in '
                                 '?code=, issued by the permission-gated '
                                 'provisioning link; spent on any attempt, 410 '
                                 'once used or expired)'),
    ('_bootstrap_allowed', 'none (first run only: refused once an admin exists)'),
    ('verify_jwt_in_request', 'jwt'),
)

# Routes whose gate is inline AND conditional, where the call sweep above would
# name the wrong thing. One line each, and the line is the point: these are the
# handlers where "what authenticates this?" has an answer no sweep can derive.
INLINE_AUTH = {
    # The key may be absent, or invalid, and the call still proceeds: a PC
    # installed without the shared key must still be able to reach an admin. A
    # valid key only decides whether auto-enroll may issue on a clean match.
    'endpoints.enroll_endpoint': 'none (per-IP rate limited and capped; a valid '
                                 'fleet key only decides auto-enroll)',
    # The unguessable token in the path IS the authorisation - the recipient of
    # a secure send has no account here.
    'tools.peek_secret': 'none (the unguessable token in the path is the '
                         'authorisation)',
    'tools.redeem_secret': 'none (the unguessable token in the path is the '
                           'authorisation; redeeming spends the secret)',
    # This IS the auth surface, so a token cannot gate it. The bounds it carries
    # instead are a per-IP rate limit and the 5-strike account lockout.
    'auth.login': 'none (the login surface itself; rate limited, 5-strike lockout)',
}

# Verbs that carry a body, for the seeded params line.
BODY_VERBS = ('POST', 'PUT', 'PATCH')


def _literal_set(filename, name):
    """One module-level set literal, read without importing the module.

    Imported, tests/test_core/test_authz.py builds a second app at import time,
    which rebinds the SQLAlchemy extension out from under the one this script
    is walking.
    """
    import ast
    try:
        with open(filename, 'r', encoding='utf-8') as handle:
            tree = ast.parse(handle.read())
    except (OSError, SyntaxError):
        return set()
    for node in tree.body:
        if not isinstance(node, ast.Assign):
            continue
        for target in node.targets:
            if isinstance(target, ast.Name) and target.id == name:
                try:
                    return set(ast.literal_eval(node.value))
                except ValueError:
                    return set()
    return set()


def _reviewed_exceptions():
    """(endpoints, blueprints) already reviewed as needing no permission.

    Read from the two sweeps in tests/ rather than restated here, so the
    published contract cannot disagree with the gate that enforces it. A route
    on one of those lists was argued about in front of somebody, in a file whose
    whole job is to make that argument visible; a route on neither, with no
    gate, is marked for review instead. That is the difference between a
    decision and an oversight, which is what those lists exist to keep.
    """
    tests = os.path.join(REPO, 'tests', 'test_core')
    reads = os.path.join(tests, 'test_public_read_surface.py')
    writes = os.path.join(tests, 'test_authz.py')
    endpoints = (_literal_set(reads, 'PUBLIC_ENDPOINTS')
                 | _literal_set(writes, 'EXEMPT_ENDPOINTS'))
    return endpoints, _literal_set(writes, 'EXEMPT_BLUEPRINTS')


def _source_index(filename, _cache={}):
    """{function name: (decorator sources, call sources)} for one file.

    Read from the source, not from the live object: functools.wraps copies the
    name and the docstring onto the wrapper and nothing else, so by the time a
    view is in url_map its decorators are invisible. _requires_permission
    survives only because require_permission deliberately stamps it on.
    """
    import ast
    if filename in _cache:
        return _cache[filename]
    index = {}
    try:
        with open(filename, 'r', encoding='utf-8') as handle:
            tree = ast.parse(handle.read())
    except (OSError, SyntaxError):
        _cache[filename] = index
        return index
    for node in ast.walk(tree):
        if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            continue
        index[node.name] = (
            [ast.unparse(item) for item in node.decorator_list],
            [ast.unparse(item) for item in ast.walk(node)
             if isinstance(item, ast.Call)],
        )
    _cache[filename] = index
    return index


def _view_source(view):
    """(decorators, calls) for the undecorated view behind a url_map entry."""
    import inspect
    original = inspect.unwrap(view)
    filename = inspect.getsourcefile(original) or ''
    return _source_index(filename).get(original.__name__, ([], []))


def _auth_for(view, endpoint, reviewed, blueprints):
    """The honest auth string for one route, derived in gate order.

    Nothing here guesses. Every branch names a mechanism that is in the code - a
    marker the decorator stamped on, a decorator in the source, a call the
    handler makes, or a name on a reviewed exemption list - and a route that
    matches none of them says exactly that.
    """
    marker = getattr(view, '_requires_permission', None)
    if marker:
        roles = [name.split(':', 1)[1] for name in marker
                 if name.startswith('role:')]
        if roles:
            return 'jwt + role:' + ' or '.join(sorted(roles))
        return 'permission:' + ' or '.join(marker)

    decorators, calls = _view_source(view)
    for text in decorators:
        name = text.split('(', 1)[0].strip().split('.')[-1]
        if name in DECORATOR_AUTH:
            return DECORATOR_AUTH[name]
        if name == 'jwt_required':
            if 'refresh=True' in text:
                return 'jwt (refresh token)'
            return JWT_OPTIONAL if 'optional=True' in text else 'jwt'

    if endpoint in INLINE_AUTH:
        return INLINE_AUTH[endpoint]

    for call in calls:
        tail = call.split('(', 1)[0].split('.')[-1]
        for needle, auth in HANDLER_AUTH:
            if tail != needle:
                continue
            # Optional verification is a read that works signed out, not a gate.
            # Only the required form authenticates anybody.
            if needle == 'verify_jwt_in_request' and 'optional' in call:
                return JWT_OPTIONAL
            return auth

    blueprint, _, shortname = endpoint.partition('.')
    if endpoint in reviewed or shortname in reviewed or blueprint in blueprints:
        return 'none (reviewed: named in the public-read or mutation sweep)'
    return NEEDS_REVIEW


def _surface_for(view, endpoint):
    """The tag this route is grouped under: one surface per plugin, and one per
    core blueprint.

    Derived, not preserved: a route that moves into a plugin moves tag with it,
    which is the whole reason the reader is looking at the tag.
    """
    import inspect
    module = getattr(inspect.unwrap(view), '__module__', '') or ''
    if module.startswith('plugins.'):
        return 'plugin-' + module.split('.')[1]
    return 'core-' + endpoint.partition('.')[0]


def _seed_purpose(view, endpoint):
    """First paragraph of the view's docstring, as a starting point.

    A docstring says what the handler does; the inventory wants what the
    endpoint is FOR, which is usually more. This seeds the entry so a new route
    is never published blank, and the prose is preserved from then on.
    """
    import inspect
    doc = inspect.getdoc(inspect.unwrap(view)) or ''
    first = ' '.join(doc.split('\n\n')[0].split())
    return first or 'UNDOCUMENTED: %s has no docstring.' % endpoint


def _path_arguments(rule):
    """[(name, converter)] for one rule, in the order they appear."""
    return [(name, converter or 'string') for converter, name
            in re.findall(r'<(?:([^:>]+):)?([^>]+)>', rule)]


def _seed_params(rule, methods):
    parts = []
    arguments = _path_arguments(rule)
    if arguments:
        parts.append('path: ' + ', '.join(
            '%s (%s)' % (name, converter) for name, converter in arguments))
    if any(method in BODY_VERBS for method in methods):
        parts.append('body: JSON (fields not yet documented)')
    # Path arguments are in the rule; query arguments are read out of the
    # handler and cannot be derived, so an entry that has neither says it does
    # not know rather than claiming the route takes nothing.
    return '; '.join(parts) if parts else 'not documented'


def _seed_example(rule, methods, auth):
    """A curl that is the right SHAPE for this route: its verb, a filled path
    and the header its auth actually wants.

    Placeholder ids, because the generator has no database - but a reader who
    substitutes one gets a call that works, which is more than a blank field
    gives them.
    """
    def fill(match):
        converter = (match.group(1) or '').split('(')[0]
        if converter.startswith(('int', 'float')):
            return '1'
        return 'example.png' if converter == 'path' else 'x'

    url = 'http://localhost:5001' + re.sub(r'<(?:([^:>]+):)?[^>]+>', fill, rule)
    method = sorted(methods)[0] if methods else 'GET'
    command = 'curl' if method == 'GET' else 'curl -X ' + method
    header = ''
    if auth.startswith(('permission:', 'jwt')):
        header = ' -H "Authorization: Bearer $TOK"'
    elif auth.startswith('api-key'):
        # The header name varies per surface (X-API-Key, X-Kiosk-Token,
        # X-Endpoint-Bootstrap-Token), and the auth string names the one this
        # route wants. A pasted example with the wrong header is worse than
        # none: it looks like it should work.
        named = re.search(r'\bX-[A-Za-z0-9-]+\b', auth)
        header = ' -H "%s: $KEY"' % (named.group(0) if named else 'X-API-Key')
    elif auth.startswith('setup-code'):
        url += '?code=$CODE'
    body = ''
    if any(verb in methods for verb in BODY_VERBS):
        body = " -H 'Content-Type: application/json' -d '{}'"
    return '%s %s%s%s' % (command, url, header, body)


def _served_routes(app):
    """Every /api rule, as (path, sorted methods, endpoint, view).

    HEAD and OPTIONS are Werkzeug's, not the product's. A rule with several
    verbs stays one entry, written 'PUT|PATCH', which is how the inventory has
    always spelled it and what the parity test splits on.
    """
    routes = []
    for rule in app.url_map.iter_rules():
        if rule.endpoint == 'static' or not rule.rule.startswith(API_PREFIX):
            continue
        methods = sorted((rule.methods or set()) - {'HEAD', 'OPTIONS'})
        if not methods:
            continue
        routes.append((rule.rule, methods, rule.endpoint,
                       app.view_functions[rule.endpoint]))
    return sorted(routes, key=lambda row: (row[0], row[1]))


def _existing_prose():
    """{(path, METHOD): entry} for whatever the inventory already says.

    Keyed per method rather than per entry so prose survives a route that gains
    or loses a verb: 'PUT' still finds the words written when the entry said
    'PUT|PATCH'.
    """
    prose = {}
    if not os.path.isfile(INVENTORY):
        return prose
    with open(INVENTORY, 'r', encoding='utf-8') as handle:
        surfaces = json.load(handle)
    for surface in surfaces:
        for entry in surface.get('endpoints', []):
            path = entry.get('path') or ''
            for method in re.split(r'[/,|]', entry.get('method') or 'GET'):
                method = method.strip().upper()
                if method:
                    prose.setdefault((path, method), entry)
    return prose


def _write_inventory(document):
    """One writer, because --refresh and --harvest both rewrite this file.

    Same indent and same trailing newline from both, so a harvest that changes
    nothing changes no bytes and the diff of a regenerated inventory is only
    what actually moved.
    """
    with open(INVENTORY, 'w', encoding='utf-8') as handle:
        json.dump(document, handle, indent=2)
        handle.write('\n')


def _inventory_app():
    """The app, with every bundled plugin enabled.

    The inventory describes the PRODUCT, not one site's switched-on subset: a
    plugin disabled on the machine that happens to run this would drop its whole
    surface out of the published contract without a word, and the next refresh
    somewhere else would put it back. So a disabled plugin is refused here
    rather than quietly omitted.
    """
    import sys
    # .env BEFORE create_app, not after: config is read while the app is built,
    # so a load underneath it changes nothing. A plain `python scripts/...` run
    # gets neither wsgi.py's load nor the flask CLI's, and falls back to the
    # hardcoded root@localhost URL - which surfaces as a bare 1045 access-denied
    # that says nothing about the real cause.
    from dotenv import load_dotenv
    load_dotenv(os.path.join(REPO, '.env'))
    os.environ.setdefault('FLASK_ENV', 'testing')
    if REPO not in sys.path:
        sys.path.insert(0, REPO)
    from shopdb import create_app
    app = create_app('testing')
    plugins = os.path.join(REPO, 'plugins')
    bundled = sorted(
        name for name in os.listdir(plugins)
        if os.path.isfile(os.path.join(plugins, name, 'manifest.json')))
    manager = app.extensions.get('plugin_manager')
    off = [name for name in bundled
           if not (manager and manager.registry.is_enabled(name))]
    if off:
        raise SystemExit(
            'these bundled plugins are disabled, so their routes would be '
            'missing\nfrom the inventory: %s\nenable them and re-run:\n'
            '  FLASK_APP=shopdb venv/bin/flask plugins enable %s'
            % (', '.join(off), off[0]))
    return app


def refresh_inventory():
    """Rewrite the inventory's derived half from the routes the app serves.

    Structure (surface, method, path, auth) is replaced outright. Prose
    (purpose, params, example) is carried over untouched wherever an entry for
    that route already exists, and seeded from the view's docstring where one
    does not.
    """
    app = _inventory_app()
    reviewed, blueprints = _reviewed_exceptions()
    prose = _existing_prose()

    surfaces = {}
    seeded = []
    unclassified = []
    for path, methods, endpoint, view in _served_routes(app):
        auth = _auth_for(view, endpoint, reviewed, blueprints)
        if auth == NEEDS_REVIEW:
            unclassified.append('%s %s (%s)' % ('|'.join(methods), path, endpoint))
        known = prose.get((path, methods[0]))
        if known is None:
            seeded.append('%s %s' % ('|'.join(methods), path))
        entry = dict(known or {})
        entry.update({
            'method': '|'.join(methods),
            'path': path,
            'auth': auth,
            'params': entry.get('params') or _seed_params(path, methods),
            'purpose': entry.get('purpose') or _seed_purpose(view, endpoint),
            'example': entry.get('example') or _seed_example(path, methods, auth),
        })
        # DERIVED, like auth: rewritten outright rather than preserved. A query
        # argument the handler no longer reads has to leave the contract, and a
        # hand-written one has nothing holding it true.
        derived = _query_parameters(view)
        if derived:
            entry['queryparams'] = derived
        else:
            entry.pop('queryparams', None)
        # Same rule: derived, not preserved. A handler that stops reading a body
        # must stop advertising one.
        if _reads_a_json_body(view):
            entry['readsbody'] = True
        else:
            entry.pop('readsbody', None)
        # Fixed key order, so a refresh that changes nothing changes no bytes.
        ordered = {key: entry[key] for key in
                   ('method', 'path', 'auth', 'params', 'purpose', 'example')}
        ordered.update({key: value for key, value in entry.items()
                        if key not in ordered})
        surfaces.setdefault(_surface_for(view, endpoint), []).append(ordered)

    document = [{'surface': name, 'endpoints': entries}
                for name, entries in sorted(surfaces.items())]
    _write_inventory(document)

    total = sum(len(surface['endpoints']) for surface in document)
    print('refreshed %s: %d entries in %d surfaces (%d new)'
          % (INVENTORY, total, len(document), len(seeded)))
    for line in seeded:
        print('  seeded  %s' % line)
    if unclassified:
        print('%d route(s) have no gate this can see. Classify each one by hand '
              'in the inventory, or gate it:' % len(unclassified))
        for line in unclassified:
            print('  review  %s' % line)


# ---------------------------------------------------------------------------
# Harvested examples, and the redaction that has to happen first
# ---------------------------------------------------------------------------
#
# 0 of 573 operations carried a machine-readable example, which is the single
# thing people mean when they ask for usable API documentation. Writing 573 of
# them by hand is not going to happen, and inventing them is forbidden: an
# example body that nobody ever sent is a lie in a contract that a generated
# client will send verbatim.
#
# The examples already exist. The test suite posts real bodies at the real app
# and asserts the real responses. So --harvest runs pytest with this file
# loaded as a plugin, records what the test client actually sent and got back,
# and folds it into the inventory. Nothing is composed here; every example is a
# transcript.
#
# Harvesting is OPTIONAL and RE-RUNNABLE. gen_openapi.py generates a spec with
# no harvest data at all (that is the normal case in a published checkout, which
# has no tests), and a second harvest overwrites the first rather than
# accumulating.

HARVEST_DIR_VAR = 'SHOPDB_OPENAPI_HARVEST_DIR'

# The keys and values that never reach the published spec.
#
# THIS FILE IS SERVED ON PRODUCTION AND PUBLISHED TO A WIKI. A harvested body is
# whatever a test happened to build, and tests are written with real-looking
# material: a login body carries a password, an agent body carries its token, a
# kiosk body carries a badge number, an endpoint body carries a hostname and an
# address. None of that is dangerous in a test file, and all of it is dangerous
# on a public page - a reader cannot tell a fixture from a fact, and neither can
# a scraper.
#
# Matched on the KEY first, because the key is the reliable signal: a field
# called 'passwordhash' is a credential whatever it happens to contain today,
# including when a fixture leaves it empty. Then on the VALUE, for the material
# that carries no name - an address in a free-text field, a JWT in a list.
SECRET_KEY_PATTERN = re.compile(
    r'pass|pwd|secret|token|api[-_ ]?key|privatekey|credential|authoriz'
    r'|displayprefix'
    r'|cookie|salt|signature|sessionid|\botp\b|passphrase|hash', re.IGNORECASE)

# A person. Not a secret, but not ours to publish either: these pages are read
# by other sites and by the public, and a name or a badge number identifies
# somebody who never agreed to appear in API documentation.
PERSON_KEY_PATTERN = re.compile(
    r'username|email|firstname|lastname|fullname|displayname|badge|employee'
    r'|createdby|updatedby|modifiedby|assignedto|checkedoutby|reservedby'
    r'|requestedby|approvedby|\bowner|contactname|phone', re.IGNORECASE)

# A machine. ADR-015: one plant's server name, its FQDN or its internal networks
# are neither ours to publish nor meaningful to any other site reading this.
HOST_KEY_PATTERN = re.compile(
    r'hostname|fqdn|ipaddress|\bip\b|macaddress|serialnumber|servicetag'
    r'|computername|machinename|sharepath|uncpath|netbios', re.IGNORECASE)

# Searched, not matched: an address or a token embedded in a free-text field is
# as published as one that is the whole value, and a sentence that contains one
# is scrubbed entire rather than partially.
# A per-request value, normalised rather than published as recorded.
#
# WHY. A transcript carries the instant it was taken and the id of the request
# that produced it, and both are different on every run. Published verbatim,
# re-harvesting rewrote 313 of 320 examples with no change of meaning - a diff
# nobody can review, over a file that is committed and published. An example
# exists to show a reader the FORMAT of a field, not the moment one call
# happened, so the format is kept exactly and the value is pinned.
ISO_PATTERN = re.compile(
    r'^(\d{4})-(\d{2})-(\d{2})([T ])(\d{2}):(\d{2}):(\d{2})'
    r'(\.\d+)?(Z|[+-]\d{2}:?\d{2})?$')
DATE_PATTERN = re.compile(r'^\d{4}-\d{2}-\d{2}$')
STABLE_DATE = '2026-01-01'
STABLE_TIME = '00:00:00'
REQUEST_ID_KEYS = ('requestid', 'request_id', 'correlationid')
STABLE_REQUEST_ID = '00000000'

EMAIL_PATTERN = re.compile(r'[^@\s]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}')
JWT_PATTERN = re.compile(r'\beyJ[A-Za-z0-9_-]{8,}')
HEX_PATTERN = re.compile(r'^[0-9a-fA-F]{32,}$')
# An opaque handle: long, no spaces, and mixing letters with digits. A sentence
# does not look like this and a slug is not this long.
#
# ':' and '/' are NOT in the class, deliberately. With them, every example URL
# in the suite ('http://localhost:5001/api/assets') matched and the examples lost
# the one string in them a reader most wants to see. A base64 token containing
# '/' therefore slips this rule - it is caught by its key instead, which is the
# reliable half of this anyway.
OPAQUE_PATTERN = re.compile(r'^(?=[^\s]*\d)(?=[^\s]*[A-Za-z])[A-Za-z0-9_.+=-]{24,}$')
# A credential riding in a URL's query string. This is the hole the key layer
# and the opaque layer BOTH miss by construction: the key is 'scripturl' or
# 'url', which names no secret, and OPAQUE_PATTERN excludes ':' and '/' so the
# surrounding URL never matches. A harvest published a real bootstrap token and
# a real kiosk setup code that way, in a file that is served on prod and copied
# to a wiki. The whole URL must not be dropped - it is the most useful string in
# the example - so only the parameter VALUE is scrubbed.
URL_CREDENTIAL_PATTERN = re.compile(
    r'(?i)\b(t|token|code|key|apikey|api_key|secret|sig|signature|password|pw|'
    r'auth|access_token|refresh_token)=([^&\s"\']+)')
IPV4_PATTERN = re.compile(r'\b(?:\d{1,3}\.){3}\d{1,3}\b')
UNC_PATTERN = re.compile(r'^\\\\[^\\]+\\')
# A path off the machine the harvest ran on. A pytest tmpdir is a different
# string on every run, and a developer's home directory is the exact material
# tests/test_docs_publishable.py exists to keep out of published pages.
LOCAL_PATH_PATTERN = re.compile(r'(^|\s)/(tmp|home|Users|var/folders)/|pytest-of-')

# Addresses that name nothing: a loopback or a documentation host is safe to
# publish and is more useful in an example than a placeholder.
PUBLISHABLE_HOSTS = ('127.0.0.1', '0.0.0.0', 'localhost', 'example.com',
                     'example.local', 'test.local')

# How big an example may get. The spec is downloaded by every reader and parsed
# by every generated client, so a 400-row list response in one operation costs
# everybody. Three rows show the shape, which is the entire job of an example.
EXAMPLE_MAX_CHARS = 2000
EXAMPLE_MAX_ITEMS = 3
EXAMPLE_MAX_DEPTH = 8
EXAMPLE_MAX_STRING = 200


def _stable_timestamp(text):
    """The same timestamp with its value pinned and its shape untouched.

    Fractional digits and the zone suffix are preserved exactly, because that is
    the part a reader is reading it for - '2026-09-18T18:30:48.498252Z' and
    '2026-09-18 18:30:48' are two different contracts.
    """
    # A bare date too: a fixture that calls date.today() churns exactly like a
    # timestamp does, and the shape is all an example was showing.
    if DATE_PATTERN.match(text):
        return STABLE_DATE
    match = ISO_PATTERN.match(text)
    if not match:
        return None
    fraction = ''
    if match.group(8):
        fraction = '.' + '0' * len(match.group(8)[1:])
    return (STABLE_DATE + match.group(4) + STABLE_TIME + fraction
            + (match.group(9) or ''))


def _redacted_string(value):
    """The marker for a string value that must not be published, or None.

    Returns None when the value is safe, so the caller can tell "checked and
    fine" from "scrubbed".
    """
    text = value.strip()
    if not text:
        return None
    if EMAIL_PATTERN.search(text):
        return '<redacted:email>'
    if JWT_PATTERN.search(text):
        return '<redacted:token>'
    if UNC_PATTERN.match(text):
        return '<redacted:sharepath>'
    if LOCAL_PATH_PATTERN.search(text):
        return '<redacted:path>'
    if HEX_PATTERN.match(text):
        return '<redacted:digest>'
    for found in IPV4_PATTERN.findall(text):
        if found not in PUBLISHABLE_HOSTS:
            return '<redacted:address>'
    if OPAQUE_PATTERN.match(text):
        return '<redacted:token>'
    scrubbed = URL_CREDENTIAL_PATTERN.sub(
        lambda match: '%s=<redacted:token>' % match.group(1), text)
    if scrubbed != text:
        return scrubbed
    return None


def redact(value, key='', depth=0):
    """Scrub secrets, people and machines out of a harvested body.

    Key-driven first, value-driven second, and applied to every level - a token
    two objects down is as published as one at the top. The marker names WHICH
    class was matched, so a reader can see that the field exists and is not
    simply empty, without learning what was in it.

    Deliberately conservative in the direction of over-redacting. A redacted
    example is still a shape a client can copy; a published badge number cannot
    be un-published.
    """
    name = (key or '').lower()
    if depth > EXAMPLE_MAX_DEPTH:
        return '<redacted:depth>'
    if isinstance(value, dict):
        return {str(itemkey): redact(itemvalue, str(itemkey), depth + 1)
                for itemkey, itemvalue in value.items()}
    if isinstance(value, list):
        # Only the first few rows survive, and the cut is recorded rather than
        # silent: a reader who sees three assets should not conclude the list
        # endpoint returns three.
        trimmed = [redact(item, name, depth + 1) for item in value[:EXAMPLE_MAX_ITEMS]]
        if len(value) > EXAMPLE_MAX_ITEMS:
            trimmed.append('<trimmed for the example: %d more>'
                           % (len(value) - EXAMPLE_MAX_ITEMS))
        return trimmed
    # A NUMBER under a key ending in 'id' is a row identifier, not a credential.
    # Without this, 'tokenid': 3 published as <redacted:secret> and the example
    # for every token route lost the one field a caller needs to address the
    # row. The exemption is deliberately narrow - numeric only, 'id' only, and
    # it does not apply to the person or host rules below, where a badge number
    # is a number and still nobody's business.
    numericid = (isinstance(value, int) and not isinstance(value, bool)
                 and name.endswith('id'))
    if name and not numericid and SECRET_KEY_PATTERN.search(name):
        return '<redacted:secret>'
    if name and PERSON_KEY_PATTERN.search(name):
        return '<redacted:person>'
    if name and HOST_KEY_PATTERN.search(name):
        return '<redacted:host>'
    if name in REQUEST_ID_KEYS:
        return STABLE_REQUEST_ID
    if isinstance(value, str):
        marker = _redacted_string(value)
        if marker:
            return marker
        pinned = _stable_timestamp(value)
        if pinned is not None:
            return pinned
        if len(value) > EXAMPLE_MAX_STRING:
            return value[:EXAMPLE_MAX_STRING] + '...'
        return value
    return value


def _example_fits(value):
    """Whether a redacted example is small enough to publish."""
    try:
        return len(json.dumps(value)) <= EXAMPLE_MAX_CHARS
    except (TypeError, ValueError):
        return False


def _best_example(samples):
    """One example out of every sample recorded for an operation.

    Deterministic, because the suite runs in parallel and the order samples
    arrive in is not stable: sort by size descending then by text, and take the
    first that fits. Largest-that-fits is the most informative, and the text
    tiebreak means two runs of the same suite produce the same spec - which is
    what makes a regenerated docs/openapi.json reviewable as a diff.
    """
    candidates = []
    for sample in samples:
        scrubbed = redact(sample)
        if not _example_fits(scrubbed):
            continue
        text = json.dumps(scrubbed, sort_keys=True)
        candidates.append((-len(text), text, scrubbed))
    if not candidates:
        return None
    candidates.sort(key=lambda row: (row[0], row[1]))
    return candidates[0][2]


# ---------------------------------------------------------------------------
# The pytest plugin half: loaded with `-p gen_openapi` by --harvest
# ---------------------------------------------------------------------------
#
# Flask.full_dispatch_request is patched rather than an after_request handler
# registered, because the app under test is built by the suite's own session
# fixture and there is no reliable moment to reach into it - the conftest
# imports create_app at module import, before any hook here could wrap it.
# Patching the class catches every app in the process however and whenever it
# was created.
#
# Recorded to one file per process, appended a line at a time: the suite runs
# under xdist, so several workers record at once, and a crash in one of them
# must not cost the others their samples.

_HARVEST_STATE = {'original': None, 'path': None}


def _harvest_record(flaskrequest, response):
    """Append one request/response transcript, if it is one worth keeping."""
    rule = getattr(flaskrequest, 'url_rule', None)
    if rule is None or not rule.rule.startswith(API_PREFIX):
        return
    status = getattr(response, 'status_code', 0)
    # A body that produced a 400 is an example of getting it wrong, and a 500 is
    # an example of nothing. Only a call that worked describes the contract.
    if status < 200 or status >= 300:
        return
    sent = None
    try:
        if flaskrequest.mimetype == 'application/json':
            sent = flaskrequest.get_json(silent=True)
    except Exception:
        sent = None
    got = None
    try:
        if (response.mimetype == 'application/json'
                and not getattr(response, 'direct_passthrough', False)
                and not getattr(response, 'is_streamed', False)):
            got = json.loads(response.get_data(as_text=True))
    except Exception:
        got = None
    if sent is None and got is None:
        return
    line = json.dumps({
        'method': flaskrequest.method,
        'rule': rule.rule,
        'request': sent,
        'response': got,
    })
    try:
        with open(_HARVEST_STATE['path'], 'a', encoding='utf-8') as handle:
            handle.write(line + '\n')
    except OSError:
        pass


def pytest_configure(config):
    """Wrap Flask's dispatch so every test-client call is transcribed.

    Inert unless --harvest set the directory, so loading this file as a plugin
    in an ordinary run changes nothing.
    """
    outdir = os.environ.get(HARVEST_DIR_VAR)
    if not outdir or _HARVEST_STATE['original'] is not None:
        return
    import flask
    original = flask.Flask.full_dispatch_request
    _HARVEST_STATE['original'] = original
    _HARVEST_STATE['path'] = os.path.join(outdir, 'samples-%d.jsonl' % os.getpid())

    def dispatch(self):
        response = original(self)
        # A fault in the recorder must never change what the suite reports. The
        # harvest is documentation; the tests are the product.
        try:
            _harvest_record(flask.request, response)
        except Exception:
            pass
        return response

    flask.Flask.full_dispatch_request = dispatch


def pytest_unconfigure(config):
    if _HARVEST_STATE['original'] is None:
        return
    import flask
    flask.Flask.full_dispatch_request = _HARVEST_STATE['original']
    _HARVEST_STATE['original'] = None


def _read_samples(outdir):
    """{(METHOD, rule): {'request': [...], 'response': [...]}} from a harvest run."""
    collected = {}
    if not os.path.isdir(outdir):
        return collected
    for name in sorted(os.listdir(outdir)):
        if not name.endswith('.jsonl'):
            continue
        with open(os.path.join(outdir, name), 'r', encoding='utf-8') as handle:
            for line in handle:
                line = line.strip()
                if not line:
                    continue
                try:
                    row = json.loads(line)
                except ValueError:
                    continue
                key = (row.get('method', '').upper(), row.get('rule', ''))
                bucket = collected.setdefault(key, {'request': [], 'response': []})
                if row.get('request') is not None:
                    bucket['request'].append(row['request'])
                if row.get('response') is not None:
                    bucket['response'].append(row['response'])
    return collected


def fold_examples(collected):
    """Write harvested examples into the inventory. Returns (request, response) counts.

    The inventory is the input to the generator, so this is where they belong -
    it keeps the spec a pure function of the inventory, and it means a harvest
    survives in the committed file for a reader who never runs the suite.

    ADDITIVE, deliberately. Only a route this run actually exercised is
    rewritten; an entry the run never reached keeps the example it has. Two
    reasons, and the second is the important one:

      A targeted harvest has to be targeted. `--harvest tests/test_core` would
      otherwise strip the examples off every route those tests do not touch,
      which is most of them - the narrow command would do the most damage.

      The suite does not exercise the same set twice. Run under xdist against a
      shared instance directory, two full runs of the same tree reached 311 and
      327 operations. Clearing on absence turned that into a churn of examples
      appearing and vanishing in a committed, published file, with nothing
      wrong in the tree.

    An example that has gone stale is still a transcript of a response this app
    really gave. A route that is gone entirely takes its entry with it, which
    tests/test_api_inventory_parity.py already enforces.
    """
    with open(INVENTORY, 'r', encoding='utf-8') as handle:
        surfaces = json.load(handle)
    requests = 0
    responses = 0
    for surface in surfaces:
        for entry in surface.get('endpoints', []):
            path = entry.get('path') or ''
            sent = []
            got = []
            for method in re.split(r'[/,|]', entry.get('method') or 'GET'):
                bucket = collected.get((method.strip().upper(), path))
                if bucket:
                    sent.extend(bucket['request'])
                    got.extend(bucket['response'])
            example = _best_example(sent)
            if example is not None:
                entry['requestexample'] = example
            example = _best_example(got)
            if example is not None:
                entry['responseexample'] = example
            requests += 1 if 'requestexample' in entry else 0
            responses += 1 if 'responseexample' in entry else 0
    _write_inventory(surfaces)
    return requests, responses


def harvest(pytestargs):
    """Run the suite with the recorder attached, then fold what it saw in."""
    import subprocess
    import tempfile
    outdir = tempfile.mkdtemp(prefix='shopdb-openapi-harvest-')
    environment = dict(os.environ)
    environment[HARVEST_DIR_VAR] = outdir
    # This file is the plugin. Putting scripts/ on the path is what makes
    # `-p gen_openapi` resolvable, in the workers as well as the controller.
    environment['PYTHONPATH'] = os.pathsep.join(
        [HERE] + ([environment['PYTHONPATH']] if environment.get('PYTHONPATH') else []))
    command = [sys.executable, '-m', 'pytest', '-p', 'gen_openapi', '-q']
    command.extend(pytestargs or ['tests'])
    print('harvesting examples: ' + ' '.join(command))
    # The suite's own exit status is not this command's: a red test still
    # produced real transcripts for every route it did reach, and refusing to
    # record them would make the documentation hostage to an unrelated failure.
    subprocess.call(command, cwd=REPO, env=environment)
    collected = _read_samples(outdir)
    requests, responses = fold_examples(collected)
    print('harvested %d operation(s); inventory now carries %d request '
          'example(s) and %d response example(s)'
          % (len(collected), requests, responses))


EXAMPLE_FIELDS = ('requestexample', 'responseexample')


def _scrub_stored_examples(surfaces):
    """Re-redact every example the inventory carries, before it is published."""
    for surface in surfaces:
        for entry in surface.get('endpoints', []):
            for field in EXAMPLE_FIELDS:
                if field in entry and entry[field] is not None:
                    entry[field] = redact(entry[field], field)
    return surfaces


def main():
    arguments = sys.argv[1:]
    if '--refresh' in arguments:
        refresh_inventory()
    if '--harvest' in arguments:
        # Everything after the flag is pytest's, so a targeted harvest is
        # `--harvest tests/test_core/test_assets.py` and needs no second flag.
        harvest(arguments[arguments.index('--harvest') + 1:])
    with open(INVENTORY, 'r', encoding='utf-8') as handle:
        surfaces = json.load(handle)
    # SCRUB ON THE WAY OUT AS WELL AS ON THE WAY IN. Harvested examples are
    # stored in the inventory and a refresh preserves them, so a scrubber fixed
    # after a harvest protects only future harvests - the stored copy keeps
    # publishing whatever the old rules missed. On 2026-09-18 that was a live
    # bootstrap token and a live kiosk setup code, both hidden in a URL. Running
    # redaction here means the published file cannot carry a secret that the
    # CURRENT rules would catch, whatever is sitting in the inventory.
    surfaces = _scrub_stored_examples(surfaces)
    spec = build(surfaces)
    with open(OUT, 'w', encoding='utf-8') as handle:
        json.dump(spec, handle, indent=1)
    ops = sum(len(v) for v in spec['paths'].values())
    print('wrote %s: %d paths, %d operations from %d surfaces'
          % (OUT, len(spec['paths']), ops, len(surfaces)))


if __name__ == '__main__':
    main()