topmark.pipeline.steps.scanner ¶

def may_proceed(
    self,
    ctx: ProcessingContext,
) -> bool:
    """Determine if processing can proceed to the scan step.

    Processing can proceed if all of the following apply:
    - A header processor is available (ctx.header_processor is not None), meaning:
        - The file was successfully resolved (ctx.status.resolve is RESOLVED)
        - The file type was resolved (ctx.file_type is not None)
        - The file type is registered to a header processor
    - ReaderStep set `ContentStatus.OK`

    Args:
        ctx: The processing context for the current file.

    Returns:
        True if processing can proceed to the scanner step, False otherwise.
    """
    if ctx.is_halted:
        return False

    return ctx.header_processor is not None and ctx.status.content == ContentStatus.OK

def run(
    self,
    ctx: ProcessingContext,
) -> None:
    """Detect and extract a TopMark header from the file image.

    Behavior with discriminated ``HeaderBounds``:
        - ``NONE`` → mark ``HeaderStatus.MISSING`` and return.
        - ``MALFORMED`` → build a minimal ``HeaderView`` (when possible), set
          ``HeaderStatus.MALFORMED``, attach a diagnostic hint, and stop flow.
        - ``SPAN`` → slice the span, build ``HeaderView``, parse fields, and set
          one of ``{DETECTED, EMPTY, MALFORMED_SOME_FIELDS, MALFORMED_ALL_FIELDS}``.

    Args:
        ctx: Processing context with the file image and a header processor.

    Raises:
        RuntimeError: If the header processor is not defined, or if the exclusive start and
            end indexes are undefined or invalid.

    Mutations:
        ProcessingContext: The same context updated with:
        - ``ctx.views.header``: a ``HeaderView`` containing range/lines/block/mapping
        - ``ctx.status.header``: one of {MISSING, EMPTY, DETECTED, MALFORMED,
          MALFORMED_SOME_FIELDS, MALFORMED_ALL_FIELDS}
    """
    logger.debug(
        "Scanning file %s of type %s for header fields",
        ctx.path,
        ctx.file_type,
    )
    logger.debug("ctx: %s", ctx)

    if ctx.header_processor is None:
        raise RuntimeError("Header processor not defined")

    if ctx.status.fs == FsStatus.EMPTY:
        # An empty file is considered to have no header, but we can still proceed
        logger.info("File %s is empty; no header to scan.", ctx.path)
        ctx.status.header = HeaderStatus.MISSING
        self._halt_if_policy_blocks(ctx)
        return

    if ctx.image_line_count() == 0:
        # Defensive guard; upstream reader should have set UNREADABLE or EMPTY_FILE
        # TODO: check if ctx.status.header must be set (is now DEFAULT)
        logger.error("scan(): No file lines available for %s", ctx.path)
        return

    # Use header_processor.get_header_bounds() to locate header start and end indices
    hb: HeaderBounds = ctx.header_processor.get_header_bounds(
        lines=ctx.iter_image_lines(),
        newline_style=ctx.newline_style,
    )

    # NONE → MISSING
    if hb.kind is BoundsKind.NONE:
        logger.info("No header found in '%s'", ctx.path)
        ctx.status.header = HeaderStatus.MISSING
        ctx.header_newline_style = _select_header_newline(
            ctx,
            header_range=None,
        )
        self._halt_if_policy_blocks(ctx)
        return

    # MALFORMED → terminal (always)
    if hb.kind is BoundsKind.MALFORMED:
        s_opt: int | None
        e_opt: int | None
        s_opt, e_opt = hb.start, hb.end
        if s_opt is not None and e_opt is not None and s_opt < e_opt:
            # materialize minimal view for diagnostics
            lines: list[str] = list(islice(ctx.iter_image_lines(), s_opt, e_opt))
            ctx.views.header = HeaderView(
                range=(s_opt, e_opt - 1),  # store inclusive range for views
                lines=lines,
                block="".join(lines),
                mapping=None,
            )
        ctx.status.header = HeaderStatus.MALFORMED
        reason: str = hb.reason or "Malformed header markers"
        ctx.diagnostics.add_warning(reason)
        ctx.hint(
            axis=Axis.HEADER,
            code=KnownCode.HEADER_MALFORMED,
            cluster=Cluster.ERROR,
            message=reason,
            terminal=True,
        )
        ctx.request_halt(reason=f"scanner: {reason}", at_step=self)
        return

    # SPAN → slice, parse, classify
    s_excl: int | None = hb.start
    e_excl: int | None = hb.end  # exclusive
    if s_excl is None or e_excl is None:
        raise RuntimeError(
            f"Exclusive start ({s_excl!r}) and end ({e_excl!r}) indexes must be defined",
        )
    if s_excl >= e_excl:
        raise RuntimeError(
            f"Exclusive start ({s_excl!r}) index > end index ({e_excl!r}) index",
        )

    # Header found and valid: ensure we have inclusive indices
    # Slice via iterator to avoid materializing the full file
    header_lines: list[str] = list(islice(ctx.iter_image_lines(), s_excl, e_excl))
    header_block: str = "".join(header_lines)

    # 1) Create the view first (mapping=None for now). Store an inclusive range for the view.
    ctx.views.header = HeaderView(
        range=(s_excl, e_excl - 1),
        lines=header_lines,
        block=header_block,
        mapping=None,
    )
    ctx.header_newline_style = _select_header_newline(
        ctx,
        header_range=(s_excl, e_excl - 1),
    )

    # 2) Parse fields (parse_fields reads from context.header)
    parse_result: HeaderParseResult = ctx.header_processor.parse_fields(ctx)

    # 3) Attach mapping and counts to the view
    ctx.views.header.mapping = parse_result.fields
    ctx.views.header.success_count = parse_result.success_count
    ctx.views.header.error_count = parse_result.error_count

    # Now we no longer write to ctx.views.header
    header_view: HeaderView = ctx.views.header
    total_count: int = header_view.success_count + header_view.error_count

    logger.debug(
        "Header extracted from lines %d to %d, found %d header lines (%d ok, %d with errors), "
        "header block:\n%s",
        s_excl + 1,
        e_excl + 1,
        total_count,
        header_view.success_count,
        header_view.error_count,
        header_view.block,
    )

    if header_view.error_count > 0:
        # At least one header field line errored
        reason = (
            f"Header markers present at line {s_excl + 1} and {e_excl + 1}, "
            f"total: {total_count}, "
            f"ok: {header_view.success_count}, errors: {header_view.error_count}, "
        )

        if header_view.success_count == 0:
            # All header lines in the header block are malformed
            ctx.status.header = HeaderStatus.MALFORMED_ALL_FIELDS
            ctx.diagnostics.add_warning(f"{reason} - header contains no valid header lines.")
            return
        else:
            # At least one remaining valid header line
            ctx.status.header = HeaderStatus.MALFORMED_SOME_FIELDS
            ctx.diagnostics.add_warning(
                f"{reason} - header contains valid and invalid header lines."
            )
            return

    if not header_view.mapping:
        ctx.status.header = HeaderStatus.EMPTY
        ctx.diagnostics.add_warning(
            f"Header markers present at line {s_excl + 1} and {e_excl + 1} but no fields."
        )
    else:
        ctx.status.header = HeaderStatus.DETECTED

    self._halt_if_policy_blocks(ctx)

    logger.debug(
        "File status: %s, resolve status: %s, content status: %s, header status: %s",
        ctx.status.fs.value,
        ctx.status.resolve.value,
        ctx.status.content.value,
        ctx.status.header.value,
    )
    logger.trace(
        "Existing header dict: %s",
        header_view.mapping,
    )

    return

def hint(
    self,
    ctx: ProcessingContext,
) -> None:
    """Attach header detection hints (non-binding).

    Args:
        ctx: The processing context.

    Raises:
        RuntimeError: If the context contains an unexpected header status value.
    """
    st: HeaderStatus = ctx.status.header

    match st:
        # May proceed to next step (always):
        case HeaderStatus.DETECTED:
            ctx.hint(
                axis=Axis.HEADER,
                code=KnownCode.HEADER_DETECTED,
                cluster=Cluster.PENDING,
                message="TopMark header detected",
            )

        case HeaderStatus.MISSING:
            ctx.hint(
                axis=Axis.HEADER,
                code=KnownCode.HEADER_MISSING,
                cluster=Cluster.PENDING,
                message="no TopMark header detected",
            )

        case HeaderStatus.EMPTY:
            ctx.hint(
                axis=Axis.HEADER,
                code=KnownCode.HEADER_EMPTY,
                cluster=Cluster.PENDING,
                message="empty TopMark header",
            )

        # May proceed to next step (policy):
        case HeaderStatus.MALFORMED_ALL_FIELDS:
            ctx.hint(
                axis=Axis.HEADER,
                code=KnownCode.HEADER_MALFORMED,
                cluster=Cluster.BLOCKED_POLICY,
                message="all header fields malformed",
            )

        case HeaderStatus.MALFORMED_SOME_FIELDS:
            ctx.hint(
                axis=Axis.HEADER,
                code=KnownCode.HEADER_MALFORMED,
                cluster=Cluster.BLOCKED_POLICY,
                message="some header fields malformed",
            )

        # Stop processing:
        case HeaderStatus.MALFORMED:
            ctx.hint(
                axis=Axis.HEADER,
                code=KnownCode.HEADER_MALFORMED,
                cluster=Cluster.SKIPPED,
                message="malformed TopMark header",
                terminal=True,
            )

        # States owned outside this step:
        case HeaderStatus.PENDING:  # pragma: no cover - BaseStep owns pending-state handling.
            # BaseStep.__call__() handles PENDING state (step did not complete)
            pass

        case _:  # pragma: no cover - exhaustive enum guard for untyped callers.
            raise RuntimeError(f"Unexpected HeaderStatus found: {st!r}")