chore: archive previous Aider worker
This commit is contained in:
Christian Meyer
307
old_2026-08-31/src/lai_automation/removal.py
Normal file
307
old_2026-08-31/src/lai_automation/removal.py
Normal file
@@ -0,0 +1,307 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import unicodedata
|
||||
from dataclasses import dataclass
|
||||
from difflib import SequenceMatcher
|
||||
from pathlib import Path
|
||||
|
||||
from .models import FailureReport
|
||||
|
||||
_REMOVAL_INTENT = re.compile(
|
||||
r"\b(?:entfern\w*|lösch\w*|loesch\w*|weglass\w*|streichen|"
|
||||
r"remove\w*|delet\w*|omit\w*|drop\w*|ohne|without)\b|"
|
||||
r"\b(?:darf|soll|muss)\b.{0,50}\b(?:nicht|nie|niemals)\b|"
|
||||
r"\b(?:must|should)\s+not\b|\bdo\s+not\b",
|
||||
re.IGNORECASE | re.DOTALL,
|
||||
)
|
||||
_NEGATED_REMOVAL = re.compile(
|
||||
r"\b(?:nicht|nie|niemals)\s+(?:entfern\w*|lösch\w*|loesch\w*|weglass\w*)\b|"
|
||||
r"\b(?:do\s+not|don't|never)\s+(?:remove\w*|delet\w*|omit\w*|drop\w*)\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_ONLY_REMOVAL = re.compile(
|
||||
r"\b(?:einfach\s+)?nur\b.{0,120}\b(?:entfern\w*|lösch\w*|loesch\w*|"
|
||||
r"weglass\w*|streichen)\b|"
|
||||
r"\b(?:just|only)\b.{0,120}\b(?:remove\w*|delet\w*|omit\w*|drop\w*)\b",
|
||||
re.IGNORECASE | re.DOTALL,
|
||||
)
|
||||
_QUOTED_FRAGMENT = re.compile(
|
||||
r'"(?P<double>[^"\n]{1,500})"|'
|
||||
r"'(?P<single>[^'\n]{1,500})'|"
|
||||
r"`(?P<backtick>[^`\n]{1,500})`|"
|
||||
r"„(?P<german>[^”\n]{1,500})”"
|
||||
)
|
||||
_TECHNICAL_IDENTIFIER = re.compile(
|
||||
r"\b(?:[A-Za-z_][A-Za-z0-9_]*\.)?[A-Za-z_][A-Za-z0-9_]*_[A-Za-z0-9_]+\b"
|
||||
)
|
||||
_QUESTION_KEY = re.compile(r"['\"]question['\"]\s*=>", re.IGNORECASE)
|
||||
_SQL_HEREDOC = re.compile(
|
||||
r"['\"]sql['\"]\s*=>\s*<<<(?P<quote>['\"]?)(?P<label>[A-Za-z_][A-Za-z0-9_]*)"
|
||||
r"(?P=quote)[^\S\r\n]*\r?\n",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RemovalDirective:
|
||||
fragment: str
|
||||
origin: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ReferenceExample:
|
||||
question: str
|
||||
sql: str
|
||||
sql_start: int
|
||||
sql_end: int
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class AlreadyFixedResult:
|
||||
filename: str
|
||||
question: str
|
||||
match_score: float
|
||||
forbidden_fragments: tuple[str, ...]
|
||||
|
||||
|
||||
def removal_directives(report: FailureReport) -> tuple[RemovalDirective, ...]:
|
||||
expected = report.expected_behavior or ""
|
||||
if not expected.strip() or not _REMOVAL_INTENT.search(expected):
|
||||
return ()
|
||||
negated_removal = _NEGATED_REMOVAL.search(expected)
|
||||
if negated_removal:
|
||||
remaining = expected[:negated_removal.start()] + expected[negated_removal.end():]
|
||||
if not _REMOVAL_INTENT.search(remaining):
|
||||
return ()
|
||||
|
||||
observed_sql = report.sql_query or ""
|
||||
directives: list[RemovalDirective] = []
|
||||
for match in _QUOTED_FRAGMENT.finditer(expected):
|
||||
candidate = next(value for value in match.groupdict().values() if value is not None)
|
||||
candidate = clean_fragment(candidate)
|
||||
if _looks_like_sql_fragment(candidate):
|
||||
directives.append(RemovalDirective(candidate, "quoted expected behavior"))
|
||||
|
||||
if not directives:
|
||||
for sentence in re.split(r"[\n.!?;]+", expected):
|
||||
if not _REMOVAL_INTENT.search(sentence):
|
||||
continue
|
||||
for identifier in _TECHNICAL_IDENTIFIER.findall(sentence):
|
||||
predicate = _predicate_containing(observed_sql, identifier)
|
||||
directives.append(RemovalDirective(predicate or identifier, "removal sentence"))
|
||||
|
||||
deduplicated: dict[str, RemovalDirective] = {}
|
||||
for directive in directives:
|
||||
key = normalize_sql(directive.fragment)
|
||||
if key:
|
||||
deduplicated.setdefault(key, directive)
|
||||
return tuple(deduplicated.values())
|
||||
|
||||
|
||||
def is_removal_only(report: FailureReport) -> bool:
|
||||
expected = report.expected_behavior or ""
|
||||
return bool(removal_directives(report) and _ONLY_REMOVAL.search(expected))
|
||||
|
||||
|
||||
def clean_fragment(fragment: str) -> str:
|
||||
result = re.sub(r"\s+", " ", fragment).strip().strip(";")
|
||||
result = re.sub(r"^(?:AND|OR|WHERE)\s+", "", result, flags=re.IGNORECASE)
|
||||
return result.strip()
|
||||
|
||||
|
||||
def normalize_sql(sql: str) -> str:
|
||||
return _canonical_sql(sql).strip(";").casefold()
|
||||
|
||||
|
||||
def sql_contains_fragment(sql: str, fragment: str) -> bool:
|
||||
normalized_fragment = normalize_sql(clean_fragment(fragment))
|
||||
if not normalized_fragment:
|
||||
return False
|
||||
return normalized_fragment in normalize_sql(sql)
|
||||
|
||||
|
||||
def remove_sql_fragment(sql: str, fragment: str) -> str:
|
||||
result = _canonical_sql(sql)
|
||||
target = _canonical_sql(clean_fragment(fragment))
|
||||
if not target:
|
||||
return result
|
||||
|
||||
while True:
|
||||
index = result.casefold().find(target.casefold())
|
||||
if index < 0:
|
||||
break
|
||||
before = result[:index]
|
||||
after = result[index + len(target):]
|
||||
leading_connector = re.search(r"\s+(?:AND|OR)\s*$", before, re.IGNORECASE)
|
||||
trailing_connector = re.match(r"^\s+(?:AND|OR)\s+", after, re.IGNORECASE)
|
||||
leading_where = re.search(r"\bWHERE\s*$", before, re.IGNORECASE)
|
||||
if leading_connector:
|
||||
result = before[:leading_connector.start()] + after
|
||||
elif trailing_connector:
|
||||
result = before + after[trailing_connector.end():]
|
||||
elif leading_where:
|
||||
result = before[:leading_where.start()] + after
|
||||
else:
|
||||
result = before + after
|
||||
result = re.sub(r"\s+", " ", result).strip()
|
||||
return result
|
||||
|
||||
|
||||
def _canonical_sql(sql: str) -> str:
|
||||
result = re.sub(r"\s+", " ", sql).strip()
|
||||
return re.sub(r"\s*([(),.=<>])\s*", r"\1", result)
|
||||
|
||||
|
||||
def extract_reference_examples(text: str) -> tuple[ReferenceExample, ...]:
|
||||
markers = list(_QUESTION_KEY.finditer(text))
|
||||
examples: list[ReferenceExample] = []
|
||||
for index, marker in enumerate(markers):
|
||||
block_end = markers[index + 1].start() if index + 1 < len(markers) else len(text)
|
||||
question = _php_string_after(text, marker.end(), block_end)
|
||||
if question is None:
|
||||
continue
|
||||
heredoc = _SQL_HEREDOC.search(text, marker.end(), block_end)
|
||||
if not heredoc:
|
||||
continue
|
||||
terminator = re.compile(
|
||||
rf"^{re.escape(heredoc.group('label'))}\s*,?\s*$",
|
||||
re.MULTILINE,
|
||||
).search(text, heredoc.end(), block_end)
|
||||
if not terminator:
|
||||
continue
|
||||
examples.append(ReferenceExample(
|
||||
question=question,
|
||||
sql=text[heredoc.end():terminator.start()].strip(),
|
||||
sql_start=heredoc.end(),
|
||||
sql_end=terminator.start(),
|
||||
))
|
||||
return tuple(examples)
|
||||
|
||||
|
||||
def find_matching_reference(
|
||||
text: str,
|
||||
prompt: str | None,
|
||||
) -> tuple[ReferenceExample | None, float]:
|
||||
if not prompt:
|
||||
return None, 0.0
|
||||
scored = sorted(
|
||||
((_question_score(prompt, example.question), example)
|
||||
for example in extract_reference_examples(text)),
|
||||
key=lambda item: item[0],
|
||||
reverse=True,
|
||||
)
|
||||
if not scored or scored[0][0] < 0.92:
|
||||
return None, 0.0
|
||||
if len(scored) > 1 and scored[1][0] >= scored[0][0] - 0.05:
|
||||
return None, 0.0
|
||||
return scored[0][1], scored[0][0]
|
||||
|
||||
|
||||
def detect_already_fixed(
|
||||
report: FailureReport,
|
||||
workspace: Path,
|
||||
routed_files: tuple[str, ...],
|
||||
) -> AlreadyFixedResult | None:
|
||||
directives = removal_directives(report)
|
||||
observed_sql = report.sql_query or ""
|
||||
if not is_removal_only(report) or not observed_sql:
|
||||
return None
|
||||
if not all(sql_contains_fragment(observed_sql, item.fragment) for item in directives):
|
||||
return None
|
||||
|
||||
matches: list[tuple[str, ReferenceExample, float]] = []
|
||||
for filename in routed_files:
|
||||
if filename.startswith("tests/"):
|
||||
continue
|
||||
path = workspace / filename
|
||||
if not path.is_file():
|
||||
continue
|
||||
example, score = find_matching_reference(path.read_text(encoding="utf-8"), report.prompt)
|
||||
if example:
|
||||
matches.append((filename, example, score))
|
||||
if len(matches) != 1:
|
||||
return None
|
||||
|
||||
filename, example, score = matches[0]
|
||||
if any(sql_contains_fragment(example.sql, item.fragment) for item in directives):
|
||||
return None
|
||||
return AlreadyFixedResult(
|
||||
filename=filename,
|
||||
question=example.question,
|
||||
match_score=score,
|
||||
forbidden_fragments=tuple(item.fragment for item in directives),
|
||||
)
|
||||
|
||||
|
||||
def without_reference_sql(text: str, example: ReferenceExample) -> str:
|
||||
return text[:example.sql_start] + "<REFERENCE_SQL>" + text[example.sql_end:]
|
||||
|
||||
|
||||
def normalize_source_layout(text: str) -> str:
|
||||
lines = [line.rstrip() for line in text.replace("\r\n", "\n").splitlines()]
|
||||
return "\n".join(lines).strip()
|
||||
|
||||
|
||||
def _looks_like_sql_fragment(value: str) -> bool:
|
||||
return bool(
|
||||
_TECHNICAL_IDENTIFIER.search(value)
|
||||
or re.search(r"\b(?:WHERE|AND|OR|JOIN|GROUP\s+BY|ORDER\s+BY)\b", value, re.IGNORECASE)
|
||||
or (re.search(r"[=<>]", value) and re.search(r"[A-Za-z_]", value))
|
||||
)
|
||||
|
||||
|
||||
def _predicate_containing(sql: str, identifier: str) -> str | None:
|
||||
clauses = re.finditer(
|
||||
r"\b(?:WHERE|AND|OR)\b\s*(?P<predicate>.*?)"
|
||||
r"(?=\s+\b(?:AND|OR|GROUP\s+BY|ORDER\s+BY|HAVING|LIMIT)\b|;|$)",
|
||||
sql,
|
||||
re.IGNORECASE | re.DOTALL,
|
||||
)
|
||||
for clause in clauses:
|
||||
predicate = clause.group("predicate").strip()
|
||||
if re.search(rf"(?<![A-Za-z0-9_]){re.escape(identifier)}(?![A-Za-z0-9_])", predicate, re.IGNORECASE):
|
||||
return predicate
|
||||
return None
|
||||
|
||||
|
||||
def _php_string_after(text: str, start: int, end: int) -> str | None:
|
||||
position = start
|
||||
while position < end and text[position].isspace():
|
||||
position += 1
|
||||
if position >= end or text[position] not in {"'", '"'}:
|
||||
return None
|
||||
quote = text[position]
|
||||
position += 1
|
||||
value: list[str] = []
|
||||
while position < end:
|
||||
character = text[position]
|
||||
if character == "\\" and position + 1 < end:
|
||||
value.append(text[position + 1])
|
||||
position += 2
|
||||
continue
|
||||
if character == quote:
|
||||
return "".join(value)
|
||||
value.append(character)
|
||||
position += 1
|
||||
return None
|
||||
|
||||
|
||||
def _normalize_question(value: str) -> str:
|
||||
normalized = unicodedata.normalize("NFKD", value.casefold().replace("ß", "ss"))
|
||||
normalized = "".join(character for character in normalized if not unicodedata.combining(character))
|
||||
return " ".join(re.findall(r"[a-z0-9]+", normalized))
|
||||
|
||||
|
||||
def _question_score(left: str, right: str) -> float:
|
||||
normalized_left = _normalize_question(left)
|
||||
normalized_right = _normalize_question(right)
|
||||
if normalized_left == normalized_right:
|
||||
return 1.0
|
||||
left_tokens = set(normalized_left.split())
|
||||
right_tokens = set(normalized_right.split())
|
||||
if min(len(left_tokens), len(right_tokens)) < 4:
|
||||
return 0.0
|
||||
overlap = len(left_tokens & right_tokens) / len(left_tokens | right_tokens)
|
||||
sequence = SequenceMatcher(None, normalized_left, normalized_right).ratio()
|
||||
return (0.7 * overlap) + (0.3 * sequence)
|
||||
Reference in New Issue
Block a user