232 lines
7.9 KiB
Python
232 lines
7.9 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from io import StringIO
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Callable, Iterable, TextIO
|
|
from band_aider.judgement.repo_exec import RepoExec
|
|
|
|
from band_aider.context.class_meta import ClassMeta, ContentType
|
|
from band_aider.judgement.code_parser import CodeParser
|
|
from band_aider.judgement.lexical_anchor_check import LexicalAnchorCheck
|
|
from band_aider.judgement.source_discovery_strategy import (
|
|
SourceDiscoveryStrategy,
|
|
searchable_beat_texts,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
LexicalCheckFactory = Callable[[Iterable[str], Iterable[str]], LexicalAnchorCheck]
|
|
|
|
|
|
def create_lexical_anchor_check(
|
|
anchors: Iterable[str],
|
|
beats: Iterable[str],
|
|
) -> LexicalAnchorCheck:
|
|
"""Create one fresh lexical anchor check for strict parser prefiltering."""
|
|
|
|
return LexicalAnchorCheck(
|
|
changes=[],
|
|
anchors=anchors,
|
|
candidate_beats=beats,
|
|
)
|
|
|
|
|
|
@dataclass(init=False)
|
|
class StrictParser(CodeParser):
|
|
"""StrictParser gates parser delegation through lexical anchor matching only."""
|
|
|
|
_inner: CodeParser = field(init=False, repr=False)
|
|
anchors: tuple[str, ...]
|
|
source_discovery_strategy: SourceDiscoveryStrategy
|
|
lexical_check_factory: LexicalCheckFactory
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
inner: CodeParser,
|
|
anchors: Iterable[str],
|
|
source_discovery_strategy: SourceDiscoveryStrategy,
|
|
lexical_check_factory: LexicalCheckFactory | None = None,
|
|
) -> None:
|
|
"""Initialize one strict parser wrapper with injected anchors, strategy, and check factory."""
|
|
|
|
self._inner = inner
|
|
self.anchors = tuple(anchors)
|
|
self.source_discovery_strategy = source_discovery_strategy
|
|
self.lexical_check_factory = lexical_check_factory or create_lexical_anchor_check
|
|
|
|
@property
|
|
def inner(self) -> CodeParser:
|
|
"""Return the wrapped parser."""
|
|
|
|
return self._inner
|
|
|
|
# Forward the execution adapter to callers that require it
|
|
@property
|
|
def repo_exec(self) -> RepoExec:
|
|
return self._inner.repo_exec
|
|
|
|
def __getstate__(self) -> dict[str, object]:
|
|
"""Return durable parser state while preserving strict parser configuration."""
|
|
|
|
return dict(self.__dict__)
|
|
|
|
def parse_file(
|
|
self,
|
|
file_name: str,
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> ClassMeta:
|
|
"""Parse one file path into ClassMeta metadata."""
|
|
|
|
source = self.repo_exec.read_text(Path(file_name), encoding="utf-8")
|
|
return self._parse_source(file_name, source, prompt, content_type)
|
|
|
|
def parse_stream(
|
|
self,
|
|
file_name: str,
|
|
stream: TextIO,
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> ClassMeta:
|
|
"""Parse one readable stream into ClassMeta metadata."""
|
|
|
|
source = stream.read()
|
|
return self._parse_source(file_name, source, prompt, content_type)
|
|
|
|
def parse_files(
|
|
self,
|
|
file_names: list[str],
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> list[ClassMeta]:
|
|
"""Parse multiple file paths into ClassMeta metadata."""
|
|
|
|
parsed: list[ClassMeta] = []
|
|
for file_name in file_names:
|
|
source = self.repo_exec.read_text(Path(file_name), encoding="utf-8")
|
|
parsed.append(self._parse_source(file_name, source, prompt, content_type))
|
|
return parsed
|
|
|
|
def _parse_source(
|
|
self,
|
|
file_name: str,
|
|
source: str,
|
|
prompt: str,
|
|
content_type: ContentType,
|
|
) -> ClassMeta:
|
|
"""Route one source string through strict parse handling."""
|
|
|
|
if not self.anchors:
|
|
self._log_decision(
|
|
file_name=file_name,
|
|
beats=0,
|
|
lexical_status="skipped",
|
|
lexical_result=False,
|
|
decision="skip",
|
|
reason="empty_anchors",
|
|
)
|
|
return self._skip_meta(file_name, content_type)
|
|
|
|
beats = self._searchable_beats(file_name, source)
|
|
if not beats:
|
|
self._log_decision(
|
|
file_name=file_name,
|
|
beats=0,
|
|
lexical_status="skipped",
|
|
lexical_result=False,
|
|
decision="skip",
|
|
reason="empty_searchable_beats",
|
|
)
|
|
return self._skip_meta(file_name, content_type)
|
|
|
|
lexical_check = self.lexical_check_factory(self.anchors, beats)
|
|
lexical_result = lexical_check()
|
|
decision = "forward" if lexical_result else "skip"
|
|
self._log_decision(
|
|
file_name=file_name,
|
|
beats=len(beats),
|
|
lexical_status="complete" if not lexical_check.error else "error",
|
|
lexical_result=lexical_result,
|
|
decision=decision,
|
|
lexical_check=lexical_check,
|
|
)
|
|
if lexical_result:
|
|
return self.inner.parse_stream(file_name, StringIO(source), prompt, content_type)
|
|
return self._skip_meta(file_name, content_type)
|
|
|
|
def _searchable_beats(self, file_name: str, source: str) -> tuple[str, ...]:
|
|
"""Return language-strategy searchable beats while failing closed on extraction errors."""
|
|
|
|
try:
|
|
beats = searchable_beat_texts(
|
|
self.source_discovery_strategy.searchable_beats(file_name, source)
|
|
)
|
|
except Exception as error:
|
|
logger.exception(
|
|
"Prefilter scope=strict file=%s beat_extraction=error error=%s: %s "
|
|
"semantic=not_applicable decision=skip",
|
|
file_name,
|
|
type(error).__name__,
|
|
error,
|
|
)
|
|
return ()
|
|
return beats
|
|
|
|
def _skip_meta(self, file_name: str, content_type: ContentType) -> ClassMeta:
|
|
"""Return conservative metadata when the prefilter decides to skip parsing."""
|
|
|
|
return ClassMeta(
|
|
location=str(file_name),
|
|
summary=f"Search prefilter skipped advisory parse for {file_name}.",
|
|
relevance=[],
|
|
relevant_symbols={},
|
|
content_type=content_type,
|
|
)
|
|
|
|
def _log_decision(
|
|
self,
|
|
*,
|
|
file_name: str,
|
|
beats: int,
|
|
lexical_status: str,
|
|
lexical_result: bool,
|
|
decision: str,
|
|
reason: str = "",
|
|
lexical_check: LexicalAnchorCheck | None = None,
|
|
) -> None:
|
|
"""Log one structured strict prefilter decision."""
|
|
|
|
best_anchor = lexical_check.best_anchor if lexical_check is not None else ""
|
|
best_candidate = lexical_check.best_candidate if lexical_check is not None else ""
|
|
edit_distance = lexical_check.edit_distance if lexical_check is not None else None
|
|
allowed_edit_distance = lexical_check.allowed_edit_distance if lexical_check is not None else None
|
|
comparison_count = lexical_check.comparison_count if lexical_check is not None else 0
|
|
check_reason = lexical_check.reason if lexical_check is not None else ""
|
|
error = lexical_check.error if lexical_check is not None else ""
|
|
|
|
logger.info(
|
|
"Prefilter scope=strict file=%s anchors=%s beats=%s lexical=%s "
|
|
"lexical_status=%s lexical_anchor=%s lexical_candidate=%s edit_distance=%s "
|
|
"allowed_distance=%s comparisons=%s semantic=not_applicable reason=%s "
|
|
"error=%s decision=%s",
|
|
file_name,
|
|
len(self.anchors),
|
|
beats,
|
|
lexical_result,
|
|
lexical_status,
|
|
best_anchor or "none",
|
|
best_candidate or "none",
|
|
edit_distance if edit_distance is not None else "none",
|
|
allowed_edit_distance if allowed_edit_distance is not None else "none",
|
|
comparison_count,
|
|
reason or check_reason or "none",
|
|
error or "none",
|
|
decision,
|
|
)
|