#!/usr/bin/env python3
"""Scrape Console v2: live-översikt + historik + kontroller (localhost-only).
GET / /state /history /log?src=multi&n=50 · POST /control (127.0.0.1 only)."""
import json, os, re, shutil, subprocess, time
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlparse, parse_qs
import data_views
import dash_views
import scraper_views
import ops_views
HOME = os.path.expanduser("~")
SF = os.path.join(HOME, "Documents/SF_Extractor")
CI = os.path.join(HOME, ".local/share/carinfo")
HIST = os.path.join(HOME, ".local/share/scrape-console/history.jsonl")
CACHE = {"ts": 0, "data": None}
def sys_show(unit, *props):
try:
out = subprocess.run(
["systemctl", "--user", "show", unit, "-p", ",".join(props)],
capture_output=True,
text=True,
timeout=10,
).stdout
d = {}
for line in out.splitlines():
if "=" in line:
k, v = line.split("=", 1)
d[k] = v
return d
except Exception:
return {}
def svc(unit, label, desc):
p = sys_show(
unit,
"ActiveState",
"MainPID",
"MemoryCurrent",
"CPUUsageNSec",
"ExecMainStartTimestamp",
)
mem = int(p.get("MemoryCurrent") or 0)
return {
"unit": unit,
"label": label,
"desc": desc,
"active": p.get("ActiveState", "?"),
"pid": p.get("MainPID", "?"),
"mem_mb": round(mem / 1024 / 1024, 1),
"cpu_s": round(int(p.get("CPUUsageNSec") or 0) / 1e9, 1),
"started": (p.get("ExecMainStartTimestamp") or "")[4:16],
}
def tail(path, n=8):
try:
with open(path, errors="replace") as f:
return "".join(f.readlines()[-n:])
except Exception as e:
return f"(kan inte läsa {path}: {e})"
def sf_progress():
out = {}
try:
with open(os.path.join(SF, "logs/multi_scrape.log"), errors="replace") as f:
for line in f:
m = re.match(
r"\[(.+?)\]\s+n=(\d+)\s+ok=(\d+)\s+fail=(\d+)\s+@(\d+)/(\d+)",
line.strip(),
)
if m:
src, n, ok, fail, pos, total = m.groups()
out[src] = {
"n": int(n),
"ok": int(ok),
"fail": int(fail),
"pos": int(pos),
"total": int(total),
}
except Exception as e:
out["_error"] = str(e)
return out
def du(path):
try:
return round(
sum(
os.path.getsize(os.path.join(dp, f))
for dp, _, fns in os.walk(path)
for f in fns
)
/ 1024
/ 1024,
1,
)
except Exception:
return -1
def collect():
try:
d = json.load(open(os.path.join(CI, "ckpt.json")))
ci_ckpt = {k: (len(v) if isinstance(v, list) else v) for k, v in d.items()}
except Exception as e:
ci_ckpt = {"_error": str(e)[:100]}
try:
spool = len(os.listdir(os.path.join(SF, "spool")))
except Exception:
spool = -1
prog = sf_progress()
mem = {
s["unit"]: s["mem_mb"]
for s in [
svc("sf-multi-scrape.service", "", ""),
svc("carinfo-crawl.service", "", ""),
]
}
data = {
"ts": time.strftime("%H:%M:%S"),
"scrapers": [
dict(
svc(
"sf-multi-scrape.service",
"SF multi-scrape",
"fordon: biluppgifter/fordonskollen/fordonssok/car.info",
),
progress=prog,
spool_files=spool,
data_mb={"full_data": du(os.path.join(SF, "full_data"))},
log=tail(os.path.join(SF, "logs/multi_scrape.log"), 6),
),
dict(
svc(
"carinfo-crawl.service", "car.info specs", "VEH-HIST-030, resumable"
),
ckpt=ci_ckpt,
log=tail(os.path.join(CI, "crawl.log"), 6),
),
dict(svc("siax-enrich.service", "SIAX enrich", "LinkedIn jobbsidor")),
dict(
svc(
"siax-laptop-scanner.service",
"SIAX laptop-scanner",
"jobb, residential IP",
)
),
],
"cron": {
"p0lytics_health": tail("/tmp/p0lytics-health.log", 2).strip(),
"p0lytics_rate": tail("/tmp/p0lytics-rate.log", 2).strip(),
},
}
# historik (1 prov/min)
try:
last = 0
if os.path.exists(HIST):
with open(HIST, errors="replace") as f:
for line in f:
pass
try:
last = json.loads(line).get("t", 0)
except Exception:
pass
now = int(time.time())
if now - last >= 55:
snap = {
"t": now,
"ok": {k: v["ok"] for k, v in prog.items() if not k.startswith("_")},
"mem": mem,
}
with open(HIST, "a") as f:
f.write(json.dumps(snap) + "\n")
# kapa till ~1500 rader
with open(HIST, errors="replace") as f:
lines = f.readlines()
if len(lines) > 1500:
open(HIST, "w").writelines(lines[-1500:])
except Exception:
pass
return data
HTML = """
Scrape Console live
"""
class H(BaseHTTPRequestHandler):
def log_message(self, *a):
pass
def _send(self, body, ctype):
self.send_response(200)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def do_GET(self):
u = urlparse(self.path)
if u.path == "/state":
if time.time() - CACHE["ts"] > 10:
CACHE["data"] = collect()
CACHE["ts"] = time.time()
self._send(json.dumps(CACHE["data"]).encode(), "application/json")
elif u.path == "/history":
try:
with open(HIST, errors="replace") as f:
lines = f.readlines()
data = [json.loads(l) for l in lines[-90:] if l.strip()]
except Exception:
data = []
self._send(json.dumps(data).encode(), "application/json")
elif u.path == "/log":
q = parse_qs(u.query)
src = q.get("src", ["multi"])[0]
n = min(int(q.get("n", ["50"])[0]), 200)
path = {
"multi": os.path.join(SF, "logs/multi_scrape.log"),
"carinfo": os.path.join(CI, "crawl.log"),
}.get(src, "")
self._send(tail(path, n).encode(), "text/plain; charset=utf-8")
elif u.path == "/data-page":
self._send(data_views.PAGE.encode(), "text/html; charset=utf-8")
elif u.path == "/dash":
self._send(dash_views.DASH_PAGE.encode(), "text/html; charset=utf-8")
elif u.path == "/datasets":
counts = data_views.counts()
out = [
{
"id": d["id"],
"label": d["label"],
"scraper": data_views.SCRAPER_OF.get(d["id"], ""),
"rows": counts.get(d["id"], {}).get("rows", -1),
}
for d in data_views.DATASETS
]
self._send(json.dumps(out).encode(), "application/json")
elif u.path == "/data":
q = parse_qs(u.query)
res = data_views.query(
q.get("ds", [""])[0],
q.get("q", [""])[0],
int(q.get("page", ["1"])[0]),
sort=q.get("sort", [None])[0],
sdir=q.get("dir", ["asc"])[0],
)
self._send(json.dumps(res, ensure_ascii=False).encode(), "application/json")
elif u.path == "/record":
q = parse_qs(u.query)
res = data_views.record_at(
q.get("ds", [""])[0], int(q.get("n", ["0"])[0]), q.get("q", [""])[0]
)
self._send(json.dumps(res, ensure_ascii=False).encode(), "application/json")
elif u.path == "/export":
q = parse_qs(u.query)
body, ctype, fname = data_views.export(
q.get("ds", [""])[0], q.get("q", [""])[0], q.get("fmt", ["csv"])[0]
)
self.send_response(200)
self.send_header("Content-Type", ctype)
self.send_header("Content-Disposition", "attachment; filename=" + fname)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
elif u.path.startswith("/scraper/"):
sid = u.path.split("/scraper/")[-1].split("?")[0]
sc = scraper_views.SCRAPERS.get(sid)
if not sc:
self.send_response(404); self.end_headers(); return
html = scraper_views.DETAIL.replace("__ID__", sid).replace(
"__UNIT__", sc["unit"]).replace("__LABEL__", sc["label"])
self._send(html.encode(), "text/html; charset=utf-8")
elif u.path == "/scraper-meta":
q = parse_qs(u.query)
sid = q.get("id", [""])[0]
sc = scraper_views.SCRAPERS.get(sid, {})
self._send(json.dumps({"key": sid, "label": sc.get("label", sid),
"desc": sc.get("desc", ""), "cmd": sc.get("cmd", ""),
"spec": sc.get("spec", [])}).encode(),
"application/json")
elif u.path == "/ops":
self._send(json.dumps(ops_views.ops()).encode(), "application/json")
else:
self._send(HTML.encode(), "text/html; charset=utf-8")
def do_POST(self):
if self.path != "/control":
self.send_response(404)
self.end_headers()
return
if self.client_address[0] not in ("127.0.0.1", "::1"):
self.send_response(403)
self.end_headers()
self.wfile.write(b"localhost only")
return
try:
d = json.loads(self.rfile.read(int(self.headers["Content-Length"] or 0)))
unit, action = d["unit"], d["action"]
assert re.match(r"^[a-z0-9@._-]+\.service$", unit) and action in (
"start",
"stop",
"restart",
)
r = subprocess.run(
["systemctl", "--user", action, unit],
capture_output=True,
text=True,
timeout=60,
)
msg = f"{action} {unit}: {'OK' if r.returncode == 0 else r.stderr[:200]}"
except Exception as e:
msg = f"fel: {e}"
body = msg.encode()
self.send_response(200)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
if __name__ == "__main__":
HTTPServer((os.getenv("HOST", "127.0.0.1"), int(os.getenv("PORT", "8899"))), H).serve_forever()