Spaces:
Sleeping
Sleeping
| import google.generativeai as genai | |
| import os | |
| from dotenv import load_dotenv | |
| load_dotenv() | |
| GEMINI_API_KEY = os.getenv("GEMINI_API_KEY") | |
| # if GEMINI_API_KEY: | |
| # genai.configure(api_key=GEMINI_API_KEY) | |
| # model = genai.GenerativeModel('gemini-2.0-flash') | |
| # else: | |
| model = None | |
| print("Gemini LLM Disabled by user request.") | |
| # print("Warning: GEMINI_API_KEY not found in .env. LLM features will be disabled.") | |
| def generate_company_description(company_name: str) -> str: | |
| """ | |
| Generates a brief 2-3 sentence description of the company using Gemini. | |
| """ | |
| if not model: | |
| return "AI description unavailable (API Key missing)." | |
| try: | |
| prompt = f"Provide a factual, neutral 2-3 sentence description of the company '{company_name}', focusing on its industry and main products. Do not mention sentiment or controversies." | |
| response = model.generate_content(prompt) | |
| return response.text.strip() | |
| except Exception as e: | |
| print(f"Error generating description for {company_name}: {e}") | |
| return "AI description unavailable due to an error." | |
| def generate_ai_recommendations(company_name: str, analysis_data: dict) -> dict: | |
| """ | |
| Generates tailored recommendations for Customers, Investors, and Leadership based on the analysis. | |
| """ | |
| if not model: | |
| return { | |
| "for_customers": ["Review provided evidence links."], | |
| "for_investors": ["Analyze financial risks mentioned in report."], | |
| "for_company_leadership": ["Address flagged contradictions."] | |
| } | |
| try: | |
| # Construct a summary context for the LLM | |
| context = f""" | |
| Company: {company_name} | |
| Greenwashing Risk: {'High' if analysis_data.get('greenwashingLabel') == 1 else 'Low'} | |
| Reason: {analysis_data.get('internal_documents_analysis', {}).get('major_findings', ['N/A'])[0]} | |
| Contradictions: {len(analysis_data.get('contradictions_detected', []))} found. | |
| Sentiment: {analysis_data.get('external_summary', {}).get('public_sentiment', 'N/A')} | |
| """ | |
| prompt = f""" | |
| Based on the following analysis of '{company_name}', provide 3 specific, actionable recommendations for each group (Customers, Investors, Leadership). | |
| Focus on greenwashing, transparency, and sustainability accountability. | |
| Analysis Context: | |
| {context} | |
| Output purely as JSON format with keys: "for_customers", "for_investors", "for_company_leadership". Each key should have a list of strings. | |
| Do not allow Markdown code blocks. Just raw JSON. | |
| """ | |
| response = model.generate_content(prompt) | |
| text = response.text.strip() | |
| # Clean potential markdown wrapping | |
| if text.startswith("```json"): | |
| text = text[7:] | |
| if text.endswith("```"): | |
| text = text[:-3] | |
| import json | |
| return json.loads(text) | |
| except Exception as e: | |
| print(f"Error generating recommendations for {company_name}: {e}") | |
| # Fallback | |
| return { | |
| "for_customers": ["Review provided evidence links.", "Cross-check claims."], | |
| "for_investors": ["Monitor reputational risks.", "Demand clearer impact reports."], | |
| "for_company_leadership": ["Address detected contradictions.", "Improve transparency."] | |
| } | |
| def generate_combined_insights(company_name: str, analysis_data: dict) -> dict: | |
| """ | |
| Combines description and recommendations into a single API call to reduce rate limit usage. | |
| Returns: { "description": str, "recommendations": dict } | |
| """ | |
| if not model: | |
| return { | |
| "description": "AI description unavailable (API Key missing).", | |
| "recommendations": generate_ai_recommendations(company_name, analysis_data) # Fallback to default | |
| } | |
| try: | |
| context = f""" | |
| Company: {company_name} | |
| Greenwashing Risk: {'High' if analysis_data.get('greenwashingLabel') == 1 else 'Low'} | |
| Reason: {analysis_data.get('internal_documents_analysis', {}).get('major_findings', ['N/A'])[0]} | |
| """ | |
| prompt = f""" | |
| Analyze '{company_name}' based on this context: | |
| {context} | |
| Provide 2 outputs in a single JSON object: | |
| 1. "description": A factual 2-sentence description of the company. | |
| 2. "recommendations": A dictionary with keys "for_customers", "for_investors", "for_company_leadership", containing 3 actionable tips for each. | |
| Output purely JSON. No markdown. | |
| """ | |
| response = model.generate_content(prompt) | |
| text = response.text.strip() | |
| if text.startswith("```json"): text = text[7:] | |
| if text.endswith("```"): text = text[:-3] | |
| return json.loads(text) | |
| except Exception as e: | |
| print(f"Error generating combined insights for {company_name}: {e}") | |
| return { | |
| "description": "AI description unavailable due to high traffic.", | |
| "recommendations": { | |
| "for_customers": ["Review evidence links."], | |
| "for_investors": ["Analyze risks."], | |
| "for_company_leadership": ["Address contradictions."] | |
| } | |
| } | |
| def generate_batch_insights(companies_data: list) -> dict: | |
| """ | |
| Generates insights for a batch of companies (up to 10-15 recommended) in a SINGLE prompt. | |
| Input: list of {name, context: str} | |
| Output: dict { company_name: { "description": ..., "recommendations": ... } } | |
| """ | |
| import json | |
| from .hugchat_client import generate_hugchat_response | |
| # Try HuggingChat if Gemini is disabled | |
| if not model: | |
| # Construct Prompt for HuggingChat | |
| batch_context = "" | |
| for i, c in enumerate(companies_data): | |
| batch_context += f"\n--- Company {i+1}: {c['name']} ---\n{c['context']}\n" | |
| prompt = f""" | |
| You are a sustainability analyst. Analyze these {len(companies_data)} companies. | |
| {batch_context} | |
| Return a valid JSON OBJECT where keys are company names. | |
| For each company, provide: | |
| 1. "description": A factual 2-sentence summary. | |
| 2. "recommendations": Object with keys "for_customers", "for_investors", "for_company_leadership" (list of 3 tips each). | |
| Example JSON Structure: | |
| {{ | |
| "Company Name": {{ | |
| "description": "...", | |
| "recommendations": {{ "for_customers": [...], ... }} | |
| }} | |
| }} | |
| IMPORTANT: Output ONLY valid JSON. No Markdown. No Intro. | |
| """ | |
| print("Using HuggingChat for Batch Analysis...") | |
| response_text = generate_hugchat_response(prompt) | |
| try: | |
| # clean json | |
| text = response_text.strip() | |
| if text.startswith("```json"): text = text[7:] | |
| if text.endswith("```"): text = text[:-3] | |
| if "{" not in text: raise Exception("Invalid JSON format") | |
| return json.loads(text) | |
| except Exception as e: | |
| print(f"HuggingChat Parsing Error: {e}") | |
| # Fallthrogh to fallback | |
| pass | |
| if not model and not 'response_text' in locals(): | |
| # Return fallback for all | |
| return {c['name']: { | |
| "description": "AI unavailable (Key missing)", | |
| "recommendations": { | |
| "for_customers": ["Review evidence."], | |
| "for_investors": ["Check risks."], | |
| "for_company_leadership": ["Monitor compliance."] | |
| } | |
| } for c in companies_data} | |
| try: | |
| # ... (Gemini Logic remains as backup if re-enabled) ... | |
| # Construct simplified context list | |
| batch_context = "" | |
| for i, c in enumerate(companies_data): | |
| batch_context += f"\n--- Company {i+1}: {c['name']} ---\n{c['context']}\n" | |
| prompt = f""" | |
| Analyze the following {len(companies_data)} companies based on the provided contexts. | |
| {batch_context} | |
| For EACH company, provide: | |
| 1. "description": A factual 2-sentence summary. | |
| 2. "recommendations": 3 specific actionable tips per group (Customers, Investors, Leadership). | |
| Output purely as a JSON OBJECT where keys are the exact company names and values are the insight objects. | |
| Example: | |
| {{ | |
| "Company A": {{ "description": "...", "recommendations": {{ ... }} }}, | |
| "Company B": ... | |
| }} | |
| No markdown formatting. Just JSON. | |
| """ | |
| response = model.generate_content(prompt) | |
| text = response.text.strip() | |
| if text.startswith("```json"): text = text[7:] | |
| if text.endswith("```"): text = text[:-3] | |
| results = json.loads(text) | |
| return results | |
| except Exception as e: | |
| print(f"Batch generation error: {e}") | |
| return {} | |