scrape-console/ops_views.py

#!/usr/bin/env python3
"""Ops-vyer för Scrape Console (Octoparse-paritet): runs-historik, scheman,
proxy-pool, pipeline-steg och alerts. Endast stdlib. Endpoint: /ops."""

import json, os, re, shutil, subprocess, time
from collections import Counter

HOME = os.path.expanduser("~")
SF = os.path.join(HOME, "Documents/SF_Extractor")
CI = os.path.join(HOME, ".local/share/carinfo")
CACHE = {"ts": 0, "data": None}

UNITS = [
    "sf-multi-scrape.service",
    "carinfo-crawl.service",
    "siax-enrich.service",
    "siax-laptop-scanner.service",
]


def _runs(unit, days=3):
    try:
        out = subprocess.run(
            [
                "journalctl",
                "--user",
                "-u",
                unit,
                "--since",
                f"{days} days ago",
                "-o",
                "short-iso",
                "--no-pager",
            ],
            capture_output=True,
            text=True,
            timeout=30,
        ).stdout.splitlines()
    except Exception:
        return []
    runs, cur = [], None
    for line in out:
        m = re.match(
            r"(\S+) \S+ systemd\[\d+\]: (Started|Stopped|.*Failed with result '([^']+)')",
            line,
        )
        if not m:
            continue
        ts, ev = m.group(1), m.group(2)
        if ev == "Started":
            cur = {"start": ts[11:16], "day": ts[:10]}
        elif cur is not None and (ev == "Stopped" or "Failed" in ev):
            cur["end"] = ts[11:16]
            cur["result"] = m.group(3) if "Failed" in ev else "ok"
            runs.append(cur)
            cur = None
    if cur is not None:
        cur["end"] = "pågår"
        cur["result"] = "running"
        runs.append(cur)
    return runs[-12:]


def _schedules():
    out = []
    try:
        txt = subprocess.run(
            ["systemctl", "--user", "list-timers", "--no-pager"],
            capture_output=True,
            text=True,
            timeout=15,
        ).stdout
        for line in txt.splitlines():
            for key in ("p0lytics", "sf-scrape", "proton", "opencode-db", "sloso"):
                if key in line:
                    parts = line.split()
                    out.append(
                        {
                            "next": parts[0] + " " + parts[1]
                            if len(parts) > 2
                            else parts[0],
                            "unit": next((p for p in parts if ".timer" in p), "?"),
                        }
                    )
                    break
    except Exception:
        pass
    try:
        cron = subprocess.run(
            ["crontab", "-l"], capture_output=True, text=True, timeout=10
        ).stdout
        for line in cron.splitlines():
            if (
                line.strip()
                and not line.startswith("#")
                and any(k in line for k in ("p0lytics", "plane-sync", "valkompass"))
            ):
                out.append({"next": "cron", "unit": line[:110]})
    except Exception:
        pass
    return out


def _proxy():
    try:
        d = json.load(open(os.path.join(SF, "proxies/pool.json")))
        recs = d.get("records", [])
        now = time.time()
        fresh = sum(
            1
            for r in recs
            if r.get("lastChecked") and now - _ts(r["lastChecked"]) < 3600
        )
        healthy = sum(1 for r in recs if (r.get("consecutiveFailures") or 0) == 0)
        lat = [r["latencyMsEwma"] for r in recs if r.get("latencyMsEwma")]
        lat.sort()
        countries = Counter(r.get("country", "?") for r in recs).most_common(5)
        return {
            "total": len(recs),
            "fresh_1h": fresh,
            "healthy": healthy,
            "lat_p50": round(lat[len(lat) // 2]) if lat else -1,
            "top_countries": countries,
        }
    except Exception as e:
        return {"error": str(e)[:100]}


def _ts(iso):
    try:
        return (
            time.mktime(time.strptime(iso[:19], "%Y-%m-%dT%H:%M:%S"))
            - time.altzone * 0
            + (7200 if iso.endswith("Z") else 0)
        )
    except Exception:
        return 0


def _wc(path):
    try:
        with open(path, "rb") as f:
            return sum(1 for _ in f)
    except Exception:
        return -1


def _pipeline():
    stages = {}
    b = os.path.join(SF, "full_data/biluppgifter")
    stages["biluppgifter frontier (väntar)"] = _wc(os.path.join(b, "frontier.txt"))
    stages["biluppgifter crawled"] = _wc(os.path.join(b, "crawled.txt"))
    stages["biluppgifter prio-kö"] = _wc(os.path.join(b, "pending_priority.txt"))
    m = os.path.join(SF, "full_data/multi")
    try:
        for src in sorted(os.listdir(m)):
            cp = os.path.join(m, src, "crawled.txt")
            if os.path.exists(cp):
                stages[f"multi/{src} crawled"] = _wc(cp)
    except Exception:
        pass
    return stages


def ops():
    if time.time() - CACHE["ts"] < 300:
        return CACHE["data"]
    runs = {u: _runs(u) for u in UNITS}
    data = {
        "runs": runs,
        "schedules": _schedules(),
        "proxy": _proxy(),
        "pipeline": _pipeline(),
        "alerts": [],
        "disk_free_gb": 0,
    }
    try:
        data["disk_free_gb"] = round(shutil.disk_usage("/").free / 1e9, 1)
    except Exception:
        pass
    A = data["alerts"]
    if data["disk_free_gb"] and data["disk_free_gb"] < 6:
        A.append(
            {
                "level": "crit",
                "text": f"Disken nästan full: {data['disk_free_gb']} GB fritt",
            }
        )
    try:
        clog = (
            open(os.path.join(CI, "crawl.log"), errors="replace")
            .read()
            .splitlines()[-3:]
        )
        if clog and all("cooldown" in l for l in clog):
            A.append(
                {
                    "level": "warn",
                    "text": "car.info stampar i cooldown-loop (tunna modeller)",
                }
            )
    except Exception:
        pass
    try:
        if "not found" in open("/tmp/p0lytics-health.log", errors="replace").read():
            A.append(
                {
                    "level": "err",
                    "text": "p0lytics-cron trasig (.venv saknas) — ingen health-data",
                }
            )
    except Exception:
        pass
    CACHE.update(ts=time.time(), data=data)
    return data