181 lines
6.8 KiB
Python
181 lines
6.8 KiB
Python
"""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)
|