249 lines
8.1 KiB
Python
Executable File
249 lines
8.1 KiB
Python
Executable File
#!/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()
|