Spaces:
Running
Running
| from __future__ import annotations | |
| import json | |
| import os | |
| from datetime import datetime, timezone | |
| from typing import Any, Dict, List, Optional | |
| WORKLOG_VERSION = "worklog_v1_system_trace_2026_06_10" | |
| WORKLOG_FILE = os.getenv("SKYOS_WORKLOG_FILE", "skyos_worklog.jsonl") | |
| DEFAULT_RECOVERY_RULES = [ | |
| { | |
| "rule_id": "no_append_fixes_core_python", | |
| "title": "Keine Append-Fixes bei Python-Core-Dateien", | |
| "description": ( | |
| "Core-Dateien werden nicht durch Anhängen von Codeblöcken repariert. " | |
| "Änderungen erfolgen als vollständige Datei oder als klar definierter Austauschblock." | |
| ), | |
| "applies_to": [ | |
| "app.py", | |
| "router.py", | |
| "router_publish.py", | |
| "publish_gateway.py", | |
| "sheets_gateway.py", | |
| "crm_core.py", | |
| ], | |
| "reason": "Append-Fixes können doppelte Funktionen, leere def-Blöcke oder Syntaxfehler erzeugen.", | |
| "status": "active", | |
| }, | |
| { | |
| "rule_id": "split_large_router_files", | |
| "title": "Große Router-Dateien auslagern", | |
| "description": ( | |
| "Neue Funktionsbereiche werden in zweite Router oder Gateway-Dateien ausgelagert, " | |
| "statt router.py immer weiter aufzublähen." | |
| ), | |
| "applies_to": [ | |
| "router.py", | |
| "router_publish.py", | |
| ], | |
| "reason": "Zu große Dateien sind auf Handy/Hugging-Face-Editor fehleranfällig.", | |
| "status": "active", | |
| }, | |
| { | |
| "rule_id": "safe_import_fallback", | |
| "title": "Module dürfen den Haupt-Router nicht abstürzen lassen", | |
| "description": ( | |
| "Optionale Module werden mit try/except geladen. Wenn ein Modul kaputt ist, " | |
| "muss der Router trotzdem starten und den Fehler im Status anzeigen." | |
| ), | |
| "applies_to": [ | |
| "router.py", | |
| "router_publish.py", | |
| ], | |
| "reason": "Ein kaputtes Zusatzmodul darf nicht das ganze SkyOS stoppen.", | |
| "status": "active", | |
| }, | |
| ] | |
| # ============================================================ | |
| # CORE HELPERS | |
| # ============================================================ | |
| def now_iso() -> str: | |
| # KORRIGIERT: Python 3.13 konforme Zeitzonen-Generierung zur Vermeidung von Build-Warnungen | |
| return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") | |
| def safe_str(value: Any) -> str: | |
| if value is None: | |
| return "" | |
| return str(value) | |
| def normalize_entry(entry: Dict[str, Any]) -> Dict[str, Any]: | |
| return { | |
| "timestamp": entry.get("timestamp") or now_iso(), | |
| "action": entry.get("action", "unknown_action"), | |
| "module": entry.get("module", "unknown_module"), | |
| "file": entry.get("file"), | |
| "reason": entry.get("reason"), | |
| "result": entry.get("result"), | |
| "error": entry.get("error"), | |
| "recovery_action": entry.get("recovery_action"), | |
| "next_step": entry.get("next_step"), | |
| "risk_level": entry.get("risk_level", "low"), | |
| "status": entry.get("status", "logged"), | |
| "metadata": entry.get("metadata", {}), | |
| } | |
| def append_worklog_entry(entry: Dict[str, Any]) -> Dict[str, Any]: | |
| normalized = normalize_entry(entry) | |
| try: | |
| with open(WORKLOG_FILE, "a", encoding="utf-8") as file: | |
| file.write(json.dumps(normalized, ensure_ascii=False) + "\n") | |
| return { | |
| "status": "success", | |
| "module": "worklog", | |
| "action": "append_worklog_entry", | |
| "file": WORKLOG_FILE, | |
| "entry": normalized, | |
| } | |
| except Exception as exc: | |
| return { | |
| "status": "error", | |
| "module": "worklog", | |
| "action": "append_worklog_entry", | |
| "file": WORKLOG_FILE, | |
| "message": str(exc), | |
| "entry": normalized, | |
| } | |
| def read_worklog_entries(limit: int = 20) -> Dict[str, Any]: | |
| if limit <= 0: | |
| limit = 20 | |
| if not os.path.exists(WORKLOG_FILE): | |
| return { | |
| "status": "success", | |
| "module": "worklog", | |
| "action": "read_worklog_entries", | |
| "file": WORKLOG_FILE, | |
| "count": 0, | |
| "entries": [], | |
| "message": "Noch keine Worklog-Datei vorhanden.", | |
| } | |
| entries: List[Dict[str, Any]] = [] | |
| try: | |
| with open(WORKLOG_FILE, "r", encoding="utf-8") as file: | |
| lines = file.readlines() | |
| for line in lines[-limit:]: | |
| line = line.strip() | |
| if not line: | |
| continue | |
| try: | |
| entries.append(json.loads(line)) | |
| except Exception: | |
| entries.append({ | |
| "timestamp": now_iso(), | |
| "action": "read_malformed_entry", | |
| "module": "worklog", | |
| "status": "warning", | |
| "raw": line, | |
| }) | |
| return { | |
| "status": "success", | |
| "module": "worklog", | |
| "action": "read_worklog_entries", | |
| "file": WORKLOG_FILE, | |
| "count": len(entries), | |
| "entries": entries, | |
| } | |
| except Exception as exc: | |
| return { | |
| "status": "error", | |
| "module": "worklog", | |
| "action": "read_worklog_entries", | |
| "file": WORKLOG_FILE, | |
| "message": str(exc), | |
| "entries": [], | |
| } | |
| def get_last_worklog_entry() -> Dict[str, Any]: | |
| result = read_worklog_entries(limit=1) | |
| entries = result.get("entries", []) | |
| if not entries: | |
| return { | |
| "status": "empty", | |
| "module": "worklog", | |
| "message": "Kein Worklog-Eintrag vorhanden.", | |
| } | |
| return { | |
| "status": "success", | |
| "module": "worklog", | |
| "entry": entries[-1], | |
| } | |
| # ============================================================ | |
| # SPECIALIZED LOGGING | |
| # ============================================================ | |
| def log_router_recovery( | |
| error: str, | |
| recovery_action: str, | |
| result: str, | |
| next_step: str, | |
| file: str = "router.py", | |
| ) -> Dict[str, Any]: | |
| return append_worklog_entry({ | |
| "action": "router_recovery", | |
| "module": "router", | |
| "file": file, | |
| "reason": "Router oder Zusatzrouter wurde durch Fehler instabil.", | |
| "error": error, | |
| "recovery_action": recovery_action, | |
| "result": result, | |
| "next_step": next_step, | |
| "risk_level": "medium", | |
| "status": "recovered", | |
| }) | |
| def log_file_change( | |
| file: str, | |
| reason: str, | |
| result: str, | |
| next_step: Optional[str] = None, | |
| module: str = "skyos", | |
| risk_level: str = "low", | |
| ) -> Dict[str, Any]: | |
| return append_worklog_entry({ | |
| "action": "file_change", | |
| "module": module, | |
| "file": file, | |
| "reason": reason, | |
| "result": result, | |
| "next_step": next_step, | |
| "risk_level": risk_level, | |
| "status": "changed", | |
| }) | |
| def log_import_error( | |
| module: str, | |
| file: str, | |
| error: str, | |
| recovery_action: str, | |
| next_step: Optional[str] = None, | |
| ) -> Dict[str, Any]: | |
| return append_worklog_entry({ | |
| "action": "import_error", | |
| "module": module, | |
| "file": file, | |
| "reason": "Modul konnte nicht geladen werden.", | |
| "error": error, | |
| "recovery_action": recovery_action, | |
| "result": "Fehler wurde protokolliert und kann im Status sichtbar gemacht werden.", | |
| "next_step": next_step, | |
| "risk_level": "medium", | |
| "status": "warning", | |
| }) | |
| def log_decision( | |
| decision: str, | |
| reason: str, | |
| next_step: Optional[str] = None, | |
| module: str = "skyos", | |
| ) -> Dict[str, Any]: | |
| return append_worklog_entry({ | |
| "action": "decision", | |
| "module": module, | |
| "file": None, | |
| "reason": reason, | |
| "result": decision, | |
| "next_step": next_step, | |
| "risk_level": "low", | |
| "status": "decided", | |
| }) | |
| # ============================================================ | |
| # STATUS / TRACE OUTPUT | |
| # ============================================================ | |
| def get_recovery_rules() -> Dict[str, Any]: | |
| return { | |
| "status": "success", | |
| "module": "worklog", | |
| "version": WORKLOG_VERSION, | |
| "rules": DEFAULT_RECOVERY_RULES, | |
| "count": len(DEFAULT_RECOVERY_RULES), | |
| } | |
| def format_entry(entry: Dict[str, Any]) -> str: | |
| lines = [ | |
| f"- Zeit: {entry.get('timestamp')}", | |
| f"- Aktion: {entry.get('action')}", | |
| f"- Modul: {entry.get('module')}", | |
| ] | |
| if entry.get("file"): | |
| lines.append(f"- Datei: {entry.get('file')}") | |
| if entry.get("reason"): | |
| lines.append(f"- Grund: {entry.get('reason')}") | |
| if entry.get("error"): | |
| lines.append(f"- Fehler: {entry.get('error')}") | |
| if entry.get("recovery_action"): | |
| lines.append(f"- Recovery: {entry.get('recovery_action')}") | |
| if entry.get("result"): | |
| lines.append(f"- Ergebnis: {entry.get('result')}") | |
| if entry.get("next_step"): | |
| lines.append(f"- Nächster Schritt: {entry.get('next_step')}") | |
| return "\n".join(lines) | |
| def get_worklog_status(limit: int = 5) -> str: | |
| entries_result = read_worklog_entries(limit=limit) | |
| entries = entries_result.get("entries", []) | |
| lines = [ | |
| f"Worklog Version: {WORKLOG_VERSION}", | |
| "", | |
| "Worklog Status:", | |
| f"- Datei: {WORKLOG_FILE}", | |
| f"- Einträge geladen: {len(entries)}", | |
| f"- Speicherstatus: {entries_result.get('status')}", | |
| "", | |
| "Aktive Arbeitsregeln:", | |
| ] | |
| for rule in DEFAULT_RECOVERY_RULES: | |
| lines.append(f"- {rule.get('title')}") | |
| lines.append("") | |
| lines.append("Letzte Arbeitsschritte:") | |
| if not entries: | |
| lines.append("- Noch keine Einträge vorhanden.") | |
| else: | |
| for entry in entries: | |
| lines.append("") | |
| lines.append(format_entry(entry)) | |
| return "\n".join(lines) | |
| def handle_worklog_request(message: str) -> Dict[str, Any]: | |
| text = safe_str(message).lower().strip() | |
| if "regel" in text or "rules" in text or "arbeitsregel" in text: | |
| rules = get_recovery_rules() | |
| return { | |
| "status": "success", | |
| "module": "worklog", | |
| "mode": "rules", | |
| "rules": rules, | |
| "response": get_worklog_status(limit=3), | |
| } | |
| if "letzte" in text or "trace" in text or "worklog" in text or "arbeitsweise" in text: | |
| return { | |
| "status": "success", | |
| "module": "worklog", | |
| "mode": "status", | |
| "response": get_worklog_status(limit=5), | |
| } | |
| return { | |
| "status": "success", | |
| "module": "worklog", | |
| "mode": "default", | |
| "response": get_worklog_status(limit=5), | |
| } | |
| def seed_current_recovery_context() -> Dict[str, Any]: | |
| return append_worklog_entry({ | |
| "action": "system_recovery_rule_created", | |
| "module": "worklog", | |
| "file": "worklog.py", | |
| "reason": ( | |
| "Vorherige Core-Dateien wurden durch Append-Fixes und zu große Router-Dateien instabil." | |
| ), | |
| "error": ( | |
| "IndentationError in router.py und publish_gateway.py durch unvollständige oder angehängte Codeblöcke." | |
| ), | |
| "recovery_action": ( | |
| "Router wurde in Haupt-Router und Publish-Router aufgeteilt; " | |
| "Core-Dateien sollen künftig vollständig ersetzt oder kontrolliert blockweise geändert werden." | |
| ), | |
| "result": "Arbeitsweise wird ab jetzt nachvollziehbar protokolliert.", | |
| "next_step": "worklog.py in Router-Status oder eigenen Worklog-Intent einbinden.", | |
| "risk_level": "medium", | |
| "status": "seeded", | |
| }) | |
| if __name__ == "__main__": | |
| print(get_worklog_status()) | |