#!/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