Spaces:
Sleeping
Sleeping
| """ | |
| Shared utilities for the Email Triage environment. | |
| Provides logging configuration, constants, and helper functions used | |
| across multiple modules. | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| import re | |
| from typing import Optional, Tuple | |
| from src.models import Action | |
| # ββ Logging ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| LOG_FORMAT = "%(asctime)s [%(levelname)s] %(name)s: %(message)s" | |
| LOG_DATE_FORMAT = "%Y-%m-%d %H:%M:%S" | |
| def setup_logging(level: int = logging.INFO) -> None: | |
| """Configure root logger with consistent format.""" | |
| logging.basicConfig( | |
| level=level, | |
| format=LOG_FORMAT, | |
| datefmt=LOG_DATE_FORMAT, | |
| ) | |
| # ββ Action Parsing βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| _MOVE_PATTERN = re.compile( | |
| r"move\s*\(\s*(\d+)\s*,\s*(['\"]?)(\w+)\2\s*\)", re.IGNORECASE | |
| ) | |
| _FLAG_PATTERN = re.compile(r"flag\s*\(\s*(\d+)\s*\)", re.IGNORECASE) | |
| _DELETE_PATTERN = re.compile(r"delete\s*\(\s*(\d+)\s*\)", re.IGNORECASE) | |
| _SPAM_PATTERN = re.compile(r"mark_spam\s*\(\s*(\d+)\s*\)", re.IGNORECASE) | |
| _SNOOZE_PATTERN = re.compile(r"snooze\s*\(\s*(\d+)\s*\)", re.IGNORECASE) | |
| def parse_action_string(action_str: str) -> Optional[Action]: | |
| """Parse a text action like ``move(3, finance)`` into an Action model. | |
| Supports formats: | |
| - move(id, folder) | |
| - flag(id) | |
| - delete(id) | |
| - mark_spam(id) | |
| - snooze(id) | |
| Args: | |
| action_str: Raw text from the LLM. | |
| Returns: | |
| An Action instance, or None if parsing fails. | |
| """ | |
| action_str = action_str.strip() | |
| m = _MOVE_PATTERN.search(action_str) | |
| if m: | |
| return Action( | |
| action_type="move", | |
| email_id=int(m.group(1)), | |
| target_folder=m.group(3).lower(), | |
| ) | |
| m = _FLAG_PATTERN.search(action_str) | |
| if m: | |
| return Action(action_type="flag", email_id=int(m.group(1))) | |
| m = _DELETE_PATTERN.search(action_str) | |
| if m: | |
| return Action(action_type="delete", email_id=int(m.group(1))) | |
| m = _SPAM_PATTERN.search(action_str) | |
| if m: | |
| return Action(action_type="mark_spam", email_id=int(m.group(1))) | |
| m = _SNOOZE_PATTERN.search(action_str) | |
| if m: | |
| return Action(action_type="snooze", email_id=int(m.group(1))) | |
| return None | |
| def format_action_for_log(action: Action) -> str: | |
| """Format an action into a compact log string.""" | |
| if action.action_type == "move": | |
| return f"move({action.email_id}, {action.target_folder})" | |
| return f"{action.action_type}({action.email_id})" | |