← back to Nationalrealestate
scripts/scrape-asc-appraisers.py
168 lines
#!/usr/bin/env python3
"""
TK-10560 — Scrape the ASC National Appraiser Registry (asc.gov) for California,
filter to the SoCal directory footprint (within 300mi of LA), classify commercial
relevance, and stage rows for rentv_licensed_targets (role='Appraiser').
VERIFIED PAYOFF FIRST: the ASC results carry state/city/zip/license#/license-type/
status per record (confirmed on a small sample before this full run — see TK-10560).
Scrape path (server-side; the browser spike only discovered the working param set):
GET /appraiser?field_state_name_value=California&field_license_type__value=All
&field_first_name__value=&field_last_name__value=&items_per_page=250&submit=Apply&page=N
Output: JSONL of staged rows (SoCal only) -> stdout. Read-only vs asc.gov; NO DB writes.
"""
import sys, re, json, time, html, urllib.request, os
import urllib.parse as _up
# LICENSE_TYPE env filters at the SOURCE (e.g. "Certified General" = the commercial-capable
# grade). Default "All". Filtering to Certified General yields exactly the commercial load set
# in far fewer pages (asc.gov is ~24s/page, so this is a big speedup).
_LT = os.environ.get("LICENSE_TYPE", "All")
BASE = ("https://www.asc.gov/appraiser?field_state_name_value=California"
"&field_license_type__value=" + _up.quote(_LT) + "&field_first_name__value="
"&field_last_name__value=&items_per_page=250&submit=Apply&page=")
UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 Chrome/151 Safari/537.36"
DATA_DIR = os.path.join(os.path.dirname(__file__), "..", "data")
CITY_MAP_PATH = os.path.join(DATA_DIR, "socal_city_market.psv")
ZIP_MAP_PATH = os.path.join(DATA_DIR, "socal_zip_market.psv")
def load_psv(path):
m = {}
with open(path) as f:
for line in f:
line = line.rstrip("\n")
if "|" in line:
k, v = line.split("|", 1)
m[k.strip().lower()] = v.strip()
return m
def load_city_market():
return load_psv(CITY_MAP_PATH)
def load_zip_market():
return load_psv(ZIP_MAP_PATH)
def fetch(page, tries=3):
url = BASE + str(page)
for t in range(tries):
try:
req = urllib.request.Request(url, headers={"User-Agent": UA})
with urllib.request.urlopen(req, timeout=45) as r:
return r.read().decode("utf-8", "replace")
except Exception as e:
if t == tries - 1:
sys.stderr.write(f"[fetch fail p{page}] {e}\n"); return ""
time.sleep(2 * (t + 1))
return ""
ROW_RE = re.compile(r'<tr[^>]*class="[A-Za-z0-9_\- ]+"\s*>(.*?)</tr>', re.S)
def strip_comments(s): return re.sub(r"<!--.*?-->", "", s, flags=re.S)
def txt(s): return html.unescape(re.sub(r"<[^>]+>", " ", s)).strip()
def parse_page(hbody):
rows = []
for m in ROW_RE.finditer(hbody):
cell = strip_comments(m.group(1))
nm = re.search(r'/individual-appraiser-public/(\d+)"[^>]*>([^<]+)</a>', cell)
if not nm:
continue
detail_id = nm.group(1)
name_raw = html.unescape(nm.group(2)).strip()
c1 = re.search(r'middle-appraiser-columns1"[^>]*>(.*?)</div>', cell, re.S)
state = lic = ltype = ""
if c1:
parts = [p for p in (txt(x) for x in re.split(r'<br\s*/?>', c1.group(1))) if p]
for p in parts:
if re.match(r'^#\s*[A-Z]{1,3}\d+', p):
lic = re.sub(r'^#\s*', '', p).strip()
elif p.lower() == "california":
state = "CA"
elif not ltype and not p.startswith("#"):
ltype = p
if not ltype:
for p in parts:
if p.lower() != "california" and not p.startswith("#"):
ltype = p; break
if not state:
state = "CA"
st = re.search(r'class="isactive">([^<]+)</span>\s*</div>\s*</span>', cell) \
or re.search(r'class="isactive">\s*<div[^>]*>\s*<span[^>]*>([^<]+)</span>', cell)
status = txt(st.group(1)) if st else ""
street = re.search(r'optionalstreet"[^>]*>([^<]*)</div>', cell)
city = re.search(r'optionalcity"[^>]*>([^<]*)</span>([^<]*)</div>', cell)
street_v = txt(street.group(1)) if street else ""
city_v = zip_v = state_v = ""
if city:
city_v = txt(city.group(1)).rstrip(", ").strip()
mm = re.match(r'([A-Z]{2})\s+([\d\-]+)', txt(city.group(2)))
if mm:
state_v = mm.group(1); zip_v = mm.group(2)
rows.append({"detail_id": detail_id, "name_raw": name_raw, "license_no": lic,
"license_type": ltype, "license_status": status, "address": street_v,
"city": city_v, "state": state_v or state, "zip": zip_v})
return rows
def name_lf(name_raw):
if "," in name_raw:
last, first = name_raw.split(",", 1)
return f"{first.strip()} {last.strip()}".strip()
return name_raw
def zip_socal(z):
m = re.match(r'^(\d{3})', z or "")
return bool(m) and 900 <= int(m.group(1)) <= 935
COMMERCIAL_TYPES = {"certified general"}
def main():
city_market = load_city_market()
zip_market = load_zip_market()
seen, out = set(), []
stats = {"pages": 0, "raw": 0, "socal": 0, "commercial": 0, "dup": 0}
page, empty = 0, 0
while True:
body = fetch(page)
rows = parse_page(body) if body else []
if not rows:
empty += 1
if empty >= 2:
break
page += 1; continue
empty = 0
stats["pages"] += 1; stats["raw"] += len(rows)
for r in rows:
key = r["license_no"] or r["detail_id"]
if key in seen:
stats["dup"] += 1; continue
z5 = re.match(r'^(\d{5})', r["zip"] or "")
market = zip_market.get(z5.group(1)) if z5 else None # zip-primary (misspelling-proof)
if not market: # city fallback
market = city_market.get((r["city"] or "").lower())
if not (market and zip_socal(r["zip"])):
continue
seen.add(key); stats["socal"] += 1
comm = (r["license_type"] or "").strip().lower() in COMMERCIAL_TYPES
if comm:
stats["commercial"] += 1
out.append({
"source": "ASC National Appraiser Registry", "role": "Appraiser",
"entity_name": name_lf(r["name_raw"]), "contact_name": name_lf(r["name_raw"]),
"license_no": r["license_no"], "license_type": r["license_type"],
"license_status": r["license_status"], "address": r["address"],
"city": r["city"], "county": None, "state": r["state"] or "CA",
"zip": r["zip"], "market": market, "commercial_flag": comm,
"within_300mi": True,
"source_url": f"https://www.asc.gov/individual-appraiser-public/{r['detail_id']}",
"raw": r})
if stats["pages"] % 10 == 0:
sys.stderr.write(f"[progress] page={page} raw={stats['raw']} socal={stats['socal']}\n")
page += 1; time.sleep(0.8)
for rec in out:
print(json.dumps(rec, ensure_ascii=False))
sys.stderr.write(f"[DONE] {json.dumps(stats)}\n")
if __name__ == "__main__":
main()