#!/usr/bin/env python3 """BRSR X-ray pipeline — Nifty-50 slice. Steps (run in order, or `all`): fetch - Nifty-50 constituents + NSE BRSR filing list -> data/manifest.csv download - XBRL instance documents -> data/raw/*.xml (gitignored; re-downloadable) parse - data/facts_long.csv.gz (every fact) + data/brsr_nifty50.csv (curated wide) No LLM anywhere: the XBRL is deterministic. Sources: nsearchives.nseindia.com (Nifty-50 constituent CSV, XBRL files), www.nseindia.com JSON API (filing list). Polite throttle 0.7s. """ import argparse, csv, gzip, io, json, re, sys, time, urllib.request from pathlib import Path from xml.etree import ElementTree as ET BASE = Path(__file__).resolve().parent.parent # builds/brsr-xray/ DATA = BASE / "data" RAW = DATA / "raw" UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/126.0 Safari/537.36") NIFTY_URL = "https://nsearchives.nseindia.com/content/indices/ind_nifty50list.csv" BRSR_API = "https://www.nseindia.com/api/corporate-bussiness-sustainabilitiy?index=equities" REFERER = "https://www.nseindia.com/companies-listing/corporate-filings-bussiness-sustainabilitiy-reports" # Curated fields for the wide table. tag -> column. Values taken from the # current-FY, non-dimensional ("Main") context of each instance document. CURATED = { "WhetherDetailsOfGreenHouseGasEmissionsAndItsIntensityIsApplicableToTheCompany": "ghg_applicable", "TotalScope1Emissions": "scope1_tco2e", "TotalScope2Emissions": "scope2_tco2e", "WhetherTotalScope3EmissionsAndItsIntensityIsApplicableToTheCompany": "scope3_applicable", "TotalScope3Emissions": "scope3_tco2e", "TotalScope1AndScope2EmissionsIntensityPerRupeeOfTurnover": "s1s2_intensity_per_inr", "WhetherDetailsOfTotalEnergyConsumptionAndEnergyIntensityApplicableToTheCompany": "energy_applicable", "TotalEnergyConsumedFromRenewableSources": "energy_renewable_gj", "TotalEnergyConsumedFromNonRenewableSources": "energy_nonrenewable_gj", "TotalEnergyConsumedFromRenewableAndNonRenewableSources": "energy_total_gj", "TotalVolumeOfWaterWithdrawal": "water_withdrawal_kl", "TotalVolumeOfWaterConsumption": "water_consumption_kl", "TotalWasteGenerated": "waste_generated_mt", "GrossWagesPaidToFemale": "gross_wages_female", "ComplaintsOnPOSHUpHeld": "posh_upheld", "WhetherTheCompanyHasUndertakenAssessmentOrAssuranceOfTheBRSRCore": "brsr_core_assured", "ReportingBoundary": "reporting_boundary", # "Standalone basis" | "Consolidated basis" -- # the primer's #1 comparability trap; must be # a first-class column, never assumed uniform } # Text blocks scanned (full text) for net-zero / carbon-neutral commitments. TARGET_TEXTBLOCKS = [ "SpecificCommitmentsGoalsAndTargetsSetByTheEntityWithDefinedTimelinesExplanatoryTextBlock", "PerformanceOfTheEntityAgainstTheSpecificCommitmentsGoalsAndTargetsAlongWithReasonsInCaseTheSameAreNotMetExplanatoryTextBlock", "StatementByDirectorResponsibleForTheBusinessResponsibilityReportHighlightingESGRelatedChallengesTargetsAndAchievementsExplanatoryTextBlock", ] NETZERO_RE = re.compile(r"net[\s‑-]?zero|carbon[\s-]?neutral", re.I) def get(url, binary=False, retries=3): req = urllib.request.Request(url, headers={ "User-Agent": UA, "Accept": "*/*", "Referer": REFERER, "Accept-Language": "en-US,en;q=0.9"}) for i in range(retries): try: with urllib.request.urlopen(req, timeout=60) as r: b = r.read() return b if binary else b.decode("utf-8", errors="replace") except Exception as e: if i == retries - 1: raise time.sleep(2 * (i + 1)) def step_fetch(): DATA.mkdir(parents=True, exist_ok=True) nifty = list(csv.DictReader(io.StringIO(get(NIFTY_URL)))) print(f"nifty constituents: {len(nifty)}") filings = json.loads(get(BRSR_API))["data"] print(f"brsr filings in window: {len(filings)}") by_symbol = {} for f in filings: by_symbol.setdefault(f["symbol"], []).append(f) rows, missing = [], [] for c in nifty: sym = c["Symbol"] cand = by_symbol.get(sym, []) if not cand: missing.append(sym) continue # one filing per FY (latest submission wins), newest two FYs per_fy = {} for f in sorted(cand, key=lambda z: z["submissionDate"]): per_fy[(f["fyFrom"], f["fyTo"])] = f for (fy_from, fy_to), f in sorted(per_fy.items(), reverse=True)[:2]: rows.append({ "symbol": sym, "company": c["Company Name"], "industry": c["Industry"], "isin": c["ISIN Code"], "fy_from": fy_from, "fy_to": fy_to, "submission_date": f["submissionDate"], "xbrl_url": f["xbrlFile"], "pdf_url": f["attachmentFile"], }) with open(DATA / "manifest.csv", "w", newline="") as fh: w = csv.DictWriter(fh, fieldnames=list(rows[0].keys())) w.writeheader() w.writerows(rows) print(f"manifest: {len(rows)} filings across {len({r['symbol'] for r in rows})} companies") if missing: print(f"NOT IN WINDOW ({len(missing)}): {missing}") def step_download(): RAW.mkdir(parents=True, exist_ok=True) rows = list(csv.DictReader(open(DATA / "manifest.csv"))) ok = fail = skip = 0 for r in rows: dest = RAW / f"{r['symbol']}_{r['fy_from']}-{r['fy_to']}.xml" if dest.exists() and dest.stat().st_size > 1000: skip += 1 continue try: dest.write_bytes(get(r["xbrl_url"], binary=True)) ok += 1 except Exception as e: print(f"FAIL {r['symbol']} {r['fy_from']}: {e}") fail += 1 time.sleep(0.7) print(f"downloaded {ok}, skipped {skip}, failed {fail}") def local(tag): return tag.rsplit("}", 1)[-1] def parse_one(path, fy_from, fy_to): """Return (facts, curated) for one instance document. Captures BOTH current-year (CY, contextRef ~ 'CYMain') and prior-year (PY, contextRef ~ 'PYMain') non-dimensional values for curated numeric fields. PY-as-filed-this-year is an internal-consistency triangulation source: it is the same fact as filed last year's CY, independently re-typed by the company into this year's return, and can be diffed against a separately-downloaded prior filing or an external source (e.g. Climate TRACE) covering the same period. """ tree = ET.parse(path) root = tree.getroot() ctx = {} for c in root.iter(): if local(c.tag) != "context": continue cid = c.get("id") start = end = instant = None dimensional = False for e in c.iter(): ln = local(e.tag) if ln == "startDate": start = e.text elif ln == "endDate": end = e.text elif ln == "instant": instant = e.text elif ln in ("segment", "scenario"): dimensional = True ctx[cid] = ((start, end, instant), dimensional) cy = (f"{fy_from}-04-01", f"{fy_to}-03-31") py = (f"{int(fy_from)-1}-04-01", f"{int(fy_to)-1}-03-31") facts, curated, curated_units, curated_py = {}, {}, {}, {} facts = [] netzero_hit = False for e in root.iter(): cid = e.get("contextRef") if cid is None or e.text is None: continue ns = e.tag.split("}")[0] if "in-capmkt" not in ns: continue tag = local(e.tag) (start, end, instant), dimensional = ctx.get(cid, ((None, None, None), False)) is_main_cy = (not dimensional) and ((start, end) == cy or instant == cy[1]) is_main_py = (not dimensional) and ((start, end) == py or instant == py[1]) text = e.text.strip() if tag in TARGET_TEXTBLOCKS and is_main_cy and NETZERO_RE.search(text): netzero_hit = True unit = e.get("unitRef") or "" val = re.sub(r"\s+", " ", text)[:300] facts.append({"tag": tag, "context": cid, "is_main_cy": int(is_main_cy), "is_main_py": int(is_main_py), "unit": unit, "value": val}) if tag in CURATED: col = CURATED[tag] if is_main_cy and col not in curated: curated[col] = val curated_units[col] = unit if is_main_py and col not in curated_py: curated_py[col] = val curated["netzero_or_carbonneutral_in_targets"] = netzero_hit return facts, curated, curated_units, curated_py def step_parse(): rows = list(csv.DictReader(open(DATA / "manifest.csv"))) wide, n_facts = [], 0 long_fh = gzip.open(DATA / "facts_long.csv.gz", "wt", newline="") lw = csv.writer(long_fh) lw.writerow(["symbol", "fy_from", "fy_to", "tag", "context", "is_main_cy", "is_main_py", "unit", "value"]) for r in rows: path = RAW / f"{r['symbol']}_{r['fy_from']}-{r['fy_to']}.xml" if not path.exists(): print(f"missing raw file: {path.name}") continue try: facts, curated, units, py = parse_one(path, r["fy_from"], r["fy_to"]) except Exception as e: print(f"PARSE FAIL {path.name}: {e}") continue for f in facts: lw.writerow([r["symbol"], r["fy_from"], r["fy_to"], f["tag"], f["context"], f["is_main_cy"], f["is_main_py"], f["unit"], f["value"]]) n_facts += len(facts) row = {"symbol": r["symbol"], "company": r["company"], "industry": r["industry"], "fy_from": r["fy_from"], "fy_to": r["fy_to"], "submission_date": r["submission_date"]} for col in CURATED.values(): row[col] = curated.get(col, "") row[col + "_unit"] = units.get(col, "") row[col + "_PY_as_filed"] = py.get(col, "") row["netzero_or_carbonneutral_in_targets"] = curated.get( "netzero_or_carbonneutral_in_targets", False) wide.append(row) long_fh.close() cols = list(wide[0].keys()) with open(DATA / "brsr_nifty50.csv", "w", newline="") as fh: w = csv.DictWriter(fh, fieldnames=cols) w.writeheader() w.writerows(wide) print(f"parsed {len(wide)} filings, {n_facts} facts") if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("step", choices=["fetch", "download", "parse", "all"]) a = ap.parse_args() if a.step in ("fetch", "all"): step_fetch() if a.step in ("download", "all"): step_download() if a.step in ("parse", "all"): step_parse()