#!/usr/bin/env python3 """ Script de récapitulatif DÉTAILLÉ des mises à jour système Affiche packages Debian mis à jour """ import json import sys import os import re import argparse from datetime import datetime from typing import Dict, List, Optional, Tuple import requests from pathlib import Path CRONMASTER_LOGS_DIR = "/opt/container/cronmaster/data/logs" CONFIG_FILE = os.path.join(os.path.dirname(__file__), 'system-update-config.json') LOG_FILE = os.path.join(os.path.dirname(__file__), 'system-update-recap.log') def log(message: str, level: str = "INFO"): timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") log_message = f"[{timestamp}] [{level}] {message}\n" try: with open(LOG_FILE, 'a', encoding='utf-8') as f: f.write(log_message) except Exception: pass if level in ["ERROR", "INFO"]: print(log_message.strip()) def load_config() -> Dict: try: with open(CONFIG_FILE, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: log(f"Erreur config: {e}", "ERROR") sys.exit(1) def find_latest_update_job() -> Optional[Tuple[str, str]]: log("Recherche du dernier job...") logs_path = Path(CRONMASTER_LOGS_DIR) if not logs_path.exists(): return None latest_job = None latest_time = None for job_dir in sorted(logs_path.iterdir(), reverse=True): if not job_dir.is_dir(): continue for log_file in sorted(job_dir.iterdir(), reverse=True): if not log_file.suffix == '.log': continue try: try: with open(log_file, 'r', encoding='utf-8') as f: content = f.read() except UnicodeDecodeError: with open(log_file, 'r', encoding='latin-1') as f: content = f.read() if 'apt update' in content or 'apt full-upgrade' in content or 'Mise à jour système complète' in content or 'system-full-update' in content: timestamp_match = re.search(r'Timestamp\s*:\s*(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})', content) if timestamp_match: timestamp_str = timestamp_match.group(1) timestamp = datetime.strptime(timestamp_str, "%Y-%m-%d %H:%M:%S") if latest_time is None or timestamp > latest_time: latest_time = timestamp latest_job = (job_dir.name, str(log_file)) except Exception: continue if latest_job: log(f"Job trouvé: {latest_job[0]}") return latest_job def parse_package_upgrades(content: str) -> List[Dict]: packages = [] for line in content.split('\n'): unpacking = re.search(r'Unpacking\s+([^\s]+)\s+\(([^)]+)\)\s+over\s+\(([^)]+)\)', line) if unpacking: packages.append({ 'name': unpacking.group(1), 'old_version': unpacking.group(3), 'new_version': unpacking.group(2) }) continue setting_up = re.search(r'Paramétrage de\s+([^\s]+)\s+\(([^)]+)\)', line) if setting_up: pkg_name = setting_up.group(1) pkg_version = setting_up.group(2) if not any(p['name'] == pkg_name for p in packages): packages.append({ 'name': pkg_name, 'old_version': None, 'new_version': pkg_version }) return packages def parse_update_log(log_file_path: str) -> Dict: log(f"Parsing: {log_file_path}") try: try: with open(log_file_path, 'r', encoding='utf-8') as f: content = f.read() except UnicodeDecodeError: with open(log_file_path, 'r', encoding='latin-1') as f: content = f.read() except Exception as e: log(f"Erreur lecture: {e}", "ERROR") return {} result = { 'timestamp': None, 'duration': None, 'exit_code': None, 'status': None, 'packages_upgraded': 0, 'package_details': [] } timestamp_match = re.search(r'Timestamp\s*:\s*(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})', content) if timestamp_match: result['timestamp'] = timestamp_match.group(1) duration_match = re.search(r'Duration\s*:\s*(\d+)s', content) if duration_match: result['duration'] = int(duration_match.group(1)) exit_code_match = re.search(r'Exit Code\s*:\s*(\d+)', content) if exit_code_match: result['exit_code'] = int(exit_code_match.group(1)) status_match = re.search(r'Status\s*:\s*(\w+)', content) if status_match: result['status'] = status_match.group(1) result['package_details'] = parse_package_upgrades(content) result['packages_upgraded'] = len(result['package_details']) return result def send_to_discord(config: Dict, update_info: Dict) -> bool: log("Envoi Discord...") discord_config = config.get('discord', {}) webhook_url = discord_config.get('webhook_url', '') if not webhook_url or webhook_url == "WEBHOOK_URL_HERE": log("Webhook non configuré", "ERROR") return False try: status_emoji = "✅" if update_info['status'] == 'SUCCESS' else "❌" duration_str = f"{update_info['duration']}s" if update_info['duration'] < 60 else f"{update_info['duration']//60}m {update_info['duration']%60}s" fields = [] fields.append({ "name": "📊 Status", "value": f"{status_emoji} {update_info['status']}\n⏱️ {duration_str}", "inline": True }) pkg_details = update_info.get('package_details', []) if pkg_details: pkg_list = [] for pkg in pkg_details[:10]: if pkg.get('old_version'): pkg_list.append(f"• {pkg['name']}: {pkg['old_version']} → {pkg['new_version']}") else: pkg_list.append(f"• {pkg['name']}: {pkg['new_version']}") if len(pkg_details) > 10: pkg_list.append(f"... et {len(pkg_details) - 10} autres") fields.append({ "name": f"📦 Packages Debian ({len(pkg_details)} mis à jour)", "value": "\n".join(pkg_list), "inline": False }) else: fields.append({ "name": "📦 Packages Debian", "value": "Aucune mise à jour", "inline": False }) embed = { "title": f"{status_emoji} Mise à jour système complète", "description": f"Exécutée le {update_info['timestamp']}", "color": 5814783, "fields": fields, "timestamp": datetime.now().isoformat(), "footer": {"text": "Récapitulatif détaillé système"} } response = requests.post(webhook_url, json={"embeds": [embed]}, timeout=10) response.raise_for_status() log("Envoyé ✅") return True except Exception as e: log(f"Erreur Discord: {e}", "ERROR") return False def main(): parser = argparse.ArgumentParser(description='Récap détaillé mises à jour') parser.add_argument('--job-id', type=str) parser.add_argument('--auto', action='store_true') args = parser.parse_args() config = load_config() if args.job_id: job_dir = Path(CRONMASTER_LOGS_DIR) / args.job_id log_files = sorted(job_dir.glob('*.log'), reverse=True) log_file = str(log_files[0]) if log_files else None else: result = find_latest_update_job() if not result: log("Aucun job trouvé", "ERROR") sys.exit(1) _, log_file = result if not log_file: log("Log introuvable", "ERROR") sys.exit(1) update_info = parse_update_log(log_file) if not update_info.get('timestamp'): log("Parse impossible", "ERROR") sys.exit(1) success = send_to_discord(config, update_info) sys.exit(0 if success else 1) if __name__ == "__main__": main()