""" Email parser for priority release check. Ported from priority-release-check/src/utils/parseEmail.ts """ import re from dataclasses import dataclass, field from datetime import datetime, timedelta @dataclass class Track: artist: str title: str label: str = "" notes: str = "" @dataclass class Album: artist: str title: str label: str = "" notes: str = "" @dataclass class ParsedEmail: release_date: str # YYYY-MM-DD tracks: list = field(default_factory=list) albums: list = field(default_factory=list) def extract_date(text: str) -> str: """Extract release date from email text. Scans first 20 lines for 'M/DD New Releases' or 'M/DD' pattern. Infers year from current date. """ lines = text.split("\n") match = None for line in lines[:20]: match = re.search(r"(\d{1,2})/(\d{1,2})\s+New\s+Releases", line, re.IGNORECASE) if match: break if not match: for line in lines[:20]: match = re.match(r"(\d{1,2})/(\d{1,2})\b", line) if match: break if not match: return "" month = int(match.group(1)) day = int(match.group(2)) now = datetime.now() year = now.year # If parsed date is more than 6 months in the future, assume last year try: candidate = datetime(year, month, day) if (candidate - now) > timedelta(days=180): year -= 1 except ValueError: return "" return f"{year}-{month:02d}-{day:02d}" def parse_entry(line: str): """Parse an artist/title line. Returns (artist, title, notes) or None.""" trimmed = line.strip() if not trimmed or trimmed == "NONE" or trimmed.startswith("+"): return None notes = "" working = trimmed # Strip "- Distro: ..." suffix distro_match = re.search(r"\s*-\s*(?:.+?\s*-\s*)?Distro:\s*(.+)$", working, re.IGNORECASE) if distro_match: notes = f"Distro: {distro_match.group(1).strip()}" working = working[: distro_match.start()].strip() # Known notes patterns notes_patterns = [ re.compile(r"\s+(Tracking Starting \w+)$", re.IGNORECASE), re.compile(r"\s+(Reporting \w+.*)$", re.IGNORECASE), ] for pattern in notes_patterns: m = pattern.search(working) if m: notes = m.group(1).strip() working = working[: m.start()].strip() # Try: Artist \u2013/\u2014/- "Title" m = re.match(r'^(.+?)\s*[\u2013\u2014-]\s*["\u201C](.+?)["\u201D]\s*$', working) if m: return (m.group(1).strip(), m.group(2).strip(), notes) # Try: Artist "Title" (no dash) m = re.match(r'^(.+?)\s+["\u201C](.+?)["\u201D]\s*$', working) if m: return (m.group(1).strip(), m.group(2).strip(), notes) # Try: Artist \u2013/\u2014/- Title (no quotes) m = re.match(r"^(.+?)\s*[\u2013\u2014-]\s+(.+)$", working) if m: return (m.group(1).strip(), m.group(2).strip(), notes) return None def is_label_header(line: str): """Check if line is a label header (e.g., 'Columbia:', 'RCA:'). Returns the label name or None. """ trimmed = line.strip() if "\u201C" in trimmed or '"' in trimmed: return None if "\u2013" in trimmed or "\u2014" in trimmed: return None m = re.match(r"^([^:]+):\s*(.*)$", trimmed) if m and len(m.group(1).strip()) <= 40: trailing = m.group(2).strip().upper() if not trailing or trailing == "NONE": return m.group(1).strip() return None def is_section_header(line: str): """Check if line is 'TRACKS' or 'ALBUMS'. Returns section name or None.""" upper = line.strip().upper() if upper in ("TRACKS", "TRACK"): return "tracks" if upper in ("ALBUMS", "ALBUM"): return "albums" return None def parse_email(text: str) -> ParsedEmail: """Parse full email text into structured tracks and albums.""" lines = text.split("\n") release_date = extract_date(text) tracks = [] albums = [] current_section = None current_label = "" for line in lines: trimmed = line.strip() if not trimmed: continue # Check for section headers section = is_section_header(trimmed) if section: current_section = section current_label = "" continue # Skip known non-entry lines skip_prefixes = [ "On Duty:", "Weekend reporting", "Priority titles", "Tracking Starting", "Reporting ", ] if any(trimmed.startswith(p) for p in skip_prefixes): continue if trimmed.startswith("Tracks:") and "," in trimmed: continue if trimmed.startswith("Albums:") and "," in trimmed: continue if trimmed.startswith("+"): continue if current_section is None: continue # Check for label header label = is_label_header(trimmed) if label: label_match = re.match(r"^([^:]+):\s*(.*)$", trimmed) trailing = label_match.group(2).strip().upper() if label_match else "" if trailing == "NONE": current_label = "" else: current_label = label continue if trimmed.upper() == "NONE": continue # Try to parse as an entry entry = parse_entry(trimmed) if entry: artist, title, notes = entry item_cls = Track if current_section == "tracks" else Album item = item_cls(artist=artist, title=title, label=current_label, notes=notes) if current_section == "tracks": tracks.append(item) else: albums.append(item) return ParsedEmail(release_date=release_date, tracks=tracks, albums=albums)