account-onboarding/app.py

#!/usr/bin/env python3
"""
Castor — Signup Dashboard
Check the providers you want to open accounts with, then launch them in a
browser with your identity auto-filled. No retyping your info.

It fills YOUR OWN verified info into forms YOU submit, one account at a time.
"""

import argparse
import json
import os
import re
import subprocess
import sys

from flask import Flask, request, render_template, redirect, url_for, flash

from playwright.sync_api import sync_playwright

BASE = os.path.dirname(os.path.abspath(__file__))
PROFILE_PATH = os.path.join(BASE, "autofill_profile.json")
PROVIDERS_PATH = os.path.join(BASE, "providers.json")

AUTOCOMPLETE_MAP = {
    "given-name": "first_name",
    "family-name": "last_name",
    "name": "full_name",
    "email": "email",
    "tel": "phone",
    "tel-national": "phone",
    "tel-local": "phone",
    "street-address": "address_line1",
    "address-line1": "address_line1",
    "address-line2": "address_line2",
    "address-level1": "state",
    "address-level2": "city",
    "postal-code": "zip",
    "bday": "dob",
    "organization": "employer",
    "organization-title": "employer",
}

KEYWORD_MAP = {
    "first_name": ["firstname", "givenname", "fname"],
    "last_name": ["lastname", "surname", "familyname", "lname"],
    "email": ["emailaddress", "email"],
    "phone": ["phonenumber", "phonenum", "telephone", "mobile", "cellphone", "phone"],
    "address_line1": ["streetaddress", "addressline1", "address1", "mailingaddress", "street"],
    "address_line2": ["addressline2", "address2", "suite", "apt", "unit"],
    "city": ["city", "town", "municipality"],
    "state": ["stateprovince", "province", "state"],
    "zip": ["zipcode", "postalcode", "zip", "postal"],
    "dob": ["dateofbirth", "birthdate", "birthday", "dob"],
    "ssn": ["socialsecuritynumber", "socialsecurity", "ssnnumber", "ssn", "taxpayerid", "taxid", "tin", "nationalid"],
    "employer": ["currentemployer", "employername", "employer", "companyname"],
    "income": ["annualincome", "incomeamount", "income", "salary"],
}

STATE_NAMES = {
    "AL": "Alabama", "AK": "Alaska", "AZ": "Arizona", "AR": "Arkansas", "CA": "California",
    "CO": "Colorado", "CT": "Connecticut", "DE": "Delaware", "FL": "Florida", "GA": "Georgia",
    "HI": "Hawaii", "ID": "Idaho", "IL": "Illinois", "IN": "Indiana", "IA": "Iowa",
    "KS": "Kansas", "KY": "Kentucky", "LA": "Louisiana", "ME": "Maine", "MD": "Maryland",
    "MA": "Massachusetts", "MI": "Michigan", "MN": "Minnesota", "MS": "Mississippi",
    "MO": "Missouri", "MT": "Montana", "NE": "Nebraska", "NV": "Nevada", "NH": "New Hampshire",
    "NJ": "New Jersey", "NM": "New Mexico", "NY": "New York", "NC": "North Carolina",
    "ND": "North Dakota", "OH": "Ohio", "OK": "Oklahoma", "OR": "Oregon", "PA": "Pennsylvania",
    "RI": "Rhode Island", "SC": "South Carolina", "SD": "South Dakota", "TN": "Tennessee",
    "TX": "Texas", "UT": "Utah", "VT": "Vermont", "VA": "Virginia", "WA": "Washington",
    "WV": "West Virginia", "WI": "Wisconsin", "WY": "Wyoming", "DC": "District of Columbia",
}

PROFILE_FIELDS = [
    ("first_name", "First name"), ("last_name", "Last name"), ("email", "Email"),
    ("phone", "Phone"), ("address_line1", "Address line 1"), ("address_line2", "Address line 2"),
    ("city", "City"), ("state", "State (2-letter)"), ("zip", "ZIP"),
    ("dob", "Date of birth (MM/DD/YYYY)"), ("ssn", "SSN (no dashes)"),
    ("employer", "Employer"), ("income", "Annual income"),
]

PASSWORD_LIKE = re.compile(r"password|passwd|passcode|pwd|confirmpassword", re.I)


def norm(s):
    return re.sub(r"[^a-z0-9]", "", (s or "").lower())


def load_json(path, default):
    if os.path.exists(path):
        with open(path) as fh:
            return json.load(fh)
    return default


def load_profile():
    return load_json(PROFILE_PATH, {})


def save_profile(data):
    with open(PROFILE_PATH, "w") as fh:
        json.dump(data, fh, indent=2)


def load_providers():
    return load_json(PROVIDERS_PATH, [])


# ------------------------- Playwright fill engine -------------------------
GATHER_FIELDS_JS = """
() => {
  const out = [];
  const els = document.querySelectorAll('input, select, textarea');
  els.forEach((el, idx) => {
    const tag = el.tagName.toLowerCase();
    const type = tag === 'input' ? (el.type || 'text').toLowerCase() : tag;
    if (['hidden','submit','button','reset','checkbox','radio','file','password'].includes(type)) return;
    if (el.offsetParent === null) return;
    let label = '';
    if (el.id) { const l = document.querySelector('label[for="' + CSS.escape(el.id) + '"]'); if (l) label = l.innerText; }
    if (!label && el.closest('label')) label = el.closest('label').innerText;
    if (!label && el.parentElement) label = el.parentElement.innerText.split('\\n')[0];
    out.push({ idx, tag, type, name: el.name || '', id: el.id || '', ac: el.autocomplete || '',
               ph: el.placeholder || '', al: el.getAttribute('aria-label') || '', label: (label || '').trim() });
  });
  return out;
}
"""

FILL_ONE_JS = """
(arg) => {
  const idx = arg.idx; const candidates = arg.candidates;
  const els = document.querySelectorAll('input, select, textarea');
  const el = els[idx];
  if (!el) return 'missing';
  el.scrollIntoView({ block: 'center' });
  el.style.outline = '3px solid #6c8cff'; el.style.outlineOffset = '1px';
  let ok = false;
  if (el.tagName === 'SELECT') {
    const opts = Array.from(el.options);
    outer:
    for (const want of candidates) {
      const w = String(want).toLowerCase();
      for (const o of opts) {
        const t = o.text.trim().toLowerCase();
        if (t === w || t.includes(w) || w.includes(t) || o.value.toLowerCase() === w) { el.value = o.value; ok = true; break outer; }
      }
    }
  } else {
    const proto = el.tagName === 'TEXTAREA' ? window.HTMLTextAreaElement.prototype : window.HTMLInputElement.prototype;
    Object.getOwnPropertyDescriptor(proto, 'value').set.call(el, String(candidates[0]));
    ok = true;
  }
  el.dispatchEvent(new Event('input', { bubbles: true }));
  el.dispatchEvent(new Event('change', { bubbles: true }));
  setTimeout(() => { el.style.outline = ok ? '2px solid #4ade80' : '2px solid #f87171'; }, 500);
  return ok ? 'filled' : 'no-match';
}
"""


def build_sig(f):
    return norm(" ".join([f["name"], f["id"], f["ph"], f["al"], f["label"]]))


def match_profile_key(f):
    ac = f["ac"].strip().lower()
    if ac in AUTOCOMPLETE_MAP:
        return AUTOCOMPLETE_MAP[ac]
    sig = build_sig(f)
    best_key, best_len = None, 0
    for key, kws in KEYWORD_MAP.items():
        for kw in kws:
            if kw in sig and len(kw) > best_len:
                best_key, best_len = key, len(kw)
    return best_key


def fill_candidates(profile, key):
    val = profile.get(key, "")
    if not val:
        return None
    cands = [val]
    if key == "state":
        cands.append(STATE_NAMES.get(val.upper(), val))
    if key == "full_name":
        cands = ["%s %s" % (profile.get("first_name", ""), profile.get("last_name", "")).strip()]
    return cands


def fill_page(page, profile, delay=0.2):
    fields = page.evaluate(GATHER_FIELDS_JS)
    results = []
    for f in fields:
        blob = f["name"] + " " + f["id"] + " " + f["label"] + " " + f["ac"]
        if PASSWORD_LIKE.search(blob):
            continue
        key = match_profile_key(f)
        if key is None:
            continue
        cands = fill_candidates(profile, key)
        if not cands:
            continue
        page.evaluate(FILL_ONE_JS, {"idx": f["idx"], "candidates": cands})
        results.append((key, f, cands[0]))
        if delay:
            page.wait_for_timeout(int(delay * 1000))
    return results


def launch_fill(urls, headless=False, delay=0.2):
    """Open each URL in its own tab, auto-fill, leave the browser open. Returns a summary."""
    import time
    profile = load_profile()
    summary = []
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=headless)
        context = browser.new_context()
        pages = []
        for url in urls:
            page = context.new_page()
            pages.append(page)
            page.goto(url, wait_until="domcontentloaded", timeout=60000)
            page.wait_for_timeout(1500)
            results = fill_page(page, profile, delay)
            summary.append((url, results))
        if not headless:
            # Keep the browser alive until the user closes it (signups in progress).
            while browser.is_connected():
                time.sleep(1)
        browser.close()
    return summary


# ------------------------- Flask app -------------------------
app = Flask(__name__)
app.secret_key = "castor-signup-dashboard-local-only"


@app.route("/")
def index():
    providers = load_providers()
    profile = load_profile()
    filled = sum(1 for k, v in profile.items() if v)
    total = len(PROFILE_FIELDS)
    categories = []
    seen = []
    for p in providers:
        c = p["category"]
        if c not in seen:
            seen.append(c)
            categories.append(c)
    return render_template(
        "index.html", providers=providers, categories=categories,
        profile=profile, filled=filled, total=total,
    )


@app.route("/launch", methods=["POST"])
def launch():
    ids = request.form.getlist("providers")
    providers = load_providers()
    chosen = [p for p in providers if p["id"] in ids]
    if not chosen:
        flash("Select at least one provider.", "error")
        return redirect(url_for("index"))
    urls = [p["url"] for p in chosen]
    names = [p["name"] for p in chosen]

    # Fill a report log to disk, then launch the fill engine in a visible browser.
    report = {"launched": names, "urls": urls}
    with open(os.path.join(BASE, "last_launch.json"), "w") as fh:
        json.dump(report, fh, indent=2)

    # Launch in a subprocess so the visible browser opens on this machine.
    subprocess.Popen(
        [sys.executable, os.path.join(BASE, "fill_worker.py"), "--urls", json.dumps(urls)],
        cwd=BASE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
    )

    flash("Opened %d signup page(s) — completing auto-fill in the browser now." % len(chosen), "ok")
    return redirect(url_for("index"))


@app.route("/profile", methods=["GET", "POST"])
def profile_route():
    if request.method == "POST":
        data = {}
        for key, _label in PROFILE_FIELDS:
            data[key] = request.form.get(key, "").strip()
        save_profile(data)
        flash("Profile saved.", "ok")
        return redirect(url_for("profile_route"))
    return render_template("profile.html", fields=PROFILE_FIELDS, profile=load_profile())


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--host", default="127.0.0.1")
    ap.add_argument("--port", type=int, default=5057)
    args = ap.parse_args()
    app.run(host=args.host, port=args.port, debug=False)