Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| """ | |
| builders/build_docx.py -- Enhanced Word report builder. | |
| Sections: Cover, Executive Summary, Company Profile, Industry Profile, | |
| SWOT, Industry Trends, Financial Trends, Ratios, DuPont, Peer Comparison, | |
| Key Insights, Risks, Things to Watch, Limitations, Sources. | |
| """ | |
| import json, sys, datetime | |
| from pathlib import Path | |
| from typing import Dict, List, Optional | |
| try: | |
| from docx import Document | |
| from docx.shared import Pt, Cm, RGBColor, Inches | |
| from docx.enum.text import WD_ALIGN_PARAGRAPH | |
| from docx.enum.table import WD_ALIGN_VERTICAL, WD_TABLE_ALIGNMENT | |
| from docx.oxml.ns import qn | |
| from docx.oxml import OxmlElement | |
| except ImportError: | |
| print("ERROR: pip install python-docx", file=sys.stderr); sys.exit(1) | |
| # -- Colors -------------------------------------------------------------------- | |
| NAVY = RGBColor(0x1F, 0x4E, 0x78) | |
| GRAY = RGBColor(0x59, 0x59, 0x59) | |
| GREEN_DARK = RGBColor(0x37, 0x5A, 0x3D) | |
| # -- Formatting helpers -------------------------------------------------------- | |
| def _shd(cell, hex_color): | |
| tcp = cell._tc.get_or_add_tcPr() | |
| s = OxmlElement("w:shd") | |
| s.set(qn("w:val"), "clear"); s.set(qn("w:color"), "auto"); s.set(qn("w:fill"), hex_color) | |
| tcp.append(s) | |
| def _borders(cell, color="BFBFBF", sz="4"): | |
| tcp = cell._tc.get_or_add_tcPr() | |
| tb = OxmlElement("w:tcBorders") | |
| for edge in ("top","left","bottom","right"): | |
| b = OxmlElement(f"w:{edge}") | |
| b.set(qn("w:val"),"single"); b.set(qn("w:sz"),sz); b.set(qn("w:color"),color) | |
| tb.append(b) | |
| tcp.append(tb) | |
| def _hdr_repeat(row): | |
| trp = row._tr.get_or_add_trPr() | |
| th = OxmlElement("w:tblHeader"); th.set(qn("w:val"),"true"); trp.append(th) | |
| def _add_heading(doc, text, level=1): | |
| sizes = {0:28, 1:18, 2:14, 3:12} | |
| p = doc.add_paragraph() | |
| r = p.add_run(text) | |
| r.font.size = Pt(sizes.get(level, 12)); r.font.bold = True | |
| r.font.color.rgb = NAVY; r.font.name = "Calibri" | |
| p.paragraph_format.space_before = Pt(12); p.paragraph_format.space_after = Pt(6) | |
| return p | |
| def _add_text(doc, text, size=11): | |
| if not text: return | |
| for block in [b.strip() for b in str(text).split("\n\n") if b.strip()]: | |
| p = doc.add_paragraph() | |
| r = p.add_run(block.replace("\n"," ")) | |
| r.font.size = Pt(size); r.font.name = "Calibri" | |
| p.paragraph_format.space_after = Pt(8) | |
| def _fmt_v(v, kind="money", currency="INR", units="Crores"): | |
| if v is None: return "\u2014" | |
| try: | |
| sym = "\u20b9" if currency == "INR" else ("$" if currency == "USD" else "") | |
| if kind == "money": | |
| # Data is already scaled (Crores for INR, Millions for USD) | |
| av = abs(v) | |
| if currency == "INR": | |
| if av >= 100000: return f"{sym}{v/100000:,.2f} L.Cr" | |
| return f"{sym}{v:,.2f} Cr" | |
| else: | |
| if av >= 1000: return f"{sym}{v/1000:,.2f}B" | |
| return f"{sym}{v:,.2f}M" | |
| if kind == "money_dec": return f"{v:,.2f}" | |
| if kind == "pct": return f"{v*100:.1f}%" | |
| if kind in ("ratio", "days"): return f"{v:.2f}" | |
| return str(v) | |
| except: return str(v) | |
| def _page_break(doc): | |
| p = doc.add_paragraph(); p.add_run().add_break() | |
| # -- Table builder ------------------------------------------------------------- | |
| def _table(doc, headers, rows, formats=None, currency="INR", units="Crores"): | |
| t = doc.add_table(rows=1+len(rows), cols=len(headers)) | |
| t.alignment = WD_TABLE_ALIGNMENT.LEFT | |
| hc = t.rows[0].cells | |
| for i, h in enumerate(headers): | |
| hc[i].text = "" | |
| p = hc[i].paragraphs[0] | |
| r = p.add_run(str(h)); r.font.bold = True; r.font.color.rgb = RGBColor(0xFF,0xFF,0xFF) | |
| r.font.size = Pt(10); r.font.name = "Calibri" | |
| p.alignment = WD_ALIGN_PARAGRAPH.CENTER if i > 0 else WD_ALIGN_PARAGRAPH.LEFT | |
| _shd(hc[i], "1F4E78"); _borders(hc[i]) | |
| _hdr_repeat(t.rows[0]) | |
| for ri, row in enumerate(rows): | |
| bg = "F2F2F2" if ri % 2 else "FFFFFF" | |
| cells = t.rows[ri+1].cells | |
| for ci, val in enumerate(row): | |
| cells[ci].text = "" | |
| p = cells[ci].paragraphs[0] | |
| kind = formats[ci] if formats and ci < len(formats) else "money" | |
| if ci == 0: | |
| text = str(val) if val is not None else "--" | |
| else: | |
| text = _fmt_v(val, kind, currency, units) | |
| r = p.add_run(text); r.font.size = Pt(10); r.font.name = "Calibri" | |
| p.alignment = WD_ALIGN_PARAGRAPH.LEFT if ci == 0 else WD_ALIGN_PARAGRAPH.RIGHT | |
| _shd(cells[ci], bg); _borders(cells[ci]) | |
| doc.add_paragraph().paragraph_format.space_after = Pt(6) | |
| return t | |
| def _gather(items, src, years): | |
| return [[label] + [(src.get(key) or [None]*len(years))[:len(years)][i] | |
| for i in range(len(years))] for label, key in items] | |
| # -- Section builders ---------------------------------------------------------- | |
| def _cover(doc, data): | |
| co = data.get("company", {}) | |
| units_label = data.get("units_label", data.get("units", "Crores")) | |
| currency = data.get("currency", co.get("currency", "INR")) | |
| for _ in range(3): doc.add_paragraph() | |
| p = doc.add_paragraph() | |
| r = p.add_run("Financial Statement Analysis") | |
| r.font.size = Pt(32); r.font.bold = True; r.font.color.rgb = NAVY; r.font.name = "Calibri" | |
| p = doc.add_paragraph() | |
| r = p.add_run(co.get("name","")) | |
| r.font.size = Pt(24); r.font.color.rgb = GRAY; r.font.name = "Calibri" | |
| p = doc.add_paragraph() | |
| r = p.add_run(f"{co.get('ticker','')} \u00b7 {co.get('exchange','')} \u00b7 {co.get('industry','')}") | |
| r.font.size = Pt(13); r.font.color.rgb = GRAY; r.font.italic = True; r.font.name = "Calibri" | |
| for _ in range(6): doc.add_paragraph() | |
| for text in [ | |
| f"Report date: {co.get('report_date', datetime.date.today().isoformat())}", | |
| f"Analysis window: {', '.join(data.get('years',[]))}", | |
| f"Reporting currency: {currency} | Units: {units_label}", | |
| f"Market cap: {co.get('market_cap', 'N/A')} | Share price: {co.get('share_price', 'N/A')}", | |
| f"AI models used: {', '.join(data.get('models_used',[]) or ['Data-only (no AI narrative)'])}", | |
| f"Data source: {data.get('source', 'Yahoo Finance (yahooquery)')}", | |
| ]: | |
| p = doc.add_paragraph(); r = p.add_run(text) | |
| r.font.size = Pt(11); r.font.color.rgb = GRAY; r.font.name = "Calibri" | |
| _page_break(doc) | |
| def _swot_section(doc, data): | |
| _add_heading(doc, "SWOT Analysis", 1) | |
| swot = data.get("swot", {}) | |
| configs = [ | |
| ("Strengths", "strengths", "C6EFCE", GREEN_DARK), | |
| ("Weaknesses", "weaknesses", "FFC7CE", RGBColor(0xC0,0x00,0x00)), | |
| ("Opportunities", "opportunities", "DDEBF7", NAVY), | |
| ("Threats", "threats", "FFEB9C", RGBColor(0x7F,0x60,0x00)), | |
| ] | |
| parsed_any = any(len(swot.get(k, [])) > 0 for _, k, _, _ in configs) | |
| if not parsed_any and swot.get("raw"): | |
| for line in swot["raw"].split("\n"): | |
| if line.strip(): | |
| p = doc.add_paragraph() | |
| r = p.add_run(line.strip()) | |
| r.font.size = Pt(11); r.font.name = "Calibri" | |
| doc.add_paragraph() | |
| return | |
| t = doc.add_table(rows=2, cols=2) | |
| t.alignment = WD_TABLE_ALIGNMENT.LEFT | |
| idx = 0 | |
| for row_i in range(2): | |
| for col_i in range(2): | |
| if idx >= len(configs): break | |
| label, key, bg, color = configs[idx]; idx += 1 | |
| cell = t.cell(row_i, col_i) | |
| cell.text = "" | |
| p = cell.paragraphs[0] | |
| r = p.add_run(label); r.font.bold = True; r.font.size = Pt(12) | |
| r.font.color.rgb = color; r.font.name = "Calibri" | |
| _shd(cell, bg) | |
| for item in swot.get(key, []): | |
| p2 = cell.add_paragraph(f"• {item}") | |
| for run in p2.runs: run.font.size = Pt(10); run.font.name = "Calibri" | |
| doc.add_paragraph() | |
| def _trends_section(doc, data): | |
| _add_heading(doc, "Multi-Year Financial Trends", 1) | |
| years = data.get("years", []) | |
| currency = data.get("currency", "INR") | |
| units_label = data.get("units_label", data.get("units", "Crores")) | |
| trend = data.get("trend_analysis", {}) | |
| for section_title, prose_key, items, src_key, fmts in [ | |
| ("Revenue & Profitability", "revenue_and_profitability", | |
| [("Revenue","revenue"),("Gross Profit","gross_profit"),("Operating Income","operating_income"), | |
| ("Net Income","net_income")], | |
| "income_statement", ["text"]+["money"]*len(years)), | |
| ("Balance Sheet Strength", "balance_sheet_strength", | |
| [("Total Assets","total_assets"),("Total Equity","total_equity"), | |
| ("Long-term Debt","long_term_debt"),("Current Ratio","current_ratio")], | |
| "balance_sheet", ["text"]+["money"]*len(years)), | |
| ("Cash Flow Generation", "cash_flow_generation", | |
| [("Cash from Operations","cash_from_operations"),("Capex","capex"),("Free Cash Flow","free_cash_flow")], | |
| "cash_flow", ["text"]+["money"]*len(years)), | |
| ]: | |
| _add_heading(doc, section_title, 2) | |
| _add_text(doc, trend.get(prose_key,"")) | |
| src_data = data.get(src_key, {}) | |
| if src_key == "balance_sheet": | |
| merged = dict(data.get("balance_sheet",{})); merged.update(data.get("ratios",{})) | |
| src_data = merged | |
| rows = _gather(items, src_data, years) | |
| _table(doc, [f"({units_label})"]+years, rows, fmts, currency, units_label) | |
| def _ratio_section(doc, data): | |
| _add_heading(doc, "Financial Ratio Analysis", 1) | |
| _add_text(doc, data.get("ratio_analysis_commentary","")) | |
| years = data.get("years", []) | |
| ratios = data.get("ratios", {}) | |
| currency = data.get("currency", "INR") | |
| units_label = data.get("units_label", data.get("units", "Crores")) | |
| for sec_name, items in [ | |
| ("Profitability", [("Gross Margin","gross_margin","pct"),("Operating Margin","operating_margin","pct"), | |
| ("Net Margin","net_margin","pct"),("ROE","roe","pct"),("ROIC","roic","pct")]), | |
| ("Liquidity & Solvency", [("Current Ratio","current_ratio","ratio"),("Quick Ratio","quick_ratio","ratio"), | |
| ("Debt-to-Equity","debt_to_equity","ratio"),("Net Debt/EBITDA","net_debt_to_ebitda","ratio")]), | |
| ("Efficiency", [("Asset Turnover","asset_turnover","ratio"),("DSO (days)","days_sales_outstanding","days"), | |
| ("DPO (days)","days_payables","days"),("CCC (days)","cash_conversion_cycle","days")]), | |
| ("Cash Flow Quality", [("OCF/Net Income","ocf_to_net_income","ratio"),("FCF Margin","fcf_margin","pct"), | |
| ("FCF Conversion","fcf_conversion","ratio")]), | |
| ]: | |
| _add_heading(doc, sec_name, 3) | |
| rows = [] | |
| for label, key, kind in items: | |
| vals = (ratios.get(key,[]) + [None]*len(years))[:len(years)] | |
| rows.append([label] + list(vals)) | |
| kinds = ["text"] + [items[0][2]]*len(years) | |
| _table(doc, ["Metric"]+years, rows, kinds, currency, units_label) | |
| def _dupont_section(doc, data): | |
| _add_heading(doc, "DuPont Analysis", 1) | |
| p = doc.add_paragraph() | |
| r = p.add_run("ROE = Net Margin \u00d7 Asset Turnover \u00d7 Equity Multiplier") | |
| r.font.size = Pt(11); r.font.bold = True; r.font.name = "Calibri" | |
| years = data.get("years", []) | |
| dp = data.get("dupont", {}) | |
| currency = data.get("currency", "INR") | |
| rows = [] | |
| for label, key, kind in [ | |
| ("Net Margin","net_margin","pct"),("\u00d7 Asset Turnover","asset_turnover","ratio"), | |
| ("\u00d7 Equity Multiplier","equity_multiplier","ratio"),("= ROE (DuPont)","roe_computed","pct"), | |
| ("ROE (Reported)","roe_reported","pct"), | |
| ]: | |
| vals = (dp.get(key,[]) + [None]*len(years))[:len(years)] | |
| rows.append([label]+list(vals)) | |
| _table(doc, ["Component"]+years, rows, ["text"]+["pct"]*len(years), currency) | |
| def _peer_section(doc, data): | |
| _add_heading(doc, "Peer Benchmarking", 1) | |
| _add_text(doc, data.get("peer_comparison_commentary","")) | |
| peers = data.get("peers", []) | |
| currency = data.get("currency", "INR") | |
| units_label = data.get("units_label", "Crores") | |
| co = data.get("company",{}); ratios = data.get("ratios",{}) | |
| inc = data.get("income_statement",{}); val = data.get("valuation_snapshot",{}) | |
| def last(arr): | |
| for v in reversed(arr or []): | |
| if v is not None: return v | |
| return None | |
| headers = ["Company","Revenue","Op Margin","Net Margin","ROIC","D/E","FCF Margin","P/E"] | |
| fmts = ["text","money","pct","pct","pct","ratio","pct","ratio"] | |
| rows = [[ | |
| co.get("name",co.get("ticker",""))+" (Target)", | |
| last(inc.get("revenue",[])), last(ratios.get("operating_margin",[])), | |
| last(ratios.get("net_margin",[])), last(ratios.get("roic",[])), | |
| last(ratios.get("debt_to_equity",[])), last(ratios.get("fcf_margin",[])), | |
| val.get("pe_ttm"), | |
| ]] | |
| if not peers: | |
| doc.add_paragraph("Peer comparison data not available for this run.") | |
| for p in peers: | |
| rows.append([p.get("name",p.get("ticker","")), p.get("revenue"), p.get("operating_margin"), | |
| p.get("net_margin"), p.get("roic"), p.get("debt_to_equity"), | |
| p.get("fcf_margin"), p.get("pe_ttm")]) | |
| if rows: | |
| _table(doc, headers, rows, fmts, currency, units_label) | |
| def _bullets(doc, items, heading=None, level=1): | |
| if heading: _add_heading(doc, heading, level) | |
| for item in items: | |
| if isinstance(item, dict): | |
| p = doc.add_paragraph(style="List Bullet") | |
| r1 = p.add_run(f"{item.get('category','')}: "); r1.font.bold=True; r1.font.size=Pt(11); r1.font.name="Calibri" | |
| r2 = p.add_run(item.get("description","")); r2.font.size=Pt(11); r2.font.name="Calibri" | |
| else: | |
| p = doc.add_paragraph(str(item), style="List Bullet") | |
| for r in p.runs: r.font.size=Pt(11); r.font.name="Calibri" | |
| p.paragraph_format.space_after = Pt(2) | |
| def _sources_section(doc, data): | |
| _add_heading(doc, "Sources", 1) | |
| for i, src in enumerate(data.get("sources",[]), 1): | |
| p = doc.add_paragraph() | |
| p.add_run(f"{i}. ").font.size = Pt(10) | |
| url_r = p.add_run(src.get("url","")) | |
| url_r.font.size=Pt(10); url_r.font.color.rgb=RGBColor(0x05,0x63,0xC1); url_r.font.name="Calibri" | |
| suffix = "" | |
| if src.get("description"): suffix += f" -- {src['description']}" | |
| if src.get("retrieved"): suffix += f" (retrieved {src['retrieved']})" | |
| if suffix: | |
| sr = p.add_run(suffix); sr.font.size=Pt(10); sr.font.color.rgb=GRAY; sr.font.name="Calibri" | |
| def _disclaimer(doc): | |
| p = doc.add_paragraph() | |
| r = p.add_run( | |
| "Disclaimer: This report is generated from public financial data and AI analysis for informational " | |
| "purposes only. It is not investment advice. Verify all figures against primary sources before " | |
| "making any financial decisions." | |
| ) | |
| r.font.size=Pt(9); r.font.italic=True; r.font.color.rgb=GRAY; r.font.name="Calibri" | |
| p.paragraph_format.space_before=Pt(12) | |
| # -- Orchestrator -------------------------------------------------------------- | |
| def build_document(data: dict, output_path: str) -> None: | |
| doc = Document() | |
| for section in doc.sections: | |
| section.left_margin = section.right_margin = Cm(2.0) | |
| section.top_margin = section.bottom_margin = Cm(2.0) | |
| _cover(doc, data) | |
| # Executive Summary | |
| _add_heading(doc, "Executive Summary", 1) | |
| _add_text(doc, data.get("executive_summary","")) | |
| # Company & Industry Profile | |
| _add_heading(doc, "Company Profile", 1) | |
| _add_text(doc, data.get("company_overview","")) | |
| _add_heading(doc, "Industry Profile", 1) | |
| _add_text(doc, data.get("industry_profile","")) | |
| _add_heading(doc, "Industry Trends", 1) | |
| _add_text(doc, data.get("industry_trends","")) | |
| _swot_section(doc, data) | |
| _trends_section(doc, data) | |
| # Visualizations Section | |
| try: | |
| from builders.chart_builder import generate_all_charts | |
| charts = generate_all_charts(data) | |
| if any(charts.values()): | |
| _add_heading(doc, "Financial Visualizations", 1) | |
| if charts.get("revenue_net_income"): | |
| _add_heading(doc, "Revenue & Net Income Trend", 2) | |
| doc.add_picture(charts["revenue_net_income"], width=Inches(6.0)) | |
| if charts.get("margins"): | |
| _add_heading(doc, "Profitability Margins", 2) | |
| doc.add_picture(charts["margins"], width=Inches(6.0)) | |
| if charts.get("cash_flow"): | |
| _add_heading(doc, "Cash Flow & CAPEX", 2) | |
| doc.add_picture(charts["cash_flow"], width=Inches(6.0)) | |
| except Exception as e: | |
| print(f"Warning: Could not generate DOCX charts: {e}", file=sys.stderr) | |
| _ratio_section(doc, data) | |
| _dupont_section(doc, data) | |
| _peer_section(doc, data) | |
| if data.get("key_insights"): | |
| _bullets(doc, data["key_insights"], "Key Insights", 1) | |
| if data.get("risks"): | |
| _bullets(doc, data["risks"], "Risks", 1) | |
| if data.get("things_to_watch"): | |
| _bullets(doc, data["things_to_watch"], "Things to Watch", 1) | |
| if data.get("limitations_and_caveats"): | |
| _bullets(doc, data["limitations_and_caveats"], "Limitations & Caveats", 1) | |
| _sources_section(doc, data) | |
| _disclaimer(doc) | |
| Path(output_path).parent.mkdir(parents=True, exist_ok=True) | |
| doc.save(output_path) | |
| if __name__ == "__main__": | |
| import argparse | |
| p = argparse.ArgumentParser() | |
| p.add_argument("--input","-i",required=True) | |
| p.add_argument("--output","-o",required=True) | |
| args = p.parse_args() | |
| with open(args.input, encoding="utf-8") as f: | |
| data = json.load(f) | |
| build_document(data, args.output) | |
| print(f"Wrote: {args.output}") | |