Rxrohans/PayLens-Dev
0
1"""2DateTime Parser for PayLens3Converts temporal references in queries to concrete dates for accurate web searches.4 5Author: Rohan Singh6Created: March 20267"""8 9import re10from datetime import datetime, timedelta11from typing import Dict, Optional, Tuple, List12import logging13 14logger = logging.getLogger(__name__)15 16 17class DateTimeParser:18 """Parse temporal references and convert them to concrete dates."""19 20 # Temporal patterns to detect (order matters for multi-word phrases)21 TEMPORAL_PATTERNS = {22 # Multi-word patterns first (to avoid partial matches)23 'last 30 days': (-30, 'days'),24 'last 7 days': (-7, 'days'),25 'last 3 days': (-3, 'days'),26 'last 48 hours': (-2, 'days'),27 'last 24 hours': (-1, 'days'),28 'last week': (-7, 'days'),29 'last month': (-1, 'months'),30 'last year': (-1, 'years'),31 'this week': (0, 'weeks'),32 'this month': (0, 'months'),33 'this year': (0, 'years'),34 35 # Single word patterns36 'today': (0, 'days'),37 'yesterday': (-1, 'days'),38 'tomorrow': (1, 'days'),39 40 # Weekday references41 'monday': ('weekday', 0),42 'tuesday': ('weekday', 1),43 'wednesday': ('weekday', 2),44 'thursday': ('weekday', 3),45 'friday': ('weekday', 4),46 'saturday': ('weekday', 5),47 'sunday': ('weekday', 6),48 49 # Recent time periods50 'recent': (-3, 'days'),51 'recently': (-3, 'days'),52 'latest': (0, 'days'),53 }54 55 def __init__(self):56 self.today = datetime.now()57 logger.info(f"DateTimeParser initialized with current date: {self.today.strftime('%Y-%m-%d')}")58 59 def detect_temporal_references(self, query: str) -> List[str]:60 """61 Detect temporal references in query.62 63 Args:64 query: User query string65 66 Returns:67 List of detected temporal keywords (sorted by position in query)68 """69 query_lower = query.lower()70 detected = []71 72 # Handle common variations before pattern matching73 # Map variations to canonical forms74 variations = {75 r"\btodays?\b": "today", # todays, today's → today76 r"\btoday'?s\b": "today", # today's → today77 r"\bcurrent\s+(?:rate|price|exchange|fee)": "today", # current rate → today78 r"\bpresent\s+(?:rate|price)": "today", # present rate → today79 r"\bnow\b": "today", # now → today (in financial context)80 r"\byesterdays?\b": "yesterday", # yesterdays → yesterday81 r"\byesterday'?s\b": "yesterday",82 r"\blast\s+weeks?\b": "last week",83 r"\bprevious\s+week": "last week",84 }85 86 # Normalize query by replacing variations87 normalized_query = query_lower88 for pattern, canonical in variations.items():89 if re.search(pattern, normalized_query):90 normalized_query = re.sub(pattern, canonical, normalized_query)91 logger.debug(f"Normalized '{pattern}' to '{canonical}'")92 93 # Sort patterns by length (longest first) to match multi-word phrases first94 sorted_patterns = sorted(95 self.TEMPORAL_PATTERNS.keys(),96 key=len,97 reverse=True98 )99 100 for pattern in sorted_patterns:101 # Use word boundaries to avoid partial matches102 pattern_regex = r'\b' + re.escape(pattern) + r'\b'103 if re.search(pattern_regex, normalized_query):104 if pattern not in detected: # Avoid duplicates105 detected.append(pattern)106 107 if detected:108 logger.info(f"Detected temporal references: {detected}")109 110 return detected111 112 def resolve_to_date(self, temporal_ref: str) -> Optional[str]:113 """114 Convert temporal reference to concrete date string.115 116 Args:117 temporal_ref: Temporal keyword (e.g., 'yesterday', 'last week')118 119 Returns:120 Date string in 'YYYY-MM-DD' format or None if not recognized121 """122 temporal_ref_lower = temporal_ref.lower()123 124 if temporal_ref_lower not in self.TEMPORAL_PATTERNS:125 logger.warning(f"Unknown temporal reference: {temporal_ref}")126 return None127 128 pattern_info = self.TEMPORAL_PATTERNS[temporal_ref_lower]129 130 # Handle weekday references131 if isinstance(pattern_info, tuple) and pattern_info[0] == 'weekday':132 target_weekday = pattern_info[1]133 current_weekday = self.today.weekday()134 135 # Calculate days back to last occurrence136 days_back = (current_weekday - target_weekday) % 7137 if days_back == 0:138 days_back = 7 # Go to last week if today is that day139 140 target_date = self.today - timedelta(days=days_back)141 date_str = target_date.strftime('%Y-%m-%d')142 logger.debug(f"Resolved '{temporal_ref}' to {date_str} (weekday reference)")143 return date_str144 145 # Handle relative time periods146 delta_value, delta_unit = pattern_info147 148 if delta_unit == 'days':149 target_date = self.today + timedelta(days=delta_value)150 elif delta_unit == 'weeks':151 target_date = self.today + timedelta(weeks=delta_value)152 elif delta_unit == 'months':153 # Approximate month as 30 days154 target_date = self.today + timedelta(days=delta_value * 30)155 elif delta_unit == 'years':156 # Approximate year as 365 days157 target_date = self.today + timedelta(days=delta_value * 365)158 else:159 logger.warning(f"Unknown delta unit: {delta_unit}")160 return None161 162 date_str = target_date.strftime('%Y-%m-%d')163 logger.debug(f"Resolved '{temporal_ref}' to {date_str} ({delta_value} {delta_unit})")164 return date_str165 166 def get_date_range(self, temporal_ref: str) -> Optional[Tuple[str, str]]:167 """168 Get date range for temporal reference.169 Useful for queries that need a time window.170 171 Args:172 temporal_ref: Temporal keyword173 174 Returns:175 Tuple of (start_date, end_date) in 'YYYY-MM-DD' format176 """177 temporal_ref_lower = temporal_ref.lower()178 179 if temporal_ref_lower == 'today':180 date_str = self.today.strftime('%Y-%m-%d')181 return (date_str, date_str)182 183 if temporal_ref_lower == 'this week':184 # Start of week (Monday)185 start = self.today - timedelta(days=self.today.weekday())186 # End of week (Sunday)187 end = start + timedelta(days=6)188 return (start.strftime('%Y-%m-%d'), end.strftime('%Y-%m-%d'))189 190 if temporal_ref_lower == 'this month':191 # Start of month192 start = self.today.replace(day=1)193 # Last day of month194 if self.today.month == 12:195 end = self.today.replace(day=31)196 else:197 end = (self.today.replace(month=self.today.month + 1, day=1) 198 - timedelta(days=1))199 return (start.strftime('%Y-%m-%d'), end.strftime('%Y-%m-%d'))200 201 if temporal_ref_lower == 'this year':202 start = self.today.replace(month=1, day=1)203 end = self.today.replace(month=12, day=31)204 return (start.strftime('%Y-%m-%d'), end.strftime('%Y-%m-%d'))205 206 # For other references, use single date207 date_str = self.resolve_to_date(temporal_ref_lower)208 if date_str:209 return (date_str, date_str)210 211 return None212 213 def augment_query(self, query: str) -> str:214 """215 Augment query with concrete date context for web search.216 217 Example:218 "What happened today with UPI?" 219 → "What happened today with UPI? [today = 2026-03-06]"220 221 Args:222 query: Original user query223 224 Returns:225 Augmented query with date context appended226 """227 detected = self.detect_temporal_references(query)228 229 if not detected:230 return query231 232 # Build date context233 date_contexts = []234 for ref in detected:235 date_str = self.resolve_to_date(ref)236 if date_str:237 date_contexts.append(f"{ref} = {date_str}")238 239 if date_contexts:240 context_str = " [" + ", ".join(date_contexts) + "]"241 augmented = query + context_str242 logger.info(f"Augmented query: '{query}' → '{augmented}'")243 return augmented244 245 return query246 247 def get_current_date_context(self) -> str:248 """249 Get current date context for LLM prompt.250 251 Returns:252 String with current date info formatted for LLM253 """254 weekday = self.today.strftime('%A')255 month = self.today.strftime('%B')256 day = self.today.day257 year = self.today.year258 259 # Add ordinal suffix (1st, 2nd, 3rd, 4th, etc.)260 if 10 <= day <= 20:261 suffix = 'th'262 else:263 suffix = {1: 'st', 2: 'nd', 3: 'rd'}.get(day % 10, 'th')264 265 return f"Current date: {self.today.strftime('%Y-%m-%d')} ({weekday}, {month} {day}{suffix}, {year})"266 267 def is_temporal_query(self, query: str) -> bool:268 """269 Quick check if query contains temporal references.270 271 Args:272 query: User query string273 274 Returns:275 Boolean indicating if temporal references found276 """277 return len(self.detect_temporal_references(query)) > 0278 279 280def parse_datetime_query(query: str) -> Dict[str, any]:281 """282 Utility function to parse query and extract datetime info.283 284 This is the main entry point for other modules.285 286 Args:287 query: User query string288 289 Returns:290 Dictionary with:291 - original_query: Original query292 - augmented_query: Query with date context for web search293 - temporal_refs: List of detected temporal references294 - date_context: Current date context string for LLM295 - has_temporal: Boolean indicating if temporal refs found296 297 Example:298 >>> result = parse_datetime_query("What happened today with UPI?")299 >>> print(result['has_temporal'])300 True301 >>> print(result['augmented_query'])302 "What happened today with UPI? [today = 2026-03-06]"303 """304 parser = DateTimeParser()305 detected = parser.detect_temporal_references(query)306 307 return {308 'original_query': query,309 'augmented_query': parser.augment_query(query),310 'temporal_refs': detected,311 'date_context': parser.get_current_date_context(),312 'has_temporal': len(detected) > 0313 }314 315 316# For quick testing317if __name__ == "__main__":318 # Set up logging319 logging.basicConfig(level=logging.INFO)320 321 # Test queries322 test_queries = [323 "What happened today with UPI?",324 "What changed last week with NEFT?",325 "Any updates yesterday on RBI policies?",326 "What are the latest digital payment trends?",327 "Show me news from last 7 days",328 "What are RTGS charges?" # No temporal reference329 ]330 331 print("=" * 80)332 print("DateTime Parser Test")333 print("=" * 80)334 335 parser = DateTimeParser()336 print(f"\n{parser.get_current_date_context()}\n")337 338 for query in test_queries:339 print(f"\nQuery: {query}")340 result = parse_datetime_query(query)341 342 if result['has_temporal']:343 print(f" ✓ Temporal refs: {result['temporal_refs']}")344 print(f" → Augmented: {result['augmented_query']}")345 else:346 print(f" ✗ No temporal references detected")