Spaces:
Sleeping
Sleeping
| 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(""" | |
| <style> | |
| .esg-header { | |
| background: linear-gradient(135deg, #1a2744 0%, #0d3d2b 100%); | |
| border-radius: 12px; padding: 1.5rem 2rem; margin-bottom: 1.5rem; | |
| border: 1px solid #2a4060; | |
| } | |
| .esg-header h1 { color: #7dd3b0; font-size: 1.8rem; margin: 0; } | |
| .esg-header p { color: #9ab8c8; margin: 0.3rem 0 0; font-size: 0.95rem; } | |
| .answer-card { | |
| background: #1a2535; border-radius: 10px; padding: 1.2rem 1.5rem; | |
| border-left: 4px solid #3a7bd5; margin: 0.5rem 0; | |
| } | |
| .answer-card.found { border-left-color: #2ecc71; } | |
| .answer-card.filled { border-left-color: #f0a500; } | |
| .answer-card.missing { border-left-color: #e74c3c; } | |
| .kv-row { | |
| display: flex; gap: 0.6rem; padding: 0.35rem 0; | |
| border-bottom: 1px solid #243040; font-size: 0.88rem; | |
| } | |
| .kv-key { color: #8bacc8; min-width: 220px; font-weight: 600; } | |
| .kv-value { color: #d4e8ff; } | |
| </style> | |
| """, unsafe_allow_html=True) | |
| st.markdown(""" | |
| <div class="esg-header"> | |
| <h1>🌿 ESG Indicator Analyzer</h1> | |
| <p>Powered by Gemini — intelligent extraction from annual reports, BRSR, sustainability disclosures & more</p> | |
| </div> | |
| """, 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'<div class="kv-row"><span class="kv-key">{label}</span>' | |
| f'<span class="kv-value">{inner}</span></div>' | |
| ) | |
| else: | |
| val = str(v) | |
| rows.append( | |
| f'<div class="kv-row"><span class="kv-key">{label}</span>' | |
| f'<span class="kv-value">{val}</span></div>' | |
| ) | |
| return "".join(rows) | |
| elif isinstance(data, list): | |
| items = [] | |
| for item in data: | |
| rendered = _render_json_as_kv(item, depth) | |
| if rendered: | |
| items.append(f'<li style="color:#cde;margin:2px 0">{rendered}</li>') | |
| if not items: | |
| return "" | |
| return f'<ul style="margin:4px 0;padding-left:18px">{"".join(items)}</ul>' | |
| else: | |
| val = str(data) | |
| if val.lower() in _NULL_PHRASES: | |
| return "" | |
| return f'<span style="color:#cde">{val}</span>' | |
| 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'<div class="answer-card found">{inner}</div>' | |
| # 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", "<br>") | |
| rows.append( | |
| f'<div class="kv-row"><span class="kv-key">{k}</span>' | |
| f'<span class="kv-value">{v_html}</span></div>' | |
| ) | |
| return f'<div class="answer-card found">{"".join(rows)}</div>' | |
| # 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 | |
| # ========================================== | |
| 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( | |
| '<div class="answer-card missing"><span style="color:#e74c3c">' | |
| '⚠️ Information not found in the uploaded documents.</span></div>', | |
| 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, | |
| ) |