import json from datetime import datetime import firebase_admin from firebase_admin import credentials, firestore from collections import defaultdict # 🔐 Initialize Firebase (only once) if not firebase_admin._apps: try: cred = credentials.Certificate("firebase_key.json") firebase_admin.initialize_app(cred) print("✅ Firebase connection established.") except Exception as e: print(f"❌ Firebase initialization failed: {e}") raise # === Firestore client db = firestore.client() def log_user_feedback(goal, sol1, sol2, winner): try: formatted_winner = "1" if "1" in winner else "2" # Basic input sanitation goal = goal.strip()[:500] sol1 = sol1.strip()[:500] sol2 = sol2.strip()[:500] doc = { "timestamp": datetime.utcnow().isoformat(), "goal": goal, "solution_1": sol1, "solution_2": sol2, "winner": formatted_winner, "user_feedback": "from app.py", } ref = db.collection("evo_feedback").add(doc) print(f"📅 Feedback logged (ID: {ref[1].id}) — Winner: Solution {formatted_winner}") except Exception as e: print(f"❌ Feedback logging failed: {e}") raise def get_goal_leaderboard(top_n=5): try: # Query all feedback docs feedback_docs = db.collection("evo_feedback").stream() goal_counter = defaultdict(int) for doc in feedback_docs: data = doc.to_dict() goal = data.get("goal", "").strip() if goal: goal_counter[goal] += 1 # Sort by count, descending sorted_goals = sorted(goal_counter.items(), key=lambda x: x[1], reverse=True) return sorted_goals[:top_n] except Exception as e: print(f"❌ Failed to load leaderboard: {e}") return [] def get_vote_counts(): try: docs = db.collection("evo_feedback").stream() count_1, count_2 = 0, 0 for doc in docs: winner = doc.to_dict().get("winner", "").strip() if winner == "1": count_1 += 1 elif winner == "2": count_2 += 1 return {"1": count_1, "2": count_2} except Exception as e: print(f"❌ Failed to get vote counts: {e}") return {"1": 0, "2": 0}