Includes: API (Express/TypeScript), Astro site, Python workers, document generators, FCC compliance tools, Canada CRTC formation, Ansible infrastructure, and deployment scripts. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
178 lines
5.6 KiB
Python
178 lines
5.6 KiB
Python
"""
|
|
name_search.py — Multi-state business name availability search coordinator.
|
|
|
|
Loads the appropriate state adapter and performs name availability searches.
|
|
Supports searching a single state or multiple states in parallel.
|
|
|
|
Usage:
|
|
python -m formation.name_search "My Business LLC" WY
|
|
python -m formation.name_search "My Business LLC" WY,NV,NM,TX
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import sys
|
|
import time
|
|
from dataclasses import asdict
|
|
|
|
from .base import NameSearchResult
|
|
from .states import get_adapter, STATES
|
|
|
|
LOG = logging.getLogger("formation.name_search")
|
|
|
|
|
|
async def search_name(name: str, state_code: str) -> NameSearchResult:
|
|
"""
|
|
Search for business name availability in a single state.
|
|
|
|
Loads the state adapter, launches a browser session, performs the search,
|
|
and returns a NameSearchResult.
|
|
|
|
Args:
|
|
name: The business name to search (e.g. "Acme Holdings LLC").
|
|
state_code: Two-letter state code (e.g. "WY").
|
|
|
|
Returns:
|
|
NameSearchResult with availability info.
|
|
"""
|
|
code = state_code.upper()
|
|
if code not in STATES:
|
|
return NameSearchResult(
|
|
available=False,
|
|
searched_name=name,
|
|
state_code=code,
|
|
raw_response=f"Unknown state code: {code}",
|
|
)
|
|
|
|
LOG.info("Searching name '%s' in %s (%s)...", name, code, STATES[code]["name"])
|
|
adapter = get_adapter(code)
|
|
|
|
try:
|
|
await adapter.start_browser(headless=True)
|
|
result = await adapter.search_name(name)
|
|
# Ensure state_code and searched_name are populated
|
|
result.state_code = result.state_code or code
|
|
result.searched_name = result.searched_name or name
|
|
return result
|
|
except Exception as exc:
|
|
LOG.error("Name search failed in %s: %s", code, exc, exc_info=True)
|
|
return NameSearchResult(
|
|
available=False,
|
|
searched_name=name,
|
|
state_code=code,
|
|
raw_response=f"Error: {exc}",
|
|
)
|
|
finally:
|
|
await adapter.close_browser()
|
|
|
|
|
|
async def search_multiple_states(
|
|
name: str,
|
|
state_codes: list[str],
|
|
) -> list[NameSearchResult]:
|
|
"""
|
|
Search for business name availability across multiple states in parallel.
|
|
|
|
Launches concurrent searches using asyncio.gather. Each state gets its own
|
|
browser instance so they don't interfere with each other.
|
|
|
|
Args:
|
|
name: The business name to search.
|
|
state_codes: List of two-letter state codes.
|
|
|
|
Returns:
|
|
List of NameSearchResult, one per state (order matches state_codes).
|
|
"""
|
|
LOG.info(
|
|
"Searching name '%s' across %d states: %s",
|
|
name,
|
|
len(state_codes),
|
|
", ".join(c.upper() for c in state_codes),
|
|
)
|
|
start = time.monotonic()
|
|
|
|
tasks = [search_name(name, code) for code in state_codes]
|
|
results = await asyncio.gather(*tasks, return_exceptions=False)
|
|
|
|
elapsed = time.monotonic() - start
|
|
available_in = [r.state_code for r in results if r.available]
|
|
unavailable_in = [r.state_code for r in results if not r.available]
|
|
|
|
LOG.info(
|
|
"Multi-state search complete in %.1fs — available: %s | unavailable: %s",
|
|
elapsed,
|
|
", ".join(available_in) or "(none)",
|
|
", ".join(unavailable_in) or "(none)",
|
|
)
|
|
return results
|
|
|
|
|
|
def _format_result(result: NameSearchResult) -> str:
|
|
"""Pretty-print a single search result for CLI output."""
|
|
status = "AVAILABLE" if result.available else "UNAVAILABLE"
|
|
lines = [
|
|
f" [{result.state_code}] {status} — \"{result.searched_name}\"",
|
|
]
|
|
if result.exact_match:
|
|
lines.append(f" Exact match found")
|
|
if result.similar_names:
|
|
lines.append(f" Similar names: {', '.join(result.similar_names[:5])}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
async def _main(name: str, raw_states: str) -> int:
|
|
"""CLI entry point logic."""
|
|
# Parse comma-separated or space-separated state codes
|
|
state_codes = [s.strip().upper() for s in raw_states.replace(",", " ").split() if s.strip()]
|
|
|
|
if not state_codes:
|
|
print("Error: No state codes provided.", file=sys.stderr)
|
|
return 1
|
|
|
|
# Validate state codes
|
|
invalid = [s for s in state_codes if s not in STATES]
|
|
if invalid:
|
|
print(f"Error: Unknown state code(s): {', '.join(invalid)}", file=sys.stderr)
|
|
print(f"Valid codes: {', '.join(sorted(STATES.keys()))}", file=sys.stderr)
|
|
return 1
|
|
|
|
print(f"Searching: \"{name}\"")
|
|
print(f"States: {', '.join(state_codes)}")
|
|
print("-" * 60)
|
|
|
|
if len(state_codes) == 1:
|
|
result = await search_name(name, state_codes[0])
|
|
results = [result]
|
|
else:
|
|
results = await search_multiple_states(name, state_codes)
|
|
|
|
for r in results:
|
|
print(_format_result(r))
|
|
|
|
print("-" * 60)
|
|
available_count = sum(1 for r in results if r.available)
|
|
print(f"Available in {available_count}/{len(results)} state(s).")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(name)s] %(levelname)s %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S",
|
|
)
|
|
|
|
if len(sys.argv) < 3:
|
|
print("Usage: python -m formation.name_search <business_name> <state_code(s)>")
|
|
print()
|
|
print("Examples:")
|
|
print(' python -m formation.name_search "Acme Holdings LLC" WY')
|
|
print(' python -m formation.name_search "Acme Holdings LLC" WY,NV,NM,TX')
|
|
sys.exit(1)
|
|
|
|
business_name = sys.argv[1]
|
|
states_arg = " ".join(sys.argv[2:]) # Allow "WY NV" or "WY,NV"
|
|
exit_code = asyncio.run(_main(business_name, states_arg))
|
|
sys.exit(exit_code)
|