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.
1402 lines
61 KiB
Python
1402 lines
61 KiB
Python
"""
|
|
SARS Retun VAT201 single or all pty — extract VAT201 payment breakdowns
|
|
for one or many companies over one or more periods.
|
|
|
|
This is the batch-capable superset (mirrors `SARS Retun EMP201 single or all pty.py`)
|
|
for the VAT201 return type. Same structure, different SARS sub-item
|
|
("Value Added Tax (VAT201)") and a different per-row payload.
|
|
|
|
For each (company, period) we capture:
|
|
- work page header + Status table fields (always available)
|
|
* taxpayer_name
|
|
* period (YYYYMM, e.g. "202606"; the work page shows "2026/06"
|
|
but the underlying list value is YYYYMM)
|
|
* tax_reference (VAT number — different from PAYE number)
|
|
* return_type ("VAT201")
|
|
* efiling_status ("Filed" / "Saved" / etc.)
|
|
* date_filed (ISO YYYY-MM-DD from "DD/MM/YYYY" or "YYYY/MM/DD")
|
|
* declared_amount (R, 2 decimals, e.g. "8443.12")
|
|
* version (int, e.g. 1, 2)
|
|
* last_updated_by
|
|
- form fields (only when the v1 form loads — see KNOWN ISSUE below)
|
|
* standard_rate_1 (Field 1)
|
|
* zero_rate_2 (Field 2)
|
|
* exempt_and_non_supplies_3 (Field 3)
|
|
* capital_goods_and_services_supplied_to_you_14 (Field 14)
|
|
* other_goods_and_services_supplied_to_you_15 (Field 15)
|
|
- work page → "Open" path on the FILLED row's "VAT201" link opens a
|
|
read-only form with the values that were submitted. (Clicking the
|
|
v1 link in the work page's Status table goes to
|
|
LoadHTML5Form.aspx?ITID={TPReturnPITID_of_v1}.)
|
|
|
|
Usage:
|
|
Run for ONE company: --company "RJ BULK SUPPLIES"
|
|
Run for SELECTED: --companies "X,Y,Z"
|
|
Run for ALL active: --all-active
|
|
Period selection: --periods "202606" or --periods "202601-202612"
|
|
|
|
Output:
|
|
JSON file: C:\\Users\\SJ\\Downloads\\sars_captures\\YYYY-MM-DD_HHMMSS.json
|
|
Same top-level shape as the EMP201 capture script:
|
|
{
|
|
"captured_at": "...",
|
|
"last_updated": "...",
|
|
"return_type": "VAT201",
|
|
"periods_requested": ["202606"],
|
|
"companies": [
|
|
{
|
|
"name": "RJ BULK SUPPLIES",
|
|
"login_as": "sjcoetzee1092",
|
|
"portfolio": "Group TME",
|
|
"vat_no": "4650306386",
|
|
"last_updated": "...",
|
|
"results": [
|
|
{"period":"202606", "taxpayer_name":"...",
|
|
"tax_reference":"...", "return_type":"VAT201",
|
|
"efiling_status":"Filed", "date_filed":"2026-07-30",
|
|
"declared_amount":"8443.12", "version":1,
|
|
"last_updated_by":"SG BARNARD",
|
|
"standard_rate_1":"753175",
|
|
"zero_rate_2":"20345403",
|
|
"exempt_and_non_supplies_3":"0",
|
|
"capital_goods_and_services_supplied_to_you_14":"0.00",
|
|
"other_goods_and_services_supplied_to_you_15":"89797.10"},
|
|
...
|
|
],
|
|
"not_filed": [], # periods not present in the period list
|
|
"errors": []
|
|
},
|
|
...
|
|
]
|
|
}
|
|
|
|
"Active client" definition (for --all-active):
|
|
PocketBase record on ysterbaard_companies where
|
|
is_client = true AND sars_vat_no is non-empty.
|
|
(VAT201 only applies to companies registered for VAT; we mirror
|
|
EMP201's is_client + sars_paye_no gate, but with sars_vat_no.)
|
|
|
|
KNOWN ISSUE: v1 read-only form load
|
|
For some periods the v1 form (read-only) appears to hang at
|
|
"Please wait while your tax return is generated." even after 15s.
|
|
When that happens, we capture ONLY the work-page fields and skip
|
|
the form-level fields (logged as `form_load_failed: true` in the
|
|
JSON row, NOT as a script error). The .py is conservative: we
|
|
don't raise the whole run if one period's form is slow.
|
|
"""
|
|
|
|
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
|
|
|
|
# Force UTF-8 stdout/stderr so Unicode chars in error messages (e.g. agent-browser's
|
|
# ✗ error prefix) don't crash print() under Windows cp1252.
|
|
try:
|
|
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
|
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# ── 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()
|
|
tmp_err = tempfile.NamedTemporaryFile(
|
|
mode="w", delete=False, encoding="utf-8", errors="replace",
|
|
prefix="ab_err_", suffix=".out",
|
|
)
|
|
tmp_err_path = tmp_err.name
|
|
tmp_err.close()
|
|
try:
|
|
try:
|
|
result = subprocess.run(
|
|
[AGENT_BROWSER_EXE, *args],
|
|
stdout=open(tmp_path, "w", encoding="utf-8", errors="replace"),
|
|
stderr=open(tmp_err_path, "w", encoding="utf-8", errors="replace"),
|
|
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()
|
|
with open(tmp_err_path, "r", encoding="utf-8", errors="replace") as f:
|
|
err_out = f.read().strip()
|
|
finally:
|
|
try:
|
|
os.unlink(tmp_path)
|
|
except OSError:
|
|
pass
|
|
try:
|
|
os.unlink(tmp_err_path)
|
|
except OSError:
|
|
pass
|
|
if result.returncode != 0:
|
|
raise RuntimeError(
|
|
f"agent-browser {args!r} failed (rc={result.returncode}): "
|
|
f"stdout={out!r} stderr={err_out!r}"
|
|
)
|
|
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 ────────────────────────────────────────────────
|
|
|
|
# SARS "VAT Return Every" cadence → which MM periods exist for the company.
|
|
# (See user-policy: a company's VAT Return Every setting on SARS eFiling
|
|
# controls which periods SARS generates. Periods that don't match are simply
|
|
# absent from the period list and can never be "filed" or "not filed" — SARS
|
|
# won't even let you request them.)
|
|
#
|
|
# Format reference (verified with JS Mining AND ENGINEERING = "Even months",
|
|
# RJ BULK SUPPLIES = "Every month"):
|
|
# 01 None — no VAT periods at all (company is not VAT-registered)
|
|
# 02 Every month — 01, 02, 03, 04, 05, 06, 07, 08, 09, 10, 11, 12
|
|
# 03 Even months — 02, 04, 06, 08, 10, 12 (bi-monthly: 02 = Jan-Feb, 04 = Mar-Apr, ...)
|
|
# 04 Uneven months — 01, 03, 05, 07, 09, 11 (bi-monthly, inverse of even)
|
|
# 05 Bi-annually — 02 (Sep-Feb, due end Mar) and 08 (Mar-Aug, due end Sep)
|
|
# 06 Annually — 03 (Mar-Feb, due end Mar of following year)
|
|
#
|
|
# Due-date rule (all cadences): "end of the month after the period's last month."
|
|
# Examples: 202608 (period #8 = AUGUST) due end of SEPTEMBER;
|
|
# 202609 (period #9 = SEPTEMBER) due end of OCTOBER;
|
|
# 202612 (period #12 = DECEMBER) due end of JANUARY (next year).
|
|
# Note: MM in YYYYMM is the PERIOD NUMBER (not a calendar-month name).
|
|
# Period #08 is AUGUST, period #09 is SEPTEMBER, etc.
|
|
VAT_CADENCE_PERIODS: dict = {
|
|
"None": set(),
|
|
"Every month": {1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12},
|
|
"Even months": {2, 4, 6, 8, 10, 12},
|
|
"Uneven months": {1, 3, 5, 7, 9, 11},
|
|
"Bi-annually": {2, 8},
|
|
"Annually": {3},
|
|
}
|
|
|
|
# Aliases the user might pass via --cadence (case-insensitive)
|
|
VAT_CADENCE_ALIASES: dict = {
|
|
"none": "None",
|
|
"monthly": "Every month",
|
|
"every-month": "Every month",
|
|
"every_month": "Every month",
|
|
"every month": "Every month",
|
|
"even": "Even months",
|
|
"even-months": "Even months",
|
|
"even_months": "Even months",
|
|
"even months": "Even months",
|
|
"bi-monthly": "Even months", # common confusion
|
|
"bimonthly": "Even months",
|
|
"uneven": "Uneven months",
|
|
"uneven-months": "Uneven months",
|
|
"uneven_months": "Uneven months",
|
|
"uneven months": "Uneven months",
|
|
"odd": "Uneven months",
|
|
"bi-annual": "Bi-annually",
|
|
"biannual": "Bi-annually",
|
|
"bi-annually": "Bi-annually",
|
|
"bi_annually": "Bi-annually",
|
|
"bi annually": "Bi-annually",
|
|
"twice-yearly": "Bi-annually",
|
|
"annual": "Annually",
|
|
"yearly": "Annually",
|
|
"annually": "Annually",
|
|
}
|
|
|
|
|
|
def normalize_cadence(raw: str) -> str:
|
|
"""Accept any of the aliases or canonical names and return the canonical."""
|
|
if not raw:
|
|
return ""
|
|
key = raw.strip().lower()
|
|
if key in VAT_CADENCE_ALIASES:
|
|
return VAT_CADENCE_ALIASES[key]
|
|
if raw in VAT_CADENCE_PERIODS:
|
|
return raw
|
|
raise ValueError(
|
|
f"Unknown VAT Return Every cadence {raw!r}. "
|
|
f"Valid: {sorted(VAT_CADENCE_PERIODS.keys())} "
|
|
f"(aliases: every-month, even-months, uneven-months, bi-annually, annually, none)"
|
|
)
|
|
|
|
|
|
def filter_periods_by_cadence(periods: List[str], cadence: str) -> List[str]:
|
|
"""Drop periods whose month-component isn't valid for the company's cadence.
|
|
|
|
E.g. for an "Even months" company, drop 202603, 202605, 202607, ...
|
|
Returns the filtered list. If the filtered list is empty, the caller
|
|
should warn — `--periods` had nothing valid for this cadence.
|
|
"""
|
|
if not cadence:
|
|
return periods
|
|
canon = normalize_cadence(cadence)
|
|
allowed = VAT_CADENCE_PERIODS.get(canon, set())
|
|
if not allowed:
|
|
return [] # "None" — no VAT periods at all
|
|
kept = [p for p in periods if int(p[4:6]) in allowed]
|
|
dropped = [p for p in periods if p not in kept]
|
|
if dropped:
|
|
print(
|
|
f" [cadence] {canon}: dropping periods that don't exist for this cadence: {dropped}",
|
|
flush=True,
|
|
)
|
|
if not kept:
|
|
print(
|
|
f" [cadence] WARNING: {canon} has no VAT periods at all "
|
|
f"(company is not VAT-registered, or --cadence is wrong)",
|
|
flush=True,
|
|
)
|
|
return kept
|
|
|
|
|
|
def expand_periods(spec: str) -> List[str]:
|
|
"""Expand "202608" or "202601-202612" 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_vat_clients() -> List[dict]:
|
|
"""Return all ysterbaard_companies records that are:
|
|
is_client = true AND sars_vat_no is non-empty.
|
|
|
|
VAT201 only applies to VAT-registered entities, so we filter on
|
|
sars_vat_no (mirrors EMP201's sars_paye_no gate).
|
|
"""
|
|
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_vat_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.
|
|
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
|
|
|
|
|
|
# ── 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."""
|
|
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."""
|
|
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_vat201() -> None:
|
|
"""Navigate to the VAT201 Returns History list.
|
|
|
|
IMPORTANT: the SARS sidebar has BOTH 'Returns Issued' and 'Returns
|
|
History' parent links, each with its own sub-items named after each
|
|
return type. We want 'Returns History' → 'Value Added Tax (VAT201)'
|
|
(HistoryReturns page, which lists filed returns).
|
|
|
|
Strategy:
|
|
1. Use snapshot to find 'Returns History' parent ref and click it.
|
|
If 'Returns Issued' is already expanded with a 'Value Added Tax
|
|
(VAT201)' sub-item visible, the snapshot will show BOTH the
|
|
parent and its sub-items. We then pick the VAT201 link whose
|
|
parent is 'Returns History' (not 'Returns Issued') by walking
|
|
the snapshot tree.
|
|
2. As a fallback, if step 1's DOM walker is too brittle, do an
|
|
explicit collapse-Resturns-Issued + expand-Returns-History +
|
|
click-VAT201 sequence using eval_js.
|
|
"""
|
|
label = "Value Added Tax (VAT201)"
|
|
|
|
# Step 1: snapshot, click Returns History parent — but only if it's
|
|
# currently collapsed. If it's already expanded from a prior session,
|
|
# clicking would collapse it and lose its sub-items.
|
|
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 in sidebar")
|
|
|
|
# Detect expansion via the JS-side check. The check counts the number of
|
|
# <a> elements inside the parent nav group; collapsed = 1 (just the
|
|
# parent), expanded = several (parent + sub-items).
|
|
expand_check_js = (
|
|
"(() => { "
|
|
" const a = Array.from(document.querySelectorAll('a'))"
|
|
" .find(x => x.textContent.trim() === 'Returns History'); "
|
|
" if (!a) return 'no parent'; "
|
|
"/* Count <a> in the parent nav group (parent + sub-items when expanded). */"
|
|
" const grp = a.closest('li, .mat-nav-list-item, .nav-item, mat-nav-list-item') || a.parentElement; "
|
|
" const cc = grp ? grp.querySelectorAll('a').length : 0; "
|
|
" return cc > 1 ? 'expanded' : 'collapsed'; "
|
|
"})()"
|
|
)
|
|
state = eval_js(expand_check_js, timeout=10)
|
|
print(f" [nav] Returns History state: {state!r}", flush=True)
|
|
if state == "collapsed":
|
|
print(f" [nav] clicking Returns History parent ref={parent_ref} ...", flush=True)
|
|
run_ab("click", f"@{parent_ref}", timeout=15)
|
|
elif state == "expanded":
|
|
print(f" [nav] already expanded; not clicking (ref={parent_ref})", flush=True)
|
|
# Belt-and-braces: after the click (or no-op), force-click again if still
|
|
# collapsed. This handles the rare race where state="expanded" but the
|
|
# snapshot below still misses the sub-items.
|
|
time.sleep(4)
|
|
state2 = eval_js(expand_check_js, timeout=10)
|
|
print(f" [nav] state after settle: {state2!r}", flush=True)
|
|
# Sanity check: how many VAT201 links are visible right now?
|
|
vat_count_js = (
|
|
"(() => Array.from(document.querySelectorAll('a'))"
|
|
".filter(a => a.textContent.trim() === 'Value Added Tax (VAT201)').length)()"
|
|
)
|
|
vat_count = eval_js(vat_count_js, timeout=10)
|
|
print(f" [nav] VAT201 links visible right now: {vat_count}", flush=True)
|
|
if state2 == "collapsed":
|
|
# One more click attempt with the current ref (may have rotated)
|
|
snap2 = run_ab("snapshot", "-i", timeout=15)
|
|
parent_ref2 = _find_ref(snap2, "link", "Returns History")
|
|
if parent_ref2:
|
|
print(f" [nav] retry click on ref={parent_ref2} ...", flush=True)
|
|
run_ab("click", f"@{parent_ref2}", timeout=15)
|
|
time.sleep(4)
|
|
state3 = eval_js(expand_check_js, timeout=10)
|
|
print(f" [nav] state after retry: {state3!r}", flush=True)
|
|
if state3 == "collapsed":
|
|
raise RuntimeError(
|
|
f"Returns History parent did not expand after retries (last ref={parent_ref2})"
|
|
)
|
|
|
|
# Step 2: find the VAT201 sub-item belonging to Returns History and
|
|
# click it via JS. We use JS .click() because agent-browser's click
|
|
# command doesn't reliably trigger Angular Router's navigation for
|
|
# Angular Material mat-list-item links (which have empty href and
|
|
# rely on Angular's click handler).
|
|
#
|
|
# The DOM contains TWO VAT201 links (one under Returns Issued, one
|
|
# under Returns History). We pick the one whose ancestor nav-group
|
|
# is Returns History.
|
|
print(f" [nav] click sub-item {label!r} ...", flush=True)
|
|
pick_vat_js = (
|
|
"(() => { "
|
|
" const vatLinks = Array.from(document.querySelectorAll('a'))"
|
|
" .filter(a => a.textContent.trim() === " + json.dumps(label) + "); "
|
|
" for (const a of vatLinks) { "
|
|
" let p = a.parentElement; "
|
|
" for (let i = 0; i < 8 && p; i++) { "
|
|
" const pt = (p.textContent || '').trim(); "
|
|
" if (pt.startsWith('Returns History')) { "
|
|
" a.click(); "
|
|
" return 'clicked: ' + a.outerHTML.slice(0, 100); "
|
|
" } "
|
|
" if (pt.startsWith('Returns Issued')) break; "
|
|
" p = p.parentElement; "
|
|
" } "
|
|
" } "
|
|
" return 'no history vat link (' + vatLinks.length + ' candidates)'; "
|
|
"})()"
|
|
)
|
|
res = eval_js(pick_vat_js, timeout=15)
|
|
print(f" [nav] pick_vat_js returned: {res!r}", flush=True)
|
|
if not res or not res.startswith("clicked:"):
|
|
raise RuntimeError(f"VAT201 sub-item selection failed: {res!r}")
|
|
time.sleep(4)
|
|
|
|
|
|
# ── VAT201 period + work page reads ──────────────────────────────
|
|
|
|
def get_vat201_period_open_links() -> List[dict]:
|
|
"""Read all period rows from the VAT201 Returns History list.
|
|
Returns [{period:'YYYYMM', href:'https://...LoadPage.aspx?...', complete_date:'DD/MM/YYYY'}, ...].
|
|
|
|
We match on `TaxPeriod:` text (which only appears on VAT history rows
|
|
for this return-type page) rather than the looser `20260[0-9]` used
|
|
for EMP201, because the VAT201 list can show much older periods and
|
|
`20260[0-9]` would miss pre-2026 history rows.
|
|
|
|
We poll up to ~20s for the period rows to appear (the iframe content
|
|
can take a few seconds to render after the click navigates).
|
|
"""
|
|
js = (
|
|
"(() => { const ifr = document.getElementById('main') || Array.from(document.querySelectorAll('iframe')).find(f => f.src.includes('EFDotNet')); "
|
|
"if (!ifr || !ifr.contentDocument) return 'no doc'; "
|
|
"const doc = ifr.contentDocument; "
|
|
"const rows = Array.from(doc.querySelectorAll('tr')); "
|
|
"const out = []; "
|
|
"for (const r of rows) { "
|
|
" const m = r.textContent.match(/TaxPeriod:\\s*(\\d{6})/); "
|
|
" if (!m) continue; "
|
|
" const period = m[1]; "
|
|
" const link = Array.from(r.querySelectorAll('a')).find(a => "
|
|
" a.textContent.trim() === 'Open' && a.href && a.href.includes('LoadPage')); "
|
|
" if (!link) continue; "
|
|
"/* Complete Date is the second-to-last cell (before Open link). The list view shows DD/MM/YYYY like 30/07/2026. */"
|
|
" const cells = Array.from(r.querySelectorAll('td, th')); "
|
|
" let completeDate = ''; "
|
|
" for (const c of cells) { "
|
|
" const t = c.textContent.trim(); "
|
|
" if (/^\\d{2}\\/\\d{2}\\/\\d{4}$/.test(t)) { completeDate = t; break; } "
|
|
" } "
|
|
" out.push({ period, href: link.href, complete_date: completeDate }); "
|
|
"} "
|
|
"return JSON.stringify(out); })()"
|
|
)
|
|
# Poll until the period row count stabilises (no growth for 2 consecutive
|
|
# 2-second polls) or we hit the deadline (60s). Some periods take longer than
|
|
# others to render in the iframe.
|
|
deadline = time.time() + 60
|
|
last_count = -1
|
|
stable_polls = 0
|
|
last_out: List[dict] = []
|
|
while time.time() < deadline:
|
|
try:
|
|
res = eval_js(js, timeout=10)
|
|
except Exception:
|
|
res = None
|
|
out: List[dict] = []
|
|
if res and res != "null" and res != "no doc":
|
|
try:
|
|
out = json.loads(res) if isinstance(res, str) else res
|
|
except (json.JSONDecodeError, TypeError):
|
|
out = []
|
|
if out:
|
|
last_out = out
|
|
if len(out) == last_count:
|
|
stable_polls += 1
|
|
if stable_polls >= 2:
|
|
return last_out
|
|
else:
|
|
stable_polls = 0
|
|
last_count = len(out)
|
|
time.sleep(2)
|
|
return last_out
|
|
|
|
def navigate_iframe_to(url: str) -> None:
|
|
js = (
|
|
"(() => { const f = document.getElementById('main') || "
|
|
"Array.from(document.querySelectorAll('iframe')).find(x => x.src.includes('EFDotNet')); "
|
|
f"if (f) {{ f.src = {json.dumps(url)}; return 'ok'; }} "
|
|
"return 'no sars iframe'; "
|
|
"})()"
|
|
)
|
|
eval_js(js, timeout=15)
|
|
|
|
|
|
def read_work_page_fields() -> Optional[dict]:
|
|
"""Read the work page (LoadReturn.aspx) — extracts header table,
|
|
eFiling Status table, and the v1 Status table.
|
|
|
|
Returns a dict with:
|
|
taxpayer_name, period (YYYYMM), tax_reference, return_type,
|
|
efiling_status, date_filed (YYYY/MM/DD), declared_amount,
|
|
version, last_updated_by
|
|
|
|
Or None if the work page didn't load.
|
|
"""
|
|
# Debug dump (set VAT201_DUMP_HTML=1 to write the page HTML to disk).
|
|
if os.environ.get("VAT201_DUMP_HTML") == "1":
|
|
try:
|
|
dump_js = (
|
|
"(() => { const f = document.getElementById('main'); "
|
|
"const d = f && f.contentDocument; "
|
|
"return d ? d.documentElement.outerHTML : 'no doc'; })()"
|
|
)
|
|
html = eval_js(dump_js, timeout=15) or ""
|
|
if html and html != "no doc":
|
|
dump_path = Path(os.environ.get("VAT201_DUMP_DIR", r"C:\Users\SJ\.minimax-agent\projects\.runtime")) / f"vat201-work-{datetime.now().strftime('%H%M%S')}.html"
|
|
dump_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(dump_path, "w", encoding="utf-8") as fh:
|
|
fh.write(html)
|
|
print(f" [debug] dumped work page HTML to {dump_path}", flush=True)
|
|
except Exception as e:
|
|
print(f" [debug] dump failed: {e}", flush=True)
|
|
|
|
js = (
|
|
"(() => { "
|
|
" const ifr = document.getElementById('main') || Array.from(document.querySelectorAll('iframe')).find(f => f.src.includes('EFDotNet')); "
|
|
" if (!ifr) return JSON.stringify({err: 'no iframe'}); "
|
|
" const doc = ifr.contentDocument; "
|
|
" if (!doc) return JSON.stringify({err: 'no doc'}); "
|
|
"/* HEADER TABLE: 'Taxpayer Name' / 'Tax Period' / 'Tax Reference' / 'Return Type' labels followed by value rows. */"
|
|
" function findHeaderValue(label) { "
|
|
" const labels = Array.from(doc.querySelectorAll('td, th')); "
|
|
" for (const c of labels) { "
|
|
" if (c.textContent.trim() === label) { "
|
|
" const labelRow = c.parentElement; "
|
|
" const nextRow = labelRow.nextElementSibling; "
|
|
" if (nextRow) { "
|
|
" const valTds = Array.from(nextRow.querySelectorAll('td')); "
|
|
" if (valTds.length >= 2) return valTds[1].textContent.trim(); "
|
|
" } "
|
|
" } "
|
|
" } "
|
|
" return ''; "
|
|
" } "
|
|
" const taxpayerName = findHeaderValue('Taxpayer Name'); "
|
|
" const taxPeriodDisp = findHeaderValue('Tax Period'); "
|
|
" const taxReference = findHeaderValue('Tax Reference'); "
|
|
" const returnType = findHeaderValue('Return Type') || 'VAT201'; "
|
|
" const efilingStatus = findHeaderValue('eFiling Status'); "
|
|
"/* STATUS TABLE: header row has 'Return Type | Status | Date | Declared Amount | Version | Last Updated By'. Use EXACT cell match to skip spanning rows where one cell has all labels concatenated. */"
|
|
" let statusRow = null; "
|
|
" let statusHeader = null; "
|
|
" const allTrs = Array.from(doc.querySelectorAll('tr')); "
|
|
" for (let i = 0; i < allTrs.length; i++) { "
|
|
" const tr = allTrs[i]; "
|
|
" const cellsTxt = Array.from(tr.querySelectorAll('td, th')).map(c => c.textContent.trim()); "
|
|
" const has = (s) => cellsTxt.some(c => c === s); "
|
|
" if (!has('Return Type') || !has('Status') || !has('Declared Amount') || "
|
|
" !has('Version') || !has('Last Updated By')) continue; "
|
|
" statusHeader = tr; "
|
|
"/* find v1 data row: search ALL <tr> in the doc. The v1 data row has an <a> with "
|
|
" onclick='OpenReturn(...)' as a DIRECT child of one of its own <td> cells (not "
|
|
" buried inside a nested table). Walk r.children to look only at direct-child cells. */"
|
|
" for (const r of allTrs) { "
|
|
" if (r === tr) continue; "
|
|
" let hasOpenReturnLink = false; "
|
|
" for (const c of r.children) { "
|
|
" if (c.tagName !== 'TD' && c.tagName !== 'TH') continue; "
|
|
" for (const a of c.children) { "
|
|
" if (a.tagName === 'A' && (a.getAttribute('onclick') || '').includes('OpenReturn')) { "
|
|
" hasOpenReturnLink = true; "
|
|
" break; "
|
|
" } "
|
|
" } "
|
|
" if (hasOpenReturnLink) break; "
|
|
" } "
|
|
" if (hasOpenReturnLink) { statusRow = r; break; } "
|
|
" } "
|
|
" break; "
|
|
" } "
|
|
" let status = '', dateDisp = '', declaredAmount = '', version = '', lastUpdatedBy = ''; "
|
|
" if (statusRow) { "
|
|
" const tds = Array.from(statusRow.querySelectorAll('td, th')); "
|
|
"/* Column order: [<td colspan=2>VAT201</td>, Status, Date, Declared Amount, Version, Last Updated By, <empty>] */"
|
|
" status = (tds[1]?.textContent || '').trim(); "
|
|
" dateDisp = (tds[2]?.textContent || '').trim(); "
|
|
" declaredAmount = (tds[3]?.textContent || '').trim().replace(/ |\\s+/g, ' ').replace(/^\\s+|\\s+$/g, ''); "
|
|
" version = (tds[4]?.textContent || '').trim().replace(/ |\\s+/g, ' ').replace(/^\\s+|\\s+$/g, ''); "
|
|
" lastUpdatedBy = (tds[5]?.textContent || '').trim().replace(/ |\\s+/g, ' ').replace(/^\\s+|\\s+$/g, ''); "
|
|
" } "
|
|
"/* DEBUG: surface the matched rows for offline diagnosis. */"
|
|
" if (typeof window !== 'undefined') { "
|
|
" window.__VAT_DEBUG__ = { "
|
|
" taxpayerName, taxPeriodDisp, taxReference, returnType, efilingStatus, "
|
|
" statusHeaderCells: statusHeader ? Array.from(statusHeader.querySelectorAll('td, th')).map(c => c.textContent.trim()).slice(0, 12) : null, "
|
|
" statusRowCells: statusRow ? Array.from(statusRow.querySelectorAll('td, th')).map(c => c.textContent.trim().slice(0, 50)) : null, "
|
|
" status, dateDisp, declaredAmount, version, lastUpdatedBy "
|
|
" }; "
|
|
" } "
|
|
"/* Convert '2026/06' display -> '202606'. */"
|
|
" let period = ''; "
|
|
" const pm = taxPeriodDisp.match(/^(\\d{4})\\/(\\d{2})$/); "
|
|
" if (pm) period = pm[1] + pm[2]; "
|
|
" else if (/^\\d{6}$/.test(taxPeriodDisp)) period = taxPeriodDisp; "
|
|
" else period = taxPeriodDisp; "
|
|
" return JSON.stringify({ "
|
|
" taxpayer_name: taxpayerName, "
|
|
" period: period, "
|
|
" tax_reference: taxReference, "
|
|
" return_type: returnType, "
|
|
" efiling_status: efilingStatus, "
|
|
" date_filed: dateDisp, "
|
|
" declared_amount: declaredAmount, "
|
|
" version: version, "
|
|
" last_updated_by: lastUpdatedBy, "
|
|
" }); "
|
|
"})()"
|
|
)
|
|
try:
|
|
res = eval_js(js, timeout=20)
|
|
except Exception:
|
|
return None
|
|
if not res or res == "null":
|
|
return None
|
|
try:
|
|
d = json.loads(res) if isinstance(res, str) else res
|
|
except (json.JSONDecodeError, TypeError):
|
|
return None
|
|
if not isinstance(d, dict) or d.get("err"):
|
|
return None
|
|
if os.environ.get("VAT201_DEBUG_FIELDS") == "1":
|
|
print(f" [debug] read_work_page_fields: {d}", flush=True)
|
|
try:
|
|
dbg_js = "JSON.stringify(window.__VAT_DEBUG__ || {})"
|
|
dbg = eval_js(dbg_js, timeout=10)
|
|
print(f" [debug] __VAT_DEBUG__: {dbg}", flush=True)
|
|
except Exception as e:
|
|
print(f" [debug] __VAT_DEBUG__ fetch failed: {e}", flush=True)
|
|
return d
|
|
|
|
|
|
def _period_yyyymm(work_period_display: str) -> str:
|
|
"""Convert SARS work page's period display ("2026/06") to YYYYMM ("202606").
|
|
The period list (TaxPeriod: row) already uses YYYYMM, but the work page's
|
|
"Tax Period" header is in the slash form. We keep the YYYYMM version for
|
|
the JSON output."""
|
|
m = re.match(r"^(\d{4})/(\d{2})$", work_period_display.strip())
|
|
if m:
|
|
return f"{m.group(1)}{m.group(2)}"
|
|
# Already YYYYMM?
|
|
m = re.match(r"^(\d{6})$", work_period_display.strip())
|
|
if m:
|
|
return work_period_display.strip()
|
|
return work_period_display.strip()
|
|
|
|
|
|
def _parse_date_to_iso(date_text: str) -> Optional[str]:
|
|
"""Parse 'DD/MM/YYYY' (work page Status Date) or 'YYYY/MM/DD' (some
|
|
views) into ISO 'YYYY-MM-DD'. Returns None if not parseable."""
|
|
if not date_text:
|
|
return None
|
|
date_text = date_text.strip()
|
|
m = re.match(r"^(\d{4})/(\d{2})/(\d{2})$", date_text)
|
|
if m:
|
|
return f"{m.group(1)}-{m.group(2)}-{m.group(3)}"
|
|
m = re.match(r"^(\d{2})/(\d{2})/(\d{4})$", date_text)
|
|
if m:
|
|
return f"{m.group(3)}-{m.group(2)}-{m.group(1)}"
|
|
return None
|
|
|
|
|
|
# ── VAT201 form reads ─────────────────────────────────────────────
|
|
|
|
# Field IDs are stable across periods (SARS hard-codes them in the HTML5
|
|
# form). We map each field ID to a JSON key.
|
|
VAT201_FORM_FIELD_MAP = {
|
|
"mat-input-21": "standard_rate_1", # Field 1 (Standard rate excl. capital goods)
|
|
"mat-input-23": "zero_rate_2", # Field 2 (Zero rate excl. exported goods)
|
|
"mat-input-25": "exempt_and_non_supplies_3",# Field 3 (Exempt and non-supplies)
|
|
"mat-input-37": "capital_goods_and_services_supplied_to_you_14", # Field 14
|
|
"mat-input-39": "other_goods_and_services_supplied_to_you_15", # Field 15
|
|
"mat-input-46": "total_payable", # Section C Field 38 — headline payable/refundable
|
|
"mat-input-15": "payment_reference", # Payment Reference No. (e.g. 4650306386VC2026069)
|
|
}
|
|
|
|
|
|
def _find_vat_form_doc():
|
|
"""Locate the document where the populated VAT201 v1 form lives, after
|
|
clicking the VAT201 link on the work page. The link navigates to
|
|
LoadHTML5Form.aspx — that page may either host the form inline (with
|
|
gov-expansion-panel-header elements) or wrap it in an iframe.
|
|
|
|
If the inner iframe shows the SARS JS SyntaxError page (random parse
|
|
failure), force-reload the iframe by setting its src to itself with a
|
|
cache-buster — this often clears the error on retry.
|
|
|
|
Returns (doc, source_label) on success, else (None, None)."""
|
|
js = (
|
|
"(() => { "
|
|
" const sars = document.getElementById('main') || Array.from(document.querySelectorAll('iframe')).find(f => f.src.includes('EFDotNet') || (f.src || '').includes('LoadHTML5Form') || (f.src || '').includes('LoadReturn')); "
|
|
" if (!sars) return JSON.stringify({err: 'no sars iframe', page: document.title}); "
|
|
" const sarsDoc = sars.contentDocument; "
|
|
" if (!sarsDoc) return JSON.stringify({err: 'no sars doc', page: sars.src.slice(-40)}); "
|
|
"/* 1. Inline form in the SARS iframe doc? */"
|
|
" const inlineSections = sarsDoc.querySelectorAll('gov-expansion-panel-header').length; "
|
|
" if (inlineSections >= 5) return JSON.stringify({found: 'inline', sections: inlineSections, sarsSrc: sars.src.slice(-40)}); "
|
|
"/* 2. Form inside an inner iframe (LoadHTML5Form/VAT201HTML)? */"
|
|
" const inner = sarsDoc.querySelector('iframe[src*=\"VAT201HTML\"], iframe[src*=\"HTML5\"], iframe[src*=\"LoadHTML5Form\"]') || sarsDoc.querySelector('iFrameForm') || (sarsDoc.querySelector('iframe') ); "
|
|
" if (!inner) return JSON.stringify({err: 'no inner'}); "
|
|
" if (!inner.contentDocument) return JSON.stringify({err: 'no inner doc'}); "
|
|
" const isects = inner.contentDocument.querySelectorAll('gov-expansion-panel-header').length; "
|
|
" const iinputs = inner.contentDocument.querySelectorAll('input').length; "
|
|
" const bodyText = (inner.contentDocument.body ? inner.contentDocument.body.innerText : '').slice(0, 200); "
|
|
" const innerSrc = (inner.src || '').slice(-80); "
|
|
" /* Detect SARS JS parse failure (page renders as 'spx?Uncaught SyntaxError: ...'). */ "
|
|
" const hasSyntaxError = innerSrc.includes('Uncaught') || bodyText.includes('Uncaught') || bodyText.includes('SyntaxError'); "
|
|
" if (hasSyntaxError) { "
|
|
" /* Reload the iframe with a cache-buster. Setting src to the current value forces a fresh load. */ "
|
|
" const base = inner.getAttribute('src') || inner.src; "
|
|
" if (base) { inner.src = base.split('?')[0] + '?reload=' + Date.now(); } "
|
|
" return JSON.stringify({err: 'syntax-error-reloading', isects, iinputs, innerSrc, bodyText}); "
|
|
" } "
|
|
" if (isects >= 5) { "
|
|
" return JSON.stringify({found: 'inner', sections: isects, inputs: iinputs, innerSrc, bodyText: bodyText.slice(0, 80)}); "
|
|
" } "
|
|
" return JSON.stringify({err: 'no form yet', isects, iinputs, innerSrc, bodyText: bodyText.slice(0, 100)}); "
|
|
"})()"
|
|
)
|
|
try:
|
|
res = eval_js(js, timeout=15)
|
|
except Exception:
|
|
return (None, None)
|
|
if not res or res == "null":
|
|
return (None, None)
|
|
try:
|
|
d = json.loads(res) if isinstance(res, str) else res
|
|
except (json.JSONDecodeError, TypeError):
|
|
return (None, None)
|
|
if not isinstance(d, dict) or d.get("err"):
|
|
return (None, None)
|
|
return (d, res)
|
|
|
|
|
|
def _vat_form_loaded() -> bool:
|
|
found, _ = _find_vat_form_doc()
|
|
if not found:
|
|
return False
|
|
if found.get("found") == "inline" and found.get("sections", 0) >= 5:
|
|
return True
|
|
if found.get("found") == "inner" and found.get("sections", 0) >= 5:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _vat_form_expand_all() -> None:
|
|
"""Click every collapsed gov-expansion-panel-header in the VAT form."""
|
|
js = (
|
|
"(() => { "
|
|
" const sars = document.getElementById('main') || Array.from(document.querySelectorAll('iframe')).find(f => f.src.includes('EFDotNet')); "
|
|
" if (!sars) return 0; "
|
|
" const sarsDoc = sars.contentDocument; "
|
|
" if (!sarsDoc) return 0; "
|
|
" let doc = null; "
|
|
" if (sarsDoc.querySelectorAll('gov-expansion-panel-header').length >= 5) { doc = sarsDoc; } "
|
|
" else { "
|
|
" const inner = sarsDoc.querySelector('iframe[src*=\"VAT201HTML\"], iframe[src*=\"HTML5\"]') || sarsDoc.querySelector('iFrameForm'); "
|
|
" if (inner && inner.contentDocument) doc = inner.contentDocument; "
|
|
" } "
|
|
" if (!doc) return 0; "
|
|
" const headers = Array.from(doc.querySelectorAll('gov-expansion-panel-header[aria-expanded=\"false\"]')); "
|
|
" headers.forEach(h => h.click()); "
|
|
" return headers.length; "
|
|
"})()"
|
|
)
|
|
try:
|
|
eval_js(js, timeout=15)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def read_vat201_form_fields() -> dict:
|
|
"""Read the form-level fields we care about from the v1 form, wherever
|
|
it lives (inline in the SARS iframe doc, or nested in an inner iframe)."""
|
|
js = (
|
|
"(() => { "
|
|
" const sars = document.getElementById('main') || Array.from(document.querySelectorAll('iframe')).find(f => f.src.includes('EFDotNet')); "
|
|
" if (!sars) return JSON.stringify({}); "
|
|
" const sarsDoc = sars.contentDocument; "
|
|
" if (!sarsDoc) return JSON.stringify({}); "
|
|
" let doc = null; "
|
|
" if (sarsDoc.querySelectorAll('gov-expansion-panel-header').length >= 5) { doc = sarsDoc; } "
|
|
" else { "
|
|
" const inner = sarsDoc.querySelector('iframe[src*=\"VAT201HTML\"], iframe[src*=\"HTML5\"]') || sarsDoc.querySelector('iFrameForm'); "
|
|
" if (inner && inner.contentDocument) doc = inner.contentDocument; "
|
|
" } "
|
|
" if (!doc) return JSON.stringify({}); "
|
|
" const out = {}; "
|
|
" const wanted = " + json.dumps(list(VAT201_FORM_FIELD_MAP.keys())) + "; "
|
|
" for (const id of wanted) { "
|
|
" const el = doc.getElementById(id); "
|
|
" if (el) out[id] = (el.value || '').trim(); "
|
|
" } "
|
|
" return JSON.stringify(out); "
|
|
"})()"
|
|
)
|
|
try:
|
|
res = eval_js(js, timeout=20)
|
|
except Exception:
|
|
return {}
|
|
if not res or res == "null":
|
|
return {}
|
|
try:
|
|
raw = json.loads(res) if isinstance(res, str) else res
|
|
except (json.JSONDecodeError, TypeError):
|
|
return {}
|
|
if not isinstance(raw, dict):
|
|
return {}
|
|
# Translate from field-id-keyed dict to JSON-key-keyed dict.
|
|
out = {}
|
|
for fid, value in raw.items():
|
|
key = VAT201_FORM_FIELD_MAP.get(fid)
|
|
if key:
|
|
out[key] = value
|
|
return out
|
|
|
|
|
|
def open_vat201_form_and_read() -> tuple:
|
|
"""The VAT201 v1 form is ALREADY loaded in the work page's iFrameForm
|
|
(id='iFrameForm', src='VAT201HTML/Index.html'). We expand the collapsed
|
|
panels and read form fields.
|
|
|
|
FALLBACK: if the iFrameForm renders as a SARS JS SyntaxError page,
|
|
_find_vat_form_doc() reloads the iframe; if the form is still broken
|
|
after the polling deadline, navigate the SARS iframe to LoadHTML5Form.aspx
|
|
directly (via OpenReturn's hardcoded URL pattern) and retry there.
|
|
|
|
Returns (form_fields_dict, loaded_ok_bool).
|
|
"""
|
|
deadline = time.time() + 60
|
|
loaded = False
|
|
attempts = 0
|
|
while time.time() < deadline:
|
|
found, raw = _find_vat_form_doc()
|
|
attempts += 1
|
|
if found:
|
|
loaded = True
|
|
break
|
|
if os.environ.get("VAT201_DEBUG_FIELDS") == "1" and attempts % 5 == 1:
|
|
print(f" [debug] open_vat201_form_and_read: attempt {attempts}, raw={(raw or '')[:200]}", flush=True)
|
|
time.sleep(2)
|
|
if not loaded:
|
|
# Last-ditch: navigate to LoadHTML5Form.aspx directly and retry.
|
|
if _navigate_to_loadhtml5_form():
|
|
deadline2 = time.time() + 30
|
|
while time.time() < deadline2:
|
|
found, _ = _find_vat_form_doc()
|
|
if found:
|
|
loaded = True
|
|
break
|
|
time.sleep(2)
|
|
if not loaded:
|
|
return ({}, False)
|
|
_vat_form_expand_all()
|
|
time.sleep(2)
|
|
return (read_vat201_form_fields(), True)
|
|
|
|
|
|
def _navigate_to_loadhtml5_form() -> bool:
|
|
"""Click the VAT201 OpenReturn link to navigate the SARS iframe to
|
|
LoadHTML5Form.aspx — used as a last-ditch fallback when iFrameForm
|
|
keeps rendering as a SyntaxError page.
|
|
|
|
Returns True if the click was dispatched, False if no link found.
|
|
"""
|
|
js = (
|
|
"(() => { "
|
|
" const sars = document.getElementById('main'); "
|
|
" if (!sars) return false; "
|
|
" const doc = sars.contentDocument; "
|
|
" if (!doc) return false; "
|
|
" const trs = Array.from(doc.querySelectorAll('tr')); "
|
|
" for (const r of trs) { "
|
|
" for (const c of r.children) { "
|
|
" if (c.tagName !== 'TD' && c.tagName !== 'TH') continue; "
|
|
" for (const a of c.children) { "
|
|
" if (a.tagName === 'A' && (a.getAttribute('onclick') || '').includes('OpenReturn')) { "
|
|
" a.click(); "
|
|
" return true; "
|
|
" } "
|
|
" } "
|
|
" } "
|
|
" } "
|
|
" return false; "
|
|
"})()"
|
|
)
|
|
try:
|
|
res = eval_js(js, timeout=10)
|
|
return bool(res)
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
# ── Logout helper ────────────────────────────────────────────────
|
|
|
|
def force_logout() -> None:
|
|
"""Best-effort: navigate to a logout URL so the next company can
|
|
sign in as a different SARS user."""
|
|
try:
|
|
run_ab("navigate", "https://secure.sarsefiling.co.za/app/auth/logout", timeout=15)
|
|
except Exception:
|
|
pass
|
|
time.sleep(2)
|
|
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 VAT201 data. Returns a dict ready for
|
|
the JSON output. Never raises."""
|
|
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,
|
|
"vat_no": (record.get("sars_vat_no") or "").strip(),
|
|
"last_updated": None,
|
|
"results": [],
|
|
"not_filed": [],
|
|
"errors": [],
|
|
}
|
|
|
|
try:
|
|
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_vat201()
|
|
|
|
all_links = get_vat201_period_open_links()
|
|
print(f" [capture] found {len(all_links)} periods on page", flush=True)
|
|
if all_links:
|
|
print(f" [capture] periods found: {sorted(e['period'] for e in all_links)}", flush=True)
|
|
|
|
href_by_period = {e["period"]: e["href"] for e in all_links}
|
|
|
|
for period in periods:
|
|
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]
|
|
print(f" {period}: work page ...", flush=True)
|
|
try:
|
|
navigate_iframe_to(href)
|
|
time.sleep(4)
|
|
|
|
work_fields = read_work_page_fields()
|
|
if not work_fields:
|
|
print(f" {period}: not filed (work page didn't load or VAT201 row missing)",
|
|
flush=True)
|
|
company_record["not_filed"].append(period)
|
|
continue
|
|
|
|
# Build the work-page half of the row.
|
|
# PB context fills taxpayer_name + tax_reference when the work
|
|
# page doesn't expose them (read-only mode strips labels).
|
|
vat_no = (record.get("sars_vat_no") or "").strip()
|
|
row = {
|
|
"period": period, # YYYYMM
|
|
"taxpayer_name": (work_fields.get("taxpayer_name") or name).strip(),
|
|
"tax_reference": (work_fields.get("tax_reference") or vat_no).strip(),
|
|
"return_type": (work_fields.get("return_type") or "VAT201").strip(),
|
|
"efiling_status": (work_fields.get("efiling_status") or work_fields.get("status") or "Filed").strip(),
|
|
"date_filed": _parse_date_to_iso(work_fields.get("date_filed") or work_fields.get("date") or ""),
|
|
"declared_amount": (work_fields.get("declared_amount") or "").strip().replace(",", ""),
|
|
"version": (work_fields.get("version") or "").strip(),
|
|
"last_updated_by": (work_fields.get("last_updated_by") or "").strip(),
|
|
}
|
|
|
|
# Open the v1 form and read the 5 form-level fields.
|
|
form_fields, loaded = open_vat201_form_and_read()
|
|
row["form_loaded"] = loaded
|
|
for json_key, raw_value in form_fields.items():
|
|
row[json_key] = (raw_value or "").strip()
|
|
|
|
company_record["results"].append(row)
|
|
print(f" {period}: tax_ref={row['tax_reference']} "
|
|
f"declared={row['declared_amount']} "
|
|
f"date_filed={row['date_filed']} "
|
|
f"version={row['version']} "
|
|
f"form_loaded={loaded} "
|
|
f"f1={row.get('standard_rate_1','-')} "
|
|
f"f2={row.get('zero_rate_2','-')} "
|
|
f"f3={row.get('exempt_and_non_supplies_3','-')} "
|
|
f"f14={row.get('capital_goods_and_services_supplied_to_you_14','-')} "
|
|
f"f15={row.get('other_goods_and_services_supplied_to_you_15','-')}",
|
|
flush=True)
|
|
except Exception as e:
|
|
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 VAT201 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_vat_no set")
|
|
ap.add_argument("--return-type", default="VAT201")
|
|
ap.add_argument("--periods", required=True,
|
|
help='One period "202606" or range "202601-202612"')
|
|
ap.add_argument("--cadence", default="",
|
|
help='Company\'s SARS "VAT Return Every" setting: '
|
|
'"Every month" (default), "Even months", "Uneven months", '
|
|
'"Bi-annually", "Annually", or "None". '
|
|
'Aliases: monthly, even, uneven, bi-annual, annual, none. '
|
|
'When set, --periods is filtered to periods that actually exist '
|
|
'for this cadence (e.g. 202603-202608 with Even months keeps '
|
|
'only 202604, 202606, 202608).')
|
|
args = ap.parse_args()
|
|
|
|
if args.return_type != "VAT201":
|
|
raise SystemExit(f"return-type {args.return_type!r} not yet supported; only VAT201")
|
|
|
|
periods = expand_periods(args.periods)
|
|
if args.cadence:
|
|
try:
|
|
periods = filter_periods_by_cadence(periods, args.cadence)
|
|
except ValueError as e:
|
|
raise SystemExit(str(e))
|
|
if not periods:
|
|
raise SystemExit(
|
|
f"--periods resolved to empty after applying --cadence {args.cadence!r}. "
|
|
f"Nothing to do."
|
|
)
|
|
|
|
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_vat_clients()
|
|
print(f"Mode: all-active ({len(records)} VAT 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)
|
|
|
|
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)
|
|
|
|
forms_loaded = sum(1 for c in companies_out for r in c["results"] if r.get("form_loaded"))
|
|
forms_skipped = sum(1 for c in companies_out for r in c["results"] if not r.get("form_loaded"))
|
|
|
|
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"Form fields: {forms_loaded} loaded, {forms_skipped} skipped (v1 read-only form slow to load)", flush=True)
|
|
print(f"", flush=True)
|
|
print(f"JSON: {out_path}", flush=True)
|
|
print("=" * 60, flush=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|
|
|