Spencer
4c9fecda16
Webhook listener that monitors Gitea repos for BLIGHT: triggers in markdown files, processes them via Gemini 2.5 Flash-Lite, and writes results back in-place. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
78 lines
2.3 KiB
Python
78 lines
2.3 KiB
Python
import re
|
|
import time
|
|
import logging
|
|
from ai import GeminiProvider
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
TRIGGER_PATTERN = re.compile(r"^BLIGHT:\s+(.+)$", re.MULTILINE)
|
|
FAILED_TEMPLATE = "<!-- BLIGHT_FAILED: {instruction} -->"
|
|
|
|
_MAX_RETRIES = 3
|
|
_RETRY_DELAYS = [1, 2, 4] # seconds between attempts
|
|
|
|
_provider = GeminiProvider()
|
|
|
|
|
|
def process_document(content: str) -> tuple[str, bool]:
|
|
"""Scan content for BLIGHT triggers and process each one.
|
|
|
|
Returns:
|
|
(updated_content, changed) where changed is True if any triggers
|
|
were found and the content was modified.
|
|
"""
|
|
triggers = list(TRIGGER_PATTERN.finditer(content))
|
|
if not triggers:
|
|
return content, False
|
|
|
|
# Process triggers one by one. After each replacement the string length
|
|
# may change, so we re-search on the updated content each iteration.
|
|
changed = False
|
|
for _ in range(len(triggers)):
|
|
match = TRIGGER_PATTERN.search(content)
|
|
if not match:
|
|
break
|
|
|
|
instruction = match.group(1).strip()
|
|
trigger_line = match.group(0)
|
|
logger.info("Processing trigger: %s", instruction)
|
|
|
|
replacement = _call_with_retry(content, instruction)
|
|
content = content[:match.start()] + replacement + content[match.end():]
|
|
changed = True
|
|
|
|
return content, changed
|
|
|
|
|
|
def _call_with_retry(document: str, instruction: str) -> str:
|
|
"""Call the AI provider with up to _MAX_RETRIES attempts.
|
|
|
|
Returns the AI response on success, or a BLIGHT_FAILED comment on
|
|
exhausted retries.
|
|
"""
|
|
last_error: Exception | None = None
|
|
for attempt in range(_MAX_RETRIES):
|
|
try:
|
|
return _provider.complete(document, instruction)
|
|
except Exception as exc:
|
|
last_error = exc
|
|
if attempt < _MAX_RETRIES - 1:
|
|
delay = _RETRY_DELAYS[attempt]
|
|
logger.warning(
|
|
"Attempt %d/%d failed for instruction %r: %s — retrying in %ds",
|
|
attempt + 1,
|
|
_MAX_RETRIES,
|
|
instruction,
|
|
exc,
|
|
delay,
|
|
)
|
|
time.sleep(delay)
|
|
|
|
logger.error(
|
|
"All %d attempts failed for instruction %r: %s",
|
|
_MAX_RETRIES,
|
|
instruction,
|
|
last_error,
|
|
)
|
|
return FAILED_TEMPLATE.format(instruction=instruction)
|