"""EIN Application — IRS SS-4 Online Automation. Automates the IRS online EIN application at: https://sa.www4.irs.gov/modiein/individual/index.jsp IMPORTANT: IRS online EIN is only available: Monday – Friday, 7:00 AM – 10:00 PM Eastern Time The handler checks availability before attempting. If outside hours, it queues for the next available window. Flow: 1. Check if within IRS business hours 2. Navigate to IRS EIN online application 3. Fill entity type, state, responsible party info 4. Submit and capture the EIN assignment 5. Store EIN in order intake_data + create admin todo Intake data needed: - entity_type: LLC, Corporation, Partnership, Sole Proprietor - entity_name: legal name of the entity - state: state of formation - responsible_party_name: full name of responsible party - responsible_party_ssn: SSN or ITIN (for identity) - address: street, city, state, zip - phone: contact phone """ from __future__ import annotations import asyncio import json import logging import os from datetime import datetime, timezone, timedelta from pathlib import Path LOG = logging.getLogger("workers.services.ein_application") SCREENSHOTS_DIR = Path(os.getenv("SCREENSHOTS_DIR", "/tmp/ein-screenshots")) SCREENSHOTS_DIR.mkdir(parents=True, exist_ok=True) IRS_EIN_URL = "https://sa.www4.irs.gov/modiein/individual/index.jsp" # IRS business hours: Mon-Fri 7am-10pm Eastern EASTERN_OFFSET = timedelta(hours=-5) # EST (EDT would be -4) def is_irs_available() -> bool: """Check if IRS online EIN is currently available (Mon-Fri 7am-10pm ET).""" now_utc = datetime.now(timezone.utc) # Use ET (approximate — doesn't handle DST precisely, but close enough) # EDT = UTC-4, EST = UTC-5. Use -4 (EDT) during summer months. month = now_utc.month offset = timedelta(hours=-4) if 3 <= month <= 11 else timedelta(hours=-5) now_et = now_utc + offset # Check day of week (0=Monday, 6=Sunday) if now_et.weekday() >= 5: # Saturday or Sunday return False # Check time (7am - 10pm) hour = now_et.hour return 7 <= hour < 22 def next_available_window() -> datetime: """Calculate the next available IRS window.""" now_utc = datetime.now(timezone.utc) month = now_utc.month offset = timedelta(hours=-4) if 3 <= month <= 11 else timedelta(hours=-5) now_et = now_utc + offset # If it's a weekday before 10pm, next window is 7am today or tomorrow if now_et.weekday() < 5 and now_et.hour < 22: if now_et.hour < 7: # Today at 7am ET target = now_et.replace(hour=7, minute=0, second=0, microsecond=0) else: # Already in window return now_utc else: # Next Monday at 7am ET if weekend, or tomorrow 7am if weekday after 10pm days_ahead = 1 next_day = now_et + timedelta(days=1) while next_day.weekday() >= 5: next_day += timedelta(days=1) days_ahead += 1 target = next_day.replace(hour=7, minute=0, second=0, microsecond=0) return target - offset # Convert back to UTC class EINApplicationHandler: """Handle EIN application orders.""" SERVICE_SLUG = "ein-application" SERVICE_NAME = "EIN Application (IRS SS-4)" async def process(self, order_data: dict) -> list[str]: """Entry point called by job_server.""" order_number = order_data.get("order_number", order_data.get("name", "")) return await self.handle(order_data, order_number) async def handle(self, order_data: dict, order_number: str) -> list[str]: """Process an EIN application order.""" LOG.info("[%s] Processing EIN application", order_number) intake = order_data.get("intake_data") or {} if isinstance(intake, str): intake = json.loads(intake) # Check IRS availability if not is_irs_available(): next_window = next_available_window() LOG.info("[%s] IRS offline — next window at %s UTC", order_number, next_window.isoformat()) # Create admin todo to process during business hours self._create_todo( order_number, intake, title=f"EIN Application QUEUED — {intake.get('entity_name', 'Unknown')}", description=( f"IRS online EIN not available (Mon-Fri 7am-10pm ET only).\n" f"Next available: {next_window.strftime('%A %I:%M %p ET')}.\n" f"Will auto-retry or process manually." ), priority="normal", ) return [] # Attempt automated filing is_prod = os.environ.get("NODE_ENV") == "production" or os.environ.get("ENV") == "production" if not is_prod: LOG.info("[%s] DEV MODE — skipping IRS EIN submission", order_number) self._create_todo( order_number, intake, title=f"EIN Application (DEV) — {intake.get('entity_name', 'Unknown')}", description="DEV MODE — IRS submission skipped.", priority="low", ) return [] # TODO: Playwright automation of IRS EIN form # For now, create admin todo for manual processing self._create_todo( order_number, intake, title=f"EIN Application — {intake.get('entity_name', 'Unknown')}", description=( f"Apply for EIN via IRS online (Mon-Fri 7am-10pm ET).\n" f"URL: {IRS_EIN_URL}\n" f"Entity: {intake.get('entity_name', 'N/A')}\n" f"Type: {intake.get('entity_type', 'LLC')}\n" f"State: {intake.get('formation_state', intake.get('state', 'N/A'))}\n" f"Responsible party: {intake.get('signer_name', 'N/A')}\n\n" f"EIN is issued immediately upon completion.\n" f"Update the order intake_data with the EIN once received." ), priority="high", ) return [] def _create_todo(self, order_number, intake, title, description, priority="normal"): """Create admin todo.""" try: import psycopg2 conn = psycopg2.connect(os.environ.get("DATABASE_URL", "")) with conn.cursor() as cur: cur.execute(""" INSERT INTO admin_todos ( title, category, priority, order_number, service_slug, description, data, status ) VALUES (%s, %s, %s, %s, %s, %s, %s, 'pending') """, ( title, "filing", priority, order_number, self.SERVICE_SLUG, description, json.dumps(intake), )) conn.commit() conn.close() except Exception as exc: LOG.error("[%s] Failed to create EIN todo: %s", order_number, exc)