import os from collections import OrderedDict from dataclasses import dataclass, field from datetime import datetime from typing import Any from zoneinfo import ZoneInfo from urllib.parse import urljoin import requests import urllib3 import hashlib import json from data.master_data import BRANCHES as _BRANCHES, PURCHASING_GROUPS as _PURCHASING_GROUPS urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) def _build_branch_plant_options() -> tuple[str, ...]: seen: set = set() opts: list[str] = [] for b in _BRANCHES: p = b["plant"] if p not in seen: seen.add(p) opts.append(f"[{{plant:{p}}}]") return tuple(opts) _BRANCH_PLANT_OPTIONS: tuple[str, ...] = _build_branch_plant_options() def _build_brand_options() -> tuple[str, ...]: """Build [{brand:CODE}] options for every purchasing group.""" opts = ["[{brand:}]"] # All brands first for code in sorted(_PURCHASING_GROUPS.keys()): opts.append(f"[{{brand:{code}}}]") return tuple(opts) _BRAND_OPTIONS: tuple[str, ...] = _build_brand_options() def _env_flag(name: str, default: bool = False) -> bool: value = os.getenv(name) if value is None: return default return str(value).strip().lower() in {"1", "true", "yes", "on"} @dataclass(frozen=True) class SapReportField: name: str label: str default: str = "" required: bool = False placeholder: str = "" help_text: str = "" field_type: str = "text" # "text" or "select" options: tuple[str, ...] = () def to_dict(self, resolved_default: str | None = None) -> dict: return { "name": self.name, "label": self.label, "default": resolved_default if resolved_default is not None else self.default, "required": self.required, "placeholder": self.placeholder, "help_text": self.help_text, "field_type": self.field_type, "options": list(self.options), } @dataclass(frozen=True) class SapReportDefinition: key: str label: str description: str path: str fields: tuple[SapReportField, ...] = field(default_factory=tuple) def to_dict(self, resolved_defaults: dict | None = None) -> dict: resolved_defaults = resolved_defaults or {} return { "key": self.key, "label": self.label, "description": self.description, "path": self.path, "fields": [f.to_dict(resolved_default=resolved_defaults.get(f.name)) for f in self.fields], } class SapApiService: # Human-readable labels for raw SAP API column keys _COLUMN_LABELS: dict[str, str] = { "PLANT": "Plant", "PLANTNAME": "Plant Name", "STATENAME": "State", "BRANDCODE": "Brand", "BRANDNAME": "Brand Name", "MATERIALGROUP": "Material Group", "MAT_GRP_NAME": "Product Group", "PARTNERCODE": "Partner Code", "PARTNERNAME": "Partner Name", "NETVALUE": "Net Value", "NETVAL": "Net Value", "BILLINGDATE": "Billing Date", "COMPANY_CODE": "Company", "COMPANYCODE": "Company", "SALESORG": "Sales Org", "SALESORG_NAME": "Sales Org Name", "DIVCHANNEL": "Distribution Channel", "CUSTGRP": "Customer Group", "CUSTGRPNAME": "Customer Group Name", "QUANTITY": "Quantity", "SALESQTY": "Sales Qty", "UNIT": "Unit", "CURRENCY": "Currency", "FISCALYEAR": "Fiscal Year", "FISCALPERIOD": "Fiscal Period", "MONTH": "Month", "YEAR": "Year", "REGION": "Region", "DIVISION": "Division", "MATERIAL": "Material", "MATERIALNAME": "Material Name", "DOCTYPE": "Doc Type", "ORDERTYPE": "Order Type", } def __init__(self, base_url: str | None = None, *, timeout: int | None = None, verify_tls: bool | None = None): self.base_url = (base_url or os.getenv("SAP_REPORTS_BASE_URL") or "https://wbdrpl.rptechindia.co:3223").rstrip("/") + "/" self.timeout = timeout or int(os.getenv("SAP_REPORTS_TIMEOUT_SECONDS") or 45) self.verify_tls = verify_tls if verify_tls is not None else _env_flag("SAP_REPORTS_VERIFY_TLS", default=False) self._reports = { report.key: report for report in ( SapReportDefinition( key="ageing_branch", label="Ageing (Branch)", description="Branch ageing report with plant, brand, company, and RMA filters.", path="/reporting/ageing", fields=( SapReportField("sap-client", "SAP Client", "900", True, field_type="hidden"), SapReportField("date", "Date", "[{date:__TODAY__,__TODAY__}]", True, help_text="Format: [{date:YYYYMMDD,YYYYMMDD}]"), SapReportField("plant", "Branch / Plant", "", field_type="select", options=("",) + _BRANCH_PLANT_OPTIONS, help_text="Leave blank for all branches"), SapReportField("brand", "Brand", "[{brand:ASN,HP}]", True, field_type="select", options=_BRAND_OPTIONS), SapReportField("material", "Material", "[{}]", field_type="hidden"), SapReportField("company", "Company", "[{company:RPPL}]", True, field_type="select", options=("[{company:RPPL}]", "[{company:PVPL}]", "[{company:PVLL}]", "[{company:STPL}]", "[{company:RPLL}]")), SapReportField("inrma", "In RMA", "1", field_type="checkbox"), ), ), SapReportDefinition( key="ageing_vendor", label="Ageing (Vendor)", description="Vendor stock ageing with company, brand, date, and RMA filters.", path="/reporting/venstkage", fields=( SapReportField("company", "Company", "[{company:RPPL}]", True, field_type="select", options=("[{company:RPPL}]", "[{company:PVPL}]", "[{company:PVLL}]", "[{company:STPL}]", "[{company:RPLL}]")), SapReportField("brand", "Brand", "[{brand:ASN,HP}]", True, field_type="select", options=_BRAND_OPTIONS), SapReportField("date", "Date", "[{date:__FIRSTOFMONTH__,__TODAY__}]", True, help_text="Format: [{date:YYYYMMDD,YYYYMMDD}]"), SapReportField("material", "Material", "[{material:}]", field_type="hidden"), SapReportField("inrma", "In RMA", "1", field_type="checkbox"), ), ), SapReportDefinition( key="ageing_vendor_lpd", label="Ageing (Vendor) LPD", description="Vendor ageing LPD variant with blank RMA filter preserved.", path="/reporting/venstkage", fields=( SapReportField("company", "Company", "[{company:RPPL}]", True, field_type="select", options=("[{company:RPPL}]", "[{company:PVPL}]", "[{company:PVLL}]", "[{company:STPL}]", "[{company:RPLL}]")), SapReportField("brand", "Brand", "[{brand:ASN,HP}]", True, field_type="select", options=_BRAND_OPTIONS), SapReportField("date", "Date", "[{date:__FIRSTOFMONTH__,__TODAY__}]", True, help_text="Format: [{date:YYYYMMDD,YYYYMMDD}]"), SapReportField("material", "Material", "[{material:}]", field_type="hidden"), SapReportField("inrma", "In RMA", "", field_type="checkbox"), ), ), SapReportDefinition( key="exim_order_transit", label="EXIM Order And Transit", description="EXIM order and transit status by date, brand, and company.", path="/reporting/yatplorder", fields=( SapReportField("date", "Date", "[{date:__FIRSTOFMONTH__,__TODAY__}]", help_text="Format: [{date:YYYYMMDD,YYYYMMDD}]"), SapReportField("brand", "Brand", "[{brand:ANC,ASN}]", True, field_type="select", options=_BRAND_OPTIONS), SapReportField("company", "Company", "[{company:RPPL}]", True, field_type="select", options=("[{company:RPPL}]", "[{company:PVPL}]", "[{company:PVLL}]", "[{company:STPL}]", "[{company:RPLL}]")), ), ), SapReportDefinition( key="exim_transit_report", label="EXIM Transit Report", description="EXIM transit report filtered by exact date range and company.", path="/reporting/xin", fields=( SapReportField("date", "Date", "[{date:__TODAY__,__TODAY__}]", True, help_text="Format: [{date:YYYYMMDD,YYYYMMDD}]"), SapReportField("company", "Company", "[{company:RPPL}]", True, field_type="select", options=("[{company:RPPL}]", "[{company:PVPL}]", "[{company:PVLL}]", "[{company:STPL}]", "[{company:RPLL}]")), ), ), SapReportDefinition( key="transit", label="Transit", description="Transit stock report with date, plant, brand, company, and RMA filters.", path="/reporting/transit", fields=( SapReportField("sap-client", "SAP Client", "900", True, field_type="hidden"), SapReportField("date", "Date", "[{date:__TODAY__,__TODAY__}]", True, help_text="Format: [{date:YYYYMMDD,YYYYMMDD}]"), SapReportField("plant", "Branch / Plant", "", field_type="select", options=("",) + _BRANCH_PLANT_OPTIONS, help_text="Leave blank for all branches"), SapReportField("brand", "Brand", "[{brand:ASN,HP}]", True, field_type="select", options=_BRAND_OPTIONS), SapReportField("material", "Material", "[{material:}]", field_type="hidden"), SapReportField("company", "Company", "[{company:RPPL}]", True, field_type="select", options=("[{company:RPPL}]", "[{company:PVPL}]", "[{company:PVLL}]", "[{company:STPL}]", "[{company:RPLL}]")), SapReportField("inrma", "In RMA", "", field_type="checkbox"), ), ), SapReportDefinition( key="zva05", label="ZVA05 Report", description="Open order report for company, distribution channel, division, and open-order filters.", path="/reporting/zva05", fields=( SapReportField("date", "Date", "[{date:20010101,99990101}]", True, help_text="Format: [{date:YYYYMMDD,YYYYMMDD}]"), SapReportField("company", "Company", "[{company:RPPL}]", True, field_type="select", options=("[{company:RPPL}]", "[{company:PVPL}]", "[{company:PVLL}]", "[{company:STPL}]", "[{company:RPLL}]")), SapReportField("Distributionchannel", "Distribution Channel", "[{Distributionchannel:ZB,ZC,ZE,ZO}]", True), SapReportField("Division", "Division", "[{Division:ZR}]", True), SapReportField("Openord", "Open Order", "[{Openord:X}]", True, field_type="select", options=("[{Openord:X}]", "[{Openord:}]")), ), ), SapReportDefinition( key="rebate", label="Rebate", description="Rebate query with plant, brand, billing dates, series, and material group.", path="/zrebate_rest2/rptech/rebate", fields=( SapReportField("sap-client", "SAP Client", "900", True, field_type="hidden"), SapReportField("billingdocument", "Billing Document", ""), SapReportField("plant", "Plant", "11004404", True, placeholder="Plant code"), SapReportField("brand", "Brand", "ASU", True, placeholder="e.g. ASU, ASN, HP"), SapReportField("billingdatefrom", "Billing Date From", "__FIRSTOFMONTH_DDMMYYYY__", True, help_text="Format: DD-MM-YYYY"), SapReportField("billingdateto", "Billing Date To", "__TODAY_DDMMYYYY__", True, help_text="Format: DD-MM-YYYY"), SapReportField("Basic_series", "Basic Series", ""), SapReportField("Material_group", "Material Group", ""), ), ), SapReportDefinition( key="stock", label="Stock", description="Stock report with zero-stock, RMA, and INS filters.", path="/reporting/stock", fields=( SapReportField("sap-client", "SAP Client", "900", True, field_type="hidden"), SapReportField("date", "Date", "[{date:__TODAY__,__TODAY__}]", True, help_text="Format: [{date:YYYYMMDD,YYYYMMDD}]"), SapReportField("plant", "Branch / Plant", "", field_type="select", options=("",) + _BRANCH_PLANT_OPTIONS, help_text="Leave blank for all branches"), SapReportField("brand", "Brand", "[{brand:HP}]", True, field_type="select", options=_BRAND_OPTIONS), SapReportField("material", "Material", "[{material:}]", field_type="hidden"), SapReportField("company", "Company", "[{company:RPPL}]", True, field_type="select", options=("[{company:RPPL}]", "[{company:PVPL}]", "[{company:PVLL}]", "[{company:STPL}]", "[{company:RPLL}]")), SapReportField("zerostk", "Zero Stock", "N", field_type="select", options=("N", "Y")), SapReportField("inrma", "In RMA", "1", field_type="checkbox"), SapReportField("ins", "INS", "X", field_type="select", options=("X", "")), ), ), SapReportDefinition( key="sales_brandwise", label="Sales Brandwise", description="Brandwise sales report using exact date range, company, and no-CN/DN flag.", path="/reporting/sales", fields=( SapReportField("sap-client", "SAP Client", "900", True, field_type="hidden"), SapReportField("date", "Date", "[{date:__FIRSTOFMONTH__,__TODAY__}]", True, help_text="Format: [{date:YYYYMMDD,YYYYMMDD}]"), SapReportField("plant", "Branch / Plant", "", field_type="select", options=("",) + _BRANCH_PLANT_OPTIONS, help_text="Leave blank for all branches"), SapReportField("brand", "Brand", "[{brand:ASN}]", True, field_type="select", options=_BRAND_OPTIONS), SapReportField("company", "Company", "[{company:RPPL}]", True, field_type="select", options=("[{company:RPPL}]", "[{company:PVPL}]", "[{company:PVLL}]", "[{company:STPL}]", "[{company:RPLL}]")), SapReportField("nocndn", "No CN/DN", "", field_type="select", options=("", "X")), ), ), SapReportDefinition( key="sales_plantwise", label="Sales Plantwise", description="Plantwise sales report using accounting period and plant filters.", path="/reporting/sales", fields=( SapReportField("sap-client", "SAP Client", "900", True, field_type="hidden"), SapReportField("period", "Period", "[{period:__PERIOD_MMM_YY__}]", True, help_text="Format: [{period:MMM-YY}] e.g. APR-26"), SapReportField("plant", "Plant", "[{plant:4100}]", True, field_type="select", options=_BRANCH_PLANT_OPTIONS), SapReportField("company", "Company", "[{company:RPPL}]", True, field_type="select", options=("[{company:RPPL}]", "[{company:PVPL}]", "[{company:PVLL}]", "[{company:STPL}]", "[{company:RPLL}]")), SapReportField("nocndn", "No CN/DN", "", field_type="select", options=("", "X")), ), ), SapReportDefinition( key="zme3l", label="ZME3L", description="Purchase order style report filtered by plant and document type.", path="/reporting/zme3l", fields=( SapReportField("date", "Date", "[{date:__FIRSTOFMONTH__,__TODAY__}]", True, help_text="Format: [{date:YYYYMMDD,YYYYMMDD}]"), SapReportField("plant", "Plant", "", field_type="select", options=("",) + _BRANCH_PLANT_OPTIONS, help_text="Leave blank for all branches"), SapReportField("documentype", "Document Type", "[{documentype:ZMK}]", True, field_type="hidden"), ), ), ) } def list_reports(self) -> list[SapReportDefinition]: return list(self._reports.values()) def get_report(self, report_key: str) -> SapReportDefinition | None: return self._reports.get(report_key) @staticmethod def _resolve_date_tokens(value: str) -> str: """Replace __TODAY__, __FIRSTOFMONTH__ etc. with actual dates at call time.""" today = datetime.now(tz=ZoneInfo("Asia/Kolkata")) first = today.replace(day=1) replacements = { "__TODAY__": today.strftime("%Y%m%d"), "__FIRSTOFMONTH__": first.strftime("%Y%m%d"), "__TODAY_DDMMYYYY__": today.strftime("%d-%m-%Y"), "__FIRSTOFMONTH_DDMMYYYY__": first.strftime("%d-%m-%Y"), "__PERIOD_MMM_YY__": today.strftime("%b-%y").upper(), } for token, replacement in replacements.items(): value = value.replace(token, replacement) return value def default_parameters(self, report_key: str) -> OrderedDict[str, str]: report = self.get_report(report_key) if not report: raise ValueError(f"Unknown SAP report: {report_key}") return OrderedDict( (field.name, self._resolve_date_tokens(field.default)) for field in report.fields ) def build_request_url(self, report_key: str, overrides: dict[str, Any] | None = None) -> tuple[str, OrderedDict[str, str]]: report = self.get_report(report_key) if not report: raise ValueError(f"Unknown SAP report: {report_key}") merged = self.default_parameters(report_key) for field in report.fields: if overrides and field.name in overrides: merged[field.name] = "" if overrides[field.name] is None else str(overrides[field.name]).strip() return urljoin(self.base_url, report.path.lstrip("/")), merged def execute_report(self, report_key: str, overrides: dict[str, Any] | None = None) -> dict: report = self.get_report(report_key) if not report: raise ValueError(f"Unknown SAP report: {report_key}") url, params = self.build_request_url(report_key, overrides) cache_key = f"sap_rpt:{report_key}:" + hashlib.md5( "&".join(f"{k}={v}" for k, v in sorted(params.items())).encode() ).hexdigest()[:12] stale_key = cache_key + ":stale" # Fast path: return cached result try: from extensions import redis_client as _rc cached_raw = _rc.get(cache_key) if cached_raw: cached = json.loads(cached_raw) cached["trace"] = [ {"label": "Source", "value": "SAP API (cached)"}, {"label": "Status", "value": "HIT"}, {"label": "Rows", "value": str(cached.get("row_count", 0))}, {"label": "Fetched", "value": cached.get("_fetched_at", "")}, ] return cached except Exception: pass try: qs = "&".join(f"{k}={v}" for k, v in params.items()) full_url = f"{url}?{qs}" if qs else url response = requests.get(full_url, timeout=self.timeout, verify=self.verify_tls) response.raise_for_status() except requests.exceptions.RequestException as e: error_msg = str(e) # Fallback: return stale cached data with warning try: from extensions import redis_client as _rc stale_raw = _rc.get(stale_key) if stale_raw: stale = json.loads(stale_raw) stale["trace"] = [ {"label": "Source", "value": "SAP API (stale cache)"}, {"label": "Status", "value": "SAP Unavailable"}, {"label": "Rows", "value": str(stale.get("row_count", 0))}, {"label": "Fetched", "value": stale.get("_fetched_at", "")}, ] stale["stale"] = True stale["stale_reason"] = f"SAP unreachable: {error_msg[:120]}" return stale except Exception: pass trace = [ {"label": "Source", "value": "SAP API"}, {"label": "Status", "value": "Failed"}, {"label": "Error", "value": error_msg}, {"label": "Fetched", "value": datetime.now(tz=ZoneInfo("Asia/Kolkata")).strftime("%Y-%m-%d %H:%M:%S IST")}, ] return { "report_key": report.key, "report_label": report.label, "request_url": getattr(getattr(e, 'request', None), 'url', None) or full_url, "parameters": params, "rows": [], "row_count": 0, "raw_payload": {"error": error_msg}, "trace": trace, } payload = self._parse_payload(response) rows = self._extract_rows(payload) fetched_at = datetime.now(tz=ZoneInfo("Asia/Kolkata")).strftime("%Y-%m-%d %H:%M:%S IST") trace = [ {"label": "Source", "value": "SAP API"}, {"label": "Status", "value": str(response.status_code)}, {"label": "Rows", "value": str(len(rows))}, {"label": "Fetched", "value": fetched_at}, ] result = { "report_key": report.key, "report_label": report.label, "request_url": response.url, "parameters": params, "rows": rows, "row_count": len(rows), "raw_payload": payload, "_fetched_at": fetched_at, "trace": trace, } # Store in cache: 5-min main TTL + 1-hour stale backup try: from extensions import redis_client as _rc serialized = json.dumps(result, default=str) _rc.set(cache_key, serialized, ex=300) _rc.set(stale_key, serialized, ex=3600) except Exception: pass return result def _parse_payload(self, response: requests.Response): content_type = response.headers.get("Content-Type", "").lower() if "json" in content_type: return response.json() body = response.text.strip() if body.startswith("{") or body.startswith("["): try: return response.json() except ValueError: return {"response_text": body} return {"response_text": body} def _extract_rows(self, payload) -> list[dict]: if payload is None: return [] if isinstance(payload, list): return [self._normalize_row(item) for item in payload] if isinstance(payload, dict): for key in ("result", "results", "data", "items", "rows", "records", "value"): candidate = payload.get(key) if isinstance(candidate, list): return [self._normalize_row(item) for item in candidate] if isinstance(candidate, dict): nested_rows = self._extract_rows(candidate) if nested_rows: return nested_rows for value in payload.values(): if isinstance(value, list): return [self._normalize_row(item) for item in value] if isinstance(value, dict): nested_rows = self._extract_rows(value) if nested_rows: return nested_rows return [self._normalize_row(payload)] return [{"value": payload}] def _normalize_row(self, item) -> dict: if isinstance(item, dict): return { self._COLUMN_LABELS.get(str(key), str(key).replace('_', ' ').title()): self._normalize_value(value) for key, value in item.items() } return {"value": self._normalize_value(item)} def _normalize_value(self, value): if isinstance(value, (str, int, float, bool)) or value is None: return value if isinstance(value, list): return ", ".join(str(self._normalize_value(item)) for item in value) if isinstance(value, dict): return str({key: self._normalize_value(item) for key, item in value.items()}) return str(value)