Rakseeibra/arrr-erp
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use authenticated GST dashboard XHR for return downloads

Browse Source
This commit is contained in:
A R R R Associates
2026-09-12 12:35:44 +05:30
parent 782feaeeea
commit 4d0c53b9ba
4 changed files with 172 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/modules/accounting/gst_operator_agent_runtime/README.txt
+9 -16
View File
@@ -1,4 +1,4 @@
ARRR GST Operator Agent 1.4.0
ARRR GST Operator Agent 1.5.0
Purpose
-------
@@ -25,19 +25,12 @@ Uninstall
Run uninstall_gst_operator_agent.ps1. It removes the protocol registration and also cleans legacy v1.0.x localhost listener/startup/certificate state.
V1.4.0 VERIFIED POST-DASHBOARD DOWNLOAD SEQUENCE
- After CAPTCHA/OTP login, do not deep-link to return pages.
- Close optional Aadhaar/E-KYC reminder.
- From the authenticated GST Welcome page click Return Dashboard exactly as a user would.
- Wait for return.gst.gov.in /returns/auth/ session to load.
- Only then call GSTR-1/GSTR-2B/GSTR-3B APIs using the authenticated browser context.
V1.5.0 OCTAGST-STYLE RETURN-DASHBOARD EXTRACTION
- Preserves the v1.4.0 Credential Vault login, manual CAPTCHA/OTP, natural Welcome -> Return Dashboard flow, ERP upload and failure diagnostics.
- GSTR-2B no longer transfers the browser to gstr2b.gst.gov.in.
- GSTR-2B is requested from the authenticated return.gst.gov.in Return Dashboard using the GST offline-download API pattern used by the supplied OctaGST extension.
- The agent follows GST-provided download URLs, saves returned ZIP/JSON files, and safely extracts JSON members locally.
- Same-origin GSTR-1 and GSTR-3B API calls remain unchanged.
- No automatic GST logout is performed.
- A job is completed only after every selected period/return has a saved result and ERP/client-storage upload is confirmed.
- Manual JSON/ZIP import remains available in ERP as fallback.
V1.4.0 POST-DASHBOARD REPAIR
- Does not mark a job complete merely because Return Dashboard opened.
- Requires an actual saved JSON result for every selected return and period.
- GSTR-2B now enters its module only after the natural Return Dashboard session is established, then calls the GSTR-2B API same-origin.
- Does not call GST logout after completion.
- On failure the browser remains visible for 45 seconds with the exact failure message.
- Upload to ERP/client storage happens only after all expected return files are present.
app/modules/accounting/gst_operator_agent_runtime/gst_operator_agent.py
+161 -35
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import io
import json
import re
import shutil
@@ -9,11 +10,11 @@ import traceback
import zipfile
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import parse_qs, urlsplit
from urllib.parse import parse_qs, urljoin, urlsplit
import requests
VERSION = "1.4.0"
VERSION = "1.5.0"
ROOT = Path(__file__).resolve().parent
DATA = ROOT / "data"
CONFIG_PATH = ROOT / "config.json"
@@ -26,6 +27,7 @@ GSTR1_URL = "https://return.gst.gov.in/returns/auth/api/offline/download/generat
GSTR2A_URL = "https://return.gst.gov.in/returns/auth/api/offline/download/generate?flag=0&rtn_prd={period}&rtn_typ=GSTR2A"
GSTR1_DOWNLOAD_URL = "https://return.gst.gov.in/returns/auth/api/offline/download/url?rtn_prd={period}&rtn_typ=GSTR1&file_num={file_num}"
GSTR2B_URL = "https://gstr2b.gst.gov.in/gstr2b/auth/api/gstr2b/getjson?rtnprd={period}"
GSTR2B_OFFLINE_URL = "https://return.gst.gov.in/returns/auth/api/offline/download/generate?flag=0&rtn_prd={period}&rtn_typ=GSTR2B"
ROLESTATUS_URL = "https://return.gst.gov.in/returns/auth/api/rolestatus?rtn_prd={period}"
GSTR3B_SUMMARY_URL = "https://return.gst.gov.in/returns/auth/api/gstr3b/summary?rtn_prd={period}"
GSTR3B_URL = "https://return.gst.gov.in/returns/auth/api/gstr3b/taxpayble?rtn_prd={period}"
@@ -486,32 +488,162 @@ def return_to_return_dashboard(page, progress) -> None:
raise RuntimeError(f"Could not return to GST Return Dashboard. Current page: {page.url}")
def open_gstr2b_module_from_established_session(page, progress) -> None:
"""Enter GSTR-2B only after Return Dashboard has created the return-domain session.
def _extract_download_urls(payload) -> list[str]:
"""Extract GST offline-download URLs from the return API response.
The failed v9 path called the GSTR-2B API cross-domain directly from the Return
Dashboard. This version first establishes the normal Return Dashboard session,
then opens the GSTR-2B module page and performs the JSON call same-origin from
that page, matching the browser-session pattern used by the working GST tools.
OctaGST uses the return.gst.gov.in offline download API from the already
authenticated Return Dashboard and then downloads the URLs returned in
data.url. Keep this parser tolerant of GST response-shape changes.
"""
found: list[str] = []
def visit(value):
if isinstance(value, dict):
for key, item in value.items():
if str(key).lower() in {"url", "urls", "downloadurl", "download_url"}:
if isinstance(item, str) and item.strip():
found.append(item.strip())
elif isinstance(item, list):
for x in item:
if isinstance(x, str) and x.strip():
found.append(x.strip())
visit(item)
elif isinstance(value, list):
for item in value:
visit(item)
visit(payload)
result = []
seen = set()
for item in found:
absolute = urljoin(RETURN_DASHBOARD_URL, item)
if absolute not in seen:
seen.add(absolute)
result.append(absolute)
return result
def _gst_response_message(payload) -> str:
"""Return the most useful short GST message for operator diagnostics."""
candidates = []
for obj in walk(payload):
if not isinstance(obj, dict):
continue
for key in ("message", "msg", "error", "status_desc", "statusDesc", "desc", "description"):
value = obj.get(key)
if isinstance(value, str) and value.strip():
candidates.append(value.strip())
return candidates[0][:500] if candidates else ""
def fetch_binary(page, url: str, *, referer: str = RETURN_DASHBOARD_URL) -> tuple[bytes, str]:
"""Download a GST-generated file using the authenticated browser context."""
headers = {"Accept": "*/*", "Referer": referer}
try:
response = page.context.request.get(url, headers=headers, timeout=120000)
except Exception as exc:
raise RuntimeError(f"GST generated-file download failed: {exc}") from exc
status = int(response.status or 0)
body = response.body() or b""
ctype = (response.headers.get("content-type") or "").lower()
if status >= 400:
sample = body[:500].decode("utf-8", errors="replace")
raise RuntimeError(f"GST generated-file download HTTP {status}: {sample}")
if not body:
raise RuntimeError("GST generated-file download returned an empty file.")
return body, ctype
def _store_gst_download_blob(raw_dir: Path, period: str, return_type: str, index: int, body: bytes, ctype: str) -> list[Path]:
"""Persist a GST JSON/ZIP response and safely extract JSON members from ZIPs."""
stored: list[Path] = []
is_zip = body[:4] == b"PK\x03\x04" or "zip" in (ctype or "")
if is_zip:
zip_path = raw_dir / f"{period}_{return_type}_{index}.zip"
zip_path.write_bytes(body)
stored.append(zip_path)
try:
with zipfile.ZipFile(io.BytesIO(body)) as archive:
for member in archive.infolist():
if member.is_dir() or not member.filename.lower().endswith(".json"):
continue
data = archive.read(member)
if not data:
continue
name = safe(Path(member.filename).name, f"{period}_{return_type}_{index}.json")
target = raw_dir / name
if target.exists():
target = raw_dir / f"{target.stem}_{index}{target.suffix}"
target.write_bytes(data)
stored.append(target)
except zipfile.BadZipFile as exc:
raise RuntimeError("GST returned a file marked as ZIP but it could not be opened.") from exc
return stored
text = body.decode("utf-8", errors="replace").strip()
try:
parsed = json.loads(text)
except Exception as exc:
raise RuntimeError(f"GST generated file was neither ZIP nor valid JSON: {text[:300]}") from exc
if parsed in ({}, [], None, ""):
raise RuntimeError("GST generated JSON file was empty.")
json_path = raw_dir / f"{period}_{return_type}_{index}.json"
json_path.write_text(json.dumps(parsed, ensure_ascii=False, indent=2), encoding="utf-8")
stored.append(json_path)
return stored
def download_gstr2b_octagst_style(page, raw_dir: Path, period: str, progress) -> dict:
"""Download GSTR-2B from Return Dashboard without entering gstr2b.gst.gov.in.
The supplied OctaGST extension performs return retrieval from the authenticated
Return Dashboard by XHR against return.gst.gov.in and then follows the
generated download URLs. Its GSTR-2B configuration uses apiCode GSTR2B with
the same offline-download endpoint. This avoids the cross-subdomain transfer
that GST is currently rejecting with Access Denied.
"""
verify_return_dashboard_session(page)
progress(stage="Opening GSTR-2B Module", message="Return Dashboard is ready. Opening the GST GSTR-2B module before requesting JSON.")
show_status_overlay(page, "Return Dashboard ready. Opening GSTR-2B module...", "#0f766e")
try:
page.goto(GSTR2B_PAGE_URL, wait_until="domcontentloaded", timeout=90000)
except Exception as exc:
raise RuntimeError(f"Could not enter the GSTR-2B module from the established Return Dashboard session: {exc}") from exc
page.wait_for_timeout(2500)
if is_access_denied_page(page):
raise RuntimeError(
"GST returned Access Denied while transferring from Return Dashboard to GSTR-2B. "
"The browser has been kept open temporarily for inspection."
)
url = (page.url or "").lower()
if "gstr2b.gst.gov.in" not in url:
raise RuntimeError(f"GST did not enter the GSTR-2B module. Current page: {page.url}")
show_status_overlay(page, "GSTR-2B module ready. Downloading JSON...", "#15803d")
progress(
stage=f"Downloading GSTR-2B {period}",
message=f"Requesting GSTR-2B {period} directly from the authenticated Return Dashboard session.",
)
show_status_overlay(page, f"GSTR-2B {period}: requesting return file from Return Dashboard...", "#0f766e")
response_text = page_fetch_text(page, GSTR2B_OFFLINE_URL.format(period=period), referer=RETURN_DASHBOARD_URL)
response_data = validate_gst_json(response_text, f"GSTR-2B file request {period}")
metadata_path = raw_dir / f"{period}_GSTR2B_DOWNLOAD_META.json"
metadata_path.write_text(json.dumps(response_data, ensure_ascii=False, indent=2), encoding="utf-8")
urls = _extract_download_urls(response_data)
if not urls:
message = _gst_response_message(response_data)
suffix = f" GST message: {message}" if message else ""
raise RuntimeError(
f"GSTR-2B {period}: GST did not return a downloadable file URL from the Return Dashboard API.{suffix}"
)
files: list[Path] = []
for index, url in enumerate(urls, 1):
progress(
stage=f"Downloading GSTR-2B {period}",
message=f"Downloading GSTR-2B file {index}/{len(urls)} for {period}.",
)
body, ctype = fetch_binary(page, url, referer=RETURN_DASHBOARD_URL)
files.extend(_store_gst_download_blob(raw_dir, period, "GSTR2B", index, body, ctype))
json_files = [p for p in files if p.suffix.lower() == ".json"]
primary = json_files[0] if json_files else (files[0] if files else None)
if primary is None:
raise RuntimeError(f"GSTR-2B {period}: GST download completed but no file was saved.")
show_status_overlay(page, f"GSTR-2B {period} downloaded successfully from Return Dashboard.", "#15803d")
return {
"return_type": "GSTR2B",
"path": str(primary.relative_to(raw_dir.parent.parent)),
"bytes": sum(p.stat().st_size for p in files if p.exists()),
"files": [str(p.relative_to(raw_dir.parent.parent)) for p in files],
"source": "return_dashboard_offline_download",
}
def keep_browser_visible(page, message: str, color: str, milliseconds: int) -> None:
"""Keep the visible GST browser available long enough to see success/failure."""
@@ -545,17 +677,11 @@ def download_period(page, work_root: Path, period: str, return_types: list[str],
downloaded.append({"return_type": "GSTR1", "path": str(path.relative_to(work_root)), "bytes": path.stat().st_size})
if "GSTR2B" in selected:
# Important: do not call GSTR-2B cross-domain directly from Return Dashboard.
# Establish Return Dashboard first, enter the GSTR-2B module, then call its API
# from the GSTR-2B page so the module/session cookies and origin are correct.
# OctaGST-style path: remain on the authenticated Return Dashboard and use
# the return-domain offline-download API. Do not transfer the browser to
# gstr2b.gst.gov.in, which is the step GST rejected with Access Denied.
return_to_return_dashboard(page, progress)
open_gstr2b_module_from_established_session(page, progress)
progress(stage=f"Downloading GSTR-2B {period}", message=f"Downloading GSTR-2B JSON for {period} from the established GSTR-2B session.")
content = page_fetch_text(page, GSTR2B_URL.format(period=period), referer=GSTR2B_PAGE_URL)
validate_gst_json(content, f"GSTR-2B {period}")
path = raw_dir / f"{period}_GSTR2B.json"
path.write_text(content, encoding="utf-8")
downloaded.append({"return_type": "GSTR2B", "path": str(path.relative_to(work_root)), "bytes": path.stat().st_size})
downloaded.append(download_gstr2b_octagst_style(page, raw_dir, period, progress))
if "GSTR3B" in selected:
return_to_return_dashboard(page, progress)
@@ -598,7 +724,7 @@ def download_period(page, work_root: Path, period: str, return_types: list[str],
write_json(
work_root / period / "download_manifest.json",
{"period": period, "downloaded_at_utc": now(), "source": "arrr_gst_operator_agent_1.4.0", "downloaded": downloaded},
{"period": period, "downloaded_at_utc": now(), "source": "arrr_gst_operator_agent_1.5.0", "downloaded": downloaded},
)
return downloaded
app/modules/accounting/gst_operator_agent_runtime/install_gst_operator_agent.ps1
+1 -1
View File
@@ -9,7 +9,7 @@ $Startup = Join-Path $env:APPDATA "Microsoft\Windows\Start Menu\Programs\Startup
$ThumbprintFile = Join-Path $InstallRoot "cert_thumbprint.txt"
$ProtocolKey = "HKCU:\Software\Classes\arrrgst"
Write-Host "ARRR GST Operator Agent 1.4.0 - clean installation" -ForegroundColor Cyan
Write-Host "ARRR GST Operator Agent 1.5.0 - clean installation" -ForegroundColor Cyan
Write-Host "Install root: $InstallRoot"
# Clean legacy v1.0.x listener/startup/certificate/protocol state first.