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.
871 lines
34 KiB
Python
871 lines
34 KiB
Python
"""
|
|
SARS Retun EMP201 single or all pty — extract EMP201 payment breakdowns
|
|
for one or many companies over one or more periods.
|
|
|
|
This is the batch-capable superset of `SARS Return EMP201 data.py`:
|
|
* Run for ONE company: --company "X"
|
|
* Run for SELECTED: --companies "X,Y,Z"
|
|
* Run for ALL active: --all-active
|
|
|
|
Output:
|
|
* JSON file: C:\\Users\\SJ\\Downloads\\sars_captures\\YYYY-MM-DD_HHMMSS.json
|
|
Schema:
|
|
{
|
|
"captured_at": "2026-09-10T15:30:00", # when this run started
|
|
"last_updated": "2026-09-10T15:32:14", # when this run finished
|
|
"return_type": "EMP201",
|
|
"periods_requested": ["202603","202604","202605"],
|
|
"companies": [
|
|
{
|
|
"name": "WOEL EN WERSKAF KLEUTERSKOOL",
|
|
"login_as": "bertuslouw9194",
|
|
"portfolio": "Tax Practitioner",
|
|
"paye_no": "7760830689",
|
|
"last_updated": "2026-09-10T15:31:02", # per-company, when captured
|
|
"results": [
|
|
{"period":"202603","paye":"...","sdl":"...","uif":"...","total":"...","payment_ref":"..."},
|
|
...
|
|
],
|
|
"not_filed": ["202604"], # periods the company did not submit
|
|
"errors": [] # only TRUE script/runtime failures
|
|
},
|
|
...
|
|
]
|
|
}
|
|
|
|
Per-period outcome is bucketed into three arrays, never conflated:
|
|
* "results" — captured; has parsed fields
|
|
* "not_filed" — company did not submit this period (NOT a script error)
|
|
Triggered when: (a) the period row is missing from
|
|
the period list, (b) the work page has no EMP201
|
|
link, or (c) the form page has no InputXML.
|
|
* "errors" — script/runtime failure (network, DOM, agent-browser)
|
|
|
|
* Stdout: a short summary table.
|
|
|
|
"Active client" definition (for --all-active):
|
|
PocketBase record on ysterbaard_companies where
|
|
is_client = true AND sars_paye_no is non-empty.
|
|
|
|
The PB write is NOT done here. This script only captures. A future
|
|
"save to PB" command will read the JSON and POST/PATCH records.
|
|
"""
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import List, Optional
|
|
|
|
|
|
# ── Constants ─────────────────────────────────────────────────────
|
|
|
|
CAPTURES_DIR = Path(r"C:\Users\SJ\Downloads\sars_captures")
|
|
PB_COMPANIES = "ysterbaard_companies"
|
|
DTC_USERNAME = "sjcoetzee1092"
|
|
|
|
# Loaded from .env.local by the import below.
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
from SARS_login_Select_Company import ( # noqa: E402
|
|
_pb_get, PB_URL, PB_TOKEN,
|
|
find_company, fetch_dtc_password,
|
|
)
|
|
|
|
|
|
# ── agent-browser executable discovery ─────────────────────────────
|
|
|
|
def _find_agent_browser_exe() -> str:
|
|
import shutil
|
|
for name in ("agent-browser.cmd", "agent-browser.exe",
|
|
"agent-browser.bat", "agent-browser"):
|
|
path = shutil.which(name)
|
|
if path:
|
|
return path
|
|
npm_dir = Path(os.path.expanduser("~")) / "AppData" / "Roaming" / "npm"
|
|
for name in ("agent-browser.cmd", "agent-browser.exe", "agent-browser.ps1"):
|
|
candidate = npm_dir / name
|
|
if candidate.exists():
|
|
return str(candidate)
|
|
raise RuntimeError(
|
|
"Could not find agent-browser. Install with: npm i -g agent-browser"
|
|
)
|
|
|
|
|
|
AGENT_BROWSER_EXE = _find_agent_browser_exe()
|
|
|
|
|
|
# ── subprocess wrapper ─────────────────────────────────────────────
|
|
|
|
_WRITE_COMMANDS = {
|
|
"open", "close", "navigate", "fill", "click", "wait",
|
|
"press_key", "check", "uncheck", "type", "scroll",
|
|
"drag", "double_click", "hover",
|
|
}
|
|
|
|
|
|
def run_ab(*args, timeout: int = 30) -> str:
|
|
"""Run an agent-browser command, return stdout (stripped).
|
|
|
|
Two strategies:
|
|
* "write" commands (open/close/navigate/fill/click/wait/...) — discard
|
|
stdout (DEVNULL). The temp-file redirect reliably HANGS on these.
|
|
* "read" commands (eval/snapshot/...) — redirect to a temp file.
|
|
`capture_output=True`/`stdout=PIPE` makes `eval` return `null` so
|
|
a real file is required.
|
|
"""
|
|
if args and args[0] in _WRITE_COMMANDS:
|
|
try:
|
|
result = subprocess.run(
|
|
[AGENT_BROWSER_EXE, *args],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
timeout=timeout,
|
|
)
|
|
except subprocess.TimeoutExpired as e:
|
|
raise RuntimeError(
|
|
f"agent-browser {args!r} timed out after {timeout}s"
|
|
) from e
|
|
if result.returncode != 0:
|
|
raise RuntimeError(
|
|
f"agent-browser {args!r} failed (rc={result.returncode})"
|
|
)
|
|
return ""
|
|
|
|
tmp = tempfile.NamedTemporaryFile(
|
|
mode="w", delete=False, encoding="utf-8", errors="replace",
|
|
prefix="ab_", suffix=".out",
|
|
)
|
|
tmp_path = tmp.name
|
|
tmp.close()
|
|
try:
|
|
try:
|
|
result = subprocess.run(
|
|
[AGENT_BROWSER_EXE, *args],
|
|
stdout=open(tmp_path, "w", encoding="utf-8", errors="replace"),
|
|
stderr=subprocess.DEVNULL,
|
|
timeout=timeout,
|
|
)
|
|
except subprocess.TimeoutExpired as e:
|
|
raise RuntimeError(
|
|
f"agent-browser {args!r} timed out after {timeout}s"
|
|
) from e
|
|
with open(tmp_path, "r", encoding="utf-8", errors="replace") as f:
|
|
out = f.read().strip()
|
|
finally:
|
|
try:
|
|
os.unlink(tmp_path)
|
|
except OSError:
|
|
pass
|
|
if result.returncode != 0:
|
|
raise RuntimeError(
|
|
f"agent-browser {args!r} failed (rc={result.returncode}): {out}"
|
|
)
|
|
return out
|
|
|
|
|
|
def eval_js(js: str, timeout: int = 30):
|
|
"""Run a single-line JS expression in the browser, return the result.
|
|
|
|
agent-browser eval requires single-line JS (multi-line returns null)
|
|
and chokes on shell-special characters. We base64-encode via `-b`.
|
|
"""
|
|
b64 = base64.b64encode(js.encode("utf-8")).decode("ascii")
|
|
out = run_ab("eval", "-b", b64, timeout=timeout)
|
|
if out == "null" or not out:
|
|
return None
|
|
if out.startswith('"') and out.endswith('"'):
|
|
out = json.loads(out)
|
|
return out
|
|
|
|
|
|
# ── Period parsing ────────────────────────────────────────────────
|
|
|
|
def expand_periods(spec: str) -> List[str]:
|
|
"""Expand "202608" or "202603-202608" into a list of YYYYMM strings."""
|
|
spec = spec.strip()
|
|
if "-" in spec:
|
|
start, end = spec.split("-", 1)
|
|
start, end = start.strip(), end.strip()
|
|
if (len(start) != 6 or len(end) != 6
|
|
or not start.isdigit() or not end.isdigit()):
|
|
raise ValueError(f"Invalid period range: {spec!r}")
|
|
si, ei = int(start), int(end)
|
|
if si > ei:
|
|
raise ValueError(f"Start period {start} is after end period {end}")
|
|
out = []
|
|
i = si
|
|
while i <= ei:
|
|
out.append(f"{i:06d}")
|
|
y, m = divmod(i, 100)
|
|
m += 1
|
|
if m > 12:
|
|
m = 1
|
|
y += 1
|
|
i = y * 100 + m
|
|
return out
|
|
if len(spec) != 6 or not spec.isdigit():
|
|
raise ValueError(f"Invalid period: {spec!r}")
|
|
return [spec]
|
|
|
|
|
|
# ── PocketBase lookups ─────────────────────────────────────────────
|
|
|
|
def list_active_clients() -> List[dict]:
|
|
"""Return all ysterbaard_companies records that are:
|
|
is_client = true AND sars_paye_no is non-empty.
|
|
|
|
These are the companies that --all-active will iterate over.
|
|
"""
|
|
# PB's filter syntax: is_client=true && sars_paye_no!=''
|
|
# The != '' filter on a text field needs a different operator; we
|
|
# fetch is_client=true and filter sars_paye_no in Python.
|
|
items = _pb_get(
|
|
f"/api/collections/{PB_COMPANIES}/records",
|
|
{"filter": "is_client=true", "perPage": 500},
|
|
)["items"]
|
|
return [r for r in items if (r.get("sars_paye_no") or "").strip()]
|
|
|
|
|
|
def fetch_login_info_for_record(record: dict):
|
|
"""Same logic as the single-company script's fetch_login_info, but
|
|
takes a PB record directly (no name lookup)."""
|
|
name = record.get("enterprise_name") or ""
|
|
own_username = (record.get("sars_login_username") or "").strip()
|
|
own_password = (record.get("sars_login_password") or "").strip()
|
|
|
|
if own_username == "bertuslouw9194" and own_password:
|
|
# The ONE own-login case: WOEL EN WERSKAF KLEUTERSKOOL.
|
|
# Every other company — even ones with a sars_login_username
|
|
# set for reference — is processed under DTC on the company's
|
|
# group portfolio. Matches the rule in
|
|
# `SARS_login_Select_Company.determine_portfolio`.
|
|
return name, own_username, own_password, "Tax Practitioner"
|
|
|
|
# DTC fallback
|
|
from SARS_login_Select_Company import determine_portfolio
|
|
portfolio = determine_portfolio(record)
|
|
password = fetch_dtc_password()
|
|
return name, DTC_USERNAME, password, portfolio
|
|
|
|
|
|
# ── InputXML parser ──────────────────────────────────────────────
|
|
|
|
def parse_input_xml(xml_str: str) -> dict:
|
|
"""Extract the 8 fields from the EMP201 InputXML string."""
|
|
xml_str = (
|
|
xml_str.replace("<", "<")
|
|
.replace(">", ">")
|
|
.replace(""", '"')
|
|
.replace("&", "&")
|
|
)
|
|
def get(tag):
|
|
m = re.search(rf"<{tag}>([^<]*)</{tag}>", xml_str)
|
|
return m.group(1) if m else ""
|
|
|
|
return {
|
|
"company_name": get("TRADING_NAME"),
|
|
"paye_no": get("PAYE_REF_NO"),
|
|
"payment_period": get("PAYMENT_PERIOD"),
|
|
"paye_payable": get("PAYE_AMT"),
|
|
"sdl_payable": get("SDL_AMT"),
|
|
"uif_payable": get("UIF_AMT"),
|
|
"total_payable": get("TOTAL_PAYMENT"),
|
|
"payment_ref_no": get("PAYMENT_REF_NO"),
|
|
}
|
|
|
|
|
|
# ── Snapshot parsing + session detection ──────────────────────────
|
|
|
|
def _find_ref(snap: str, role: str, name: Optional[str] = None) -> Optional[str]:
|
|
"""Parse the agent-browser snapshot for an element ref.
|
|
|
|
Snapshot format examples:
|
|
- textbox "Username" [required, ref=e9]
|
|
- button "Next" [ref=e7]
|
|
- heading "Welcome" [level=1, ref=e3]
|
|
The `ref` is the LAST attribute inside the `[...]` block, so we
|
|
match `,?\\s*ref=eN]`.
|
|
"""
|
|
if name is None:
|
|
pattern = rf"\b{re.escape(role)}\b[^\n]*?[,]?\s*ref=(e\d+)\]"
|
|
else:
|
|
pattern = (
|
|
rf"\b{re.escape(role)}\b[^\n]*?['\"]({re.escape(name)})['\"][^\n]*?"
|
|
rf"[,]?\s*ref=(e\d+)\]"
|
|
)
|
|
m = re.search(pattern, snap, re.MULTILINE)
|
|
if m:
|
|
return m.group(2) if m.lastindex and m.lastindex >= 2 else m.group(1)
|
|
return None
|
|
|
|
|
|
def _page_url() -> str:
|
|
try:
|
|
out = eval_js("window.location.href || ''", timeout=10)
|
|
return out or ""
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def _is_on_dashboard() -> bool:
|
|
return "/dashboard" in _page_url()
|
|
|
|
|
|
def _is_on_login() -> bool:
|
|
url = _page_url()
|
|
return url.endswith("/login") or "/login?" in url or url.endswith("/login/")
|
|
|
|
|
|
def _current_user_hint() -> str:
|
|
"""Best-effort: detect which user is logged in by looking at the
|
|
Taxpayer combobox label. Returns the combobox value or ''."""
|
|
try:
|
|
js = "(() => { const el = document.querySelector('mat-select[name=\"TaxPayer\"]'); return el ? (el.textContent || '').trim() : ''; })()"
|
|
out = eval_js(js, timeout=10)
|
|
return out or ""
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
# ── agent-browser flow ───────────────────────────────────────────
|
|
|
|
def login(username: str, password: str) -> None:
|
|
"""Log in to SARS eFiling (or skip if already on dashboard)."""
|
|
if _is_on_dashboard():
|
|
print(" [login] already on dashboard ...", flush=True)
|
|
return
|
|
|
|
if _is_on_login():
|
|
print(" [login] on login page (session reused) ...", flush=True)
|
|
else:
|
|
print(" [login] no usable session, starting fresh ...", flush=True)
|
|
try:
|
|
run_ab("close", "--all", timeout=8)
|
|
except Exception as e:
|
|
print(f" [login] close --all skipped ({e.__class__.__name__})", flush=True)
|
|
try:
|
|
run_ab("open", timeout=15)
|
|
except Exception as e:
|
|
print(f" [login] open failed ({e.__class__.__name__}); continuing", flush=True)
|
|
try:
|
|
run_ab("navigate", "https://secure.sarsefiling.co.za/app/login", timeout=15)
|
|
except Exception as e:
|
|
print(f" [login] navigate failed ({e.__class__.__name__})", flush=True)
|
|
run_ab("wait", "3500", timeout=15)
|
|
|
|
print(" [login] filling username ...", flush=True)
|
|
snap = run_ab("snapshot", "-i", timeout=15)
|
|
user_ref = _find_ref(snap, "textbox", "Username")
|
|
if not user_ref:
|
|
raise RuntimeError("Username textbox not found")
|
|
run_ab("fill", f"@{user_ref}", username, timeout=15)
|
|
next_ref = _find_ref(snap, "button", "Next")
|
|
if not next_ref:
|
|
raise RuntimeError("Next button not found")
|
|
run_ab("click", f"@{next_ref}", timeout=15)
|
|
run_ab("wait", "3000", timeout=15)
|
|
|
|
print(" [login] filling password ...", flush=True)
|
|
snap = run_ab("snapshot", "-i", timeout=15)
|
|
pw_ref = _find_ref(snap, "textbox", "Password")
|
|
if not pw_ref:
|
|
raise RuntimeError("Password textbox not found")
|
|
run_ab("fill", f"@{pw_ref}", password, timeout=15)
|
|
login_ref = _find_ref(snap, "button", "Login")
|
|
if not login_ref:
|
|
raise RuntimeError("Login button not found")
|
|
run_ab("click", f"@{login_ref}", timeout=15)
|
|
run_ab("wait", "5000", timeout=15)
|
|
print(" [login] done.", flush=True)
|
|
|
|
|
|
def select_dropdown_value(select_name: str, target_text: str) -> None:
|
|
"""Pick a value from an Angular Material mat-select by display text.
|
|
Uses 'Show all' if virtual scroll hides the option.
|
|
|
|
Matching: case-insensitive contains. This handles the common case
|
|
where the PB group name is `TME` but the mat-option is rendered as
|
|
`Group TME` (with prefix). Old exact-match missed that.
|
|
"""
|
|
print(f" [dashboard] open mat-select[{select_name!r}] ...", flush=True)
|
|
eval_js(
|
|
f"document.querySelector('mat-select[name=\"{select_name}\"]').click(); 'ok'",
|
|
timeout=15,
|
|
)
|
|
time.sleep(2)
|
|
needle = target_text.strip().lower()
|
|
js_pick = (
|
|
"(() => { const opts = document.querySelectorAll('.cdk-overlay-pane mat-option'); "
|
|
"for (const o of opts) { if (o.textContent.trim().toLowerCase().includes("
|
|
+ json.dumps(needle) + ")) { o.click(); return 'clicked'; } } "
|
|
"return 'not in view'; })()"
|
|
)
|
|
res = eval_js(js_pick, timeout=15)
|
|
if res == "clicked":
|
|
time.sleep(2)
|
|
return
|
|
print(f" [dashboard] target not in view, clicking 'Show all' ...", flush=True)
|
|
js_show_all = (
|
|
"(() => { const opts = document.querySelectorAll('.cdk-overlay-pane mat-option'); "
|
|
"const show = Array.from(opts).find(o => o.textContent.trim() === 'Show all'); "
|
|
"if (show) { show.click(); return 'show all clicked'; } return 'no show all'; })()"
|
|
)
|
|
eval_js(js_show_all, timeout=15)
|
|
time.sleep(2)
|
|
js_pick2 = (
|
|
"(() => { const opts = document.querySelectorAll('.cdk-overlay-pane mat-option'); "
|
|
"for (const o of opts) { if (o.textContent.trim().toLowerCase().includes("
|
|
+ json.dumps(needle) + ")) { o.click(); return 'clicked after show all'; } } "
|
|
"return 'still not found'; })()"
|
|
)
|
|
res = eval_js(js_pick2, timeout=15)
|
|
if res != "clicked after show all":
|
|
raise RuntimeError(
|
|
f"Could not select {target_text!r} in mat-select[{select_name!r}]"
|
|
)
|
|
time.sleep(2)
|
|
|
|
|
|
def navigate_to_returns_history_subitem(return_type: str) -> None:
|
|
SUBITEM_LABELS = {
|
|
"EMP201": "Employee's Tax (EMP201)",
|
|
"EMP501": "Employee s Tax (EMP501)",
|
|
}
|
|
label = SUBITEM_LABELS[return_type]
|
|
print(f" [nav] click 'Returns History' parent ...", flush=True)
|
|
snap = run_ab("snapshot", "-i", timeout=15)
|
|
parent_ref = _find_ref(snap, "link", "Returns History")
|
|
if not parent_ref:
|
|
raise RuntimeError("Returns History link not found")
|
|
run_ab("click", f"@{parent_ref}", timeout=15)
|
|
time.sleep(3)
|
|
print(f" [nav] click sub-item {label!r} ...", flush=True)
|
|
snap = run_ab("snapshot", "-i", timeout=15)
|
|
sub_ref = _find_ref(snap, "link", label)
|
|
if not sub_ref:
|
|
raise RuntimeError(f"Sub-item link {label!r} not found")
|
|
run_ab("click", f"@{sub_ref}", timeout=15)
|
|
time.sleep(4)
|
|
|
|
|
|
def get_period_open_links() -> List[dict]:
|
|
js = (
|
|
"(() => { const ifr = document.querySelectorAll('iframe')[0]; "
|
|
"const doc = ifr.contentDocument; "
|
|
"const rows = Array.from(doc.querySelectorAll('tr')); "
|
|
"const out = []; "
|
|
"for (const r of rows) { "
|
|
" const m = r.textContent.match(/20260[0-9]/); "
|
|
" if (!m) continue; "
|
|
" const period = m[0]; "
|
|
" const link = Array.from(r.querySelectorAll('a')).find(a => "
|
|
" a.textContent.trim() === 'Open' && a.href && a.href.includes('LoadPage')); "
|
|
" if (link) out.push({ period, href: link.href }); "
|
|
"} "
|
|
"return JSON.stringify(out); })()"
|
|
)
|
|
res = eval_js(js, timeout=30)
|
|
if not res or res == "null":
|
|
return []
|
|
if isinstance(res, str):
|
|
return json.loads(res)
|
|
return res
|
|
|
|
|
|
def get_period_status_texts() -> dict:
|
|
"""Read the Status column text from each period row on the Returns
|
|
History list. Returns { period: status_text }. Empty status means
|
|
the cell wasn't found (e.g. row layout changed). Best-effort: if
|
|
the eval fails the dict is just empty and date_submitted will be
|
|
left None for every period."""
|
|
js = (
|
|
"(() => { const ifr = document.querySelectorAll('iframe')[0]; "
|
|
"const doc = ifr.contentDocument; "
|
|
"const rows = Array.from(doc.querySelectorAll('tr')); "
|
|
"const out = {}; "
|
|
"for (const r of rows) { "
|
|
" const m = r.textContent.match(/20260[0-9]/); "
|
|
" if (!m) continue; "
|
|
" const period = m[0]; "
|
|
" const tds = Array.from(r.querySelectorAll('td, th')); "
|
|
" for (const td of tds) { "
|
|
" if (/Filed/.test(td.textContent)) { out[period] = td.textContent.trim(); break; } "
|
|
" } "
|
|
"} "
|
|
"return JSON.stringify(out); })()"
|
|
)
|
|
try:
|
|
res = eval_js(js, timeout=20)
|
|
except Exception:
|
|
return {}
|
|
if not res or res == "null":
|
|
return {}
|
|
try:
|
|
return json.loads(res) if isinstance(res, str) else res
|
|
except (json.JSONDecodeError, TypeError):
|
|
return {}
|
|
|
|
|
|
def parse_status_date(status_text: str) -> Optional[str]:
|
|
"""Parse 'YYYY/MM/DD' (or 'DD/MM/YYYY') out of a SARS Returns History
|
|
Status cell. The cell text looks like 'Filed through eFiling 24/06/2026'
|
|
or 'Submitted on 2026/06/24'. Returns ISO 'YYYY-MM-DD' or None."""
|
|
if not status_text:
|
|
return None
|
|
# Try YYYY/MM/DD first (some views render that way)
|
|
m = re.search(r"(\d{4})/(\d{2})/(\d{2})", status_text)
|
|
if m:
|
|
return f"{m.group(1)}-{m.group(2)}-{m.group(3)}"
|
|
# Try DD/MM/YYYY (the format the user reported)
|
|
m = re.search(r"(\d{2})/(\d{2})/(\d{4})", status_text)
|
|
if m:
|
|
return f"{m.group(3)}-{m.group(2)}-{m.group(1)}"
|
|
return None
|
|
|
|
|
|
def navigate_iframe_to(url: str) -> None:
|
|
js = f"document.querySelectorAll('iframe')[0].src = {json.dumps(url)}; 'ok'"
|
|
eval_js(js, timeout=15)
|
|
|
|
|
|
def click_emp201_link() -> bool:
|
|
js = (
|
|
"(() => { const ifr = document.querySelectorAll('iframe')[0]; "
|
|
"const doc = ifr.contentDocument; "
|
|
"const links = Array.from(doc.querySelectorAll('a')); "
|
|
"let target = links.find(a => a.href && a.href.includes('BPEMP201Form')); "
|
|
"if (!target) target = links.find(a => a.textContent.trim() === 'EMP201'); "
|
|
"if (!target) return JSON.stringify({ clicked: false }); "
|
|
"target.click(); "
|
|
"return JSON.stringify({ clicked: true, href: target.href }); })()"
|
|
)
|
|
res = eval_js(js, timeout=15)
|
|
if isinstance(res, str):
|
|
try:
|
|
res = json.loads(res)
|
|
except json.JSONDecodeError:
|
|
res = {}
|
|
return bool(res and res.get("clicked"))
|
|
|
|
|
|
def read_input_xml() -> str:
|
|
js = (
|
|
"(() => { const ifr = document.querySelectorAll('iframe')[0]; "
|
|
"const doc = ifr.contentDocument; "
|
|
"const all = Array.from(doc.querySelectorAll('input')); "
|
|
"const x = all.find(i => i.name === 'InputXML'); "
|
|
"return x ? x.value : ''; })()"
|
|
)
|
|
res = eval_js(js, timeout=15)
|
|
return res or ""
|
|
|
|
|
|
def read_status_date() -> Optional[str]:
|
|
"""Read the 'Status Date' cell from the EMP201 work page table.
|
|
|
|
SARS renders the submission date as a labelled row in a small table
|
|
on the read-only work page (the page reached by clicking "Open" on
|
|
the Returns History list). We look for any cell whose trimmed text
|
|
is exactly 'Status Date', then read the adjacent cell (same row,
|
|
next column). The date is shown as 'YYYY/MM/DD' (e.g. '2026/06/24')
|
|
on SARS; we normalise to ISO 'YYYY-MM-DD' for PB.
|
|
|
|
Returns the ISO string, or None if the cell is missing / unparseable
|
|
/ the eval transiently failed. None is not a script error — the
|
|
caller leaves the field empty rather than failing the whole period.
|
|
"""
|
|
has_cell_js = (
|
|
"(() => { "
|
|
" const ifr = document.querySelectorAll('iframe')[0]; "
|
|
" const doc = ifr && ifr.contentDocument; "
|
|
" if (!doc) return 'no_doc'; "
|
|
" const cells = Array.from(doc.querySelectorAll('td, th')); "
|
|
" const label = cells.find(c => c.textContent.trim() === 'Status Date'); "
|
|
" return label ? ('yes:' + cells.length) : ('no:' + cells.length); "
|
|
"})()"
|
|
)
|
|
try:
|
|
res = eval_js(has_cell_js, timeout=15)
|
|
except Exception:
|
|
# agent-browser eval can transiently fail (rc=1, empty output)
|
|
# when the work page is mid-navigation. The caller's retry loop
|
|
# will try again.
|
|
return None
|
|
if not res or not res.startswith("yes:"):
|
|
return None
|
|
value_js = (
|
|
"(() => { "
|
|
" const doc = document.querySelectorAll('iframe')[0].contentDocument; "
|
|
" const cells = Array.from(doc.querySelectorAll('td, th')); "
|
|
" const label = cells.find(c => c.textContent.trim() === 'Status Date'); "
|
|
" if (!label) return ''; "
|
|
" let next = label.nextElementSibling; "
|
|
" if (!next) { "
|
|
" const row = label.parentElement; "
|
|
" if (row) { "
|
|
" const tds = Array.from(row.querySelectorAll('td, th')); "
|
|
" const idx = tds.indexOf(label); "
|
|
" if (idx >= 0 && idx + 1 < tds.length) next = tds[idx + 1]; "
|
|
" } "
|
|
" } "
|
|
" return next ? next.textContent.trim() : ''; "
|
|
"})()"
|
|
)
|
|
try:
|
|
raw = eval_js(value_js, timeout=15)
|
|
except Exception:
|
|
return None
|
|
if not raw:
|
|
return None
|
|
# Normalise 'YYYY/MM/DD' → 'YYYY-MM-DD' for PB.
|
|
m = re.match(r"^(\d{4})/(\d{2})/(\d{2})$", raw)
|
|
if not m:
|
|
return None
|
|
return f"{m.group(1)}-{m.group(2)}-{m.group(3)}"
|
|
|
|
|
|
def force_logout() -> None:
|
|
"""Best-effort: navigate to a logout URL so the next company can
|
|
sign in as a different SARS user. Falls back silently if no
|
|
logout endpoint works."""
|
|
try:
|
|
run_ab("navigate", "https://secure.sarsefiling.co.za/app/auth/logout", timeout=15)
|
|
except Exception:
|
|
pass
|
|
time.sleep(2)
|
|
# If the logout URL didn't log us out, try a JavaScript-based
|
|
# session wipe + navigate to login.
|
|
try:
|
|
eval_js("document.cookie.split(';').forEach(c => { document.cookie = c.replace(/^ +/, '').replace(/=.*/, '=;expires=' + new Date().toUTCString() + ';path=/'); }); 'cleared'", timeout=10)
|
|
except Exception:
|
|
pass
|
|
try:
|
|
run_ab("navigate", "https://secure.sarsefiling.co.za/app/login", timeout=15)
|
|
except Exception:
|
|
pass
|
|
time.sleep(2)
|
|
|
|
|
|
# ── Capture flow for one company ─────────────────────────────────
|
|
|
|
def capture_one_company(record: dict, periods: List[str]) -> dict:
|
|
"""Capture one company's EMP201 data. Returns a dict ready for
|
|
the JSON output. Never raises.
|
|
|
|
Per-period outcomes are bucketed into three arrays:
|
|
results — captured; has parsed fields
|
|
not_filed — company did not submit this period (NOT a script error)
|
|
errors — script/runtime failure (network, DOM, agent-browser)
|
|
"""
|
|
started_at = datetime.now().isoformat(timespec="seconds")
|
|
name, username, password, portfolio = fetch_login_info_for_record(record)
|
|
|
|
print(f"\n=== {name} (login as {username}, portfolio {portfolio!r}) ===",
|
|
flush=True)
|
|
company_record: dict = {
|
|
"name": name,
|
|
"pb_id": record.get("id"),
|
|
"login_as": username,
|
|
"portfolio": portfolio,
|
|
"paye_no": (record.get("sars_paye_no") or "").strip(),
|
|
"last_updated": None,
|
|
"results": [],
|
|
"not_filed": [],
|
|
"errors": [],
|
|
}
|
|
|
|
try:
|
|
# If the session is logged in as a different user, log out first
|
|
current_taxpayer = _current_user_hint()
|
|
if _is_on_dashboard() and current_taxpayer and current_taxpayer != name:
|
|
print(f" [switch] session shows {current_taxpayer!r}; "
|
|
f"forcing logout before logging in as {name!r} ...", flush=True)
|
|
force_logout()
|
|
|
|
login(username, password)
|
|
select_dropdown_value("taxuser", portfolio)
|
|
select_dropdown_value("TaxPayer", name)
|
|
navigate_to_returns_history_subitem("EMP201")
|
|
|
|
all_links = get_period_open_links()
|
|
print(f" [capture] found {len(all_links)} periods on page", flush=True)
|
|
# Read the Status column text from the same list view. The
|
|
# Returns History list shows 'Filed through eFiling DD/MM/YYYY'
|
|
# for each filed period, which is exactly the date we want —
|
|
# and we already proved this eval works in `get_period_open_links`
|
|
# above, so it's reliable.
|
|
status_by_period = get_period_status_texts()
|
|
# Build per-period maps for both the work-page href (drives the
|
|
# InputXML capture) AND the Status cell text (gives us the
|
|
# SARS-side submission date without round-tripping to the work
|
|
# page — see note below).
|
|
href_by_period = {e["period"]: e["href"] for e in all_links}
|
|
|
|
for period in periods:
|
|
# Case 1: period not in the list at all → not filed
|
|
if period not in href_by_period:
|
|
print(f" {period}: not filed (no row in period list)", flush=True)
|
|
company_record["not_filed"].append(period)
|
|
continue
|
|
href = href_by_period[period]
|
|
# Submission date is captured here from the Returns History
|
|
# list's Status column (e.g. 'Filed through eFiling 24/06/2026')
|
|
# rather than the work page — the work page's eval was
|
|
# unreliable due to iframe-load timing. The list view eval
|
|
# already ran successfully above (it's the same data source
|
|
# `get_period_open_links` uses), so this is the safer read.
|
|
date_submitted = parse_status_date(status_by_period.get(period, ""))
|
|
print(f" {period}: work page ...", flush=True)
|
|
try:
|
|
navigate_iframe_to(href)
|
|
time.sleep(3.5)
|
|
# Case 2: work page exists but no EMP201 link → not filed
|
|
if not click_emp201_link():
|
|
print(f" {period}: not filed (no EMP201 link on work page)",
|
|
flush=True)
|
|
company_record["not_filed"].append(period)
|
|
continue
|
|
xml = ""
|
|
for _ in range(8):
|
|
time.sleep(2)
|
|
xml = read_input_xml()
|
|
if xml:
|
|
break
|
|
# Case 3: EMP201 link clicked but no InputXML → not filed
|
|
# (e.g. period was opened but never submitted)
|
|
if not xml:
|
|
print(f" {period}: not filed (no InputXML on form page)",
|
|
flush=True)
|
|
company_record["not_filed"].append(period)
|
|
continue
|
|
fields = parse_input_xml(xml)
|
|
fields["period"] = period
|
|
fields["date_submitted"] = date_submitted
|
|
company_record["results"].append(fields)
|
|
print(f" {period}: PAYE={fields['paye_payable']} "
|
|
f"SDL={fields['sdl_payable']} UIF={fields['uif_payable']} "
|
|
f"Total={fields['total_payable']} "
|
|
f"Ref={fields['payment_ref_no']}", flush=True)
|
|
except Exception as e:
|
|
# Anything else (network, agent-browser, JS error) is a
|
|
# genuine error, not "not filed".
|
|
msg = f"{period}: {type(e).__name__}: {e}"
|
|
print(f" {msg}", flush=True)
|
|
company_record["errors"].append(msg)
|
|
|
|
except Exception as e:
|
|
msg = f"{type(e).__name__}: {e}"
|
|
print(f" [error] {msg}", flush=True)
|
|
company_record["errors"].append(msg)
|
|
|
|
company_record["last_updated"] = datetime.now().isoformat(timespec="seconds")
|
|
print(f" [done] {name}: {len(company_record['results'])} captured, "
|
|
f"{len(company_record['not_filed'])} not filed, "
|
|
f"{len(company_record['errors'])} errors "
|
|
f"(started {started_at}, finished {company_record['last_updated']})",
|
|
flush=True)
|
|
return company_record
|
|
|
|
|
|
# ── JSON output ──────────────────────────────────────────────────
|
|
|
|
def write_json_output(payload: dict) -> Path:
|
|
CAPTURES_DIR.mkdir(parents=True, exist_ok=True)
|
|
stamp = datetime.now().strftime("%Y-%m-%d_%H%M%S")
|
|
out = CAPTURES_DIR / f"{stamp}.json"
|
|
with open(out, "w", encoding="utf-8") as f:
|
|
json.dump(payload, f, indent=2, ensure_ascii=False)
|
|
return out
|
|
|
|
|
|
# ── Main ─────────────────────────────────────────────────────────
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(
|
|
description="Capture SARS EMP201 data for one or more companies."
|
|
)
|
|
src = ap.add_mutually_exclusive_group(required=True)
|
|
src.add_argument("--company", help='Single company name as in PB')
|
|
src.add_argument("--companies", help='Comma-separated company names')
|
|
src.add_argument("--all-active", action="store_true",
|
|
help="All PB records with is_client=true AND sars_paye_no set")
|
|
ap.add_argument("--return-type", default="EMP201")
|
|
ap.add_argument("--periods", required=True,
|
|
help='One period "202608" or range "202603-202608"')
|
|
args = ap.parse_args()
|
|
|
|
if args.return_type != "EMP201":
|
|
raise SystemExit(f"return-type {args.return_type!r} not yet supported; only EMP201")
|
|
|
|
periods = expand_periods(args.periods)
|
|
|
|
# Resolve the company list
|
|
if args.company:
|
|
record = find_company(args.company)
|
|
records = [record]
|
|
print(f"Mode: single company ({args.company!r})", flush=True)
|
|
elif args.companies:
|
|
names = [n.strip() for n in args.companies.split(",") if n.strip()]
|
|
records = [find_company(n) for n in names]
|
|
print(f"Mode: {len(records)} selected companies", flush=True)
|
|
else: # --all-active
|
|
records = list_active_clients()
|
|
print(f"Mode: all-active ({len(records)} clients found in PB)", flush=True)
|
|
|
|
if not records:
|
|
raise SystemExit("No companies to process.")
|
|
|
|
print(f"Return type: {args.return_type}", flush=True)
|
|
print(f"Periods: {periods}", flush=True)
|
|
print(f"Companies: {len(records)}", flush=True)
|
|
|
|
captured_at = datetime.now().isoformat(timespec="seconds")
|
|
companies_out = []
|
|
for i, rec in enumerate(records, 1):
|
|
print(f"\n--- Company {i}/{len(records)} ---", flush=True)
|
|
companies_out.append(capture_one_company(rec, periods))
|
|
|
|
last_updated = datetime.now().isoformat(timespec="seconds")
|
|
payload = {
|
|
"captured_at": captured_at,
|
|
"last_updated": last_updated,
|
|
"return_type": args.return_type,
|
|
"periods_requested": periods,
|
|
"companies": companies_out,
|
|
}
|
|
out_path = write_json_output(payload)
|
|
|
|
# Stdout summary — three buckets, never conflated
|
|
captured = sum(1 for c in companies_out if c["results"] and not c["errors"])
|
|
not_filed_companies = sum(1 for c in companies_out if c["not_filed"] and not c["results"] and not c["errors"])
|
|
errored = sum(1 for c in companies_out if c["errors"])
|
|
|
|
total_periods_requested = len(periods) * len(companies_out)
|
|
total_captured_periods = sum(len(c["results"]) for c in companies_out)
|
|
total_not_filed_periods = sum(len(c["not_filed"]) for c in companies_out)
|
|
total_error_periods = sum(len(c["errors"]) for c in companies_out)
|
|
|
|
print("\n" + "=" * 60, flush=True)
|
|
print(f"Companies: {len(companies_out)}", flush=True)
|
|
print(f" Captured: {captured} (had at least one period with results)", flush=True)
|
|
print(f" Not filed: {not_filed_companies} (all requested periods were unfilled)", flush=True)
|
|
print(f" Errored: {errored} (had script/runtime failures)", flush=True)
|
|
print(f"", flush=True)
|
|
print(f"Periods: {total_periods_requested} requested", flush=True)
|
|
print(f" Captured: {total_captured_periods}", flush=True)
|
|
print(f" Not filed: {total_not_filed_periods}", flush=True)
|
|
print(f" Errored: {total_error_periods}", flush=True)
|
|
print(f"", flush=True)
|
|
print(f"JSON: {out_path}", flush=True)
|
|
print("=" * 60, flush=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|