from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from google import genai
from groq import Groq
import re
import json
import os
import uuid
import threading
import math
import hashlib
from datetime import datetime, timedelta
app = FastAPI()
# ---- Dynamic API Keys ----
GROQ_API_KEYS = []
i = 1
while True:
key = os.getenv(f"GROQ_API_KEY_{i}")
if not key:
break
GROQ_API_KEYS.append(key)
i += 1
print(f"Loaded {len(GROQ_API_KEYS)} Groq API key(s).")
GEMINI_API_KEYS = []
i = 1
while True:
key = os.getenv(f"GEMINI_API_KEY_{i}")
if not key:
break
GEMINI_API_KEYS.append(key)
i += 1
print(f"Loaded {len(GEMINI_API_KEYS)} Gemini API key(s).")
GROQ_MODEL = "llama-3.3-70b-versatile"
GEMINI_MODEL = "gemini-2.0-flash"
REPLIES_FILE = "replies.json"
HISTORY_FILE = "chat_history.json"
REPORTS_DIR = "reports"
MAX_HISTORY_PER_USER = 10
SESSION_TIMEOUT_MINUTES = 30
os.makedirs(REPORTS_DIR, exist_ok=True)
print(f"Reports directory: {os.path.abspath(REPORTS_DIR)}")
replies_lock = threading.Lock()
history_lock = threading.Lock()
recent_messages = {}
recent_messages_lock = threading.Lock()
DEDUP_WINDOW_SECONDS = 10
bot_sent_replies = set()
bot_sent_lock = threading.Lock()
def remember_bot_reply(message):
with bot_sent_lock:
bot_sent_replies.add(message.strip())
if len(bot_sent_replies) > 500:
bot_sent_replies.pop()
def is_bot_echo(message):
with bot_sent_lock:
return message.strip() in bot_sent_replies
BLACKLISTED_SENDERS = ["VZ-ViCARE-S","VM-","VZ-","AD-","TM-","TA-","ALERT","NOTIF"]
def is_blacklisted_sender(phone):
for prefix in BLACKLISTED_SENDERS:
if str(phone).startswith(prefix) or str(phone).upper().startswith(prefix.upper()):
return True
return False
def is_duplicate(phone, message):
key = f"{phone}|{message}"
now = datetime.utcnow().timestamp()
with recent_messages_lock:
expired = [k for k, t in recent_messages.items() if now - t > DEDUP_WINDOW_SECONDS]
for k in expired:
del recent_messages[k]
if key in recent_messages:
return True
recent_messages[key] = now
return False
def safe_read_json(path, default):
try:
with open(path, "r") as f:
content = f.read().strip()
if not content:
return default
return json.loads(content)
except (json.JSONDecodeError, FileNotFoundError):
return default
def init_file(path, default):
if not os.path.exists(path):
with open(path, "w") as f:
json.dump(default, f)
init_file(REPLIES_FILE, {"queue": []})
init_file(HISTORY_FILE, {})
# ============================================================
# LANGUAGE DETECTION
# ============================================================
LANGUAGE_MODES = {
"Hindi": "Hindi (pure Devanagari)",
"Gujarati": "Gujarati (pure Gujarati script)",
"English": "English",
"Hinglish": "Hinglish (Hindi in Roman letters)",
"Gujlish": "Gujlish (Gujarati in Roman letters)",
"Marathish": "Marathish (Marathi in Roman letters)",
"Punjabish": "Punjabish (Punjabi in Roman letters)",
"Tamilish": "Tamilish (Tamil in Roman letters)",
"Teluguish": "Teluguish (Telugu in Roman letters)",
"Kannadish": "Kannadish (Kannada in Roman letters)",
"Bengalish": "Bengalish (Bengali in Roman letters)",
"Marathi": "Marathi (Devanagari script)",
"Punjabi": "Punjabi (Gurmukhi script)",
"Tamil": "Tamil script",
"Telugu": "Telugu script",
"Kannada": "Kannada script",
"Bengali": "Bengali script",
"Odia": "Odia script",
"Urdu": "Urdu script",
}
# Unique Gujarati Roman words — not present in Hindi
GUJLISH_UNIQUE = {
"kem","cho","chhe","che","kemcho","kemo","kevi","kevo","kevu","keva",
"shu","su","kyare","kyarey","kyan","nathi","thayu","thashe","thase",
"chadvo","chadvu","chadyo","chadhu","karvu","karvanu","javu","jaiye","javanu",
"aavvu","aavshe","aavyo","levu","lidhu","lese","devu","didhu","dese",
"joie","joia","jovu","joyu","kahvu","malvu","malse","malyu","samjvu","samjyo",
"shikhvu","shikhvo","hatu","hati","hathu","tame","tamne","tamaro","tamari",
"tamara","tamaru","ame","amne","apde","ene","ena","eni","enu","maru","mara",
"pote","rite","reete","pachi","pachhi","agau","have","havey",
"badhu","badha","badhi","ghanu","ghanuj","ochhu","saras","mathi",
"ama","tema","jema","pahad","pahaad","dungar","pak","pako","khaad",
"vij","yojna","rupiya","aabhar","aavjo","mare","mane","pan","joi",
"ben","bhen","bapu","kaka","kaki","vaav","talavdi","talav","sheri",
"mun","dukhaave","taav","tav","vaid","aushadh","varsha","mokanu","vavtar",
"lani","aaje","aje","parchhe","parmothe","saanje","sanje","ratre","tran",
"kalaak","bhaat","shaak","dudh","mirchu","subsidi","karj","vima",
}
# Unique Hindi Roman words — not present in Gujarati
HINGLISH_UNIQUE = {
"kya","hai","hain","nahi","nahin","mera","meri","mujhe","tumhe",
"aap","tum","yeh","woh","kaise","kyun","kab","kahan",
"chahiye","karein","batao","samjho","accha","theek","galat",
"bijli","sadak","yojana","subsidy","matlab","samajh","bahut","zyada",
"behen","chacha","shukriya","parso","subah","shaam","rupee",
"pehle","fasal","khad","phir","thoda","dawai",
}
def detect_language(text: str) -> dict:
script_counts = {
"Gujarati":0,"Hindi":0,"Punjabi":0,"Bengali":0,
"Tamil":0,"Telugu":0,"Kannada":0,"Odia":0,"Urdu":0,
}
for char in text:
cp = ord(char)
if 0x0A80 <= cp <= 0x0AFF: script_counts["Gujarati"] += 1
elif 0x0900 <= cp <= 0x097F: script_counts["Hindi"] += 1
elif 0x0A00 <= cp <= 0x0A7F: script_counts["Punjabi"] += 1
elif 0x0980 <= cp <= 0x09FF: script_counts["Bengali"] += 1
elif 0x0B80 <= cp <= 0x0BFF: script_counts["Tamil"] += 1
elif 0x0C00 <= cp <= 0x0C7F: script_counts["Telugu"] += 1
elif 0x0C80 <= cp <= 0x0CFF: script_counts["Kannada"] += 1
elif 0x0B00 <= cp <= 0x0B7F: script_counts["Odia"] += 1
elif 0x0600 <= cp <= 0x06FF: script_counts["Urdu"] += 1
max_script = max(script_counts, key=script_counts.get)
if script_counts[max_script] > 0:
return {"mode":max_script,"label":LANGUAGE_MODES.get(max_script,max_script),
"script":"native","base":max_script}
words = set(re.findall(r'\b[a-zA-Z]+\b', text.lower()))
gujlish_hits = len(words & GUJLISH_UNIQUE)
hinglish_hits = len(words & HINGLISH_UNIQUE)
print(f"[LANG] guj={gujlish_hits} hin={hinglish_hits} words={words}")
if gujlish_hits == 0 and hinglish_hits == 0:
return {"mode":"English","label":LANGUAGE_MODES["English"],"script":"roman","base":"English"}
if gujlish_hits > hinglish_hits:
return {"mode":"Gujlish","label":LANGUAGE_MODES["Gujlish"],"script":"roman","base":"Gujarati"}
if hinglish_hits > gujlish_hits:
return {"mode":"Hinglish","label":LANGUAGE_MODES["Hinglish"],"script":"roman","base":"Hindi"}
return {"mode":"Gujlish","label":LANGUAGE_MODES["Gujlish"],"script":"roman","base":"Gujarati"}
def get_reply_language_instruction(lang_info: dict) -> str:
mode = lang_info["mode"]
instructions = {
"Hindi": "Reply in Hindi using Devanagari script. Simple and clear.",
"Gujarati": "Reply in Gujarati using Gujarati script. Simple and clear.",
"Marathi": "Reply in Marathi using Devanagari script.",
"Punjabi": "Reply in Punjabi using Gurmukhi script.",
"Bengali": "Reply in Bengali using Bengali script.",
"Tamil": "Reply in Tamil using Tamil script.",
"Telugu": "Reply in Telugu using Telugu script.",
"Kannada": "Reply in Kannada using Kannada script.",
"Odia": "Reply in Odia using Odia script.",
"Urdu": "Reply in Urdu using Roman script.",
"Hinglish": "User writes Hinglish (Hindi in Roman). Reply in SAME Hinglish. NO Devanagari. Example: 'Fasal ke liye urea khad use karo, 2 bag per bigha.'",
"Gujlish": "User writes Gujlish (Gujarati in Roman). Reply in SAME Gujlish. NO Gujarati script. Example: 'Pak mate urea khaad vapro, bigha ma 2 bag kaafi chhe.'",
"Marathish": "Reply in Marathi Roman letters only.",
"Punjabish": "Reply in Punjabi Roman letters only.",
"Tamilish": "Reply in Tamil Roman letters only.",
"Teluguish": "Reply in Telugu Roman letters only.",
"Kannadish": "Reply in Kannada Roman letters only.",
"Bengalish": "Reply in Bengali Roman letters only.",
"English": "Reply in simple clear English.",
}
return instructions.get(mode, f"Reply in {mode}. Keep it simple.")
# Map language mode to its full native script name for report header
LANGUAGE_DISPLAY_NAME = {
"Hindi": "हिंदी",
"Gujarati": "ગુજરાતી",
"Marathi": "मराठी",
"Punjabi": "ਪੰਜਾਬੀ",
"Bengali": "বাংলা",
"Tamil": "தமிழ்",
"Telugu": "తెలుగు",
"Kannada": "ಕನ್ನಡ",
"Odia": "ଓଡ଼ିଆ",
"Urdu": "اردو",
"Hinglish": "Hinglish",
"Gujlish": "Gujlish",
"Marathish": "Marathish",
"Punjabish": "Punjabish",
"Tamilish": "Tamilish",
"Teluguish": "Teluguish",
"Kannadish": "Kannadish",
"Bengalish": "Bengalish",
"English": "English",
}
# ---- SMS Cleaner ----
def clean_for_sms(text, limit=155):
text = re.sub(r"[*_`#]+", "", text)
text = text.replace("\r\n"," ").replace("\n"," ").replace("\r"," ").replace("\t"," ")
text = re.sub(r"\s*[-•●]\s*", " ", text)
text = re.sub(r" +", " ", text).strip()
if len(text) <= limit:
return text
for punct in ["।",".", "!", "?", ";", ","]:
idx = text.rfind(punct, 0, limit)
if idx != -1:
return text[:idx+1].strip()
idx = text.rfind(" ", 0, limit)
return text[:idx].strip() if idx != -1 else text[:limit]
# ---- Session Helpers ----
def load_history():
with history_lock:
return safe_read_json(HISTORY_FILE, {})
def save_history(history):
with history_lock:
with open(HISTORY_FILE, "w") as f:
json.dump(history, f, ensure_ascii=False, indent=2)
def get_user_session(phone):
history = load_history()
user_data = history.get(phone)
if user_data:
last_active = datetime.fromisoformat(user_data["last_active"])
if datetime.utcnow() - last_active > timedelta(minutes=SESSION_TIMEOUT_MINUTES):
del history[phone]
save_history(history)
return {"messages":[],"last_active":datetime.utcnow().isoformat(),"language":"English"}
return user_data
return {"messages":[],"last_active":datetime.utcnow().isoformat(),"language":"English"}
def update_user_session(phone, user_message, bot_reply, language_mode):
history = load_history()
session = history.get(phone, {"messages":[],"last_active":None,"language":language_mode})
session["messages"].append({"role":"user","text":user_message})
session["messages"].append({"role":"assistant","text":bot_reply})
session["last_active"] = datetime.utcnow().isoformat()
session["language"] = language_mode
max_entries = MAX_HISTORY_PER_USER * 2
if len(session["messages"]) > max_entries:
session["messages"] = session["messages"][-max_entries:]
history[phone] = session
save_history(history)
def build_conversation_context(messages):
if not messages:
return "No previous conversation."
return "\n".join(
f"{'User' if e['role']=='user' else 'Assistant'}: {e['text']}"
for e in messages
)
# ---- AI Providers ----
def try_groq(prompt, max_tokens=80):
for index, api_key in enumerate(GROQ_API_KEYS):
try:
print(f"[Groq] Trying key #{index+1}...")
client = Groq(api_key=api_key)
response = client.chat.completions.create(
model=GROQ_MODEL,
messages=[{"role":"user","content":prompt}],
max_tokens=max_tokens,
)
raw = response.choices[0].message.content
print(f"[Groq] OK #{index+1}")
return raw
except Exception as e:
msg = str(e)
if "429" in msg or "quota" in msg.lower() or "rate" in msg.lower() or "limit" in msg.lower():
print(f"[Groq] Key #{index+1} rate limited.")
else:
print(f"[Groq] Key #{index+1} error: {msg}")
return None
def try_gemini(prompt):
for index, api_key in enumerate(GEMINI_API_KEYS):
try:
print(f"[Gemini] Trying key #{index+1}...")
client = genai.Client(api_key=api_key)
response = client.models.generate_content(model=GEMINI_MODEL, contents=prompt)
raw = response.text
print(f"[Gemini] OK #{index+1}")
return raw
except Exception as e:
msg = str(e)
if "429" in msg or "quota" in msg.lower() or "rate" in msg.lower():
print(f"[Gemini] Key #{index+1} quota exceeded.")
else:
print(f"[Gemini] Key #{index+1} error: {msg}")
return None
FALLBACK_MESSAGES = {
"Hindi": "क्षमा करें, सेवा अभी उपलब्ध नहीं है।",
"Gujarati": "માફ કરશો, સેવા હાલ ઉપલબ્ધ નથી.",
"Marathi": "माफ करा, सेवा उपलब्ध नाही.",
"Punjabi": "ਮਾਫ਼ ਕਰਨਾ, ਸੇਵਾ ਉਪਲਬਧ ਨਹੀਂ।",
"Bengali": "দুঃখিত, পরিষেবা নেই।",
"Tamil": "மன்னிக்கவும், சேவை இல்லை.",
"Telugu": "క్షమించండి, సేవ లేదు.",
"Kannada": "ಕ್ಷಮಿಸಿ, ಸೇವೆ ಲಭ್ಯವಿಲ್ಲ.",
"Odia": "କ୍ଷମା, ସେବା ଉପଲବ୍ଧ ନୁହେଁ।",
"Urdu": "Maafi, service nahi hai.",
"Hinglish": "Sorry bhai, service nahi. Baad mein try karo.",
"Gujlish": "Maaf karo, service nathi. Pachhi try karo.",
"Marathish": "Sorry, service nahi. Nantare try kara.",
"Punjabish": "Sorry, service nahi. Baad vich try karo.",
"Tamilish": "Sorry, service illai. Apuram try pannunga.",
"Teluguish": "Sorry, service ledu. Tarvata try cheyandi.",
"Kannadish": "Sorry, service illa. Nachche try madi.",
"Bengalish": "Sorry, service nei. Pore try koro.",
"English": "Sorry, service unavailable. Please try again later.",
}
def generate_reply(prompt, language_mode, max_tokens=80):
if GROQ_API_KEYS:
result = try_groq(prompt, max_tokens=max_tokens)
if result:
return clean_for_sms(result)
if GEMINI_API_KEYS:
result = try_gemini(prompt)
if result:
return clean_for_sms(result)
return FALLBACK_MESSAGES.get(language_mode, FALLBACK_MESSAGES["English"])
# ---- Reply Queue ----
def load_queue():
with replies_lock:
return safe_read_json(REPLIES_FILE, {"queue":[]}).get("queue",[])
def save_queue(queue):
with replies_lock:
with open(REPLIES_FILE, "w") as f:
json.dump({"queue":queue}, f, ensure_ascii=False, indent=2)
def enqueue_reply(phone, message, language_mode):
queue = load_queue()
queue.append({
"id":str(uuid.uuid4()), "phone":phone, "message":message,
"language":language_mode, "created_at":datetime.utcnow().isoformat(), "sent":False
})
save_queue(queue)
def dequeue_unsent_replies():
queue = load_queue()
unsent = [r for r in queue if not r["sent"]]
sent_ids = {r["id"] for r in unsent}
for item in queue:
if item["id"] in sent_ids:
item["sent"] = True
sent = [r for r in queue if r["sent"]]
unsent_remaining = [r for r in queue if not r["sent"]]
save_queue(unsent_remaining + sent[-100:])
return unsent
# ============================================================
# SVG VISUAL BUILDERS — Only render when data is meaningful
# ============================================================
def build_svg_bar_chart(labels, values, unit="", title="", width=340, height=220):
if not values or len(values) < 2 or max(values) == 0:
return ""
max_val = max(values)
n = len(values)
bar_width = min(44, (width - 70) // n)
gap = max(6, (width - 70 - bar_width * n) // (n + 1))
colors = ["#2E7D32","#1565C0","#E65100","#AD1457","#6A1B9A","#00695C"]
bars = ""
for i, (label, value) in enumerate(zip(labels, values)):
bar_h = max(4, int((value / max_val) * (height - 70)))
x = 55 + gap + i * (bar_width + gap)
y = height - 50 - bar_h
color = colors[i % len(colors)]
bars += (
f''
f'{value}{unit}'
f'{str(label)[:10]}'
)
grid = ""
for step in range(5):
y_pos = height - 50 - int((step / 4) * (height - 70))
val = int((step / 4) * max_val)
grid += (
f''
f'{val}'
)
title_el = f'{title}' if title else ""
return (
f''
)
def build_svg_pie_chart(labels, values, title="", width=300):
if not values or len(values) < 2:
return ""
total = sum(values)
if total == 0:
return ""
colors = ["#2E7D32","#1565C0","#E65100","#AD1457","#6A1B9A","#00695C","#F57F17","#37474F"]
cx, cy = width // 2, width // 2
r = width // 2 - 35
angle = -math.pi / 2
slices = ""
legends = ""
for i, (label, value) in enumerate(zip(labels, values)):
sweep = (value / total) * 2 * math.pi
x1, y1 = cx + r * math.cos(angle), cy + r * math.sin(angle)
x2, y2 = cx + r * math.cos(angle + sweep), cy + r * math.sin(angle + sweep)
large_arc = 1 if sweep > math.pi else 0
color = colors[i % len(colors)]
pct = round(value / total * 100)
slices += (
f''
)
mid = angle + sweep / 2
lx = cx + (r * 0.62) * math.cos(mid)
ly = cy + (r * 0.62) * math.sin(mid)
if pct > 7:
slices += (
f'{pct}%'
)
legends += (
f''
f''
f'{str(label)[:20]} — {value}'
)
angle += sweep
total_h = width + 20 + len(labels) * 24
title_el = f'{title}' if title else ""
return (
f''
)
def build_svg_flowchart(steps, title="", width=340):
"""Vertical step-by-step flowchart — ideal for processes like farming steps"""
if not steps or len(steps) < 2:
return ""
box_h = 44
box_w = width - 40
gap = 28
colors = ["#1B5E20","#1A237E","#BF360C","#880E4F","#4A148C","#006064"]
total_h = len(steps) * (box_h + gap) + 30
boxes = ""
for i, step in enumerate(steps):
y = 30 + i * (box_h + gap)
color = colors[i % len(colors)]
label = str(step)[:45]
# Box
boxes += (
f''
# Step number circle
f''
f'{i+1}'
# Step label
f'{label}'
)
# Arrow between steps
if i < len(steps) - 1:
arrow_y = y + box_h + 2
mid_x = width // 2
boxes += (
f''
f''
)
title_el = f'{title}' if title else ""
return (
f''
)
def build_svg_comparison(items, title="", width=340):
"""
Visual comparison card — for comparing options side by side.
items = [{"label":"Option A","score":80,"note":"Good for sandy soil"},...]
"""
if not items or len(items) < 2:
return ""
bar_h = 28
gap = 14
colors = ["#2E7D32","#1565C0","#E65100","#AD1457"]
total_h = 40 + len(items) * (bar_h + gap)
max_score = max(it.get("score", 50) for it in items)
bars = ""
for i, item in enumerate(items):
y = 34 + i * (bar_h + gap)
score = item.get("score", 50)
label = str(item.get("label",""))[:18]
note = str(item.get("note",""))[:30]
color = colors[i % len(colors)]
bw = int((score / max_score) * (width - 120))
bars += (
f'{label}'
f''
f'{score}'
f'{note}'
)
title_el = f'{title}' if title else ""
return (
f''
)
def render_visual(section: dict) -> str:
"""
Renders the correct visual based on section type.
Only renders if data is present and meaningful.
"""
vtype = section.get("visual_type","")
if vtype == "bar_chart":
return build_svg_bar_chart(
section.get("labels",[]),
section.get("values",[]),
section.get("unit",""),
section.get("visual_title",""),
)
elif vtype == "pie_chart":
return build_svg_pie_chart(
section.get("labels",[]),
section.get("values",[]),
section.get("visual_title",""),
)
elif vtype == "flowchart":
return build_svg_flowchart(
section.get("steps",[]),
section.get("visual_title",""),
)
elif vtype == "comparison":
return build_svg_comparison(
section.get("items",[]),
section.get("visual_title",""),
)
return ""
# ============================================================
# SMART REPORT PROMPT
# ============================================================
def generate_detailed_prompt(message: str, language_mode: str, short_reply: str) -> str:
"""
Instructs AI to:
1. Write ALL content in the user's exact language
2. Only add visuals when they genuinely add value
3. Choose the RIGHT type of visual for the content
4. Be detailed, practical, and helpful — not vague
"""
lang_instruction = {
"Hindi": "Write EVERYTHING in Hindi using Devanagari script (हिंदी में लिखें).",
"Gujarati": "Write EVERYTHING in Gujarati using Gujarati script (ગુજરાતીમાં લખો).",
"Marathi": "Write EVERYTHING in Marathi using Devanagari script.",
"Punjabi": "Write EVERYTHING in Punjabi using Gurmukhi script.",
"Bengali": "Write EVERYTHING in Bengali script.",
"Tamil": "Write EVERYTHING in Tamil script.",
"Telugu": "Write EVERYTHING in Telugu script.",
"Kannada": "Write EVERYTHING in Kannada script.",
"Odia": "Write EVERYTHING in Odia script.",
"Urdu": "Write EVERYTHING in Urdu Roman script.",
"Hinglish": "Write EVERYTHING in Hinglish (Hindi words in Roman/English letters). No Devanagari.",
"Gujlish": "Write EVERYTHING in Gujlish (Gujarati words in Roman/English letters). No Gujarati script.",
"Marathish": "Write EVERYTHING in Marathi Roman letters.",
"Punjabish": "Write EVERYTHING in Punjabi Roman letters.",
"Tamilish": "Write EVERYTHING in Tamil Roman letters.",
"Teluguish": "Write EVERYTHING in Telugu Roman letters.",
"Kannadish": "Write EVERYTHING in Kannada Roman letters.",
"Bengalish": "Write EVERYTHING in Bengali Roman letters.",
"English": "Write EVERYTHING in simple clear English.",
}.get(language_mode, "Write in English.")
return f"""You are an expert rural assistant for Indian villagers and farmers.
USER QUESTION: "{message}"
SHORT SMS ANSWER ALREADY SENT: "{short_reply}"
LANGUAGE RULE (CRITICAL): {lang_instruction}
Every single word in title, summary, content, labels, tips — ALL must be in {language_mode}.
Only exception: numbers and units (kg, ml, %, Rs) can stay as-is.
YOUR TASK: Create a DETAILED, PRACTICAL, HELPFUL report that a rural Indian user can actually use.
The report must feel like advice from a knowledgeable, caring local expert — not a generic textbook.
VISUAL RULES (VERY IMPORTANT):
- Only add a visual if it genuinely helps understanding. NOT every section needs one.
- For step-by-step processes (how to do something): use "flowchart"
- For comparing quantities or amounts over time/categories: use "bar_chart"
- For showing proportions or share of something: use "pie_chart"
- For comparing 2-4 options with scores: use "comparison"
- For plain explanations with no data: set visual_type to "none"
- NEVER add a chart just to fill space. If data is not meaningful, set visual_type to "none".
Return ONLY a valid JSON object. No markdown, no explanation text before or after.
JSON STRUCTURE:
{{
"title": "specific descriptive title in {language_mode}",
"summary": "3-4 sentence genuinely helpful summary in {language_mode} — include key facts, numbers, and actionable advice",
"sections": [
{{
"heading": "section heading in {language_mode}",
"content": "detailed, practical paragraph in {language_mode} — include real numbers, timing, dosage, method etc. Minimum 3-4 sentences.",
"visual_type": "flowchart | bar_chart | pie_chart | comparison | none",
"visual_title": "chart title in {language_mode} (only if visual_type is not none)",
"steps": ["step 1", "step 2", "step 3"],
"labels": ["label1", "label2"],
"values": [10, 20],
"unit": "kg",
"items": [{{"label": "Option A", "score": 80, "note": "short note"}}]
}}
],
"tips": [
"Practical tip 1 in {language_mode} — be specific",
"Practical tip 2 in {language_mode} — be specific",
"Practical tip 3 in {language_mode} — be specific"
],
"warning": "Important caution or warning in {language_mode} (if relevant, else empty string)",
"language": "{language_mode}"
}}
SECTION GUIDELINES based on question type:
- Farming/agriculture: sections = [preparation, method/process with flowchart, quantities with bar_chart, timeline, expected yield]
- Health: sections = [symptoms/cause, remedy, dosage, when to see doctor]
- Machine/repair: sections = [diagnosis, repair steps with flowchart, tools needed, prevention]
- Government scheme: sections = [eligibility, documents needed, how to apply with flowchart, benefit amount with bar_chart]
- General knowledge: sections = [explanation, examples, practical use]
Include 3-5 sections. Make each section GENUINELY USEFUL with specific numbers, timings, and local context for Indian rural users."""
# ============================================================
# HTML REPORT BUILDER
# ============================================================
def build_report_html(data: dict, report_id: str, language_mode: str) -> str:
lang_display = LANGUAGE_DISPLAY_NAME.get(language_mode, language_mode)
# Determine text direction
rtl_langs = {"Urdu"}
dir_attr = 'dir="rtl"' if language_mode in rtl_langs else ''
# Build sections
sections_html = ""
for section in data.get("sections", []):
heading = section.get("heading", "")
content = section.get("content", "")
visual_type = section.get("visual_type", "none")
# Render visual only if meaningful
visual_html = ""
if visual_type and visual_type != "none":
svg = render_visual(section)
if svg:
visual_html = f'
{svg}
'
sections_html += f"""
{heading}
{content}
{visual_html}
"""
# Tips
tips = data.get("tips", [])
tips_html = ""
if tips:
items = "".join(
f'
{i+1}{tip}
'
for i, tip in enumerate(tips)
)
tips_html = f"""
💡 {'Useful Tips' if language_mode=='English' else 'Tips'}