11d0178ab1
Adds the full Next.js app (app/, components/, lib/, scripts/, assets/, docs/, Dockerfile, compose, configs) to version control. Security: - CIPC BizPortal customer-code logins (Login 1 + Login 3) now read CIPC_LOGIN1_CODE / CIPC_LOGIN1_PASSWORD / CIPC_LOGIN3_CODE / CIPC_LOGIN3_PASSWORD from .env.local instead of source. - BizPortal personal login in apply-bizportal route reads BIZPORTAL_ID_NO / BIZPORTAL_PASSWORD. - Added .env.example as the committed template. .gitignore excludes .env*, node_modules, .next, logs, sqlite, Playwright storage_state.json (live CIPC cookies), scratch dirs, client PDFs and debug screenshots.
647 lines
25 KiB
Python
647 lines
25 KiB
Python
"""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()
|