hermes-agent/homelab-scripts
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial import: monitoring, power-management, maintenance, web scripts

Browse Source
This commit is contained in:
hermes-agent
2026-08-05 12:26:50 +02:00
commit 1936ded08f
10 changed files with 1306 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+35
View File
@@ -0,0 +1,35 @@
# homelab-scripts
Scripts de monitoring, maintenance et automatisation du homelab Keller.
## Structure
```
homelab-scripts/
├── monitoring/ # Métriques et rapports
│ ├── pve-monitor.py # Collecte métriques PVE toutes les 15min
│ ├── pve-report.py # Génération rapports PDF hebdo/mensuel
│ └── hermes-metrics.py # Métriques Hermes → Prometheus (1min)
├── power-management/ # Gestion d'alimentation
│ ├── host-power-start.py # Allumage serveur (11h)
│ └── host-power-stop.py # Extinction serveur (22h)
├── maintenance/ # Tâches de fond
│ ├── archive-old-sessions.sh # Archive sessions Hermes (3h)
│ └── check-obsolete.sh # Rapport éléments obsolètes (lundi)
└── web/ # Surveillance sites
├── site-test.py # Audit sécurité/UX laurentkeller.org
└── suivi-quotidien.py # Suivi quotidien Mon-site-web-photos
```
## Déploiement
Les scripts sont exécutés par les cron jobs Hermes. Pour les utiliser hors Hermes :
```bash
git clone https://git.keller-laurent.org/hermes-agent/homelab-scripts.git
cd homelab-scripts
```
## Licence
AGPL-3.0
maintenance/archive-old-sessions.sh
Executable
+46
View File
@@ -0,0 +1,46 @@
#!/usr/bin/env bash
# Archive sessions older than 7 days to S3, keep archives 30 days, then delete.
set -euo pipefail
ARCHIVE_DIR="${HERMES_HOME:-$HOME/.hermes}/archives"
RETENTION_DAYS=30
AGE_DAYS=7
S3_BUCKET="hermes-archives"
S3_ENDPOINT="https://s3.keller-laurent.org"
DATE=$(date +%F)
ARCHIVE_FILE="sessions-${DATE}.jsonl"
mkdir -p "$ARCHIVE_DIR"
# 1. Export sessions older than AGE_DAYS to a dated archive file
if hermes sessions export "$ARCHIVE_DIR/$ARCHIVE_FILE" 2>/dev/null; then
echo "✅ Exported sessions to $ARCHIVE_DIR/$ARCHIVE_FILE"
# 2. Upload to S3
mc cp "$ARCHIVE_DIR/$ARCHIVE_FILE" "minio/$S3_BUCKET/" 2>/dev/null && \
echo "✅ Uploaded to s3://$S3_BUCKET/$ARCHIVE_FILE"
else
echo "ℹ️ No sessions to export (or export failed)"
fi
# 3. Prune sessions older than AGE_DAYS from the DB
yes | hermes sessions prune --older-than "$AGE_DAYS" 2>/dev/null && \
echo "✅ Pruned sessions older than ${AGE_DAYS} days" || \
echo "ℹ️ No sessions to prune"
# 4. Delete local archive files older than RETENTION_DAYS
find "$ARCHIVE_DIR" -name 'sessions-*.jsonl' -type f -mtime +"$RETENTION_DAYS" -delete 2>/dev/null
echo "✅ Cleaned local archives older than ${RETENTION_DAYS} days"
# 5. Clean old archives from S3 (older than RETENTION_DAYS)
mc find "minio/$S3_BUCKET" --name "sessions-*.jsonl" --older-than "${RETENTION_DAYS}d" \
--exec "mc rm {}" 2>/dev/null || true
echo "✅ Cleaned S3 archives older than ${RETENTION_DAYS} days"
# 6. Report
echo ""
echo "=== Archives dans s3://$S3_BUCKET ==="
mc ls "minio/$S3_BUCKET/" 2>/dev/null || echo "(empty)"
echo ""
echo "=== Archives locales ==="
ls -lh "$ARCHIVE_DIR"/ 2>/dev/null || echo "(empty)"
maintenance/check-obsolete.sh
+88
View File
@@ -0,0 +1,88 @@
#!/bin/bash
# Script d'analyse hebdomadaire des éléments obsolètes sur le système
# Génère un rapport et l'envoie par email à admin@keller-laurent.org
set -euo pipefail
REPORT=$(mktemp)
trap 'rm -f "$REPORT"' EXIT
{
echo "═══════════════════════════════════════════════"
echo " RAPPORT HEBDOMADAIRE — ÉLÉMENTS OBSOLÈTES"
echo " $(date '+%Y-%m-%d %H:%M') — $(hostname)"
echo "═══════════════════════════════════════════════"
echo ""
# ── 1. Paquets orphelins (autoremove) ──
echo "── 1. PAQUETS ORPHELINS (autoremove) ──────────"
orphans=$(apt-get --just-print autoremove 2>&1 | grep "Remv" | sed 's/Remv //' | sed 's/ \[.*//' || true)
orphan_count=$(echo "$orphans" | grep -c . || true)
if [ "$orphan_count" -eq 0 ]; then
echo " Aucun paquet orphelin."
else
orphan_size=$(dpkg-query -Wf '${Package}\t${Installed-Size}\n' $orphans 2>/dev/null | awk '{sum+=$2} END {printf "%.0f", sum/1024}')
echo " Paquets: $orphan_count | Taille: ${orphan_size:-0} MB"
echo ""
echo " Top 10 plus gros :"
dpkg-query -Wf '${Package}\t${Installed-Size}\n' $orphans 2>/dev/null | sort -t$'\t' -k2 -rn | head -10 | awk '{printf " %-40s %d MB\n", $1, $2/1024}'
fi
echo ""
# ── 2. Anciens noyaux ──
echo "── 2. ANCIENS NOYAUX ──────────────────────────"
current_kernel=$(uname -r)
installed_kernels=$(dpkg -l 'linux-image-*' 2>/dev/null | grep '^ii' | awk '{print $2}' | sort -V || true)
old_kernels=$(echo "$installed_kernels" | grep -v "$(echo $current_kernel | sed 's/-[^-]*$//')" || true)
old_count=$(echo "$old_kernels" | grep -c . || true)
echo " Noyau actif : $current_kernel"
if [ "$old_count" -eq 0 ]; then
echo " Aucun ancien noyau."
else
echo " Anciens noyaux installés : $old_count"
echo "$old_kernels" | sed 's/^/ /'
fi
echo ""
# ── 3. Cache APT ──
echo "── 3. CACHE APT ───────────────────────────────"
apt_cache=$(du -sh /var/cache/apt/archives/ 2>/dev/null | awk '{print $1}')
echo " Taille : $apt_cache"
echo ""
# ── 4. Journaux systemd ──
echo "── 4. JOURNAUX SYSTEMD ────────────────────────"
journal_size=$(journalctl --disk-usage 2>&1 | grep -oP '[\d.]+[GMK]' || echo "?")
echo " Taille : $journal_size"
echo ""
# ── 5. Conteneurs/images Docker inutilisés ──
if command -v docker &>/dev/null; then
echo "── 5. DOCKER ──────────────────────────────────"
unused_containers=$(docker container ls -a --filter status=exited --filter status=created -q 2>/dev/null | wc -l)
dangling_images=$(docker images -f dangling=true -q 2>/dev/null | wc -l)
unused_volumes=$(docker volume ls -f dangling=true -q 2>/dev/null | wc -l)
echo " Conteneurs arrêtés : $unused_containers"
echo " Images dangling : $dangling_images"
echo " Volumes orphelins : $unused_volumes"
echo ""
fi
# ── 6. Espace disque ──
echo "── 6. ESPACE DISQUE ───────────────────────────"
df -h / | tail -1 | awk '{printf " Utilisé: %s / %s (%s libre)\n", $3, $2, $4}'
echo ""
# ── 7. Fichiers temporaires volumineux ──
echo "── 7. FICHIERS TEMPORAIRES VOLUMINEUX ─────────"
tmp_size=$(du -sh /tmp 2>/dev/null | awk '{print $1}')
echo " /tmp : $tmp_size"
echo ""
echo "═══════════════════════════════════════════════"
echo " FIN DU RAPPORT"
echo "═══════════════════════════════════════════════"
} > "$REPORT"
# Output for cron delivery
cat "$REPORT"
monitoring/hermes-metrics.py
+204
View File
@@ -0,0 +1,204 @@
#!/usr/bin/env python3
"""
Export Hermes Agent metrics to Prometheus textfile collector format.
Reads ~/.hermes/state.db and writes /var/lib/node_exporter/textfile_collector/hermes.prom.
Designed to run as a cron job (every 60s).
Metrics produced:
hermes_sessions_total{source,model} — session count
hermes_messages_total{source} — message count
hermes_tokens_input_total{source,model} — input tokens
hermes_tokens_output_total{source,model} — output tokens
hermes_tokens_cache_read_total{source,model} — cache read tokens
hermes_tokens_cache_write_total{source,model} — cache write tokens
hermes_tokens_reasoning_total{source,model} — reasoning tokens
hermes_cost_usd_total{source,model} — estimated cost in USD
hermes_api_calls_total{source,model} — API call count
hermes_sessions_active{source} — active sessions (not ended)
hermes_sessions_ended{source} — ended sessions in period
hermes_errors_total{source} — sessions with errors
hermes_db_size_bytes — state.db file size
hermes_scrape_duration_seconds — how long the scrape took
"""
import json
import os
import sqlite3
import time
from collections import defaultdict
from pathlib import Path
# --- Config ----------------------------------------------------------------
HERMES_HOME = Path(os.environ.get("HERMES_HOME", os.path.expanduser("~/.hermes")))
DB_PATH = HERMES_HOME / "state.db"
OUTPUT_PATH = Path("/var/lib/node_exporter/textfile_collector/hermes.prom")
# ---------------------------------------------------------------------------
# Labels that are safe to expose (no PII)
SAFE_SOURCES = {"cli", "webui", "telegram", "discord", "slack", "whatsapp",
"signal", "matrix", "teams", "email", "cron", "gateway", "acp"}
def _safe_source(source: str) -> str:
"""Normalise source to a known label or 'other'."""
s = source.lower().strip()
return s if s in SAFE_SOURCES else "other"
def _fmt_metric(name: str, value, labels: dict = None) -> str:
"""Format one Prometheus metric line."""
if labels:
label_str = ",".join(f'{k}="{v}"' for k, v in sorted(labels.items()))
return f"{name}{{{label_str}}} {value}"
return f"{name} {value}"
def scrape() -> str:
"""Read state.db and return Prometheus text."""
start = time.monotonic()
lines = []
lines.append("# HELP hermes_sessions_total Total Hermes sessions created")
lines.append("# TYPE hermes_sessions_total counter")
lines.append("# HELP hermes_messages_total Total messages across all sessions")
lines.append("# TYPE hermes_messages_total counter")
lines.append("# HELP hermes_tokens_input_total Total input tokens consumed")
lines.append("# TYPE hermes_tokens_input_total counter")
lines.append("# HELP hermes_tokens_output_total Total output tokens generated")
lines.append("# TYPE hermes_tokens_output_total counter")
lines.append("# HELP hermes_tokens_cache_read_total Total cache read tokens")
lines.append("# TYPE hermes_tokens_cache_read_total counter")
lines.append("# HELP hermes_tokens_cache_write_total Total cache write tokens")
lines.append("# TYPE hermes_tokens_cache_write_total counter")
lines.append("# HELP hermes_tokens_reasoning_total Total reasoning tokens")
lines.append("# TYPE hermes_tokens_reasoning_total counter")
lines.append("# HELP hermes_cost_usd_total Total estimated cost in USD")
lines.append("# TYPE hermes_cost_usd_total counter")
lines.append("# HELP hermes_api_calls_total Total API calls made")
lines.append("# TYPE hermes_api_calls_total counter")
lines.append("# HELP hermes_sessions_active Currently active sessions")
lines.append("# TYPE hermes_sessions_active gauge")
lines.append("# HELP hermes_sessions_ended Sessions ended in the last hour")
lines.append("# TYPE hermes_sessions_ended gauge")
lines.append("# HELP hermes_errors_total Sessions with non-nil end_reason (errors)")
lines.append("# TYPE hermes_errors_total counter")
lines.append("# HELP hermes_db_size_bytes Size of state.db on disk")
lines.append("# TYPE hermes_db_size_bytes gauge")
lines.append("# HELP hermes_scrape_duration_seconds Time spent scraping metrics")
lines.append("# TYPE hermes_scrape_duration_seconds gauge")
if not DB_PATH.exists():
lines.append(_fmt_metric("hermes_db_size_bytes", 0))
lines.append(_fmt_metric("hermes_scrape_duration_seconds", time.monotonic() - start))
return "\n".join(lines) + "\n"
# DB file size
lines.append(_fmt_metric("hermes_db_size_bytes", DB_PATH.stat().st_size))
try:
conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True)
conn.row_factory = sqlite3.Row
cur = conn.cursor()
now = time.time()
one_hour_ago = now - 3600
# --- Per-source, per-model aggregates (all time) ---
cur.execute("""
SELECT
COALESCE(NULLIF(source, ''), 'unknown') AS source,
COALESCE(NULLIF(model, ''), 'unknown') AS model,
COUNT(*) AS sessions,
SUM(message_count) AS messages,
SUM(input_tokens) AS input_tokens,
SUM(output_tokens) AS output_tokens,
SUM(cache_read_tokens) AS cache_read_tokens,
SUM(cache_write_tokens) AS cache_write_tokens,
SUM(reasoning_tokens) AS reasoning_tokens,
SUM(estimated_cost_usd) AS cost,
SUM(api_call_count) AS api_calls
FROM sessions
GROUP BY source, model
""")
for row in cur.fetchall():
src = _safe_source(row["source"])
model = row["model"]
labels = {"source": src, "model": model}
lines.append(_fmt_metric("hermes_sessions_total", row["sessions"] or 0, labels))
lines.append(_fmt_metric("hermes_messages_total", row["messages"] or 0, labels))
lines.append(_fmt_metric("hermes_tokens_input_total", row["input_tokens"] or 0, labels))
lines.append(_fmt_metric("hermes_tokens_output_total", row["output_tokens"] or 0, labels))
lines.append(_fmt_metric("hermes_tokens_cache_read_total", row["cache_read_tokens"] or 0, labels))
lines.append(_fmt_metric("hermes_tokens_cache_write_total", row["cache_write_tokens"] or 0, labels))
lines.append(_fmt_metric("hermes_tokens_reasoning_total", row["reasoning_tokens"] or 0, labels))
lines.append(_fmt_metric("hermes_cost_usd_total", row["cost"] or 0, labels))
lines.append(_fmt_metric("hermes_api_calls_total", row["api_calls"] or 0, labels))
# --- Active sessions (not ended) ---
cur.execute("""
SELECT COALESCE(NULLIF(source, ''), 'unknown') AS source, COUNT(*) AS cnt
FROM sessions WHERE ended_at IS NULL
GROUP BY source
""")
for row in cur.fetchall():
lines.append(_fmt_metric("hermes_sessions_active", row["cnt"],
{"source": _safe_source(row["source"])}))
# --- Sessions ended in last hour ---
cur.execute("""
SELECT COALESCE(NULLIF(source, ''), 'unknown') AS source, COUNT(*) AS cnt
FROM sessions WHERE ended_at >= ? AND ended_at IS NOT NULL
GROUP BY source
""", (one_hour_ago,))
for row in cur.fetchall():
lines.append(_fmt_metric("hermes_sessions_ended", row["cnt"],
{"source": _safe_source(row["source"])}))
# --- Sessions with error end_reason ---
cur.execute("""
SELECT COALESCE(NULLIF(source, ''), 'unknown') AS source, COUNT(*) AS cnt
FROM sessions
WHERE end_reason IS NOT NULL
AND end_reason NOT IN ('completed', 'user_exit', 'timeout', 'compression_split', '')
GROUP BY source
""")
for row in cur.fetchall():
lines.append(_fmt_metric("hermes_errors_total", row["cnt"],
{"source": _safe_source(row["source"])}))
conn.close()
except sqlite3.Error as e:
lines.append(f"# ERROR scraping state.db: {e}")
elapsed = time.monotonic() - start
lines.append(_fmt_metric("hermes_scrape_duration_seconds", elapsed))
return "\n".join(lines) + "\n"
def main():
try:
output = scrape()
OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
# Atomic write via temp file + rename
tmp = OUTPUT_PATH.with_suffix(".prom.tmp")
tmp.write_text(output)
tmp.rename(OUTPUT_PATH)
except Exception as e:
# Write an error metric so Prometheus can alert on it
error_output = (
"# HELP hermes_scrape_error 1 if the last scrape failed\n"
f"# TYPE hermes_scrape_error gauge\n"
f"hermes_scrape_error 1.0\n"
)
try:
OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
tmp = OUTPUT_PATH.with_suffix(".prom.tmp")
tmp.write_text(error_output)
tmp.rename(OUTPUT_PATH)
except Exception:
pass
raise
if __name__ == "__main__":
main()
monitoring/pve-monitor.py
+125
View File
@@ -0,0 +1,125 @@
#!/usr/bin/env python3
"""Collecte les métriques PVE (RAM, swap, load, disque, état VMs/CTs) et les
ajoute à un fichier CSV mensuel. Conçu pour être appelé toutes les 15 min
via cron (no_agent=True)."""
import csv, os, subprocess, json, sys
from datetime import datetime
OUTPUT_DIR = os.path.expanduser("~/.hermes/cron/output/pve-monitor")
os.makedirs(OUTPUT_DIR, exist_ok=True)
# Fichier CSV mensuel (ex: metrics-2026-07.csv)
now = datetime.now()
MONTH_CSV = os.path.join(OUTPUT_DIR, f"metrics-{now:%Y-%m}.csv")
# Champs du CSV (même format que l'ancien système)
FIELDS = [
"timestamp", "epoch",
"mem_total_mb", "mem_used_mb", "mem_avail_mb", "mem_pct",
"swap_total_mb", "swap_used_mb", "swap_pct",
"load_1", "load_5", "load_15",
"disk_total_gb", "disk_used_gb", "disk_avail_gb", "disk_pct",
"arc_size_mb",
]
def ssh_cmd(host, cmd):
"""Exécute une commande SSH et retourne stdout."""
r = subprocess.run(
["ssh", "-o", "ConnectTimeout=5", "-o", "StrictHostKeyChecking=no",
f"root@{host}", cmd],
capture_output=True, text=True, timeout=15
)
if r.returncode != 0:
raise RuntimeError(f"SSH error on {host}: {r.stderr.strip()}")
return r.stdout.strip()
def collect():
row = {}
row["timestamp"] = now.strftime("%Y-%m-%d %H:%M:%S")
row["epoch"] = str(int(now.timestamp()))
try:
# RAM / Swap
out = ssh_cmd("192.168.0.20", r"free -m | awk '/^Mem:/{print $2,$3,$7} /^Swap:/{print $2,$3}'")
lines = out.split("\n")
mem_total, mem_used, mem_avail = lines[0].split()
swap_total, swap_used = lines[1].split() if len(lines) > 1 else ("0", "0")
row["mem_total_mb"] = mem_total
row["mem_used_mb"] = mem_used
row["mem_avail_mb"] = mem_avail
row["mem_pct"] = f"{float(mem_used) / float(mem_total) * 100:.1f}" if float(mem_total) > 0 else "0"
row["swap_total_mb"] = swap_total
row["swap_used_mb"] = swap_used
row["swap_pct"] = f"{float(swap_used) / float(swap_total) * 100:.1f}" if float(swap_total) > 0 else "0"
# Load average
load = ssh_cmd("192.168.0.20", "cat /proc/loadavg | awk '{print $1,$2,$3}'")
l1, l5, l15 = load.split()
row["load_1"] = l1
row["load_5"] = l5
row["load_15"] = l15
# Disque /
disk = ssh_cmd("192.168.0.20", r"df -BG / | awk 'NR==2{print $2,$3,$4,$5}'")
parts = disk.replace("G", "").replace("%", "").split()
row["disk_total_gb"] = parts[0]
row["disk_used_gb"] = parts[1]
row["disk_avail_gb"] = parts[2]
row["disk_pct"] = parts[3]
# ARC ZFS
arc = ssh_cmd("192.168.0.20", r"cat /proc/spl/kstat/zfs/arcstats 2>/dev/null | grep -E '^size ' | awk '{print $3}' || echo 0")
row["arc_size_mb"] = f"{int(arc) // (1024*1024)}" if arc.isdigit() else "0"
# VMs
vm_out = ssh_cmd("192.168.0.20", "qm list 2>/dev/null | awk 'NR>1{print $1,$3}'")
for line in vm_out.split("\n"):
if not line.strip():
continue
parts = line.split()
vid = parts[0]
vstatus = parts[1] if len(parts) > 1 else "unknown"
# Récupérer la RAM allouée
mem = ssh_cmd("192.168.0.20", f"qm config {vid} 2>/dev/null | grep -i '^memory:' | awk '{{print $2}}' || echo 0")
row[f"vm_{vid}_mem_gb"] = f"{int(mem) // 1024}" if mem.isdigit() else "0"
row[f"vm_{vid}_status"] = vstatus
if f"vm_{vid}_mem_gb" not in FIELDS:
FIELDS.append(f"vm_{vid}_mem_gb")
if f"vm_{vid}_status" not in FIELDS:
FIELDS.append(f"vm_{vid}_status")
# CTs
ct_out = ssh_cmd("192.168.0.20", "pct list 2>/dev/null | awk 'NR>1{print $1,$2}'")
for line in ct_out.split("\n"):
if not line.strip():
continue
parts = line.split()
cid = parts[0]
cstatus = parts[1] if len(parts) > 1 else "unknown"
mem = ssh_cmd("192.168.0.20", f"pct config {cid} 2>/dev/null | grep -i '^memory:' | awk '{{print $2}}' || echo 0")
row[f"ct_{cid}_mem_gb"] = f"{int(mem) // 1024}" if mem.isdigit() else "0"
row[f"ct_{cid}_status"] = cstatus
if f"ct_{cid}_mem_gb" not in FIELDS:
FIELDS.append(f"ct_{cid}_mem_gb")
if f"ct_{cid}_status" not in FIELDS:
FIELDS.append(f"ct_{cid}_status")
except Exception as e:
print(f"ERREUR collecte: {e}", file=sys.stderr)
# Écrire quand même une ligne avec timestamp pour marquer l'échec
for f in FIELDS:
row.setdefault(f, "ERROR")
# Écrire dans le CSV mensuel
write_header = not os.path.isfile(MONTH_CSV)
with open(MONTH_CSV, "a", newline="") as f:
writer = csv.DictWriter(f, fieldnames=FIELDS)
if write_header:
writer.writeheader()
writer.writerow(row)
print(f"OK {now:%H:%M} — {os.path.basename(MONTH_CSV)} ({len(row)} champs)")
if __name__ == "__main__":
collect()
monitoring/pve-report.py
+248
View File
@@ -0,0 +1,248 @@
#!/usr/bin/env python3
"""Génère un rapport PDF de monitoring PVE pour une période donnée.
Usage:
python3 pve-report.py weekly # 7 derniers jours
python3 pve-report.py monthly # mois en cours complet
python3 pve-report.py range 2026-07-01 2026-07-31 # période custom
Le PDF est écrit dans ~/.hermes/cron/output/pve-monitor/
Le chemin du PDF est imprimé sur stdout."""
import csv, os, sys
from datetime import datetime, timedelta
from reportlab.lib import colors
from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.units import mm
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle, PageBreak
from reportlab.graphics.shapes import Drawing
from reportlab.graphics.charts.lineplots import LinePlot
OUTPUT_DIR = os.path.expanduser("~/.hermes/cron/output/pve-monitor")
def find_csv():
"""Trouve le CSV mensuel le plus récent dans OUTPUT_DIR."""
files = [f for f in os.listdir(OUTPUT_DIR) if f.startswith("metrics-") and f.endswith(".csv")]
if not files:
return None
files.sort(reverse=True)
return os.path.join(OUTPUT_DIR, files[0])
def load_data(csv_path, start_dt=None, end_dt=None):
"""Charge les lignes du CSV, filtre par période si spécifiée."""
with open(csv_path) as f:
reader = csv.DictReader(f)
rows = list(reader)
if not rows:
return []
if start_dt or end_dt:
filtered = []
for r in rows:
try:
ts = datetime.strptime(r["timestamp"], "%Y-%m-%d %H:%M:%S")
except (ValueError, KeyError):
continue
if start_dt and ts < start_dt:
continue
if end_dt and ts > end_dt:
continue
filtered.append(r)
return filtered
return rows
def safe_float(val, default=0.0):
try:
return float(val)
except (ValueError, TypeError):
return default
def generate_pdf(rows, period_label, pdf_filename):
"""Génère le PDF avec résumé, graphiques et tableau VMs/CTs."""
if len(rows) < 2:
print(f"ERREUR: Pas assez de données ({len(rows)} lignes)", file=sys.stderr)
return None
# Stats
mem_pcts = [safe_float(r.get("mem_pct", "0")) for r in rows]
swap_pcts = [safe_float(r.get("swap_pct", "0")) for r in rows]
loads = [safe_float(r.get("load_1", "0")) for r in rows]
disk_pcts = [safe_float(r.get("disk_pct", "0")) for r in rows]
first_ts = rows[0]["timestamp"]
last_ts = rows[-1]["timestamp"]
pdf_path = os.path.join(OUTPUT_DIR, pdf_filename)
doc = SimpleDocTemplate(
pdf_path, pagesize=A4,
title=f"Rapport Monitoring PVE - {period_label}",
author="Hermes Agent"
)
styles = getSampleStyleSheet()
title_style = ParagraphStyle("Title2", parent=styles["Title"], fontSize=18, spaceAfter=20)
subtitle_style = ParagraphStyle("Sub", parent=styles["Normal"], fontSize=10, textColor=colors.grey)
h2_style = ParagraphStyle("H2", parent=styles["Heading2"], fontSize=14, spaceBefore=20, spaceAfter=10)
normal = styles["Normal"]
elements = []
# Titre
elements.append(Paragraph(f"Rapport Monitoring PVE — {period_label}", title_style))
elements.append(Paragraph(f"Période: {first_ts} → {last_ts}", subtitle_style))
elements.append(Paragraph(f"Points collectés: {len(rows)} (toutes les 15 min)", subtitle_style))
elements.append(Spacer(1, 8*mm))
# Résumé
elements.append(Paragraph("Résumé de la période", h2_style))
summary_data = [
["Métrique", "Min", "Moyen", "Max", "Actuel"],
["RAM utilisée",
f"{min(mem_pcts):.1f}%", f"{sum(mem_pcts)/len(mem_pcts):.1f}%",
f"{max(mem_pcts):.1f}%", f"{mem_pcts[-1]:.1f}%"],
["Swap utilisé",
f"{min(swap_pcts):.1f}%", f"{sum(swap_pcts)/len(swap_pcts):.1f}%",
f"{max(swap_pcts):.1f}%", f"{swap_pcts[-1]:.1f}%"],
["Load (1 min)",
f"{min(loads):.2f}", f"{sum(loads)/len(loads):.2f}",
f"{max(loads):.2f}", f"{loads[-1]:.2f}"],
["Disque /",
f"{min(disk_pcts):.1f}%", f"{sum(disk_pcts)/len(disk_pcts):.1f}%",
f"{max(disk_pcts):.1f}%", f"{disk_pcts[-1]:.1f}%"],
]
t = Table(summary_data, colWidths=[80*mm, 30*mm, 30*mm, 30*mm, 30*mm])
t.setStyle(TableStyle([
("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#1a1a2e")),
("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
("FONTSIZE", (0, 0), (-1, -1), 9),
("ALIGN", (1, 0), (-1, -1), "CENTER"),
("GRID", (0, 0), (-1, -1), 0.5, colors.grey),
("ROWBACKGROUNDS", (0, 1), (-1, -1), [colors.white, colors.HexColor("#f0f0f5")]),
]))
elements.append(t)
elements.append(Spacer(1, 8*mm))
# Graphique RAM + Swap
elements.append(Paragraph("Évolution RAM et Swap", h2_style))
drawing = Drawing(500, 200)
lp = LinePlot()
lp.x = 50
lp.y = 30
lp.width = 400
lp.height = 150
lp.data = [
(list(range(len(rows))), mem_pcts),
(list(range(len(rows))), swap_pcts),
]
lp.lines[0].strokeColor = colors.HexColor("#4361ee")
lp.lines[0].strokeWidth = 2
lp.lines[1].strokeColor = colors.HexColor("#e63946")
lp.lines[1].strokeWidth = 2
n = len(rows)
step = max(1, n // 5)
lp.xValueAxis.valueSteps = list(range(0, n, step)) + ([n-1] if n > 1 else [])
lp.xValueAxis.labels.boxAnchor = 'ne'
lp.xValueAxis.labels.angle = 45
lp.xValueAxis.labels.fontSize = 7
lp.yValueAxis.valueSteps = [0, 25, 50, 75, 100]
lp.yValueAxis.labels.fontSize = 7
drawing.add(lp)
elements.append(drawing)
elements.append(Paragraph("■ RAM (%) ■ Swap (%)",
ParagraphStyle("Legend", parent=normal, fontSize=8, textColor=colors.grey)))
elements.append(Spacer(1, 5*mm))
# Graphique Load
elements.append(Paragraph("Charge CPU (Load Average 1 min)", h2_style))
drawing2 = Drawing(500, 200)
lp2 = LinePlot()
lp2.x = 50
lp2.y = 30
lp2.width = 400
lp2.height = 150
lp2.data = [(list(range(len(rows))), loads)]
lp2.lines[0].strokeColor = colors.HexColor("#2ec4b6")
lp2.lines[0].strokeWidth = 2
lp2.xValueAxis.valueSteps = list(range(0, n, step)) + ([n-1] if n > 1 else [])
lp2.xValueAxis.labels.boxAnchor = 'ne'
lp2.xValueAxis.labels.angle = 45
lp2.xValueAxis.labels.fontSize = 7
lp2.yValueAxis.labels.fontSize = 7
drawing2.add(lp2)
elements.append(drawing2)
elements.append(Spacer(1, 8*mm))
# Tableau des VMs/CTs (dernière ligne)
elements.append(Paragraph("État des VMs et Conteneurs (fin de période)", h2_style))
last = rows[-1]
vm_ct_headers = ["Ressource", "RAM (GB)", "Statut"]
vm_ct_data = [vm_ct_headers]
for key in sorted(last.keys()):
if key.endswith("_mem_gb"):
rid = key.split("_")[0] + " " + key.split("_")[1]
mem = last.get(key, "N/A")
status_key = key.replace("_mem_gb", "_status")
status = last.get(status_key, "N/A")
vm_ct_data.append([rid, str(mem), status])
t2 = Table(vm_ct_data, colWidths=[60*mm, 40*mm, 40*mm])
t2.setStyle(TableStyle([
("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#1a1a2e")),
("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
("FONTSIZE", (0, 0), (-1, -1), 8),
("ALIGN", (1, 0), (-1, -1), "CENTER"),
("GRID", (0, 0), (-1, -1), 0.5, colors.grey),
("ROWBACKGROUNDS", (0, 1), (-1, -1), [colors.white, colors.HexColor("#f0f0f5")]),
]))
elements.append(t2)
elements.append(Spacer(1, 8*mm))
# Footer
elements.append(Paragraph(
f"Généré automatiquement par Hermes Agent le {datetime.now().strftime('%d/%m/%Y à %H:%M')}",
ParagraphStyle("Footer", parent=normal, fontSize=8, textColor=colors.grey, alignment=1)))
doc.build(elements)
print(f"PDF généré: {pdf_path}")
return pdf_path
def main():
if len(sys.argv) < 2:
print("Usage: pve-report.py weekly|monthly", file=sys.stderr)
sys.exit(1)
mode = sys.argv[1]
csv_path = find_csv()
if not csv_path:
print("ERREUR: Aucun fichier CSV trouvé", file=sys.stderr)
sys.exit(1)
now = datetime.now()
if mode == "weekly":
start_dt = now - timedelta(days=7)
end_dt = now
period_label = f"S {start_dt.strftime('%d/%m')} → {end_dt.strftime('%d/%m/%Y')}"
pdf_name = f"rapport-pve-hebdo-{now.strftime('%Y-%m-%d')}.pdf"
elif mode == "monthly":
start_dt = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
end_dt = now
period_label = f"Mensuel {now.strftime('%B %Y')}"
pdf_name = f"rapport-pve-mensuel-{now.strftime('%Y-%m')}.pdf"
else:
print(f"ERREUR: Mode inconnu '{mode}'", file=sys.stderr)
sys.exit(1)
rows = load_data(csv_path, start_dt, end_dt)
if not rows:
print(f"ERREUR: Aucune donnée pour la période {period_label}", file=sys.stderr)
sys.exit(1)
pdf_path = generate_pdf(rows, period_label, pdf_name)
if pdf_path:
print(pdf_path)
if __name__ == "__main__":
main()
power-management/host-power-start.py
+73
View File
@@ -0,0 +1,73 @@
#!/usr/bin/env python3
"""Wrapper: power on pve2 (DL380p Gen8) via iLO4 Redfish API.
Fire-and-forget — sends power-on, does NOT wait for boot.
Used by cron job server-start (no_agent mode, 120s timeout)."""
import os, sys, json, subprocess
ENV_FILE = os.path.expanduser("~/.hermes/.env")
def get_env(key):
try:
with open(ENV_FILE) as f:
for line in f:
if line.startswith(key + "="):
return line.strip().split("=", 1)[1]
except FileNotFoundError:
pass
return None
def ilo_token():
url = get_env("ILO4_URL")
user = get_env("ILO4_USER")
pwd = get_env("ILO4_PASSWORD")
payload = '{"UserName":"%s","Password":"%s"}' % (user, pwd)
cmd = ["curl", "-s", "-k", "--max-time", "10",
"-D", "/dev/stderr", "-X", "POST",
"-H", "Content-Type: application/json",
"-d", payload,
url + "/redfish/v1/SessionService/Sessions/"]
r = subprocess.run(cmd, capture_output=True, timeout=15)
stderr = r.stderr.decode("utf-8", errors="replace")
for line in stderr.split("\n"):
if "X-Auth-Token" in line:
return line.split(":")[1].strip()
return None
url = get_env("ILO4_URL")
token = ilo_token()
if not token:
print("FAIL: iLO auth failed")
sys.exit(1)
hdr = "X-Auth-Token: " + token
r = subprocess.run([
"curl", "-s", "-k", "--max-time", "15",
"-X", "POST",
"-H", hdr,
"-H", "Content-Type: application/json",
"-d", '{"ResetType":"On"}',
url + "/redfish/v1/Systems/1/Actions/ComputerSystem.Reset/"
], capture_output=True, timeout=20)
# Check curl exit code
if r.returncode != 0:
print(f"FAIL: curl returned {r.returncode}")
sys.exit(1)
# Parse iLO response - check for actual error messages
# iLO wraps success in "ExtendedError" type with "Success" MessageID
stdout = r.stdout.decode("utf-8", errors="replace")
if stdout.strip():
try:
resp = json.loads(stdout)
msgs = resp.get("Messages", []) or (resp.get("error", {}) or {}).get("@Message.ExtendedInfo", [])
for m in msgs:
mid = m.get("MessageID", "")
if mid and "Success" not in mid and "OK" not in mid:
print(f"FAIL: iLO error: {mid}")
sys.exit(1)
except json.JSONDecodeError:
pass
print("Power-on command sent via iLO")
sys.exit(0)
power-management/host-power-stop.py
+8
View File
@@ -0,0 +1,8 @@
#!/usr/bin/env python3
"""Wrapper: graceful shutdown of pve2 (DL380p Gen8) via Proxmox API.
Silent on success, sends email on failure.
Used by cron job server-stop (no_agent mode)."""
import sys, os
script = os.path.join(os.path.dirname(__file__), "host-power.py")
os.execvp(sys.executable, [sys.executable, script, "stop"])
web/site-test.py
Executable
+355
View File
@@ -0,0 +1,355 @@
#!/usr/bin/env python3
"""
Test automatisé de sécurité et UX — laurentkeller.org
Tourne en cron, archive les résultats, détecte les régressions.
"""
import json
import subprocess
import sys
import os
from datetime import datetime, timezone
from pathlib import Path
# Configuration
SITE = "https://laurentkeller.org"
REPORT_DIR = Path(os.path.expanduser("~/.hermes/cron/output/site-tests"))
REPORT_DIR.mkdir(parents=True, exist_ok=True)
HISTORY_FILE = REPORT_DIR / "history.json"
def curl(url, **kwargs):
"""Wrapper curl avec timeout."""
cmd = ["curl", "-s", "--connect-timeout", "5", "--max-time", "10"]
for k, v in kwargs.items():
if v is True:
cmd.append(f"--{k.replace('_', '-')}")
elif v:
cmd.append(f"--{k.replace('_', '-')}")
cmd.append(str(v))
cmd.append(url)
try:
r = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
return {"exit": r.returncode, "stdout": r.stdout, "stderr": r.stderr}
except subprocess.TimeoutExpired:
return {"exit": -1, "stdout": "", "stderr": "timeout"}
def test_headers():
"""Teste les en-têtes de sécurité."""
r = curl(SITE, head=True)
headers_raw = r["stdout"].lower()
results = {
"strict-transport-security": "max-age=31536000; includesubdomains" in headers_raw,
"x-frame-options": "deny" in headers_raw,
"x-content-type-options": "nosniff" in headers_raw,
"content-security-policy": "default-src 'self'" in headers_raw,
"upgrade-insecure-requests": "upgrade-insecure-requests" in headers_raw,
"permissions-policy": "camera=(), microphone=(), geolocation=()" in headers_raw,
"referrer-policy": "strict-origin-when-cross-origin" in headers_raw,
"x-powered-by": "next.js" in headers_raw,
}
# Extraire le CSP complet
csp_line = [l for l in r["stdout"].split("\n") if "content-security-policy:" in l.lower()]
results["csp_img_src"] = "s3.keller-laurent.org" in (csp_line[0] if csp_line else "")
return results
def test_http_redirect():
"""Teste la redirection HTTP → HTTPS."""
r = curl("http://laurentkeller.org/", head=True, location=False)
has_redirect = "301" in r["stderr"] or "location: https://" in r["stdout"].lower()
return has_redirect
def test_tls():
"""Teste la version TLS."""
r = subprocess.run(
["openssl", "s_client", "-servername", "laurentkeller.org",
"-connect", "laurentkeller.org:443", "-tlsextdebug"],
capture_output=True, text=True, input="", timeout=10
)
tls_version = "unknown"
for line in r.stdout.split("\n"):
if "Protocol" in line:
tls_version = line.strip()
break
return tls_version
def test_sensitive_files():
"""Teste l'accès aux fichiers sensibles."""
paths = [
"/.env", "/.env.local", "/package.json", "/.git/config",
"/.git/HEAD", "/admin", "/api/auth-debug", "/config.json",
"/backup", "/wp-admin",
]
results = {}
for path in paths:
r = curl(f"{SITE}{path}", head=True)
code = r["stdout"].split("\n")[0] if r["stdout"] else "000"
# Extraire le code HTTP
if "HTTP/" in code:
code = code.split()[1]
elif r["exit"] != 0:
code = "000"
else:
code = "200" # fallback
results[path] = code
return results
def test_methods():
"""Teste les méthodes HTTP non autorisées."""
methods = ["PUT", "DELETE", "PATCH", "OPTIONS", "TRACE"]
results = {}
for method in methods:
r = subprocess.run(
["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}",
"-X", method, "--connect-timeout", "5", "--max-time", "10",
f"{SITE}/galerie"],
capture_output=True, text=True, timeout=12
)
results[method] = r.stdout.strip()
return results
def test_xss():
"""Teste XSS basique."""
r = curl(f"{SITE}/galerie?q=<script>alert(1)</script>")
blocked = "<script>alert" not in r["stdout"]
return blocked
def test_path_traversal():
"""Teste path traversal."""
r = curl(f"{SITE}/../../../etc/passwd", head=True)
code = r["stdout"].split("\n")[0] if r["stdout"] else "000"
if "HTTP/" in code:
code = code.split()[1]
return code
def test_cors():
"""Teste CORS."""
r = subprocess.run(
["curl", "-sI", "-H", "Origin: https://evil.com",
"--connect-timeout", "5", "--max-time", "10", SITE],
capture_output=True, text=True, timeout=12
)
has_cors = "access-control-allow-origin" in r.stdout.lower()
return not has_cors # True = pas de CORS ouvert = OK
def test_pages():
"""Teste l'accessibilité des pages."""
pages = [
"/", "/galerie", "/albums", "/agenda", "/wallpapers",
"/contact", "/mentions-legales", "/confidentialite",
]
results = {}
for page in pages:
r = curl(f"{SITE}{page}")
code = r["stdout"].split("\n")[0] if r["stdout"] else "000"
if "HTTP/" in code:
code = code.split()[1]
elif r["exit"] != 0:
code = "000"
else:
code = "200"
# Vérifier le contenu
content_ok = True
if page == "/albums" and "Aucun album" in r["stdout"]:
content_ok = False
if page == "/agenda" and "Aucun evenement" in r["stdout"]:
content_ok = False
if page == "/wallpapers" and "Aucun wallpaper" in r["stdout"]:
content_ok = False
if page == "/mentions-legales" and "A definir" in r["stdout"]:
content_ok = False
results[page] = {"code": code, "content_ok": content_ok}
return results
def test_contact_form():
"""Teste le formulaire contact."""
r = subprocess.run(
["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}",
"-X", "POST", "-H", "Content-Type: application/x-www-form-urlencoded",
"-d", "name=Test&email=test@test.com&message=Test automatique",
"--connect-timeout", "5", "--max-time", "10",
f"{SITE}/contact"],
capture_output=True, text=True, timeout=12
)
return r.stdout.strip()
def test_robots_sitemap():
"""Teste robots.txt et sitemap.xml."""
r_robots = curl(f"{SITE}/robots.txt")
r_sitemap = curl(f"{SITE}/sitemap.xml")
return {
"robots_ok": "Sitemap:" in r_robots["stdout"],
"sitemap_ok": "<urlset" in r_sitemap["stdout"],
}
def load_history():
"""Charge l'historique des tests."""
if HISTORY_FILE.exists():
with open(HISTORY_FILE) as f:
return json.load(f)
return {"tests": []}
def save_history(history):
"""Sauvegarde l'historique."""
with open(HISTORY_FILE, "w") as f:
json.dump(history, f, indent=2, default=str)
def compute_score(results):
"""Calcule un score de sécurité sur 100."""
score = 100
deductions = {
"headers": {
"strict-transport-security": 10,
"x-frame-options": 5,
"x-content-type-options": 5,
"content-security-policy": 15,
"upgrade-insecure-requests": 5,
"permissions-policy": 5,
"referrer-policy": 5,
},
"http_redirect": 10,
"xss_blocked": 10,
"cors_closed": 5,
"path_traversal": 5,
}
for h, ok in results.get("headers", {}).items():
if h in deductions["headers"] and not ok:
score -= deductions["headers"][h]
if not results.get("http_redirect"):
score -= deductions["http_redirect"]
if not results.get("xss_blocked"):
score -= deductions["xss_blocked"]
if not results.get("cors_closed"):
score -= deductions["cors_closed"]
if results.get("path_traversal") not in ("404", "405", "000"):
score -= deductions["path_traversal"]
return max(0, score)
def main():
print("🔍 Lancement des tests...")
results = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"url": SITE,
}
# Tests
print(" Headers...", end=" ")
results["headers"] = test_headers()
print("OK" if all(results["headers"].values()) else "PROBLEMES")
print(" HTTP→HTTPS...", end=" ")
results["http_redirect"] = test_http_redirect()
print("OK" if results["http_redirect"] else "FAIL")
print(" TLS...", end=" ")
results["tls"] = test_tls()
print(results["tls"])
print(" Fichiers sensibles...", end=" ")
results["sensitive_files"] = test_sensitive_files()
safe = all(v in ("404", "405", "000", "302") for v in results["sensitive_files"].values())
print("OK" if safe else "PROBLEMES")
print(" Méthodes HTTP...", end=" ")
results["methods"] = test_methods()
methods_ok = all(v == "405" for k, v in results["methods"].items() if k in ("PUT", "DELETE", "PATCH", "TRACE"))
print("OK" if methods_ok else "PROBLEMES")
print(" XSS...", end=" ")
results["xss_blocked"] = test_xss()
print("OK" if results["xss_blocked"] else "FAIL")
print(" Path traversal...", end=" ")
results["path_traversal"] = test_path_traversal()
print(results["path_traversal"])
print(" CORS...", end=" ")
results["cors_closed"] = test_cors()
print("OK" if results["cors_closed"] else "OUVERT!")
print(" Pages...", end=" ")
results["pages"] = test_pages()
pages_ok = all(v["code"] == "200" for v in results["pages"].values())
print("OK" if pages_ok else "PROBLEMES")
print(" Formulaire contact...", end=" ")
results["contact_form"] = test_contact_form()
print(results["contact_form"])
print(" Robots/Sitemap...", end=" ")
results["seo"] = test_robots_sitemap()
print("OK" if all(results["seo"].values()) else "PROBLEMES")
# Score
results["score"] = compute_score(results)
# Pages vides / problèmes UX
empty_pages = [p for p, v in results["pages"].items() if not v["content_ok"]]
results["empty_pages"] = empty_pages
results["methods_anomalies"] = {k: v for k, v in results["methods"].items() if v not in ("405", "400")}
# Historique
history = load_history()
history["tests"].append(results)
# Garder max 52 entrées (1 an si hebdo)
if len(history["tests"]) > 52:
history["tests"] = history["tests"][-52:]
save_history(history)
# Rapport
print(f"\n{'='*50}")
print(f"📊 SCORE DE SÉCURITÉ : {results['score']}/100")
print(f"{'='*50}")
if results["score"] < 80:
print("⚠️ Score bas — actions correctives nécessaires")
elif results["score"] >= 95:
print("✅ Excellent")
else:
print("👍 Correct")
# Régression ?
if len(history["tests"]) >= 2:
prev = history["tests"][-2]
diff = results["score"] - prev["score"]
if diff < 0:
print(f"📉 Régression de {abs(diff)} points depuis le dernier test")
elif diff > 0:
print(f"📈 Amélioration de {diff} points depuis le dernier test")
# Problèmes détectés
issues = []
if not all(results["headers"].values()):
issues.append("Headers de sécurité incomplets")
if not results["http_redirect"]:
issues.append("Pas de redirection HTTP→HTTPS")
if not results["xss_blocked"]:
issues.append("Vulnérabilité XSS détectée")
if not results["cors_closed"]:
issues.append("CORS ouvert")
if results["methods_anomalies"]:
issues.append(f"Méthodes HTTP anormales : {results['methods_anomalies']}")
if empty_pages:
issues.append(f"Pages vides : {empty_pages}")
if results["contact_form"] != "200":
issues.append(f"Formulaire contact anormal : {results['contact_form']}")
if issues:
print(f"\n🔴 Problèmes détectés ({len(issues)}) :")
for i, issue in enumerate(issues, 1):
print(f" {i}. {issue}")
else:
print("\n✅ Aucun problème détecté")
# Sauvegarde du rapport individuel
report_file = REPORT_DIR / f"rapport-{datetime.now().strftime('%Y-%m-%d_%H-%M')}.json"
with open(report_file, "w") as f:
json.dump(results, f, indent=2, default=str)
print(f"\n📁 Rapport sauvegardé : {report_file}")
return results
if __name__ == "__main__":
results = main()
# Sortie JSON pour le cron
print(f"\n---JSON---\n{json.dumps({'score': results['score'], 'issues': len([k for k,v in results.get('headers',{}).items() if not v]) + (0 if results.get('http_redirect') else 1) + (0 if results.get('xss_blocked') else 1)}, indent=2)}\n---ENDJSON---")
web/suivi-quotidien.py
Executable
+124
View File
@@ -0,0 +1,124 @@
#!/usr/bin/env python3
"""Suivi quotidien du projet Mon-site-web-photos v4.
Utilise gh CLI pour interroger le board GitHub et produit un résumé.
À exécuter via cronjob avec no_agent=True.
"""
import json
import subprocess
import sys
from datetime import datetime, timezone
def gh_run(*args: str) -> dict:
"""Run gh CLI and return parsed JSON."""
result = subprocess.run(
["gh", *args],
capture_output=True, text=True, timeout=30,
env={**__import__("os").environ, "GH_PROMPT_DISABLED": "1"},
)
if result.returncode != 0:
print(f"ERREUR gh {' '.join(args)}: {result.stderr.strip()}", file=sys.stderr)
return {}
return json.loads(result.stdout)
def get_issue_state(number: int) -> dict:
"""Get issue state and labels."""
return gh_run("issue", "view", str(number), "--json", "number,title,state,labels,updatedAt")
def format_daily_report() -> str:
"""Generate the daily status report."""
board = gh_run("project", "item-list", "2", "--owner", "lolo1580", "--limit", "50", "--format", "json")
items = board.get("items", [])
if not items:
return "⚠️ Impossible de récupérer les données du board."
# Collect issue numbers from board items
issue_numbers = set()
for item in items:
content = item.get("content", {})
num = content.get("number")
if num:
issue_numbers.add(num)
# Get details for each issue
issues = []
for num in sorted(issue_numbers):
detail = get_issue_state(num)
if detail and detail.get("number"):
# Merge board-level labels (more reliable)
board_labels = []
for item in items:
if item.get("content", {}).get("number") == num:
board_labels = item.get("labels", [])
break
detail["_board_labels"] = board_labels
issues.append(detail)
# Categorize
bugs = [i for i in issues if "bug" in i.get("_board_labels", [])]
features = [i for i in issues if "feature" in i.get("_board_labels", [])]
critical = [i for i in issues if "critical" in i.get("_board_labels", [])]
high = [i for i in issues if "high" in i.get("_board_labels", [])]
open_issues = [i for i in issues if i.get("state") == "OPEN"]
closed_issues = [i for i in issues if i.get("state") == "CLOSED"]
now = datetime.now(timezone.utc).strftime("%d/%m/%Y %H:%M UTC")
lines = [
f"📋 **Suivi quotidien — Mon-site-web-photos v4**",
f"_{now}_",
"",
"---",
"",
f"### 📊 Résumé",
"",
f"| Indicateur | Valeur |",
f"|------------|--------|",
f"| Issues ouvertes | {len(open_issues)} |",
f"| Issues fermées | {len(closed_issues)} |",
f"| Bugs | {len(bugs)} |",
f"| Features | {len(features)} |",
f"| 🔴 Critiques | {len(critical)} |",
f"| 🟡 Hautes | {len(high)} |",
"",
]
if critical:
lines.append("### 🔴 Bugs critiques")
for i in critical:
lines.append(f"- **#{i['number']}** {i['title']}")
lines.append("")
if open_issues:
lines.append("### 📌 Issues ouvertes")
lines.append("")
lines.append("| # | Titre | Type | Priorité |")
lines.append("|---|------|------|----------|")
for i in open_issues:
labels = i.get("_board_labels", [])
i_type = "Bug" if "bug" in labels else "Feature" if "feature" in labels else "Autre"
priority = "🔴" if "critical" in labels else "🟡" if "high" in labels else "🟢" if "medium" in labels else "⚪"
lines.append(f"| #{i['number']} | {i['title'][:55]} | {i_type} | {priority} |")
lines.append("")
if closed_issues:
lines.append("### ✅ Récemment fermés")
for i in closed_issues:
lines.append(f"- **#{i['number']}** {i['title']}")
lines.append("")
lines.append("---")
lines.append(f"🔗 **Board** : https://github.com/lolo1580/Mon-site-web-photos/projects/2")
lines.append(f"📁 **Repo** : https://github.com/lolo1580/Mon-site-web-photos")
return "\n".join(lines)
if __name__ == "__main__":
report = format_daily_report()
print(report)