adolforeyna/FamReynaBrain
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

brain backup 2026-09-14

Browse Source
This commit is contained in:
Adolfo Reyna
2026-09-14 22:38:47 -04:00
parent ca346cc6e7
commit f0ead4f6fa
147 changed files with 73451 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
projects/network_usage/pihole_daily_report.py
+451
View File
@@ -0,0 +1,451 @@
#!/usr/bin/env python3
"""Create Pi-hole daily reports and retain a short Deco identity history.
Pi-hole is queried remotely over the user's existing SSH access. Deco snapshots
and completed reports are stored locally in a separate SQLite database. No
Pi-hole credentials or router session material is read or written here.
"""
from __future__ import annotations
import argparse
import json
import os
import shlex
import sqlite3
import subprocess
import sys
import time as time_module
from collections import defaultdict
from datetime import date, datetime, time, timedelta, timezone
from pathlib import Path
from typing import Dict, List, Optional
from zoneinfo import ZoneInfo
TZ = ZoneInfo("America/New_York")
DEFAULT_SERVER = "adolforeyna@192.168.68.110"
DEFAULT_PIHOLE_DB = "/home/adolforeyna/pihole/etc-pihole/pihole-FTL.db"
DEFAULT_STORAGE_DB = str(Path.home() / "brain/projects/network_usage/network_usage.sqlite3")
BLOCKED_STATUSES = (1, 4, 5, 6, 7, 8, 9, 10, 11, 18)
STATUS_NAMES = {
1: "gravity",
4: "regex",
5: "denylist",
6: "external-blocked-ip",
7: "external-blocked-null",
8: "external-blocked-nxra",
9: "gravity-cname",
10: "regex-cname",
11: "denylist-cname",
18: "external-blocked-ede15",
}
def parse_day(value: str, today: date) -> date:
if value == "today":
return today
if value == "yesterday":
return today - timedelta(days=1)
return date.fromisoformat(value)
def run_json(command: List[str], timeout: int) -> dict:
proc = subprocess.run(command, capture_output=True, text=True, timeout=timeout)
if proc.returncode:
detail = proc.stderr.strip() or proc.stdout.strip() or "exit %s" % proc.returncode
raise RuntimeError("command failed: %s" % detail[:500])
try:
return json.loads(proc.stdout)
except json.JSONDecodeError as exc:
raise RuntimeError("command returned invalid JSON: %s" % exc) from exc
def get_deco_clients(cli: str) -> List[dict]:
payload = run_json([cli, "deco", "clients", "--json"], timeout=120)
if not payload.get("ok"):
raise RuntimeError("Deco client lookup failed: %s" % payload)
return payload.get("result", {}).get("clients", [])
def connect_storage(path: str) -> sqlite3.Connection:
storage = Path(path).expanduser()
storage.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(storage), timeout=10)
conn.execute("PRAGMA foreign_keys=ON")
conn.execute("PRAGMA busy_timeout=10000")
conn.execute("PRAGMA journal_mode=WAL")
conn.executescript(
"""
CREATE TABLE IF NOT EXISTS deco_snapshots (
snapshot_id INTEGER PRIMARY KEY AUTOINCREMENT,
captured_at TEXT NOT NULL,
captured_epoch INTEGER NOT NULL,
client_count INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_deco_snapshots_epoch
ON deco_snapshots(captured_epoch);
CREATE TABLE IF NOT EXISTS deco_clients (
snapshot_id INTEGER NOT NULL REFERENCES deco_snapshots(snapshot_id) ON DELETE CASCADE,
ip TEXT NOT NULL,
mac TEXT,
hostname TEXT,
connection TEXT,
interface TEXT,
active INTEGER,
linked_deco_mac TEXT,
linked_deco_name TEXT,
linked_deco_role TEXT,
raw_json TEXT NOT NULL,
PRIMARY KEY(snapshot_id, ip, mac)
);
CREATE INDEX IF NOT EXISTS idx_deco_clients_ip
ON deco_clients(ip);
CREATE TABLE IF NOT EXISTS daily_reports (
report_day TEXT PRIMARY KEY,
generated_at TEXT NOT NULL,
report_json TEXT NOT NULL,
report_markdown TEXT NOT NULL
);
"""
)
conn.commit()
return conn
def save_deco_snapshot(storage_path: str, clients: List[dict], captured: Optional[datetime] = None) -> dict:
captured = captured or datetime.now(timezone.utc)
captured_epoch = int(captured.timestamp())
captured_text = captured.isoformat()
conn = connect_storage(storage_path)
try:
with conn:
cur = conn.execute(
"INSERT INTO deco_snapshots(captured_at,captured_epoch,client_count) VALUES(?,?,?)",
(captured_text, captured_epoch, len(clients)),
)
snapshot_id = cur.lastrowid
rows = []
for client in clients:
ip = str(client.get("ip") or "").strip()
if not ip:
continue
rows.append(
(
snapshot_id,
ip,
client.get("mac"),
client.get("hostname"),
client.get("connection"),
client.get("interface"),
1 if client.get("active") else 0,
client.get("linkedDecoMac"),
client.get("linkedDecoName"),
client.get("linkedDecoRole"),
json.dumps(client, separators=(",", ":"), sort_keys=True),
)
)
conn.executemany(
"""INSERT OR REPLACE INTO deco_clients
(snapshot_id,ip,mac,hostname,connection,interface,active,
linked_deco_mac,linked_deco_name,linked_deco_role,raw_json)
VALUES(?,?,?,?,?,?,?,?,?,?,?)""",
rows,
)
return {"snapshot_id": snapshot_id, "captured_at": captured_text, "clients": len(rows)}
finally:
conn.close()
def load_snapshot_map(storage_path: str, start_epoch: int, end_epoch: int) -> dict:
"""Return latest identity per IP observed during the report day.
If an IP was associated with multiple MACs during the day, the latest
record is selected but the mapping is explicitly marked ambiguous.
"""
conn = connect_storage(storage_path)
try:
rows = conn.execute(
"""SELECT s.captured_at,c.ip,c.mac,c.hostname,c.connection,c.interface,
c.active,c.linked_deco_mac,c.linked_deco_name,c.linked_deco_role
FROM deco_clients c JOIN deco_snapshots s ON s.snapshot_id=c.snapshot_id
WHERE s.captured_epoch>=? AND s.captured_epoch<?
ORDER BY s.captured_epoch DESC""",
(start_epoch, end_epoch),
).fetchall()
finally:
conn.close()
by_ip = {}
identities = defaultdict(set)
for row in rows:
captured_at, ip, mac, hostname, connection, interface, active, linked_mac, linked_name, linked_role = row
identities[ip].add(mac or "(unknown-mac)")
if ip in by_ip:
continue
by_ip[ip] = {
"ip": ip,
"mac": mac,
"hostname": hostname,
"connection": connection,
"interface": interface,
"active": bool(active),
"linkedDecoMac": linked_mac,
"linkedDecoName": linked_name,
"linkedDecoRole": linked_role,
"mapping_status": "historical_snapshot",
"snapshot_captured_at": captured_at,
}
for ip, deco in by_ip.items():
if len(identities[ip]) > 1:
deco["mapping_status"] = "multiple_identities_seen"
deco["mapping_identity_count"] = len(identities[ip])
return by_ip
def prune_old_snapshots(storage_path: str, before_epoch: int) -> int:
conn = connect_storage(storage_path)
try:
with conn:
cur = conn.execute("DELETE FROM deco_snapshots WHERE captured_epoch<?", (before_epoch,))
return cur.rowcount
finally:
conn.close()
def save_daily_report(storage_path: str, report: dict, markdown_text: str) -> None:
conn = connect_storage(storage_path)
try:
with conn:
conn.execute(
"""INSERT INTO daily_reports(report_day,generated_at,report_json,report_markdown)
VALUES(?,?,?,?)
ON CONFLICT(report_day) DO UPDATE SET
generated_at=excluded.generated_at,
report_json=excluded.report_json,
report_markdown=excluded.report_markdown""",
(
report["day"],
report["generated_at"],
json.dumps(report, indent=2),
markdown_text,
),
)
finally:
conn.close()
def get_pihole_data(server: str, db_path: str, start_epoch: int, end_epoch: int) -> dict:
blocked = ",".join(str(x) for x in BLOCKED_STATUSES)
remote = f'''import sqlite3,json
p={db_path!r}; start={start_epoch}; end={end_epoch}; blocked=({blocked},)
c=sqlite3.connect("file:"+p+"?mode=ro",uri=True)
c.row_factory=sqlite3.Row
base="""from query_storage q
join domain_by_id d on d.id=q.domain
join client_by_id cb on cb.id=q.client
where q.timestamp>=? and q.timestamp<?"""
def rows(sql, params=(start,end)):
return [dict(r) for r in c.execute(sql, params)]
client_sql="""select cb.ip as ip, coalesce(nullif(cb.name,''),'') as pihole_name,
count(*) as queries, sum(case when q.status in ({blocked}) then 1 else 0 end) as blocked,
count(distinct q.domain) as unique_domains, max(q.timestamp) as last_query
{{base}} group by cb.id order by queries desc""".format(blocked=','.join(str(x) for x in blocked),base=base)
domain_sql="""select cb.ip as ip, d.domain as domain, count(*) as queries,
sum(case when q.status in ({blocked}) then 1 else 0 end) as blocked,
sum(case when q.status not in ({blocked}) then 1 else 0 end) as allowed
{{base}} group by cb.ip,d.domain order by cb.ip,queries desc""".format(blocked=','.join(str(x) for x in blocked),base=base)
status_sql="""select q.status as status, count(*) as queries
from query_storage q where q.timestamp>=? and q.timestamp<? group by q.status order by q.status"""
print(json.dumps({{"clients":rows(client_sql),"domains":rows(domain_sql),"statuses":rows(status_sql),"window":{{"start":start,"end":end}}}},separators=(",",":")))'''
remote_cmd = "python3 -c " + shlex.quote(remote)
proc = subprocess.run(
["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=8", server, remote_cmd],
capture_output=True,
text=True,
timeout=180,
)
if proc.returncode:
detail = proc.stderr.strip() or proc.stdout.strip() or "exit %s" % proc.returncode
raise RuntimeError("Pi-hole query failed: %s" % detail[:500])
try:
return json.loads(proc.stdout)
except json.JSONDecodeError as exc:
raise RuntimeError("Pi-hole query returned invalid JSON: %s" % exc) from exc
def enrich(pihole: dict, deco_clients: List[dict], historical_map: dict) -> List[dict]:
by_ip = dict(historical_map)
for client in deco_clients:
ip = str(client.get("ip") or "")
if ip and ip not in by_ip:
copy = dict(client)
copy["mapping_status"] = "live_deco_lookup"
by_ip[ip] = copy
domains_by_ip = defaultdict(list)
for row in pihole["domains"]:
domains_by_ip[row["ip"]].append(row)
result = []
for row in pihole["clients"]:
ip = row["ip"]
deco = by_ip.get(ip)
visited = sorted(
(x for x in domains_by_ip[ip] if not x["blocked"]),
key=lambda x: (x["allowed"], x["queries"], x["domain"]),
reverse=True,
)
blocked = sorted(
(x for x in domains_by_ip[ip] if x["blocked"]),
key=lambda x: (x["blocked"], x["queries"], x["domain"]),
reverse=True,
)
result.append({
"ip": ip,
"pihole_name": row["pihole_name"],
"queries": row["queries"],
"blocked_queries": row["blocked"],
"block_rate_percent": round(100 * row["blocked"] / row["queries"], 2) if row["queries"] else 0,
"unique_domains": row["unique_domains"],
"last_query_epoch": row["last_query"],
"deco": deco or {
"hostname": None,
"mac": None,
"linkedDecoName": None,
"active": False,
"mapping_status": "not_in_snapshot_or_live_deco_list",
},
"top_visited": visited[:10],
"top_blocked": blocked[:10],
})
return result
def markdown(report: dict) -> str:
lines = [
f"# Pi-hole DNS usage — {report['day']}",
"",
f"Window: {report['window']['start_local']} to {report['window']['end_local']} ({report['timezone']})",
f"Generated: {report['generated_at']}",
f"Pi-hole: {report['pihole']['server']} (FTL database, read-only)",
f"Deco mapping source: {report['deco_mapping_source']}",
"",
"Blocked status codes counted: " + ", ".join(f"{k}={v}" for k, v in STATUS_NAMES.items()),
"",
"## Device summary",
"",
"| Device | IP | Deco/location | Queries | Blocked | Block rate | Top visited | Top blocked |",
"|---|---|---|---:|---:|---:|---|---|",
]
for d in report["devices"]:
deco = d["deco"]
label = deco.get("hostname") or d["pihole_name"] or "Unknown device"
location = deco.get("linkedDecoName") or "not mapped"
visited = d["top_visited"][0]["domain"] if d["top_visited"] else "—"
blocked = d["top_blocked"][0]["domain"] if d["top_blocked"] else "—"
lines.append(f"| {label} | {d['ip']} | {location} | {d['queries']:,} | {d['blocked_queries']:,} | {d['block_rate_percent']:.2f}% | `{visited}` | `{blocked}` |")
lines += ["", "## Per-device detail", ""]
for d in report["devices"]:
deco = d["deco"]
label = deco.get("hostname") or d["pihole_name"] or "Unknown device"
location = deco.get("linkedDecoName") or "not mapped"
lines += [
f"### {label} ({d['ip']})",
f"Location: {location}; MAC: {deco.get('mac') or 'unknown'}; mapping: {deco.get('mapping_status', 'unknown')}.",
f"Queries: {d['queries']:,}; blocked: {d['blocked_queries']:,}; block rate: {d['block_rate_percent']:.2f}%; unique domains: {d['unique_domains']:,}.",
"",
"Top visited (not blocked):",
]
lines += [f"- {x['domain']} — {x['queries']:,}" for x in d["top_visited"]] or ["- none"]
lines.append("Top blocked:")
lines += [f"- {x['domain']} — {x['blocked']:,} blocked of {x['queries']:,}" for x in d["top_blocked"]] or ["- none"]
lines.append("")
return "\n".join(lines)
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--date", default="today", help="today, yesterday, or YYYY-MM-DD")
parser.add_argument("--output-dir", default=str(Path.home() / "brain/projects/network_usage/reports"))
parser.add_argument("--storage-db", default=os.environ.get("NETWORK_USAGE_DB", DEFAULT_STORAGE_DB))
parser.add_argument("--server", default=os.environ.get("PIHOLE_SSH", DEFAULT_SERVER))
parser.add_argument("--db-path", default=os.environ.get("PIHOLE_DB_PATH", DEFAULT_PIHOLE_DB))
parser.add_argument("--deco-cli", default=os.environ.get("REYNA_CLI", "reyna-cli"))
parser.add_argument("--snapshot-only", action="store_true")
parser.add_argument("--prune", action="store_true", help="after saving the report, remove snapshots before today")
args = parser.parse_args()
if args.snapshot_only:
clients = get_deco_clients(args.deco_cli)
result = save_deco_snapshot(args.storage_db, clients)
print(json.dumps({"ok": True, "job": "deco_snapshot", "storage_db": args.storage_db, **result}, separators=(",", ":")))
return 0
today = datetime.now(TZ).date()
day = parse_day(args.date, today)
start_dt = datetime.combine(day, time.min, tzinfo=TZ)
end_dt = start_dt + timedelta(days=1)
now = datetime.now(TZ)
start_epoch = int(start_dt.timestamp())
end_epoch = int(end_dt.timestamp())
data = get_pihole_data(args.server, args.db_path, start_epoch, end_epoch)
historical_map = load_snapshot_map(args.storage_db, start_epoch, end_epoch)
live_clients = []
if day == today or not historical_map:
live_clients = get_deco_clients(args.deco_cli)
devices = enrich(data, live_clients, historical_map)
if historical_map:
mapping_source = "historical Deco snapshots plus live fallback"
else:
mapping_source = "live Deco lookup"
report = {
"schema_version": 2,
"day": day.isoformat(),
"timezone": str(TZ),
"generated_at": now.isoformat(),
"deco_mapping_source": mapping_source,
"window": {
"start_local": start_dt.isoformat(),
"end_local": end_dt.isoformat(),
"start_epoch": start_epoch,
"end_epoch": end_epoch,
},
"pihole": {
"server": args.server,
"database": args.db_path,
"blocked_statuses": list(BLOCKED_STATUSES),
"status_names": STATUS_NAMES,
"status_counts": data["statuses"],
},
"storage": {
"database": args.storage_db,
"historical_snapshot_client_ips": len(historical_map),
"live_deco_client_count": len(live_clients),
},
"deco_live_client_count": len(live_clients),
"devices": devices,
}
report_text = markdown(report)
save_daily_report(args.storage_db, report, report_text)
output_dir = Path(args.output_dir).expanduser()
output_dir.mkdir(parents=True, exist_ok=True)
json_path = output_dir / f"{day.isoformat()}.json"
md_path = output_dir / f"{day.isoformat()}.md"
json_path.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
md_path.write_text(report_text + "\n", encoding="utf-8")
pruned = 0
if args.prune:
current_start = int(datetime.combine(today, time.min, tzinfo=TZ).timestamp())
pruned = prune_old_snapshots(args.storage_db, current_start)
print(json.dumps({"ok": True, "day": report["day"], "devices": len(devices), "historical_snapshot_ips": len(historical_map), "live_deco_clients": len(live_clients), "pruned_snapshots": pruned, "storage_db": args.storage_db, "json": str(json_path), "markdown": str(md_path)}, separators=(",", ":")))
return 0
if __name__ == "__main__":
try:
raise SystemExit(main())
except Exception as exc:
print(json.dumps({"ok": False, "error": str(exc)}))
raise SystemExit(1)