import streamlit as st import pandas as pd import tempfile import os import io import re import time import json import chardet import requests from pathlib import Path from urllib.parse import urlparse, unquote from concurrent.futures import ThreadPoolExecutor, as_completed from collections import Counter import google.genai as genai from google.genai import types as genai_types st.set_page_config(page_title="ESG Analyzer — Gemini", page_icon="🌿", layout="wide") st.markdown(""" """, unsafe_allow_html=True) st.markdown("""

🌿 ESG Indicator Analyzer

""", unsafe_allow_html=True) # ========================================== # SUPPORTED FORMATS # ========================================== SUPPORTED_EXTENSIONS = [ ".pdf", ".docx", ".doc", ".xlsx", ".xls", ".csv", ".tsv", ".pptx", ".ppt", ".txt", ".md", ".html", ".htm", ".rtf", ".odt", ".json", ".xml" ] MIME_TO_EXT = { "application/pdf": ".pdf", "application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx", "application/msword": ".doc", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": ".xlsx", "application/vnd.ms-excel": ".xls", "text/csv": ".csv", "text/plain": ".txt", "text/html": ".html", "application/json": ".json", } # ========================================== # NULL / EMPTY VALUE CLEANER # ========================================== # Phrases that count as "empty" answers _NULL_PHRASES = [ "null", "none", "n/a", "na", "not applicable", "not disclosed", "not available", "not reported", "not provided", "not found", "not mentioned", "not specified", "no information", "no data", "data not available", "information not found", "—", "–", "-", '""', "''", ] _NULL_LINE_RE = re.compile( r'^\s*\*{0,2}[^:\n]{1,120}\*{0,2}\s*:\s*' r'(?:' + '|'.join(re.escape(p) for p in _NULL_PHRASES) + r')' r'\s*$', re.IGNORECASE | re.MULTILINE, ) _JSON_NULL_RE = re.compile( r'^\s*"[^"]+"\s*:\s*(?:null|"null"|""|\'\')\s*,?\s*$', re.IGNORECASE | re.MULTILINE, ) _BRACE_ONLY_RE = re.compile(r'^\s*[\{\}\[\]]\s*$', re.MULTILINE) def clean_answer_nulls(text: str) -> str: """Remove null/N/A/empty lines from LLM output so only real data shows.""" if not text: return text cleaned = _NULL_LINE_RE.sub("", text) cleaned = _JSON_NULL_RE.sub("", cleaned) cleaned = _BRACE_ONLY_RE.sub("", cleaned) # Remove leftover JSON fences cleaned = re.sub(r'```(?:json)?\s*', '', cleaned) cleaned = cleaned.replace('```', '') # Collapse 3+ blank lines → single blank line cleaned = re.sub(r'\n{3,}', '\n\n', cleaned) return cleaned.strip() # ========================================== # ANSWER FORMATTING HELPERS # ========================================== def _render_json_as_kv(data, depth=0): """Recursively convert a JSON dict to HTML key-value rows, skipping null values.""" if isinstance(data, dict): rows = [] for k, v in data.items(): # Skip null / empty values entirely if v is None or v == "" or str(v).lower() in _NULL_PHRASES: continue label = k.replace("_", " ").title() if isinstance(v, (dict, list)): inner = _render_json_as_kv(v, depth + 1) if inner: # only add if inner has content rows.append( f'

{label}' f'{inner}

' ) else: val = str(v) rows.append( f'

{label}' f'{val}

' ) return "".join(rows) elif isinstance(data, list): items = [] for item in data: rendered = _render_json_as_kv(item, depth) if rendered: items.append(f'

{rendered}

') if not items: return "" return f'

{"".join(items)}' else: val = str(data) if val.lower() in _NULL_PHRASES: return "" return f'{val}' def _try_parse_json(text): clean = re.sub(r"```(?:json)?\s*", "", text).strip().rstrip("`").strip() try: return json.loads(clean) except Exception: pass m = re.search(r"\{[\s\S]+\}", clean) if m: try: return json.loads(m.group()) except Exception: pass return None def _parse_kv_text(text): """Parse '- **Key:** value' or '- Key: value' lines into (key, value) pairs, skipping any line whose value is null/N/A/empty.""" pairs = [] lines = text.strip().splitlines() cur_key, cur_val = None, [] def _flush(): if cur_key: val = " ".join(cur_val).strip() if val and val.lower() not in _NULL_PHRASES: pairs.append((cur_key, val)) for line in lines: stripped = line.strip() if not stripped or stripped.startswith("|"): continue m = re.match(r"^\s*[-•*]?\s*\*{0,2}([^:]{3,80}?)\*{0,2}:\s*(.*)", stripped) if m: _flush() cur_key = m.group(1).strip() cur_val = [m.group(2).strip()] if m.group(2).strip() else [] elif cur_key: cur_val.append(stripped) _flush() return pairs def _has_markdown_table(text): lines = text.strip().splitlines() pipe_lines = [l for l in lines if "|" in l] return len(pipe_lines) >= 2 def format_answer_html(answer_text): """Convert raw answer to clean HTML. Returns HTML string, '__MARKDOWN__', or None.""" if not answer_text or is_not_found_str(answer_text): return None text = clean_answer_nulls(answer_text.strip()) if not text: return None # 1. JSON block → key-value HTML (skipping nulls inside) if text.lstrip().startswith("{") or "```json" in text or '```\n{' in text: parsed = _try_parse_json(text) if parsed: inner = _render_json_as_kv(parsed) if inner: return f'

{inner}

' # 2. Markdown table → let Streamlit render natively if _has_markdown_table(text): return "__MARKDOWN__" # 3. Key-value block → HTML rows (nulls already stripped by _parse_kv_text) kv = _parse_kv_text(text) if len(kv) >= 2: rows = [] for k, v in kv: v_html = v.replace("\n", "
") rows.append( f'

{k}' f'{v_html}

' ) return f'

{"".join(rows)}

' # 4. Plain prose return "__MARKDOWN__" # ========================================== # INDICATOR DOC HINTS # ========================================== INDICATOR_DOC_HINTS = { "scope 1": ["Sustainability Report","BRSR","GHG Inventory Report","CDP Climate Disclosure"], "scope 2": ["Sustainability Report","BRSR","GHG Inventory Report","CDP Climate Disclosure"], "scope 3": ["Sustainability Report","CDP Climate Disclosure","Value Chain Emissions Report"], "ghg": ["Sustainability Report","GHG Inventory Report","CDP Climate Disclosure","BRSR"], "carbon": ["Sustainability Report","CDP Climate Disclosure","Net Zero Report","BRSR"], "energy": ["Sustainability Report","BRSR","Annual Report","Energy Audit Report"], "renewable":["Sustainability Report","BRSR","Renewable Energy Certificate (REC) Records"], "water": ["Sustainability Report","BRSR","Water Audit Report","CDP Water Disclosure"], "effluent":["Sustainability Report","BRSR","ETP Records","Pollution Control Board Returns"], "waste": ["Sustainability Report","BRSR","Waste Management Records"], "hazardous":["CPCB/SPCB Hazardous Waste Returns","Sustainability Report","BRSR"], "nox": ["Pollution Control Board Returns","Environmental Compliance Report","BRSR"], "sox": ["Pollution Control Board Returns","Environmental Compliance Report","BRSR"], "biodiversity":["Biodiversity Assessment Report","Sustainability Report","EIA Report"], "employee":["BRSR","Annual Report","HR Policy Document","Sustainability Report"], "worker": ["BRSR","Annual Report","HR Policy Document","Sustainability Report"], "diversity":["BRSR","Sustainability Report","Annual Report","DEI Report"], "health and safety":["BRSR","Sustainability Report","OHS Management Report"], "training":["BRSR","Sustainability Report","Learning & Development Report"], "governance":["Annual Report","Corporate Governance Report","BRSR"], "supply chain":["Supplier Sustainability Report","Annual Report","BRSR"], "community":["CSR Report","Annual Report","BRSR","Social Impact Assessment Report"], "climate": ["CDP Climate Disclosure","Climate Transition Plan","Sustainability Report"], "tcfd": ["TCFD Report","CDP Climate Disclosure","Sustainability Report"], "gri": ["GRI Index","Sustainability Report"], "risk": ["Annual Report","Risk Management Report","TCFD Report"], } CATEGORY_DOC_FALLBACK = { "General & Organizational Profile": ["Annual Report","BRSR"], "Sustainability Management & Reporting": ["Sustainability Report","BRSR","GRI Index"], "Governance & Ethics": ["Annual Report","Corporate Governance Report","BRSR"], "Risk & Opportunity Management": ["Annual Report","TCFD Report","Risk Management Report"], "GHG Emissions & Climate Change": ["Sustainability Report","GHG Inventory Report","CDP Climate Disclosure"], "Energy": ["Sustainability Report","BRSR","Energy Audit Report"], "Water & Effluents": ["Sustainability Report","BRSR","CDP Water Disclosure"], "Waste & Materials": ["Sustainability Report","BRSR","Waste Management Records"], "Air Quality": ["Pollution Control Board Returns","Environmental Compliance Report","BRSR"], "Biodiversity & Land Use": ["Biodiversity Assessment Report","EIA Report","Sustainability Report"], "Labor & Human Rights": ["BRSR","Annual Report","HR Policy Document"], "Occupational Health & Safety": ["BRSR","OHS Report","ISO 45001 Certificate"], "Diversity, Equity & Inclusion": ["BRSR","DEI Report","Sustainability Report"], "Training & Skill Development": ["BRSR","L&D Report","Sustainability Report"], "Community & Social Impact": ["CSR Report","Annual Report","BRSR"], "Customer & Product Responsibility": ["Annual Report","BRSR","Product Stewardship Report"], "Economic Performance": ["Annual Report","Financial Statements","TCFD Report"], "Legal & Environmental Compliance": ["Annual Report","BRSR","Compliance Report"], } def recommend_document_for_indicator(indicator_name, indicator_id, category=""): text = (indicator_name + " " + indicator_id).lower() scores = {} for keyword, docs in INDICATOR_DOC_HINTS.items(): if keyword in text: for rank, doc in enumerate(docs): scores[doc] = scores.get(doc, 0) + (len(docs) - rank + 1) cat_docs = CATEGORY_DOC_FALLBACK.get(category, ["Annual Report","BRSR","Sustainability Report"]) for rank, doc in enumerate(cat_docs): scores[doc] = scores.get(doc, 0) + max(0, 2 - rank) if not scores: return cat_docs[:3] return [d for d, _ in sorted(scores.items(), key=lambda x: -x[1])[:4]] # ========================================== # UNIT EXTRACTION # ========================================== UNIT_PATTERNS = [ r'\b(\d[\d,\.]*)\s*(tCO2e|tco2e|tonnes?\s*CO2e?|MT\s*CO2e?)', r'\b(\d[\d,\.]*)\s*(MWh|kWh|GWh|TJ|GJ|MJ)', r'\b(\d[\d,\.]*)\s*(KL|ML|GL|m3|litres?|kilolitres?)', r'\b(\d[\d,\.]*)\s*(MT|tonnes?|tons?|kg)', r'(?:INR|Rs\.?|₹|USD|\$)\s*(\d[\d,\.]*)\s*(crore|lakh|million|billion)?', r'\b(\d[\d,\.]*)\s*(crore|lakh|million|billion)', r'\b(\d[\d,\.]*)\s*(%|percent)', r'\b(\d[\d,\.]*)\s*(employees?|workers?|sites?)', ] def extract_value_and_unit(answer): if not answer or is_not_found_str(answer): return "", "" for pattern in UNIT_PATTERNS: match = re.search(pattern, answer, re.IGNORECASE) if match: groups = [g for g in match.groups() if g] if len(groups) >= 2: return groups[0].replace(",", ""), groups[1] num_match = re.search(r'\b(\d[\d,\.]+)\b', answer) return (num_match.group(1), "") if num_match else ("", "") def is_not_found_str(answer): if not answer: return True a = answer.strip().lower() phrases = [ "information not found", "not found", "not available", "not disclosed", "not reported", "not provided", "no information", "data not available", "not mentioned", "not specified", "n/a", ] if any(a == p for p in phrases): return True if len(a) < 80 and any(p in a for p in phrases): return True return False def is_not_found(answer): return is_not_found_str(answer) # ========================================== # DOCUMENT PARSERS # ========================================== def extract_text_from_pdf(file_bytes): try: import pdfplumber with pdfplumber.open(io.BytesIO(file_bytes)) as pdf: pages = [page.extract_text() for page in pdf.pages if page.extract_text()] if pages: return "\n\n".join(pages) except Exception: pass try: import PyPDF2 reader = PyPDF2.PdfReader(io.BytesIO(file_bytes)) return "\n\n".join(p.extract_text() or "" for p in reader.pages) except Exception as e: return f"[PDF extraction error: {e}]" def extract_text_from_docx(file_bytes): try: from docx import Document doc = Document(io.BytesIO(file_bytes)) parts = [p.text for p in doc.paragraphs if p.text.strip()] for table in doc.tables: for row in table.rows: row_text = " | ".join(c.text.strip() for c in row.cells if c.text.strip()) if row_text: parts.append(row_text) return "\n\n".join(parts) except Exception as e: return f"[DOCX extraction error: {e}]" def extract_text_from_xlsx(file_bytes): try: xl = pd.ExcelFile(io.BytesIO(file_bytes)) sheets = [] for name in xl.sheet_names: df = xl.parse(name) sheets.append(f"=== Sheet: {name} ===\n{df.to_string(index=False)}") return "\n\n".join(sheets) except Exception as e: return f"[XLSX extraction error: {e}]" def extract_text_from_csv(file_bytes): try: enc = chardet.detect(file_bytes).get("encoding","utf-8") or "utf-8" df = pd.read_csv(io.BytesIO(file_bytes), encoding=enc) return df.to_string(index=False) except Exception as e: return f"[CSV extraction error: {e}]" def extract_text_from_pptx(file_bytes): try: from pptx import Presentation prs = Presentation(io.BytesIO(file_bytes)) slides = [] for i, slide in enumerate(prs.slides, 1): texts = [s.text.strip() for s in slide.shapes if hasattr(s,"text") and s.text.strip()] if texts: slides.append(f"--- Slide {i} ---\n" + "\n".join(texts)) return "\n\n".join(slides) except Exception as e: return f"[PPTX extraction error: {e}]" def extract_text_from_txt(file_bytes): enc = chardet.detect(file_bytes).get("encoding","utf-8") or "utf-8" return file_bytes.decode(enc, errors="replace") def extract_text_from_html(file_bytes): try: from bs4 import BeautifulSoup soup = BeautifulSoup(file_bytes, "html.parser") for tag in soup(["script","style","meta","link"]): tag.decompose() return soup.get_text(separator="\n", strip=True) except Exception: return extract_text_from_txt(file_bytes) def extract_text_from_rtf(file_bytes): try: from striprtf.striprtf import rtf_to_text return rtf_to_text(file_bytes.decode("latin-1", errors="replace")) except Exception as e: return f"[RTF extraction error: {e}]" def extract_text_from_odt(file_bytes): try: from odf.opendocument import load from odf.text import P doc = load(io.BytesIO(file_bytes)) paragraphs = [] for p in doc.getElementsByType(P): t = "".join(n.data for n in p.childNodes if n.nodeType == n.TEXT_NODE) if t.strip(): paragraphs.append(t.strip()) return "\n\n".join(paragraphs) except Exception as e: return f"[ODT extraction error: {e}]" def extract_text_from_json(file_bytes): try: enc = chardet.detect(file_bytes).get("encoding","utf-8") or "utf-8" data = json.loads(file_bytes.decode(enc)) return json.dumps(data, indent=2, ensure_ascii=False) except Exception as e: return f"[JSON extraction error: {e}]" def extract_text_from_xml(file_bytes): try: from bs4 import BeautifulSoup return BeautifulSoup(file_bytes, "xml").get_text(separator="\n", strip=True) except Exception: return extract_text_from_txt(file_bytes) EXT_PARSER_MAP = { ".pdf": extract_text_from_pdf, ".docx": extract_text_from_docx, ".doc": extract_text_from_docx, ".xlsx": extract_text_from_xlsx, ".xls": extract_text_from_xlsx, ".csv": extract_text_from_csv, ".tsv": extract_text_from_csv, ".pptx": extract_text_from_pptx, ".ppt": extract_text_from_pptx, ".txt": extract_text_from_txt, ".md": extract_text_from_txt, ".html": extract_text_from_html, ".htm": extract_text_from_html, ".rtf": extract_text_from_rtf, ".odt": extract_text_from_odt, ".json": extract_text_from_json, ".xml": extract_text_from_xml, } def detect_and_extract(file_bytes, filename): ext = Path(filename.lower()).suffix parser = EXT_PARSER_MAP.get(ext) return parser(file_bytes) if parser else extract_text_from_txt(file_bytes) # ========================================== # GEMINI CLIENT HELPERS # ========================================== def _get_client(): return genai.Client(api_key=st.session_state.get("_api_key","")) def _make_client(api_key): return genai.Client(api_key=api_key) def _poll_until_ready(client, gfile, display_name, timeout_seconds=300): deadline = time.time() + timeout_seconds while gfile.state and gfile.state.name == "PROCESSING": if time.time() > deadline: raise TimeoutError(f"Gemini file '{display_name}' still PROCESSING after {timeout_seconds}s.") time.sleep(3) gfile = client.files.get(name=gfile.name) if gfile.state and gfile.state.name == "FAILED": raise RuntimeError(f"Gemini rejected '{display_name}' (state=FAILED).") return gfile def _upload_with_retry(upload_fn, max_retries=4): wait = 3 last_err = None for attempt in range(max_retries): try: return upload_fn() except Exception as e: last_err = e err_str = str(e).lower() retryable = any(x in err_str for x in [ "503","service unavailable","500","internal server error", "429","quota","rate","resource exhausted", "timed out","timeout","read timed out","deadline exceeded", ]) if retryable and attempt < max_retries - 1: st.toast(f"Upload error (attempt {attempt+1}/{max_retries}) — retrying in {wait}s…", icon="🔄") time.sleep(wait) wait = min(wait * 2, 30) else: raise raise last_err def upload_text_as_gemini_file(client, text, display_name): with tempfile.NamedTemporaryFile(suffix=".txt", delete=False, mode="w", encoding="utf-8") as tmp: tmp.write(text) tmp_path = tmp.name try: def _do(): gfile = client.files.upload( file=tmp_path, config=genai_types.UploadFileConfig(mime_type="text/plain", display_name=display_name), ) return _poll_until_ready(client, gfile, display_name, timeout_seconds=120) return _upload_with_retry(_do, max_retries=4) finally: try: os.unlink(tmp_path) except OSError: pass def upload_pdf_as_gemini_file(client, file_bytes, display_name): with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp: tmp.write(file_bytes) tmp_path = tmp.name try: def _do(): gfile = client.files.upload( file=tmp_path, config=genai_types.UploadFileConfig(mime_type="application/pdf", display_name=display_name), ) return _poll_until_ready(client, gfile, display_name, timeout_seconds=300) return _upload_with_retry(_do, max_retries=4) finally: try: os.unlink(tmp_path) except OSError: pass def upload_file_smart(client, file_bytes, filename): ext = Path(filename.lower()).suffix if ext == ".pdf": return upload_pdf_as_gemini_file(client, file_bytes, filename), "PDF native" else: text = detect_and_extract(file_bytes, filename) return upload_text_as_gemini_file(client, text, filename), f"{ext.upper()} extracted" # ========================================== # AWS S3 / URL HELPERS # ========================================== def is_s3_url(url): if url.startswith("s3://"): return True host = urlparse(url).netloc.lower() return "s3.amazonaws.com" in host or bool(re.match(r".+\.s3[.-]", host)) def parse_s3_url(url): if url.startswith("s3://"): parts = url[5:].split("/", 1) return parts[0], (parts[1] if len(parts) > 1 else "") parsed = urlparse(url) host = parsed.netloc.lower() path = parsed.path.lstrip("/") if re.match(r"(.+)\.s3[.-]", host): return host.split(".")[0], unquote(path) parts = path.split("/", 1) return parts[0], unquote(parts[1]) if len(parts) > 1 else "" def fetch_from_s3(url, aws_key, aws_secret, aws_region): raw_path = unquote(urlparse(url).path if not url.startswith("s3://") else "/" + url[5:].split("/",1)[-1]) filename = raw_path.split("/")[-1] or "s3_document" if url.startswith(("https://","http://")): try: resp = requests.get(url, timeout=60) if resp.status_code == 200: return resp.content, filename except requests.exceptions.RequestException: pass bucket, key = parse_s3_url(url) filename = unquote(key.split("/")[-1]) or "s3_document" import boto3 s3 = boto3.client("s3", aws_access_key_id=aws_key or None, aws_secret_access_key=aws_secret or None, region_name=aws_region or "us-east-1") buf = io.BytesIO() s3.download_fileobj(bucket, key, buf) return buf.getvalue(), filename def fetch_from_public_url(url): resp = requests.get(url, timeout=60) resp.raise_for_status() path = unquote(urlparse(url).path) filename = path.split("/")[-1] or "document" if "." not in filename: ct = resp.headers.get("Content-Type","").split(";")[0].strip() ext = MIME_TO_EXT.get(ct,"") if ext: filename += ext return resp.content, filename # ========================================== # SIDEBAR # ========================================== with st.sidebar: st.markdown("### ⚙️ Configuration") api_key = os.environ.get("GEMINI_API_KEY","") if not api_key: api_key = st.text_input("🔑 Gemini API Key", type="password", placeholder="AIza…", help="aistudio.google.com/apikey") else: st.success("✅ API key loaded from secrets") st.markdown("---") selected_model = st.selectbox( "🤖 Gemini Model", options=["gemini-2.5-flash-lite","gemini-2.5-flash","gemini-2.5-pro", "gemini-2.0-flash","gemini-2.0-flash-lite"], index=0, ) st.markdown("---") parallel_workers = st.slider("⚡ Parallel requests", min_value=1, max_value=15, value=5, step=1) with st.expander("ℹ️ Features", expanded=False): st.markdown(""" - 🌿 Clean formatted output — no raw JSON, no null fields - 📊 Key-value cards, tables, prose — only real data shown - 🚀 Parallel batch extraction - 🔁 Iterative gap fill - ➕ Add new documents after results - 📌 Smart document recommendations """) if not api_key: st.info("⬅️ Enter your Gemini API key in the sidebar to begin.") st.stop() st.session_state["_api_key"] = api_key try: _test_client = genai.Client(api_key=api_key) list(_test_client.models.list()) with st.sidebar: st.success("✅ Gemini API connected") except Exception as e: st.error(f"❌ API key error: {e}") st.stop() with st.sidebar: st.markdown("---") with st.expander("🪣 AWS S3 Credentials", expanded=False): aws_key = os.environ.get("AWS_ACCESS_KEY_ID","") aws_secret = os.environ.get("AWS_SECRET_ACCESS_KEY","") aws_region = os.environ.get("AWS_REGION","us-east-1") if not aws_key: aws_key = st.text_input("AWS Access Key ID", type="password", key="aws_key") else: st.success("✅ AWS key loaded") if not aws_secret: aws_secret = st.text_input("AWS Secret Access Key", type="password", key="aws_secret") if not aws_region: aws_region = st.text_input("AWS Region", value="us-east-1", key="aws_region") # ========================================== # SESSION STATE # ========================================== for _key, _default in [ ("uploaded_files", {}), ("text_files", {}), ("current_filenames", []), ("s3_docs", {}), ("url_docs", {}), ("batch_results", {}), ("extra_docs", {}), ]: if _key not in st.session_state: st.session_state[_key] = _default # ========================================== # BUILD DOC LOOKUP # ========================================== def build_doc_lookup(): per_doc = {} for name, gf in st.session_state.uploaded_files.items(): per_doc[name] = gf for name, gf in st.session_state.text_files.items(): per_doc[name] = gf for doc in st.session_state.s3_docs.values(): if doc.get("gfile"): per_doc[doc["name"]] = doc["gfile"] for doc in st.session_state.url_docs.values(): if doc.get("gfile"): per_doc[doc["name"]] = doc["gfile"] for name, gf in st.session_state.extra_docs.items(): per_doc[name] = gf return per_doc # ========================================== # STEP 1 — UPLOAD DOCUMENTS # ========================================== st.markdown("## 📂 Step 1 — Upload Documents") st.info( "💡 Upload all available ESG documents for best results: " "Annual Report, BRSR, Sustainability Report, CDP Disclosures, GHG Inventory, " "OHS Reports, EPR Certificates, ISO Certificates, CSR Reports." ) tab_upload, tab_s3, tab_url = st.tabs(["📁 Upload Files", "🪣 AWS S3", "🔗 Public URL"]) with tab_upload: st.caption(f"Supported: {', '.join(SUPPORTED_EXTENSIONS)}") uploaded_files = st.file_uploader( "Upload documents", type=[e.lstrip(".") for e in SUPPORTED_EXTENSIONS], accept_multiple_files=True, key="file_uploader", ) if uploaded_files: new_names = [f.name for f in uploaded_files] if new_names != st.session_state.current_filenames: _del_client = _get_client() for gf in list(st.session_state.uploaded_files.values()) + list(st.session_state.text_files.values()): try: _del_client.files.delete(name=gf.name) except Exception: pass st.session_state.uploaded_files = {} st.session_state.text_files = {} st.session_state.current_filenames = new_names st.session_state.batch_results = {} _up_client = _get_client() for uf in uploaded_files: if uf.name in st.session_state.uploaded_files or uf.name in st.session_state.text_files: continue ext = Path(uf.name.lower()).suffix file_bytes = uf.getvalue() size_kb = len(file_bytes) // 1024 with st.spinner(f"Uploading {uf.name} ({size_kb} KB)…"): try: gfile, method = upload_file_smart(_up_client, file_bytes, uf.name) if ext == ".pdf": st.session_state.uploaded_files[uf.name] = gfile else: st.session_state.text_files[uf.name] = gfile st.success(f"✅ {uf.name} — {method} ({size_kb} KB)") except Exception as e: st.error(f"❌ Error processing `{uf.name}`: {str(e)[:300]}") with tab_s3: st.markdown("Add documents from your AWS S3 bucket.") s3_url_input = st.text_input("S3 URL", placeholder="https://my-bucket.s3.amazonaws.com/report.pdf", key="s3_url_input") col_add, col_clear = st.columns(2) with col_add: add_s3 = st.button("➕ Add S3 Document", type="primary", key="add_s3_btn") with col_clear: if st.button("🗑️ Clear S3 Docs", key="clear_s3_btn"): _del_client = _get_client() for doc in st.session_state.s3_docs.values(): if doc.get("gfile"): try: _del_client.files.delete(name=doc["gfile"].name) except Exception: pass st.session_state.s3_docs = {} st.success("Cleared.") if add_s3 and s3_url_input.strip(): url = s3_url_input.strip() if url not in st.session_state.s3_docs: with st.spinner("Fetching from S3…"): try: if not is_s3_url(url): st.error("Not a valid S3 URL.") else: file_bytes, filename = fetch_from_s3(url, aws_key, aws_secret, aws_region) gfile, method = upload_file_smart(_get_client(), file_bytes, filename) st.session_state.s3_docs[url] = {"name": filename, "gfile": gfile, "size_kb": len(file_bytes)//1024} st.success(f"✅ {filename} loaded ({method})") except Exception as e: st.error(f"❌ {str(e)[:300]}") if st.session_state.s3_docs: for url, doc in st.session_state.s3_docs.items(): st.caption(f"• {doc['name']} ({doc['size_kb']} KB)") with tab_url: st.markdown("Paste any public HTTPS URL to a document.") pub_url_input = st.text_input("Public URL", placeholder="https://example.com/annual-report.pdf", key="pub_url_input") col_add2, col_clear2 = st.columns(2) with col_add2: add_pub = st.button("➕ Add URL Document", type="primary", key="add_pub_btn") with col_clear2: if st.button("🗑️ Clear URL Docs", key="clear_pub_btn"): _del_client = _get_client() for doc in st.session_state.url_docs.values(): if doc.get("gfile"): try: _del_client.files.delete(name=doc["gfile"].name) except Exception: pass st.session_state.url_docs = {} st.success("Cleared.") if add_pub and pub_url_input.strip(): url = pub_url_input.strip() if url not in st.session_state.url_docs: with st.spinner("Downloading…"): try: file_bytes, filename = fetch_from_public_url(url) gfile, method = upload_file_smart(_get_client(), file_bytes, filename) st.session_state.url_docs[url] = {"name": filename, "gfile": gfile, "size_kb": len(file_bytes)//1024} st.success(f"✅ {filename} loaded ({method})") except Exception as e: st.error(f"❌ {str(e)[:300]}") if st.session_state.url_docs: for url, doc in st.session_state.url_docs.items(): st.caption(f"• {doc['name']} ({doc['size_kb']} KB)") # ========================================== # DOC LOOKUP # ========================================== per_doc_files = build_doc_lookup() all_gemini_files = list(per_doc_files.values()) all_doc_names = list(per_doc_files.keys()) if not all_gemini_files: st.info("⬆️ Add at least one document above to start analyzing.") st.stop() with st.sidebar: st.markdown("---") st.markdown(f"**📚 {len(all_doc_names)} document(s) loaded**") for dn in all_doc_names: st.caption(f"• {dn}") # ========================================== # SYSTEM PROMPT # ========================================== SYSTEM_PROMPT = """You are an expert ESG data analyst extracting ESG indicator data from corporate documents. OUTPUT FORMAT RULES — follow strictly: - For structured data with multiple fields: use a **Key:** Value list, one field per line, with bold keys. - For tabular data: use a well-formatted markdown table with | separators and a header separator row. - For narrative/qualitative answers: write clear, readable prose paragraphs. - NEVER output raw JSON, code blocks, JSON fences, or triple backticks. ⚠️ CRITICAL — NULL / EMPTY FIELD RULE: - If a field's value is unknown, not found, not disclosed, null, N/A, or not applicable — OMIT that field entirely. - Do NOT write "null", "N/A", "Not disclosed", "Not available", "Not reported", "—", or any placeholder. - Only include a field in your response if you found real, concrete data for it in the documents. - An answer with 3 real fields is better than an answer with 10 fields where 7 are empty/null. SOURCE CITATION: - Always end each answer with: (Source: Document Name, Section/Page) EXTRACTION RULES: 1. Search ALL documents — main sections, appendices, tables, footnotes, data annexures. 2. Use synonyms and related terms if exact labels are missing. 3. Include PARTIAL data with a clear note on what specifically is missing. 4. Provide COMPLETE, untruncated answers — never cut off data mid-sentence. 5. Output "Information not found" ONLY if genuinely absent from every document. 6. NEVER hallucinate or infer data not present in the documents.""" def clean_placeholder(text): text = re.sub(r'Context:\s*"""\s*\{\{REPORT_CONTEXT\}\}\s*"""\s*', '', text, flags=re.DOTALL) text = re.sub(r'\{\{REPORT_CONTEXT\}\}', '', text) return text.strip() def _rewrite_prompt_for_clean_output(prompt: str) -> str: """ Fully strip JSON / null instructions from indicator prompts. Replace with clean readable format instructions. """ CLEAN_KV_INSTRUCTION = ( "Present ONLY fields where real data exists, as a clean **Key:** Value list.\n" "OMIT any field that is null, unknown, not found, or not applicable — " "do NOT write null, N/A, or any placeholder for missing fields." ) # ── Remove entire "Output as a strict JSON object { ... }" blocks ── prompt = re.sub( r'Output as a strict JSON object\s*\{[\s\S]*?\}', CLEAN_KV_INSTRUCTION, prompt, flags=re.DOTALL, ) # ── Remove "Output as a strict JSON object" without a following block ── prompt = re.sub( r'Output as a strict JSON object[^\n]*', CLEAN_KV_INSTRUCTION, prompt, ) # ── Remove bare JSON field definition lines: "field": "type_or_null" ── prompt = re.sub( r'^\s*"[^"]+"\s*:\s*"[^"]*(?:null|or null|numeric_value)[^"]*"\s*,?\s*$', '', prompt, flags=re.MULTILINE, ) # ── Remove nested JSON blocks like "gas_breakdown_tCO2e": { ... } ── prompt = re.sub( r'"[^"]+"\s*:\s*\{[\s\S]*?\}', '', prompt, flags=re.DOTALL, ) # ── Replace "Output as a key-value list" variants ── for old in [ "Output as a key-value list:", "Output as a key-value list.", "Output as a key-value list", ]: prompt = prompt.replace(old, CLEAN_KV_INSTRUCTION) # ── Remove leftover lone braces / brackets on their own lines ── prompt = re.sub(r'^\s*[\{\}\[\]]\s*$', '', prompt, flags=re.MULTILINE) # ── Remove JSON code fences ── prompt = re.sub(r'```(?:json)?\s*', '', prompt) prompt = prompt.replace('```', '') # ── Collapse excess blank lines ── prompt = re.sub(r'\n{3,}', '\n\n', prompt) return prompt.strip() def generate_answer(question, gemini_files, model_name="gemini-2.5-flash-lite", raw_prompt=False, api_key=""): if raw_prompt: prompt = _rewrite_prompt_for_clean_output(clean_placeholder(question)) else: prompt = ( f"QUESTION: {question}\n\n" "Search all documents thoroughly and provide a complete, well-formatted answer.\n" "IMPORTANT: Only include fields where real data exists — omit null/missing fields entirely.\n" "Do NOT write 'null', 'N/A', 'Not disclosed', or any empty placeholder." ) contents = [] for gf in gemini_files: contents.append(genai_types.Part(file_data=genai_types.FileData(file_uri=gf.uri, mime_type=gf.mime_type))) contents.append(prompt) client = _make_client(api_key) config = genai_types.GenerateContentConfig( system_instruction=SYSTEM_PROMPT, temperature=0.0, max_output_tokens=8192, ) wait = 2 for attempt in range(7): try: resp = client.models.generate_content( model=model_name, contents=contents, config=config, ) # Post-process: strip any null values the model still emitted return clean_answer_nulls(resp.text.strip()) except Exception as e: err = str(e) if any(x in err.lower() for x in [ "429","quota","rate","resource exhausted","503","unavailable", "deadline exceeded","500","internal","timed out","timeout" ]) and attempt < 6: time.sleep(wait) wait = min(wait * 2, 60) else: return f"Error: {err[:200]}" return "Error: max retries exceeded." # ========================================== # LOAD INDICATORS # ========================================== @st.cache_resource(show_spinner=False) def load_indicators(): search_paths = ["questions.txt", "/mnt/user-data/uploads/questions.txt"] content = None for p in search_paths: if os.path.exists(p): with open(p, "r", encoding="utf-8") as f: content = f.read() break if not content: return [] indicator_re = re.compile(r'^\s*(\d+)\.\s+(IMP-[A-Z0-9]+-[A-Z0-9]+)\s+\|\s+(.+)$', re.MULTILINE) module_re = re.compile(r'^Module\s+\d+:\s+(.+?)\s+$(IMP-M\d+)$', re.MULTILINE) module_map = {m.group(2): m.group(1).strip() for m in module_re.finditer(content)} matches = list(indicator_re.finditer(content)) indicators = [] for idx, match in enumerate(matches): imp_id = match.group(2) name = match.group(3).strip() block_start = match.end() block_end = matches[idx+1].start() if idx+1 < len(matches) else len(content) block = content[block_start:block_end] module_code = "-".join(imp_id.split("-")[:2]) category = module_map.get(module_code, module_code) task_match = re.search(r'Task:\s*(.+?)(?=\n\s*\n\s*$|\Z)', block, re.DOTALL) if task_match: task_text = task_match.group(1).strip() else: dir_match = re.search(r'LLM Extraction Directive:\s*(.+)', block) task_text = dir_match.group(1).strip() if dir_match else f"Extract all available data for {name}." task_text = clean_placeholder(task_text) task_text = _rewrite_prompt_for_clean_output(task_text) full_prompt = ( f"Indicator: {imp_id} — {name}\n" f"Category: {category}\n\n" f"{task_text}\n\n" f"EXTRACTION RULES:\n" f"1. Search ALL documents — main sections, appendices, tables, footnotes.\n" f"2. Use synonyms if exact terms are missing.\n" f"3. Include PARTIAL data with a note on what is missing.\n" f"4. Cite the source section or page number.\n" f"5. Provide COMPLETE, UNTRUNCATED answers.\n" f"6. Output 'Information not found' ONLY if genuinely absent from every document.\n" f"7. Do NOT hallucinate data.\n" f"8. OMIT any field that is null, unknown, or not found — do not write null or N/A." ) indicators.append({ "id": imp_id, "seq": match.group(1), "category": category, "indicator": name, "question": full_prompt }) return indicators indicators = load_indicators() # ========================================== # SINGLE QUESTION # ========================================== st.markdown("---") st.markdown("## 💬 Ask a Single Question") question_input = st.text_input( "Ask anything about the document(s):", placeholder="e.g. What are the total Scope 1 GHG emissions?" ) if st.button("🔍 Get Answer", type="primary") and question_input: with st.spinner("Gemini is reading the documents…"): answer = generate_answer( question_input, all_gemini_files, model_name=selected_model, api_key=st.session_state.get("_api_key",""), ) st.markdown("### 🎯 Answer") if is_not_found(answer): st.warning("Information not found in the uploaded documents.") else: fmt = format_answer_html(answer) if fmt and fmt != "__MARKDOWN__": st.markdown(fmt, unsafe_allow_html=True) else: st.markdown(answer) # ========================================== # BATCH ANALYSIS # ========================================== st.markdown("---") st.markdown("## 📋 Step 2 — Batch Analysis") if not indicators: st.warning("⚠️ `questions.txt` not found. Place it in the app root directory.") st.stop() all_categories = sorted(set(ind["category"] for ind in indicators)) selected_categories = st.multiselect("Filter by category (leave empty = run all):", options=all_categories) filtered = [ind for ind in indicators if ind["category"] in selected_categories] if selected_categories else indicators st.write(f"**{len(filtered)} indicators** | **{len(all_doc_names)} document(s) loaded**") col1, col2 = st.columns([3, 1]) with col1: est = max(1, len(filtered) / parallel_workers) * 3 st.caption(f"Estimated: ~{int(est//60)}m {int(est%60)}s at {parallel_workers} workers") with col2: run_batch = st.button("🚀 Run Batch", type="primary", use_container_width=True) if run_batch: st.session_state.batch_results = {} progress_bar = st.progress(0) status_text = st.empty() completed = 0 total = len(filtered) _batch_api_key = st.session_state.get("_api_key","") with ThreadPoolExecutor(max_workers=parallel_workers) as executor: future_to_ind = { executor.submit( generate_answer, ind["question"], all_gemini_files, selected_model, True, _batch_api_key, ): ind for ind in filtered } for future in as_completed(future_to_ind): ind = future_to_ind[future] completed += 1 progress_bar.progress(completed / total) status_text.write(f"✅ {completed}/{total} — **{ind['id']}**: {ind['indicator'][:55]}…") try: answer_text = future.result() except Exception as e: answer_text = f"Error: {str(e)[:100]}" value_str, unit_str = extract_value_and_unit(answer_text) rec_docs = ( recommend_document_for_indicator(ind["indicator"], ind["id"], ind["category"]) if is_not_found(answer_text) else [] ) st.session_state.batch_results[ind["id"]] = { **ind, "answer": answer_text, "value": value_str, "unit": unit_str, "source": "All documents", "filled_from": "", "recommended_docs": rec_docs, } progress_bar.progress(1.0) status_text.write(f"🎉 Batch complete! {total} indicators processed.") # ========================================== # RESULTS + GAP TOOLS # ========================================== if st.session_state.batch_results: results_list = [st.session_state.batch_results[ind["id"]] for ind in filtered if ind["id"] in st.session_state.batch_results] not_found_ids = [r["id"] for r in results_list if is_not_found(r["answer"])] found_count = len(results_list) - len(not_found_ids) not_found_count = len(not_found_ids) st.markdown("---") m1, m2, m3 = st.columns(3) m1.metric("✅ Found", found_count) m2.metric("❌ Not Found", not_found_count) m3.metric("📄 Documents", len(all_doc_names)) if results_list: pct = found_count / len(results_list) st.progress(pct, text=f"{found_count}/{len(results_list)} indicators filled ({pct*100:.1f}%)") # ── NOT FOUND + RECOMMENDATIONS ────────────────────────────── if not_found_count > 0: st.markdown("---") st.markdown(f"### ❌ {not_found_count} Indicators — Information Not Found") not_found_rows = [r for r in results_list if is_not_found(r["answer"])] rec_records = [] for r in not_found_rows: docs = r.get("recommended_docs", []) rec_records.append({ "ID": r["id"], "Category": r["category"], "Indicator": r["indicator"], "📌 Recommended Documents": " → ".join(docs) if docs else "Annual Report / BRSR", }) st.dataframe(pd.DataFrame(rec_records), use_container_width=True, height=min(480, 45 + len(rec_records) * 38)) st.markdown("**📊 Document types most needed to fill gaps:**") doc_counter = Counter() for r in not_found_rows: for doc in r.get("recommended_docs", []): doc_counter[doc] += 1 top_docs = doc_counter.most_common(8) if top_docs: cols = st.columns(min(len(top_docs), 4)) for i, (doc, cnt) in enumerate(top_docs): cols[i % 4].metric(doc.strip(), f"{cnt} indicators") with st.expander("📋 Full document priority list", expanded=False): for doc, cnt in doc_counter.most_common(): needing = [r["id"] for r in not_found_rows if doc in r.get("recommended_docs", [])] st.markdown(f"**{doc}** ({cnt} indicators): {', '.join(needing)}") # ── UPLOAD NEW DOCUMENT ── st.markdown("---") st.markdown("### ➕ Upload a New Document to Fill These Gaps") st.caption("Existing results are preserved — only missing indicators are re-queried.") new_doc_file = st.file_uploader( "Upload new document", type=[e.lstrip(".") for e in SUPPORTED_EXTENSIONS], accept_multiple_files=False, key="gap_fill_uploader", ) col_up, col_gf = st.columns([2, 1]) with col_up: if new_doc_file: st.caption(f"Selected: **{new_doc_file.name}** ({len(new_doc_file.getvalue())//1024} KB)") with col_gf: do_upload_and_fill = st.button( "⬆️ Upload & Fill Gaps", type="primary", use_container_width=True, disabled=(new_doc_file is None), ) if do_upload_and_fill and new_doc_file: fname = new_doc_file.name _gf_client = _get_client() if fname not in st.session_state.extra_docs: with st.spinner(f"Uploading {fname} to Gemini…"): try: gfile, method = upload_file_smart(_gf_client, new_doc_file.getvalue(), fname) st.session_state.extra_docs[fname] = gfile st.success(f"✅ {fname} uploaded ({method})") except Exception as e: st.error(f"❌ Upload failed: {str(e)[:300]}") st.stop() else: gfile = st.session_state.extra_docs[fname] st.info(f"ℹ️ {fname} already uploaded — re-using.") gap_indicators = [ind for ind in filtered if ind["id"] in not_found_ids] st.markdown(f"**🔁 Gap fill: {len(gap_indicators)} missing indicators using `{fname}`…**") gf_progress = st.progress(0) gf_status = st.empty() gf_completed = 0 gf_filled = 0 _gapfill_api_key = st.session_state.get("_api_key","") with ThreadPoolExecutor(max_workers=parallel_workers) as executor: future_to_ind = { executor.submit(generate_answer, ind["question"], [gfile], selected_model, True, _gapfill_api_key): ind for ind in gap_indicators } for future in as_completed(future_to_ind): ind = future_to_ind[future] gf_completed += 1 gf_progress.progress(gf_completed / len(gap_indicators)) try: new_answer = future.result() except Exception as e: new_answer = f"Error: {str(e)[:100]}" if not is_not_found(new_answer): value_str, unit_str = extract_value_and_unit(new_answer) st.session_state.batch_results[ind["id"]].update({ "answer": new_answer, "value": value_str, "unit": unit_str, "filled_from": fname, "recommended_docs": [], }) gf_filled += 1 gf_status.write(f"✅ {gf_completed}/{len(gap_indicators)} — **{ind['id']}** filled from `{fname}`") else: gf_status.write(f"⬜ {gf_completed}/{len(gap_indicators)} — **{ind['id']}** still not found") gf_progress.progress(1.0) if gf_filled > 0: st.success(f"🎉 Filled **{gf_filled}** new indicators from `{fname}`!") still_missing = len(gap_indicators) - gf_filled if still_missing > 0: st.warning(f"⚠️ **{still_missing}** still not found. See recommendations above.") st.rerun() # ── AUTO GAP FILL ────────────────────────────────────────────── if not_found_count > 0 and len(all_doc_names) > 1: st.markdown("---") st.markdown("### 🔁 Auto Gap Fill — Across All Loaded Documents") st.markdown( f"Tries all **{len(all_doc_names)}** loaded documents one by one " f"to fill the **{not_found_count}** missing indicators." ) col_gf1, col_gf2 = st.columns([3, 1]) with col_gf1: gf_est = max(1, not_found_count / parallel_workers) * 3 * len(all_doc_names) st.caption(f"Estimated: ~{int(gf_est//60)}m {int(gf_est%60)}s worst case") with col_gf2: run_auto_gf = st.button("🔁 Auto Fill All", type="primary", use_container_width=True) if run_auto_gf: remaining_ids = set(not_found_ids) _autogf_api_key = st.session_state.get("_api_key","") for doc_idx, doc_name in enumerate(all_doc_names): if not remaining_ids: st.success("🎉 All gaps filled!") break doc_gfile = per_doc_files[doc_name] gap_indicators = [ind for ind in filtered if ind["id"] in remaining_ids] st.markdown(f"**📄 Pass {doc_idx+1}/{len(all_doc_names)}: `{doc_name}`** — {len(gap_indicators)} gaps…") gf_progress = st.progress(0) gf_status = st.empty() gf_completed = 0 gf_filled = 0 with ThreadPoolExecutor(max_workers=parallel_workers) as executor: future_to_ind = { executor.submit(generate_answer, ind["question"], [doc_gfile], selected_model, True, _autogf_api_key): ind for ind in gap_indicators } for future in as_completed(future_to_ind): ind = future_to_ind[future] gf_completed += 1 gf_progress.progress(gf_completed / len(gap_indicators)) try: new_answer = future.result() except Exception as e: new_answer = f"Error: {str(e)[:100]}" if not is_not_found(new_answer): value_str, unit_str = extract_value_and_unit(new_answer) st.session_state.batch_results[ind["id"]].update({ "answer": new_answer, "value": value_str, "unit": unit_str, "filled_from": doc_name, "recommended_docs": [], }) remaining_ids.discard(ind["id"]) gf_filled += 1 gf_status.write(f"✅ {gf_completed}/{len(gap_indicators)} — **{ind['id']}** filled") else: gf_status.write(f"⬜ {gf_completed}/{len(gap_indicators)} — **{ind['id']}** still not found") gf_progress.progress(1.0) st.caption(f"→ Filled **{gf_filled}** from `{doc_name}`. Remaining: {len(remaining_ids)}") if remaining_ids: st.warning(f"⚠️ **{len(remaining_ids)} indicators** genuinely absent from all documents.") else: st.success("🎉 All indicators filled!") # ── FINAL RESULTS TABLE ──────────────────────────────────────── st.markdown("---") st.markdown("### 📊 Final Results") final_results = [ st.session_state.batch_results[ind["id"]] for ind in filtered if ind["id"] in st.session_state.batch_results ] df = pd.DataFrame([{ "ID": r["id"], "Category": r["category"], "Indicator": r["indicator"], "Answer": r["answer"], "Value": r.get("value",""), "Unit": r.get("unit",""), "Source": r["source"], "Filled From": r.get("filled_from",""), "Recommended Documents": " → ".join(r.get("recommended_docs",[])), } for r in final_results]) def highlight_row(row): if is_not_found(row["Answer"]): return ["background-color: #3d1a1a; color: #ffcccc"] * len(row) elif row["Filled From"]: return ["background-color: #3d2e00; color: #ffe599"] * len(row) return ["background-color: #1a3d2a; color: #ccffcc"] * len(row) df_display = df.copy() df_display["Answer"] = df_display["Answer"].str[:200].fillna("") + "…" st.dataframe(df_display.style.apply(highlight_row, axis=1), use_container_width=True, height=480) st.markdown("🟩 **Initially found** | 🟨 **Gap-filled** | 🟥 **Not found**") # ── INDIVIDUAL ANSWER INSPECTOR ───────────────────────────────── st.markdown("---") st.markdown("### 🔎 Inspect Individual Answers") col_f1, col_f2 = st.columns([2,1]) with col_f1: search_term = st.text_input( "Search", placeholder="e.g. Scope 1, water, employee…", label_visibility="collapsed" ) with col_f2: not_found_only = st.checkbox("Show only 'Not Found'", value=False) show_results = final_results if not_found_only: show_results = [r for r in show_results if is_not_found(r["answer"])] if search_term: st_lower = search_term.lower() show_results = [r for r in show_results if st_lower in r["indicator"].lower() or st_lower in r["id"].lower() or st_lower in r["category"].lower()] st.caption(f"Showing {len(show_results)} of {len(final_results)} indicators") for r in show_results: if is_not_found(r["answer"]): icon = "❌" elif r.get("filled_from"): icon = "🟨" else: icon = "✅" with st.expander(f"{icon} **{r['id']}** — {r['indicator']}", expanded=False): col_a, col_b = st.columns([3, 1]) with col_a: st.markdown(f"**Category:** {r['category']}") if r.get("filled_from"): st.markdown(f"**Filled from:** `{r['filled_from']}`") with col_b: if r.get("value"): st.metric("Extracted Value", f"{r['value']} {r.get('unit','')}") st.markdown("---") if is_not_found(r["answer"]): st.markdown( '

' '⚠️ Information not found in the uploaded documents.

', unsafe_allow_html=True ) docs = r.get("recommended_docs", []) if docs: st.markdown("**📌 Recommended documents to obtain this data:**") for i, doc in enumerate(docs, 1): st.markdown(f" {i}. {doc}") else: fmt = format_answer_html(r["answer"]) if fmt and fmt != "__MARKDOWN__": st.markdown(fmt, unsafe_allow_html=True) else: st.markdown(r["answer"]) # ── SUMMARY ────────────────────────────────────────────────────── st.markdown("---") final_not_found = sum(1 for r in final_results if is_not_found(r["answer"])) final_found = len(final_results) - final_not_found gap_filled_ct = sum(1 for r in final_results if r.get("filled_from")) has_units = sum(1 for r in final_results if r.get("unit")) c1, c2, c3, c4, c5 = st.columns(5) c1.metric("✅ Total Found", final_found) c2.metric("🔁 Gap Filled", gap_filled_ct) c3.metric("❌ Still Missing", final_not_found) c4.metric("📐 With Units", has_units) c5.metric("📊 Total", len(final_results)) st.download_button( label="📥 Download Full Results (CSV)", data=df.to_csv(index=False), file_name="esg_indicators_results.csv", mime="text/csv", type="primary", use_container_width=True, )