Spaces:
Sleeping
Sleeping
File size: 2,822 Bytes
73608eb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 | """
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})"
|