# retoor import logging import re from datetime import datetime from typing import Any, Optional logger = logging.getLogger(__name__) DATE_LABEL = "Current date" _MONTH_NAMES = ( "january|february|march|april|may|june|july|august|september|october|" "november|december|jan|feb|mar|apr|jun|jul|aug|sep|sept|oct|nov|dec" ) _DATE_PATTERNS: tuple[re.Pattern[str], ...] = ( re.compile(r"\b\d{4}-\d{1,2}-\d{1,2}\b"), re.compile(r"\b\d{1,2}/\d{1,2}/\d{4}\b"), re.compile(r"\b\d{1,2}-\d{1,2}-\d{4}\b"), re.compile(r"\b\d{1,2}\.\d{1,2}\.\d{4}\b"), re.compile( rf"\b\d{{1,2}}\s+(?:{_MONTH_NAMES})\.?\s+\d{{4}}\b", re.IGNORECASE, ), re.compile( rf"\b(?:{_MONTH_NAMES})\.?\s+\d{{1,2}}(?:st|nd|rd|th)?,?\s+\d{{4}}\b", re.IGNORECASE, ), ) def contains_date(text: str) -> bool: return any(pattern.search(text) for pattern in _DATE_PATTERNS) def current_date_eu() -> str: return datetime.now().strftime("%d/%m/%Y") def system_message_text(content: Any) -> str: if isinstance(content, str): return content if isinstance(content, list): parts: list[str] = [] for block in content: if isinstance(block, dict) and block.get("type") == "text": value = block.get("text") if isinstance(value, str): parts.append(value) return "\n".join(parts) return "" def compose_system_content( preamble: str, client_content: Any, date_eu: str ) -> str: preamble_text = (preamble or "").strip() client_text = system_message_text(client_content) sections: list[str] = [] if preamble_text: sections.append(preamble_text) combined_for_date = f"{preamble_text}\n{client_text}" if not contains_date(combined_for_date): sections.append(f"{DATE_LABEL}: {date_eu}") if client_text: sections.append(client_text) return "\n\n".join(sections) def apply_system_directives( messages: list, preamble: str, date_eu: Optional[str] = None ) -> list: date_value = date_eu or current_date_eu() preamble_text = (preamble or "").strip() result: list = list(messages) system_index = next( ( index for index, message in enumerate(result) if isinstance(message, dict) and message.get("role") == "system" ), None, ) if system_index is not None: original = result[system_index] composed = compose_system_content( preamble_text, original.get("content"), date_value ) updated = dict(original) updated["content"] = composed result[system_index] = updated logger.debug( "Gateway composed existing system message (preamble=%s, date=%s)", bool(preamble_text), date_value, ) return result if not preamble_text: return result composed = compose_system_content(preamble_text, "", date_value) if not composed.strip(): return result result.insert(0, {"role": "system", "content": composed}) logger.info( "Gateway injected a system message (preamble=%s, date=%s)", bool(preamble_text), date_value, ) return result