"""Bizportal Search.py — CIPC BizProfile lookup, auto-rotation, PB integration. Usage: python "Bizportal Search.py" "JS MINING AND ENGINEERING" python "Bizportal Search.py" "COMPANY A" "COMPANY B" "COMPANY C" python "Bizportal Search.py" --batch companies.txt For each company name: 1. Looks up the PB record (ysterbaard_companies) by enterprise_name (case-insensitive substring). 2. Reads enterprise_no (K-number) from PB. 3. Logs into CIPC BizPortal (Login 1; credentials read from .env.local) and searches bizprofile.aspx by Enterprise No. 4. Extracts: Company Details, Director Details, Filed Annual Returns, Outstanding Annual Returns. (Enterprise History is intentionally skipped per SJ's instruction.) 5. PATCHes the PB record's `bizportal_data_json` field with the parsed data. 6. Prints a markdown summary to stdout for the chat. Auto-rotation: When "Searches Conducted Today: N" reaches 100 on Login 1, logs out and logs in as Login 3. When Login 3 also reaches 100, stops and reports to the operator to resume tomorrow. Subscription expiry: Parses "Subscription Expiry Date: YYYY/MM/DD ..." from the page and refuses to search if the subscription has expired (clear message to operator). Verified end-to-end on 2026-09-14 with K2018435959 (JS MINING AND ENGINEERING). """ import argparse import base64 import datetime as dt import json import os import re import subprocess import sys import time import urllib.error import urllib.request from pathlib import Path # ------------------------------------------------------------------ paths AGENT_BROWSER = Path(os.environ.get("AGENT_BROWSER_BIN", r"C:\Users\SJ\AppData\Roaming\npm\agent-browser.cmd")) WORKFLOW_ROOT = Path(os.environ.get("WORKFLOW_ROOT", r"C:\Users\SJ\.minimax-agent\projects\Workflow")) ENV_FILE = WORKFLOW_ROOT / ".env.local" PB_URL = "https://pb.agent.apphub.co.za" PB_COLLECTION = "ysterbaard_companies" # ------------------------------------------------------------------ credentials # Login 1 — CIPC Customer Code (default, primary) # Login 3 — CIPC Customer Code (fallback when Login 1 hits 100) # Login 2 was removed 2026-09-14 by SJ as inactive. # The codes/passwords live in .env.local under CIPC_LOGIN*_CODE / # CIPC_LOGIN*_PASSWORD — never in source. See .env.example. def load_env_value(key: str) -> str: """Read a single key out of .env.local.""" if not ENV_FILE.exists(): raise SystemExit(f"Missing env file: {ENV_FILE}") for line in ENV_FILE.read_text(encoding="utf-8").splitlines(): if line.strip().startswith(f"{key}="): return line.split("=", 1)[1].strip() raise SystemExit(f"{key} not found in {ENV_FILE}") LOGIN1_CODE = load_env_value("CIPC_LOGIN1_CODE") LOGIN3_CODE = load_env_value("CIPC_LOGIN3_CODE") CREDENTIALS = [ { "name": f"Login 1 ({LOGIN1_CODE})", "customer_code": LOGIN1_CODE, "password": load_env_value("CIPC_LOGIN1_PASSWORD"), }, { "name": f"Login 3 ({LOGIN3_CODE})", "customer_code": LOGIN3_CODE, "password": load_env_value("CIPC_LOGIN3_PASSWORD"), }, ] # ------------------------------------------------------------------ constants LOGIN_URL = "https://www.bizportal.gov.za/login.aspx" BIZPROFILE_URL = "https://www.bizportal.gov.za/bizprofile.aspx" SESSION_NAME = "biz-search" # agent-browser --session name SEARCH_LIMIT = 100 DEFAULT_SEARCH_WAIT = 35 # seconds, time for the result pane to render after firing SEARCH LOGIN_WAIT = 15 TOGGLE_WAIT = 2 DROPDOWN_SETTLE_WAIT = 5 POSTBACK_NAV_WAIT = 5 # =================================================================== PB layer def load_pb_token() -> str: if not ENV_FILE.exists(): raise SystemExit(f"Missing env file: {ENV_FILE}") for line in ENV_FILE.read_text(encoding="utf-8").splitlines(): if line.startswith("PB_TOKEN="): return line.split("=", 1)[1].strip() raise SystemExit(f"PB_TOKEN not found in {ENV_FILE}") def pb_request(method: str, path: str, token: str, body: dict | None = None) -> dict: url = PB_URL + path headers = {"Authorization": token} data = None if body is not None: headers["Content-Type"] = "application/json" data = json.dumps(body).encode("utf-8") req = urllib.request.Request(url, data=data, headers=headers, method=method) try: resp = urllib.request.urlopen(req, timeout=30) raw = resp.read().decode("utf-8") except urllib.error.HTTPError as e: raise SystemExit(f"PB {method} {path} HTTP {e.code}: {e.read().decode('utf-8')[:300]}") return json.loads(raw) if raw else {} def lookup_company_by_name(name: str, token: str) -> dict | None: """Case-insensitive substring search on enterprise_name. Returns the PB record or None.""" needle = name.strip().upper() url = ( f"/api/collections/{PB_COLLECTION}/records" f"?perPage=200&fields=id,enterprise_name,enterprise_no" ) data = pb_request("GET", url, token) items = data.get("items", []) # 1. Exact match for it in items: if (it.get("enterprise_name") or "").strip().upper() == needle: return it # 2. Substring match (any item containing the needle) matches = [it for it in items if needle in (it.get("enterprise_name") or "").upper()] if len(matches) == 1: return matches[0] if len(matches) > 1: print(f"WARNING: '{name}' matched {len(matches)} records, picking first: " f"{matches[0].get('enterprise_name')} ({matches[0].get('enterprise_no')})") return matches[0] return None def patch_bizportal_json(record_id: str, payload: dict, token: str) -> dict: body = { "bizportal_data_json": payload, "last_checked": dt.date.today().isoformat(), } return pb_request("PATCH", f"/api/collections/{PB_COLLECTION}/records/{record_id}", token, body) # =================================================================== agent-browser wrapper def ab(*args, timeout: int = 60) -> str: """Run an agent-browser CLI call as a standalone subprocess, wait for it to finish.""" cmd = [str(AGENT_BROWSER), "--session", SESSION_NAME, *args] r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) out = (r.stdout or "").strip() err = (r.stderr or "").strip() return out if out else err def ab_b64(js: str, *extra_args, timeout: int = 60) -> str: """Run an agent-browser eval with base64-encoded JS to dodge shell quoting issues. agent-browser wraps eval results in JSON quotes (with \\n escapes). We strip the outer quotes and unescape common sequences so callers get a clean Python string. """ b64 = base64.b64encode(js.encode("utf-8")).decode("ascii") raw = ab("eval", "-b", b64, *extra_args, timeout=timeout) # If result is JSON-quoted, unquote it. if raw.startswith('"') and raw.endswith('"'): try: return json.loads(raw) except json.JSONDecodeError: return raw[1:-1] return raw def ab_eval(js: str) -> str: """eval that returns the JS expression's value, JSON-decoded.""" return ab_b64(js) def ab_sleep(ms: int): """Short pause between calls. Uses native time.sleep — agent-browser's `wait` command appears to disrupt BizPortal's autopostback in some way (we observed data not loading when wait was used, but loading when time.sleep was used). """ time.sleep(ms / 1000.0) # =================================================================== BizPortal flow def login(creds: dict) -> str: """Logs into BizPortal. Returns the URL after login (should be user_profile.aspx).""" print(f"\n--- Logging in as {creds['name']} ---") ab("open", LOGIN_URL) ab_sleep(4000) # Toggle SA ID -> CIPC Customer Code (always click — page may have auto-toggled either way) ab_eval("document.getElementById('cntMain_chkId_ToggleButton').click(); 'T'") ab_sleep(TOGGLE_WAIT * 1000) # Verify CIPC Customer Code field is now visible (re-inspect if not) chk = ab_eval("!!document.getElementById('cntMain_txtCustCode') ? 'OK' : 'MISSING'") if "OK" not in chk: # Try toggling again print(f" toggle check: {chk} — toggling again") ab_eval("document.getElementById('cntMain_chkId_ToggleButton').click(); 'T2'") ab_sleep(TOGGLE_WAIT * 1000) # Fill + click Login login_js = ( "var el=document.getElementById('cntMain_txtCustCode');" "el.value='" + creds['customer_code'] + "';" "el.dispatchEvent(new Event('input',{bubbles:true}));" "var el2=document.getElementById('cntMain_txtPassword');" "el2.value='" + creds['password'] + "';" "el2.dispatchEvent(new Event('input',{bubbles:true}));" "document.getElementById('cntMain_btnLogin').click();" "'L'" ) ab_eval(login_js) ab_sleep(LOGIN_WAIT * 1000) return ab_eval("window.location.href") def navigate_to_bizprofile() -> str: """Standalone open (must be standalone, not via window.location.href in eval).""" ab("open", BIZPROFILE_URL) ab_sleep(POSTBACK_NAV_WAIT * 1000) return ab_eval("window.location.href") def read_counter() -> int: """Reads 'Searches Conducted Today: N' from the bizprofile page.""" body = ab_eval("document.body.innerText.substring(0, 3000)") m = re.search(r"Searches Conducted Today:\s*(\d+)", body) return int(m.group(1)) if m else -1 def read_subscription_expiry() -> str | None: """Returns YYYY/MM/DD from 'Subscription Expiry Date: YYYY/MM/DD ...'.""" body = ab_eval("document.body.innerText.substring(0, 3000)") m = re.search(r"Subscription Expiry Date:\s*(\d{4}/\d{2}/\d{2})", body) return m.group(1) if m else None def search_by_enterprise_no(k_number: str) -> str: """Set dropdown to EntNo, fill K-number, fire search via __doPostBack. Returns body text.""" # NOTE: keep this function body as minimal as possible. We observed empirically # 2026-09-14 that ANY extra agent-browser eval call between dropdown-set and # search-fire (even a JSON.stringify state check) causes BizPortal to silently # fail to render the result pane. So: set -> wait -> fire -> wait -> read. ab_eval( "var sel=document.getElementById('cntMain_drpSearchOptions');" "sel.value='EntNo';" "sel.dispatchEvent(new Event('change'));" "'OK'" ) time.sleep(5) ab_eval( "document.getElementById('cntMain_txtSearchCIPC').value='" + k_number + "';" "__doPostBack('ctl00$cntMain$btnSearch','');" "'F'" ) time.sleep(35) # 12000 chars covers COMPANY + DIRECTOR + AUDITOR + B-BBEE + INFO REG + OTHER DETAILS. return ab_eval("document.body.innerText.substring(0,12000)") # =================================================================== parsing def parse_result(body: str) -> dict: """Parse BizPortal body text. Returns dict with company details, directors, ARs (NO history). NBSP-safe: BizPortal uses U+00A0 (non-breaking space) between section blocks (e.g. ' DIRECTOR DETAILS' with 2 NBSPs before the label). Python's default \\s does NOT match NBSP, so we use [\\s\\u00a0] (or pass re.UNICODE which makes \\s include NBSP for str patterns — but the explicit version is safer). """ out: dict = {} # Strip U+00C2 artifacts (UTF-8 double-encoding of NBSPs) so regex sees clean NBSPs. body = body.replace("\u00c2", "") # NBSP-safe shorthand: WS = whitespace OR NBSP WS = r"[\s\u00a0]" NL = r"\n" ANY = rf"(?:.|\n)" # match any char including newlines # ---- Company Details block ---- company_match = re.search( rf"COMPANY DETAILS{WS}*{NL}(.*?)(?:{NL}{WS}*DIRECTOR DETAILS|{NL}{WS}*AUDITOR|\Z)", body, re.DOTALL, ) company_block = company_match.group(1).strip() if company_match else "" cd: dict = {} LABELS = [ ("Enterprise Number", "enterprise_number"), ("Enterprise Name", "enterprise_name"), ("Enterprise Type", "enterprise_type"), ("Enterprise Status", "enterprise_status"), ("Compliance Notice", "compliance_notice"), ("Registration Date", "registration_date"), ("Physical Address", "physical_address"), ("Postal Address", "postal_address"), ] # Walk line-by-line: when we hit a label, the next line is its value. Some values # span multiple lines (Physical/Postal Address: 4 lines). lines = company_block.splitlines() i = 0 while i < len(lines): line = lines[i].strip() if not line: i += 1 continue matched_label = None for label, key in LABELS: if line == label or line.startswith(label + " "): matched_label = (label, key) break if matched_label: label, key = matched_label value_lines: list[str] = [] j = i + 1 while j < len(lines): vline = lines[j].strip() if not vline: j += 1 continue if any(vline == lab or vline.startswith(lab + " ") for lab, _ in LABELS): break value_lines.append(vline) j += 1 # SA BizPortal addresses have 5 lines: street, suburb, city, province, postal_code if len(value_lines) >= 5: break cd[key] = " ".join(value_lines) i = j else: i += 1 out["company_details"] = cd # ---- Director Details ---- director_section = "" m = re.search(rf"DIRECTOR DETAILS{WS}*{NL}(.*?)(?:{NL}{WS}*AUDITOR|{NL}{WS}*ENTERPRISE HISTORY|\Z)", body, re.DOTALL) if m: director_section = m.group(1) directors = [] for line in director_section.splitlines(): line = line.strip() # SA ID format: YYMMDD + 7 chars (digits or X for redacted). 13 chars total. if re.match(r"^[X0-9]{13}\b", line): parts = re.split(r"\t+|\s{2,}", line) if len(parts) >= 5: directors.append({ "id_number": parts[0].strip(), "names": parts[1].strip(), "surname": parts[2].strip(), "type": parts[3].strip(), "status": " ".join(parts[4:]).strip(), }) elif len(parts) == 4: directors.append({ "id_number": parts[0].strip(), "names": parts[1].strip(), "surname": parts[2].strip(), "type": parts[3].strip(), "status": "", }) out["director_details"] = directors # ---- Filed ARs ---- filed = [] m = re.search(rf"FILED ANNUAL RETURNS{WS}*{NL}(.*?)(?:{NL}{WS}*OUTSTANDING ANNUAL RETURNS|\Z)", body, re.DOTALL) if m: for line in m.group(1).splitlines(): parts = re.split(r"\t+|\s{2,}", line.strip()) if len(parts) >= 3 and parts[0].isdigit() and re.match(r"^-?[\d,.]+$", parts[1]): filed.append({ "year": int(parts[0]), "amount_paid": parts[1], "date_filed": parts[2], }) out["filed_annual_returns"] = filed # ---- Outstanding ARs ---- outstanding = [] m = re.search( rf"OUTSTANDING ANNUAL RETURNS{WS}*{NL}(.*?)(?:{NL}{WS}*ENTERPRISE HISTORY|\Z)", body, re.DOTALL, ) if m: for line in m.group(1).splitlines(): parts = re.split(r"\t+|\s{2,}", line.strip()) if len(parts) >= 3 and parts[0].isdigit(): outstanding.append({ "year": int(parts[0]), "month": parts[1], "non_compliance_date": parts[2], }) out["outstanding_annual_returns"] = outstanding # ---- Other Details (SARS Tax Number, UIF, Compensation Fund) ---- # Block ends at "ABOUT BIZPORTAL" or end of string. other: dict = {} m = re.search( rf"OTHER DETAILS{WS}*{NL}(.*?)(?:{NL}{WS}*ABOUT BIZPORTAL|\Z)", body, re.DOTALL, ) if m: other_block = m.group(1) OTHER_LABELS = [ ("SARS Tax Number", "sars_tax_number"), ("UIF Registration Number", "uif_registration_number"), ("Compensation Fund Registration Number", "compensation_fund_registration_number"), ] lines = other_block.splitlines() i = 0 while i < len(lines): line = lines[i].strip() if not line: i += 1 continue matched = None for label, key in OTHER_LABELS: if line == label or line.startswith(label + " "): matched = (label, key) break if matched: label, key = matched # value is on the next non-empty line j = i + 1 while j < len(lines) and not lines[j].strip(): j += 1 if j < len(lines): val = lines[j].strip() if val.upper() == "NOT AVAILABLE": other[key] = None else: other[key] = val i = j else: i += 1 out["other_details"] = other return out # =================================================================== output formatting def format_markdown(parsed: dict, k_number: str) -> str: cd = parsed.get("company_details", {}) lines = [] lines.append(f"## {cd.get('enterprise_name', k_number)} — BizProfile results ({k_number})") lines.append("") lines.append("### COMPANY DETAILS") for label, key in [ ("Enterprise Number", "enterprise_number"), ("Enterprise Name", "enterprise_name"), ("Enterprise Type", "enterprise_type"), ("Enterprise Status", "enterprise_status"), ("Compliance Notice", "compliance_notice"), ("Registration Date", "registration_date"), ("CIPC Physical address", "physical_address"), ("CIPC Postal address", "postal_address"), ]: if cd.get(key): lines.append(f"- {label}: {cd[key]}") lines.append("") directors = parsed.get("director_details", []) if directors: lines.append("### DIRECTOR DETAILS") lines.append("| ID Number | Name(s) | Surname | Type | Status |") lines.append("|---|---|---|---|---|") for d in directors: lines.append(f"| {d['id_number']} | {d['names']} | {d['surname']} | {d['type']} | {d['status']} |") lines.append("") filed = parsed.get("filed_annual_returns", []) if filed: lines.append("### FILED ANNUAL RETURNS") lines.append("| AR Year | Amount Paid | Date Filed |") lines.append("|---|---|---|") for f in filed: lines.append(f"| {f['year']} | {f['amount_paid']} | {f['date_filed']} |") lines.append("") outstanding = parsed.get("outstanding_annual_returns", []) if outstanding: lines.append("### OUTSTANDING ANNUAL RETURNS") lines.append("| AR Year | AR Month | Non-Compliance Date |") lines.append("|---|---|---|") for o in outstanding: lines.append(f"| {o['year']} | {o['month']} | {o['non_compliance_date']} |") lines.append("") other = parsed.get("other_details") or {} if other: lines.append("### OTHER DETAILS") for label, key in [ ("SARS Tax Number", "sars_tax_number"), ("UIF Registration Number", "uif_registration_number"), ("Compensation Fund Registration Number", "compensation_fund_registration_number"), ]: val = other.get(key) if val is None or val == "": display = "Not available" else: display = val lines.append(f"- {label}: {display}") lines.append("") return "\n".join(lines) # =================================================================== main def process_company(name: str, token: str, login_idx: list[int], force_relogin: bool = False) -> int: """Process one company. Returns 0 on success, 1 on lookup failure, 2 on stop (limit hit).""" print(f"\n========== {name} ==========") rec = lookup_company_by_name(name, token) if not rec: print(f" ERROR: no PB record matched '{name}'") return 1 k_number = rec.get("enterprise_no") if not k_number: print(f" ERROR: PB record for '{name}' has no enterprise_no") return 1 print(f" PB record: id={rec['id']}, name='{rec.get('enterprise_name')}', K={k_number}") # Check if we need to login cur_url = ab_eval("window.location.href") if force_relogin or "user_profile" not in cur_url and "bizprofile" not in cur_url: # Need to login creds = CREDENTIALS[login_idx[0]] login_url = login(creds) if "user_profile.aspx" not in login_url: # Login may have failed — try next credential set print(f" login failed for {creds['name']}, URL after login: {login_url}") if login_idx[0] < len(CREDENTIALS) - 1: login_idx[0] += 1 return process_company(name, token, login_idx, force_relogin=True) print(" ERROR: all credentials exhausted") return 2 print(f" logged in OK ({creds['name']})") else: print(f" already on {cur_url}") # Navigate to bizprofile (skip if already there — re-opening in same session disrupts page state) if "bizprofile.aspx" not in cur_url: bp_url = navigate_to_bizprofile() if "bizprofile.aspx" not in bp_url: print(f" ERROR: bizprofile nav failed: {bp_url}") return 2 else: print(f" already on bizprofile, skipping nav") # Search body = search_by_enterprise_no(k_number) parsed = parse_result(body) # Verify we got real data (sanity) if not parsed.get("company_details", {}).get("enterprise_number"): print(" WARN: parsed body has no company_number — search may have failed silently.") print(" Body excerpt:", body[:400]) return 2 # bail; do not waste a search slot on rotation # Save to PB patch_body = { **parsed, "searched_at": dt.datetime.now().isoformat(timespec="seconds"), "searched_k_number": k_number, "searched_via_login": CREDENTIALS[login_idx[0]]["name"], } try: patch_bizportal_json(rec["id"], patch_body, token) print(f" PATCHed PB record {rec['id']} (bizportal_data_json + last_checked)") except SystemExit as e: print(f" PB PATCH failed: {e}") # Print to chat print() print(format_markdown(parsed, k_number)) # Re-check counter for the next iteration new_counter = read_counter() if new_counter >= SEARCH_LIMIT: print(f"\n counter now {new_counter} — switching login for next company") if login_idx[0] < len(CREDENTIALS) - 1: login_idx[0] += 1 # Log out to clear session cleanly try: ab_eval("document.getElementById('ctl00$btnLogoutNav') ? document.getElementById('ctl00$btnLogoutNav').click() : 'NO_LOGOUT_BTN'; 'L'") ab_sleep(3000) except Exception: pass return 0 def main(): global SESSION_NAME # noqa: PLW0603 (intentional module-global write) parser = argparse.ArgumentParser(description="CIPC BizProfile search with auto-rotation + PB save.") parser.add_argument("names", nargs="*", help="One or more company names (substring match on PB enterprise_name).") parser.add_argument("--batch", help="Path to a text file with one company name per line.") parser.add_argument("--session", default=SESSION_NAME, help="agent-browser session name.") args = parser.parse_args() SESSION_NAME = args.session names: list[str] = list(args.names) if args.batch: batch_path = Path(args.batch) if not batch_path.exists(): print(f"Batch file not found: {batch_path}") sys.exit(1) for line in batch_path.read_text(encoding="utf-8").splitlines(): line = line.strip() if line and not line.startswith("#"): names.append(line) if not names: parser.error("Provide company names as args or via --batch.") token = load_pb_token() login_idx = [0] # mutable container so recursion can update failures = 0 for name in names: rc = process_company(name, token, login_idx) if rc == 2: print(f"\nSTOP: hard stop reached on '{name}'.") sys.exit(2) if rc == 1: failures += 1 if failures: print(f"\nDone with {failures} lookup failures out of {len(names)}.") sys.exit(1) print(f"\nDone. {len(names)} company/companies processed.") if __name__ == "__main__": main()