💰 Potential Savings
${savings:,.2f}
Delete + Hibernate
Total Spend
${total_spend:,.2f}
Analyzed: ${analyzed_spend:,.2f}
Orphan Spend
${unallocated_cost:,.2f}
No Team Tag
Waste CPU Avg
{avg_waste_util:.1f}%
Idle Resources
💼 Cost by Action
{action_chart}
🔧 Cost by Service
{service_chart}
🏆 Team Waste Ranking
{team_chart}
🛡️ Triage Details
| ResourceID |
Service |
Cost |
Action |
Reason |
""".format(
total_spend=total_spend,
analyzed_spend=analyzed_spend,
savings=savings,
unallocated_cost=unallocated_cost,
avg_waste_util=avg_waste_util,
action_chart=action_chart_html,
service_chart=service_chart_html,
team_chart=team_chart_html
)
action_colors = {"KEEP": "#10b981", "DELETE": "#ef4444", "HIBERNATE": "#f59e0b", "AUDIT": "#f97316"}
for result in results:
res = result["resource"]
cls = result["classification"]
color = action_colors.get(cls['action'], "#6b7280")
report += f"""
| {res['ResourceID']} |
{res['Service']} |
${res['Cost_Monthly']:.2f} |
{cls['action']} |
{cls['reason']} |
"""
report += "
"
return report
def generate_emails(results):
"""Generates individual email drafts per team."""
team_data = {}
# Group resources by team
for result in results:
resource = result["resource"]
classification = result["classification"]
# Constraint: Do NOT generate emails for resources marked "KEEP"
if classification["action"] == "KEEP":
continue
# 1. Trust AI first
team_name = "Unknown"
# Check if AI provided inferred_team (Pydantic model field)
if classification.get("inferred_team"):
val = classification["inferred_team"]
# Handle Enum or String
if hasattr(val, 'value'):
team_name = val.value
else:
team_name = str(val)
# 2. Fallback to existing Tag (if AI returned Unknown)
if team_name == "Unknown":
tags = json.loads(resource.get('Tags', '{}'))
tag_team = get_tag(tags, ['team', 'owner_team'], "")
if tag_team:
team_name = tag_team.strip().title()
# Grouping Logic
if team_name not in team_data:
team_data[team_name] = {'resources': [], 'poc_emails': set()}
team_data[team_name]['resources'].append(result)
# Extract POC email (use OwnerEmail as POC)
poc_email = resource.get('OwnerEmail', '')
if poc_email and '@' in poc_email:
team_data[team_name]['poc_emails'].add(poc_email)
# Generate individual email drafts
if not team_data:
return "# No Action Required\n\nAll resources are optimally configured. No emails to send."
output = f"# 📧 Team Email Drafts ({len(team_data)} Teams)\n\n"
output += "_Copy and paste each email below to send to the team POC_\n\n"
output += "---\n\n"
# Generate ONE email draft per Team
for team in sorted(team_data.keys()):
items = team_data[team]['resources']
poc_emails = team_data[team]['poc_emails']
# Calculate total savings for this team
savings = sum(r['resource']['Cost_Monthly'] for r in items)
# Count actions
action_counts = {}
for item in items:
action = item['classification']['action']
action_counts[action] = action_counts.get(action, 0) + 1
# Email header
output += f"## 📮 Email Draft for {team}\n\n"
output += "```\n"
output += f"To: {', '.join(sorted(poc_emails)) if poc_emails else 'team-email@company.com'}\n"
output += f"Subject: [Action Required] Cloud FinOps Audit - ${savings:,.2f} Potential Savings\n"
output += "\n"
# Email body
output += f"Dear {team},\n\n"
output += f"Our automated FinOps audit has identified {len(items)} cloud resources under your team's ownership "
output += f"that can be optimized to save ${savings:,.2f} per month.\n\n"
output += "**Summary:**\n"
action_summary = " | ".join([f"{count} {action}" for action, count in sorted(action_counts.items())])
output += f"- {len(items)} resources flagged: {action_summary}\n"
output += f"- Total potential savings: ${savings:,.2f}/month\n\n"
output += "**Top Resources by Cost:**\n\n"
# Show top 5 resources in table format
top_resources = sorted(items, key=lambda x: x['resource']['Cost_Monthly'], reverse=True)[:5]
for i, item in enumerate(top_resources, 1):
res = item['resource']
cls = item['classification']
action_emoji = {
'DELETE': '🔴',
'HIBERNATE': '🟡',
'AUDIT': '🟠'
}.get(cls['action'], '⚪')
output += f"{i}. {res['ResourceID']} ({res['Service']})\n"
output += f" Cost: ${res['Cost_Monthly']:.2f}/mo | Action: {action_emoji} {cls['action']}\n"
output += f" Reason: {cls['reason'][:80]}{'...' if len(cls['reason']) > 80 else ''}\n\n"
if len(items) > 5:
output += f"...and {len(items) - 5} more resources (see full report)\n\n"
output += "**Next Steps:**\n"
output += f"1. Review the flagged resources in the attached report\n"
output += f"2. Confirm or reject the recommended actions\n"
output += f"3. Contact finops@company.com with questions\n\n"
output += "Best regards,\n"
output += "Cloud FinOps Team\n"
output += "```\n\n"
output += "---\n\n"
return output
# --- 8. MAIN ORCHESTRATION ---
def run_audit(custom_query="", progress_callback=None, log_callback=None):
"""Main function called by app.py."""
import time
logger = ReasoningLogger(log_callback)
try:
logger.log("🚀", "Starting Sentinal Cloud Audit...")
if progress_callback: progress_callback(0.1, "🚀 Initializing...")
# Load Data
if not os.path.exists("billing_export.csv"):
return json.dumps({"report": "Error: billing_export.csv missing", "emails": ""})
df = pd.read_csv("billing_export.csv")
logger.log("📊", f"Loaded {len(df)} resources.")
# --- STEP 1: INTENT EXTRACTION ---
if custom_query:
intent = extract_query_intent(custom_query, logger)
df = filter_dataframe(df, intent, logger)
if len(df) == 0:
return json.dumps({"report": "
No resources matched your query.
", "emails": "", "reasoning": logger.get_logs()})
# --- STEP 2: HYBRID ANALYSIS ---
if progress_callback: progress_callback(0.3, "⚡ Running Python Heuristics...")
simple_results, complex_cases = classify_simple_cases(df, logger)
all_results = []
for sr in simple_results:
all_results.append({
"resource": sr['resource'],
"classification": {
"action": sr['action'],
"reason": sr['reason'],
"policy_rule": sr['policy_rule'],
"inferred_team": sr.get('inferred_team', TeamName.UNKNOWN)
}
})
# --- STEP 3: AI REASONING ---
if complex_cases:
if progress_callback: progress_callback(0.6, f"🤖 AI Analyzing {len(complex_cases)} complex cases...")
logger.log("📚", "Checking Policy via RAG...")
policy_ctx = query_policy("cloud resource lifecycle policy including deletion, hibernation, and retention rules")
# OPTIMIZATION: Prioritize Top 150 Most Expensive Resources
# Instead of batching everything (slow/expensive), we focus AI on high-impact items.
complex_cases.sort(key=lambda x: x.get('Cost_Monthly', 0), reverse=True)
TOP_N = 100
high_impact_batch = complex_cases[:TOP_N]
skipped_count = len(complex_cases) - len(high_impact_batch)
if skipped_count > 0:
logger.log("⚡", f"Prioritizing Top {TOP_N} spenders. Skipped {skipped_count} low-cost items.")
if progress_callback:
progress_callback(0.7, f"🤖 AI Analyzing Top {len(high_impact_batch)} High-Impact Cases...")
try:
resources_json = json.dumps(high_impact_batch)
ai_response = classify_resources(resources_json, policy_ctx)
for cls in ai_response.classifications:
res = next((r for r in high_impact_batch if r['ResourceID'] == cls.resource_id), None)
if res:
all_results.append({
"resource": res,
"classification": {
"action": cls.action,
"reason": cls.reason,
"policy_rule": cls.policy_rule,
"inferred_team": cls.inferred_team
}
})
except Exception as e:
logger.log("⚠️", f"AI Analysis Error: {e}")
# --- STEP 4: SAFETY CHECKS ---
logger.log("🛡️", "Applying final safety guardrails...")
final_results = []
for res in all_results:
safe_action = apply_safety_rules(res['resource'], res['classification'], logger)
res['classification']['action'] = safe_action
final_results.append(res)
# --- STEP 5: REPORTING ---
if progress_callback: progress_callback(0.9, "📝 Generatng Report...")
total_spend = df['Cost_Monthly'].sum()
report = generate_report(final_results, len(df), total_spend)
emails = generate_emails(final_results)
logger.log("✅", "Audit Complete.")
return json.dumps({
"report": report,
"emails": emails,
"reasoning": logger.get_logs()
})
except Exception as e:
import traceback
traceback.print_exc()
logger.log("❌", f"Fatal Error: {e}")
return json.dumps({"report": f"Error: {e}", "emails": "", "reasoning": logger.get_logs()})
if __name__ == "__main__":
# Test run
print(run_audit("find expensive databases"))
