#!/usr/bin/env python3 """Evening Stand-Down — data collector for the 6:45pm weekday briefing. Fetches today's job activity, cron health, and Perth weather. Saves JSON data to ~/.hermes/dashboard/evening/ for the interactive HTML page. Outputs a chat-summary line for cron delivery.""" import json import os import sqlite3 import sys import urllib.request from datetime import datetime, timezone, timedelta PERTH_TZ = timezone(timedelta(hours=8)) # AWST DASHBOARD_DIR = os.path.expanduser("~/.hermes/dashboard") EVENING_DIR = os.path.join(DASHBOARD_DIR, "evening") def fetch_json(url): try: with urllib.request.urlopen(url, timeout=10) as r: return json.loads(r.read()) except Exception as e: return {"error": str(e)} def get_job_stats(): """Call the dashboard API for today's data.""" stats = fetch_json("http://localhost:9099/api/stats") jobs_resp = fetch_json("http://localhost:9099/api/jobs?limit=200") if "error" in jobs_resp: return {"error": jobs_resp["error"], "stats": stats} today = datetime.now(PERTH_TZ).strftime("%Y-%m-%d") jobs_list = jobs_resp.get("jobs", []) today_added = [j for j in jobs_list if j.get("date_added") == today] strong = [j for j in jobs_list if j.get("strong_match")] urgent_closing = [j for j in jobs_list if j.get("closing_status") in ("urgent", "today")] closing_all = [j for j in jobs_list if j.get("closing_date") or (j.get("closing_status") and j["closing_status"] not in ("none", "", None))] closing = sorted(closing_all, key=lambda j: j.get("closing_date", "9999-99-99"))[:15] by_status = {} for j in jobs_list: s = j.get("status", "Backlog") by_status[s] = by_status.get(s, 0) + 1 salaries_today = [j["salary"] for j in today_added if j.get("salary")] return { "total": stats.get("total", 0), "this_week": stats.get("this_week", 0), "strong_matches": len(strong), "today_added": len(today_added), "today_roles": [ { "name": j.get("name", ""), "role": j.get("role", ""), "salary": j.get("salary"), "source": j.get("source", ""), "url": j.get("url", ""), } for j in today_added[:8] ], "urgent_closing": len(urgent_closing), "closing_soon": [ { "name": j.get("name", ""), "role": j.get("role", ""), "closing_date": j.get("closing_date", ""), "closing_status": j.get("closing_status", ""), "salary": j.get("salary"), "source": j.get("source", ""), "strong_match": j.get("strong_match", False), "url": j.get("url", ""), } for j in closing ], "status_breakdown": by_status, "salary_range": {"min": min(salaries_today), "max": max(salaries_today)} if salaries_today else None, "refreshed_at": stats.get("refreshed_at", ""), } def get_cron_health(): """Check recent cron execution health from SQLite.""" db_path = os.path.expanduser("~/.hermes/cron/executions.db") if not os.path.exists(db_path): return {"error": "No executions DB found"} try: conn = sqlite3.connect(db_path) c = conn.cursor() yesterday = (datetime.now() - timedelta(days=1)).isoformat() c.execute(""" SELECT job_id, status, started_at, error FROM executions WHERE started_at > ? ORDER BY started_at DESC LIMIT 30 """, (yesterday,)) rows = c.fetchall() conn.close() errors = [r for r in rows if r[1] in ("error", "timeout")] return { "recent_runs": len(rows), "success": sum(1 for r in rows if r[1] == "success"), "errors": len(errors), "running": sum(1 for r in rows if r[1] == "running"), "error_details": [{"job_id": e[0], "error": (e[3] or "")[:150]} for e in errors[:3]], } except Exception as e: return {"error": str(e)} def get_weather(): """Fetch Perth weather via wttr.in (free, no key).""" try: with urllib.request.urlopen("https://wttr.in/Perth?format=j1", timeout=10) as r: data = json.loads(r.read()) current = data.get("current_condition", [{}])[0] forecast = data.get("weather", []) tomorrow = forecast[1] if len(forecast) > 1 else forecast[0] if forecast else {} return { "temp_c": current.get("temp_C", "?"), "feels_like": current.get("FeelsLikeC", "?"), "condition": current.get("weatherDesc", [{}])[0].get("value", "?"), "wind_kmh": current.get("windspeedKmph", "?"), "humidity": current.get("humidity", "?"), "tomorrow_high": tomorrow.get("maxtempC", "?"), "tomorrow_low": tomorrow.get("mintempC", "?"), "tomorrow_condition": ( tomorrow.get("hourly", [{}])[0].get("weatherDesc", [{}])[0].get("value", "?") if tomorrow.get("hourly") else "?" ), } except Exception as e: return {"error": str(e)} def main(): now = datetime.now(PERTH_TZ) is_weekday = now.weekday() < 5 jobs = get_job_stats() cron = get_cron_health() weather = get_weather() output = { "timestamp": now.isoformat(), "day": now.strftime("%A"), "date": now.strftime("%d %B %Y"), "jobs": jobs, "cron_health": cron, "weather": weather, } # Save JSON data for the interactive HTML page os.makedirs(EVENING_DIR, exist_ok=True) # Today's dated file date_str = now.strftime("%Y-%m-%d") dated_path = os.path.join(EVENING_DIR, f"{date_str}.json") with open(dated_path, "w") as f: json.dump(output, f, indent=2) # Overwrite latest.json latest_path = os.path.join(EVENING_DIR, "latest.json") with open(latest_path, "w") as f: json.dump(output, f, indent=2) # Output chat-friendly summary for cron delivery new_count = jobs.get("today_added", 0) if isinstance(jobs, dict) else 0 strong_count = jobs.get("strong_matches", 0) if isinstance(jobs, dict) else 0 urgent_count = jobs.get("urgent_closing", 0) if isinstance(jobs, dict) else 0 total = jobs.get("total", 0) if isinstance(jobs, dict) else 0 cron_ok = cron.get("errors", 0) == 0 if isinstance(cron, dict) else True temp = weather.get("temp_c", "?") if isinstance(weather, dict) else "?" condition = weather.get("condition", "?") if isinstance(weather, dict) else "?" print(f"🌆 Evening Stand-Down · {now.strftime('%A %d %B')}") print(f"📊 {new_count} new · {strong_count} strong · {urgent_count} urgent · {total} total") print(f"🌤️ {temp}°C · {condition}") print(f"{'✅' if cron_ok else '⚠️'} Cron: {'all healthy' if cron_ok else 'errors detected'}") print(f"🔗 https://hermes.kangaroo-eel.ts.net/jobs/evening.html") if __name__ == "__main__": main()