mission-control-v2/scripts/blogwatcher-logger.py

#!/usr/bin/env python3
"""
Blogwatcher Logging-Skript

Protokolliert Blogwatcher-Aktivitäten im JSON-Lines-Format.
Logdatei: /home/hitonabi/.hermes/profiles/werkstatt/logs/blogwatcher.log
"""

import argparse
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path


def get_db_stats(db_path: str) -> dict:
    """Holt aktuelle DB-Statistiken (blog_count, article_count, unread_count)."""
    import sqlite3
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute("SELECT COUNT(*) FROM blogs")
    blog_count = cursor.fetchone()[0]
    cursor.execute("SELECT COUNT(*) FROM articles")
    article_count = cursor.fetchone()[0]
    cursor.execute("SELECT COUNT(*) FROM articles WHERE is_read = 0")
    unread_count = cursor.fetchone()[0]
    conn.close()
    return {
        "blog_count": blog_count,
        "article_count": article_count,
        "unread_count": unread_count,
    }


def ensure_log_dir(log_path: Path) -> None:
    """Erstellt Log-Verzeichnis, falls nötig."""
    log_path.parent.mkdir(parents=True, exist_ok=True)


def write_log_entry(log_path: Path, entry: dict) -> None:
    """Schreibt einen JSON-Lines-Eintrag in die Logdatei."""
    ensure_log_dir(log_path)
    with open(log_path, "a", encoding="utf-8") as f:
        f.write(json.dumps(entry, ensure_ascii=False) + "\n")


def get_timestamp() -> str:
    """Gibt aktuellen Timestamp im ISO 8601 UTC Format zurück."""
    return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")


def parse_bool(value: str) -> bool:
    """Parsed einen String zu Boolean (true/false, 1/0, yes/no)."""
    if isinstance(value, bool):
        return value
    return value.lower() in ("true", "1", "yes")


def main():
    parser = argparse.ArgumentParser(
        description="Blogwatcher Logging-Skript - protokolliert Aktivitäten im JSON-Lines-Format"
    )
    parser.add_argument(
        "--action",
        required=True,
        choices=[
            "scan", "add_blog", "remove_blog", "read_article", "unread_article",
            "read_all", "list_blogs", "list_articles", "import_opml"
        ],
        help="Aktion, die protokolliert wird"
    )
    parser.add_argument(
        "--duration",
        type=int,
        default=0,
        help="Dauer der Aktion in Millisekunden (0 bei Listenoperationen)"
    )
    parser.add_argument(
        "--success",
        type=str,
        default="true",
        help="Erfolg der Aktion (true/false, 1/0, yes/no)"
    )
    parser.add_argument(
        "--blog-count",
        type=int,
        default=None,
        help="Anzahl verfolgter Blogs (default: aus DB abgefragt)"
    )
    parser.add_argument(
        "--article-count",
        type=int,
        default=None,
        help="Anzahl gespeicherter Artikel (default: aus DB abgefragt)"
    )
    parser.add_argument(
        "--unread-count",
        type=int,
        default=None,
        help="Anzahl ungelesener Artikel (default: aus DB abgefragt)"
    )
    parser.add_argument(
        "--error-message",
        type=str,
        default="",
        help="Fehlerdetail bei success=false (optional)"
    )
    parser.add_argument(
        "--feed-url",
        type=str,
        default="",
        help="Feed-URL bei scan/add_blog (optional)"
    )
    parser.add_argument(
        "--blog-name",
        type=str,
        default="",
        help="Name des Blogs (optional)"
    )
    parser.add_argument(
        "--article-id",
        type=int,
        default=None,
        help="ID des Artikels bei read/unread (optional)"
    )
    parser.add_argument(
        "--source",
        type=str,
        default="cli",
        choices=["cli", "cron", "api_server", "telegram"],
        help="Quelle der Aktion (default: cli)"
    )
    parser.add_argument(
        "--session-id",
        type=str,
        default="",
        help="Session-ID (optional, aus Hermes)"
    )

    args = parser.parse_args()

    # DB-Stats abrufen, falls nicht übergeben
    db_path = "/home/hitonabi/.blogwatcher-cli/blogwatcher-cli.db"
    stats = get_db_stats(db_path)

    blog_count = args.blog_count if args.blog_count is not None else stats["blog_count"]
    article_count = args.article_count if args.article_count is not None else stats["article_count"]
    unread_count = args.unread_count if args.unread_count is not None else stats["unread_count"]

    # Log-Eintrag zusammenstellen
    entry = {
        "timestamp": get_timestamp(),
        "action": args.action,
        "blog_count": blog_count,
        "article_count": article_count,
        "unread_count": unread_count,
        "duration_ms": args.duration,
        "success": parse_bool(args.success),
        "source": args.source,
        "session_id": args.session_id,
    }

    # Optionale Felder nur aufnehmen, wenn gesetzt
    if args.error_message:
        entry["error_message"] = args.error_message
    if args.feed_url:
        entry["feed_url"] = args.feed_url
    if args.blog_name:
        entry["blog_name"] = args.blog_name
    if args.article_id is not None:
        entry["article_id"] = args.article_id

    # Log-Pfad definieren
    log_path = Path.home() / ".hermes" / "profiles" / "werkstatt" / "logs" / "blogwatcher.log"

    # Eintrag schreiben
    try:
        write_log_entry(log_path, entry)
    except Exception as e:
        print(f"ERROR: Could not write log entry: {e}", file=sys.stderr)
        sys.exit(1)

    # Bei Fehler Warnung an stderr
    if not parse_bool(args.success) and args.error_message:
        print(f"WARN: Action '{args.action}' failed: {args.error_message}", file=sys.stderr)

    sys.exit(0)


if __name__ == "__main__":
    main()