A R R R Associates
187 lines
12 KiB
Python
187 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import calendar
|
|
import re
|
|
from datetime import date
|
|
from pathlib import Path
|
|
from urllib.parse import urlencode
|
|
|
|
from fastapi import APIRouter, Form, Request
|
|
from fastapi.responses import RedirectResponse
|
|
from sqlalchemy import select
|
|
|
|
from app.core.db.common import CommonSessionLocal
|
|
from app.core.security.csrf import get_or_create_csrf_token, validate_csrf
|
|
from app.core.templating import templates
|
|
from app.modules.accounting.agent_bridge import request_agent_command
|
|
from app.modules.accounting.ui import _find_visible_client, _require_partner, _visible_clients, _node_online
|
|
from app.modules.core.rbac.deps import get_user_permissions, get_user_roles
|
|
from app.modules.credential_vault.crypto import decrypt_value
|
|
from app.modules.credential_vault.models import CredentialVaultEntry
|
|
from app.modules.credential_vault.service import can_view_entry, log_access
|
|
from app.modules.documents.services import build_document_scope, client_folder_parts, get_active_storage_node_for_branch, sanitize_segment
|
|
from app.modules.registrations.models import ClientRegistration, RegistrationType
|
|
|
|
router = APIRouter(prefix="/tools/accounting/gst-reconciliation", tags=["accounting-gst-reconciliation-ui"])
|
|
|
|
|
|
def _fy_bounds(fy: str) -> tuple[date, date]:
|
|
m = re.fullmatch(r"(\d{4})-(\d{2})", str(fy or "").strip())
|
|
if not m:
|
|
raise ValueError("Invalid financial year.")
|
|
y = int(m.group(1))
|
|
return date(y, 4, 1), date(y + 1, 3, 31)
|
|
|
|
|
|
def _fy_for_period(period: str) -> str:
|
|
digits = re.sub(r"\D", "", period or "")
|
|
if len(digits) != 6:
|
|
raise ValueError("Return period must be MMYYYY.")
|
|
month, year = int(digits[:2]), int(digits[2:])
|
|
if month < 1 or month > 12:
|
|
raise ValueError("Invalid GST return month.")
|
|
sy = year if month >= 4 else year - 1
|
|
return f"{sy}-{str(sy+1)[-2:]}"
|
|
|
|
|
|
def _period_bounds(period: str) -> tuple[str, str]:
|
|
digits = re.sub(r"\D", "", period or "")
|
|
month, year = int(digits[:2]), int(digits[2:])
|
|
last = calendar.monthrange(year, month)[1]
|
|
return date(year, month, 1).isoformat(), date(year, month, last).isoformat()
|
|
|
|
|
|
def _gst_regs(db, tenant_id: int, client_id: int):
|
|
return db.execute(
|
|
select(ClientRegistration, RegistrationType)
|
|
.join(RegistrationType, RegistrationType.id == ClientRegistration.registration_type_id)
|
|
.where(ClientRegistration.tenant_id == tenant_id, ClientRegistration.client_id == client_id)
|
|
.order_by(ClientRegistration.id.asc())
|
|
).all()
|
|
|
|
|
|
def _is_gstin(reg, typ) -> bool:
|
|
code = str(getattr(typ, "code", "") or "").upper().strip()
|
|
num = re.sub(r"\s+", "", str(getattr(reg, "registration_number", "") or "").upper())
|
|
return code == "GSTIN" and len(num) == 15
|
|
|
|
|
|
def _vault_entries(db, user, request, tenant_id: int, client_id: int, registration_id: int | None = None):
|
|
q = select(CredentialVaultEntry).where(
|
|
CredentialVaultEntry.tenant_id == tenant_id,
|
|
CredentialVaultEntry.client_id == client_id,
|
|
CredentialVaultEntry.status != "archived",
|
|
)
|
|
if registration_id:
|
|
q = q.where(CredentialVaultEntry.registration_id == registration_id)
|
|
rows = db.execute(q.order_by(CredentialVaultEntry.title.asc())).scalars().all()
|
|
visible = [r for r in rows if can_view_entry(db, user, r, getattr(user, "branch_id", None))]
|
|
gst = [r for r in visible if "gst" in (str(r.title or "") + " " + str(r.category or "") + " " + str(r.portal_url or "")).lower()]
|
|
return gst or visible
|
|
|
|
|
|
def _storage_payload(client, fy: str, gstin: str) -> tuple[str, str]:
|
|
fy_folder = sanitize_segment(f"FY{fy}", "FY")
|
|
letter, client_folder = client_folder_parts(client, int(client.id))
|
|
root = Path(fy_folder) / "Clients" / letter / client_folder
|
|
accounting = root / "Accounting"
|
|
gst = root / "GST" / sanitize_segment(f"GSTIN_{gstin}", "GSTIN")
|
|
return accounting.as_posix(), gst.as_posix()
|
|
|
|
|
|
def _redirect(client_id: int, **params):
|
|
data = {"client_id": client_id, **{k: v for k, v in params.items() if v not in (None, "")}}
|
|
return RedirectResponse("/tools/accounting/gst-reconciliation?" + urlencode(data), status_code=303)
|
|
|
|
|
|
@router.get("")
|
|
def page(request: Request, client_id: int | None = None, registration_id: int | None = None, period: str = "", message: str = "", error: str = ""):
|
|
db = CommonSessionLocal()
|
|
try:
|
|
user, response = _require_partner(request, db, "accounting.learning.view")
|
|
if response:
|
|
return response
|
|
clients, scope = _visible_clients(db, request, user)
|
|
selected = next((c for c in clients if client_id and int(c.id) == int(client_id)), None)
|
|
registrations=[]; selected_reg=None; credentials=[]; node=None; status={}; analysis={}
|
|
if selected:
|
|
registrations=[(r,t) for r,t in _gst_regs(db,scope.tenant_id,selected.id) if _is_gstin(r,t)]
|
|
selected_reg=next((r for r,t in registrations if registration_id and int(r.id)==int(registration_id)),None)
|
|
if not selected_reg and registrations:
|
|
selected_reg=registrations[0][0]
|
|
credentials=_vault_entries(db,user,request,scope.tenant_id,selected.id,int(selected_reg.id) if selected_reg else None)
|
|
node=get_active_storage_node_for_branch(db,scope.tenant_id,scope.branch_id)
|
|
if selected_reg and period and node and _node_online(node):
|
|
gstin=re.sub(r"\s+","",str(selected_reg.registration_number or "").upper())
|
|
try:
|
|
status=(request_agent_command(node.node_code,"gst_return_download_status",{"client_id":selected.id,"gstin":gstin,"period":re.sub(r"\D","",period)},timeout_seconds=8).get("result") or {}).get("job") or {}
|
|
except Exception:
|
|
status={}
|
|
try:
|
|
job_result=status.get("result") or {}
|
|
# analysis is loaded only after an explicit Analyze action; status result is download manifest.
|
|
analysis={}
|
|
except Exception:
|
|
pass
|
|
return templates.TemplateResponse("modules/accounting/templates/accounting/gst_reconciliation.html",{
|
|
"request":request,"current_user":user,"current_user_roles":get_user_roles(db,user.id),"current_user_permissions":get_user_permissions(db,user.id),"csrf_token":get_or_create_csrf_token(request),
|
|
"clients":clients,"selected_client":selected,"registrations":registrations,"selected_registration":selected_reg,"credentials":credentials,"node":node,"node_online":_node_online(node) if node else False,"period":period,"status":status,"analysis":analysis,"message":message,"error":error,"title":"GST Return Reconciliation",
|
|
})
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("/download/start")
|
|
def start_download(request: Request, client_id: int=Form(...), registration_id: int=Form(...), credential_id: int=Form(...), period: str=Form(...), include_2a: str=Form(""), csrf_token: str=Form(...)):
|
|
validate_csrf(request,csrf_token)
|
|
db=CommonSessionLocal()
|
|
try:
|
|
user,response=_require_partner(request,db,"accounting.learning.manage")
|
|
if response: return response
|
|
client,_,scope=_find_visible_client(db,request,user,client_id)
|
|
if not client: return _redirect(client_id,error="Client is not available in your scope.")
|
|
pair=next(((r,t) for r,t in _gst_regs(db,scope.tenant_id,client.id) if int(r.id)==registration_id and _is_gstin(r,t)),None)
|
|
if not pair: return _redirect(client_id,error="Select a valid GSTIN registration.")
|
|
reg,_=pair; gstin=re.sub(r"\s+","",str(reg.registration_number or "").upper())
|
|
cred=db.get(CredentialVaultEntry,credential_id)
|
|
if not cred or int(cred.client_id or 0)!=int(client.id) or (cred.registration_id and int(cred.registration_id)!=int(reg.id)) or not can_view_entry(db,user,cred,scope.branch_id):
|
|
return _redirect(client_id,registration_id=registration_id,period=period,error="Selected GST credential is not available for this client/registration.")
|
|
username=decrypt_value(cred.tenant_id,cred.username_encrypted) or ""; password=decrypt_value(cred.tenant_id,cred.secret_encrypted) or ""
|
|
if not username or not password: return _redirect(client_id,registration_id=registration_id,period=period,error="GST username/password is missing in Credential Vault.")
|
|
fy=_fy_for_period(period); accounting_dir,gst_dir=_storage_payload(client,fy,gstin)
|
|
node=get_active_storage_node_for_branch(db,scope.tenant_id,scope.branch_id)
|
|
if not node or not _node_online(node): return _redirect(client_id,registration_id=registration_id,period=period,error="Local Storage Agent is offline.")
|
|
result=request_agent_command(node.node_code,"gst_return_download_start",{"client_id":client.id,"client_name":client.client_name,"gstin":gstin,"financial_year":fy,"period":re.sub(r"\D","",period),"gst_relative_dir":gst_dir,"accounting_relative_dir":accounting_dir,"username":username,"password":password,"include_2a":bool(include_2a),"login_timeout_seconds":900},timeout_seconds=15)
|
|
log_access(db,request,user,cred,"use_for_gst_download",reason=f"GST return download {period}",fields="username,secret",success=bool(result.get("ok")))
|
|
db.commit()
|
|
if not result.get("ok"): return _redirect(client_id,registration_id=registration_id,period=period,error=result.get("error") or "GST download could not be started.")
|
|
return _redirect(client_id,registration_id=registration_id,period=period,message="GST browser started on the Local Storage workstation. Complete captcha/OTP there; downloaded returns will be stored in the client GST directory.")
|
|
except Exception as exc:
|
|
db.rollback(); return _redirect(client_id,registration_id=registration_id,period=period,error=str(exc))
|
|
finally: db.close()
|
|
|
|
|
|
@router.post("/analyze")
|
|
def analyze(request: Request, client_id: int=Form(...), registration_id: int=Form(...), period: str=Form(...), csrf_token: str=Form(...)):
|
|
validate_csrf(request,csrf_token)
|
|
db=CommonSessionLocal()
|
|
try:
|
|
user,response=_require_partner(request,db,"accounting.learning.manage")
|
|
if response: return response
|
|
client,_,scope=_find_visible_client(db,request,user,client_id)
|
|
if not client: return _redirect(client_id,error="Client is not available in your scope.")
|
|
pair=next(((r,t) for r,t in _gst_regs(db,scope.tenant_id,client.id) if int(r.id)==registration_id and _is_gstin(r,t)),None)
|
|
if not pair: return _redirect(client_id,error="GSTIN registration was not found.")
|
|
reg,_=pair; gstin=re.sub(r"\s+","",str(reg.registration_number or "").upper()); fy=_fy_for_period(period)
|
|
accounting_dir,gst_dir=_storage_payload(client,fy,gstin); date_from,date_to=_period_bounds(period)
|
|
node=get_active_storage_node_for_branch(db,scope.tenant_id,scope.branch_id)
|
|
if not node or not _node_online(node): return _redirect(client_id,registration_id=registration_id,period=period,error="Local Storage Agent is offline.")
|
|
res=request_agent_command(node.node_code,"gst_reconciliation_analyze",{"client_id":client.id,"gstin":gstin,"financial_year":fy,"period":re.sub(r"\D","",period),"gst_relative_dir":gst_dir,"accounting_relative_dir":accounting_dir,"date_from":date_from,"date_to":date_to},timeout_seconds=25)
|
|
if not res.get("ok"): return _redirect(client_id,registration_id=registration_id,period=period,error=res.get("error") or "GST reconciliation failed.")
|
|
# Save compact analysis in session for immediate display; no GST raw data or credentials are stored on VPS.
|
|
request.session["gst_reconciliation_result"]=(res.get("result") or {}).get("analysis") or {}
|
|
return _redirect(client_id,registration_id=registration_id,period=period,message="GST Purchase, Sales and ITC reconciliation completed from local stored return data and Accounting Mirror.")
|
|
except Exception as exc:
|
|
return _redirect(client_id,registration_id=registration_id,period=period,error=str(exc))
|
|
finally: db.close()
|