Mortdecai
7bfe862f53
Add discover_printer(), _mac_of(), _port_open(), _sweep_arp(), _cache_ip() with PRINTER_MAC/PRINTER_SUBNET/CACHE_FILE config + unit tests. Also adds conftest.py to work around broken symlink that blocked pytest collection. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
539 lines
20 KiB
Python
539 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
POS Daily Briefing — Epson TM-m30
|
|
Fetches unread news from FreshRSS, summarizes via Ollama, prints ESC/POS receipt.
|
|
"""
|
|
import os
|
|
import re
|
|
import json
|
|
import socket
|
|
import subprocess
|
|
import requests
|
|
import yfinance as yf
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
# ── Config ────────────────────────────────────────────────────────────────────
|
|
CONFIG_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.json")
|
|
with open(CONFIG_PATH) as f:
|
|
config = json.load(f)
|
|
|
|
FRESHRSS_URL = config["freshrss_url"]
|
|
FRESHRSS_USER = config["freshrss_user"]
|
|
FRESHRSS_TOKEN = config["freshrss_api_key"]
|
|
OLLAMA_URL = config["ollama_url"]
|
|
OLLAMA_MODEL = config["ollama_model"]
|
|
YOURLS_URL = config["yourls_url"]
|
|
YOURLS_USER = config["yourls_user"]
|
|
YOURLS_PASS = config["yourls_pass"]
|
|
GRAFANA_URL = config["grafana_url"]
|
|
GRAFANA_TOKEN = config["grafana_token"]
|
|
REMOTE_HOST = config["remote_host"]
|
|
REMOTE_PASS = config["remote_pass"]
|
|
PRINTER_IP = config["printer_ip"]
|
|
PRINTER_PORT = int(config.get("printer_port", 9100))
|
|
|
|
from pathlib import Path
|
|
|
|
PRINTER_MAC = config.get("printer_mac", "").lower()
|
|
PRINTER_SUBNET = config.get("printer_subnet", "192.168.0")
|
|
CACHE_DIR = Path.home() / ".cache" / "pos-briefing"
|
|
CACHE_FILE = CACHE_DIR / "last_ip"
|
|
|
|
EASTERN_TZ = ZoneInfo("America/New_York")
|
|
|
|
# TM-m30 80mm paper, Font B — column width measured from test print
|
|
COLS = 57
|
|
|
|
# ── FreshRSS ──────────────────────────────────────────────────────────────────
|
|
def fetch_news():
|
|
try:
|
|
resp = requests.post(
|
|
f"{FRESHRSS_URL}/api/greader.php/accounts/ClientLogin",
|
|
data={"Email": FRESHRSS_USER, "Passwd": FRESHRSS_TOKEN},
|
|
timeout=15
|
|
)
|
|
if resp.status_code != 200:
|
|
print(f"[!] Login failed: {resp.status_code}")
|
|
return [], None
|
|
auth_token = next(
|
|
(l.split("=", 1)[1] for l in resp.text.splitlines() if l.startswith("Auth=")),
|
|
None
|
|
)
|
|
if not auth_token:
|
|
print("[!] No auth token in login response")
|
|
return [], None
|
|
except Exception as e:
|
|
print(f"[!] Login exception: {e}")
|
|
return [], None
|
|
|
|
headers = {"Authorization": f"GoogleLogin auth={auth_token}"}
|
|
try:
|
|
r = requests.get(
|
|
f"{FRESHRSS_URL}/api/greader.php/reader/api/0/stream/contents/user/-/state/com.google/reading-list",
|
|
params={"n": 100, "xt": "user/-/state/com.google/read", "output": "json"},
|
|
headers=headers,
|
|
timeout=15
|
|
)
|
|
if r.status_code != 200:
|
|
return [], auth_token
|
|
items = []
|
|
for item in r.json().get("items", []):
|
|
long_id = item.get("id", "")
|
|
summary = (item.get("summary", {}).get("content", "")
|
|
or item.get("content", {}).get("content", ""))
|
|
url = ""
|
|
if item.get("canonical"):
|
|
url = item["canonical"][0].get("href", "")
|
|
if not url and item.get("alternate"):
|
|
url = item["alternate"][0].get("href", "")
|
|
if not url:
|
|
item_id = long_id.split("/")[-1] if "/" in long_id else ""
|
|
url = f"https://fresh.sethpc.xyz/i/?c=entry&a=read&id={item_id}"
|
|
items.append({
|
|
"id": long_id,
|
|
"title": item.get("title", ""),
|
|
"summary": re.sub('<[^<]+?>', '', summary).strip(),
|
|
"url": url,
|
|
"source": item.get("origin", {}).get("title", "Unknown"),
|
|
})
|
|
return items, auth_token
|
|
except Exception as e:
|
|
print(f"[!] Fetch exception: {e}")
|
|
return [], auth_token
|
|
|
|
|
|
def mark_read(item_ids, auth_token):
|
|
if not item_ids or not auth_token:
|
|
return
|
|
headers = {"Authorization": f"GoogleLogin auth={auth_token}"}
|
|
try:
|
|
t = requests.get(
|
|
f"{FRESHRSS_URL}/api/greader.php/reader/api/0/token",
|
|
headers=headers, timeout=10
|
|
).text.strip()
|
|
requests.post(
|
|
f"{FRESHRSS_URL}/api/greader.php/reader/api/0/edit-tag",
|
|
headers=headers,
|
|
data={"i": item_ids, "a": "user/-/state/com.google/read", "T": t},
|
|
timeout=10
|
|
)
|
|
print(f"Marked {len(item_ids)} articles as read.")
|
|
except Exception as e:
|
|
print(f"[!] mark_read error: {e}")
|
|
|
|
|
|
# ── Ollama ────────────────────────────────────────────────────────────────────
|
|
def summarize_news(items, count=5, length="30-50 words"):
|
|
if not items:
|
|
return []
|
|
|
|
print(f"Phase 1: Selecting top {count} from {len(items)} items...")
|
|
titles_input = "\n".join([f"ID {i} [{item['source']}]: {item['title']}"
|
|
for i, item in enumerate(items)])
|
|
prompt_select = (
|
|
f"Select exactly {count} stories for a daily briefing:\n"
|
|
"1. Priorities: Tech, Science, World News, General, Health.\n"
|
|
"2. Variety: Max 2 stories per source.\n"
|
|
"3. Entertainment: Max 2 total.\n"
|
|
"4. Omit: Lottery, minor crime.\n"
|
|
"5. Sports: Football (College/NFL) only.\n"
|
|
f"Format: [0, 5, 12, ...]\n\nHEADLINES:\n{titles_input}"
|
|
)
|
|
selected_indices = []
|
|
try:
|
|
resp = requests.post(
|
|
f"{OLLAMA_URL}/api/generate",
|
|
json={"model": OLLAMA_MODEL, "prompt": prompt_select, "stream": False,
|
|
"options": {"num_ctx": 16384, "temperature": 0.1}, "format": "json"},
|
|
timeout=300
|
|
).json()
|
|
data = json.loads(resp.get("response", "[]"))
|
|
selected_indices = data if isinstance(data, list) else list(data.values())[0]
|
|
except Exception as e:
|
|
print(f"[!] Selection error: {e}")
|
|
selected_indices = list(range(min(count, len(items))))
|
|
|
|
candidates = []
|
|
source_counts = {}
|
|
for idx in selected_indices:
|
|
if isinstance(idx, int) and 0 <= idx < len(items):
|
|
item = items[idx]
|
|
src = item["source"]
|
|
if source_counts.get(src, 0) < 2:
|
|
candidates.append(item)
|
|
source_counts[src] = source_counts.get(src, 0) + 1
|
|
while len(candidates) < count:
|
|
for item in items:
|
|
if item not in candidates and source_counts.get(item["source"], 0) < 2:
|
|
candidates.append(item)
|
|
source_counts[item["source"]] = source_counts.get(item["source"], 0) + 1
|
|
if len(candidates) == count:
|
|
break
|
|
break
|
|
|
|
print(f"Phase 2: Summarizing {len(candidates)} items ({length})...")
|
|
content_input = "".join([
|
|
f"ARTICLE {i} ({item['title']}):\n{item['summary'][:2000]}\n\n"
|
|
for i, item in enumerate(candidates)
|
|
])
|
|
prompt_summary = (
|
|
"You are a professional news editor writing a daily briefing for a "
|
|
"receipt printer.\n"
|
|
f"For EACH article, write a concise factual summary ({length}).\n"
|
|
"No preamble, no opinion, do not write \"this article\".\n"
|
|
"LANGUAGE RULE: Use English. Translate non-English/Spanish articles.\n"
|
|
"Return STRICTLY a JSON list: [{\"id\": 0, \"summary\": \"...\"}, ...]\n\n"
|
|
f"CONTENT:\n{content_input}"
|
|
)
|
|
try:
|
|
resp = requests.post(
|
|
f"{OLLAMA_URL}/api/generate",
|
|
json={"model": OLLAMA_MODEL, "prompt": prompt_summary, "stream": False,
|
|
"options": {"num_ctx": 16384, "temperature": 0.3}},
|
|
timeout=600
|
|
).json()
|
|
raw = resp.get("response", "")
|
|
match = re.search(r'\[.*\]', raw, re.DOTALL)
|
|
if match:
|
|
summaries = json.loads(match.group(0))
|
|
else:
|
|
parsed = json.loads(raw)
|
|
summaries = parsed if isinstance(parsed, list) else list(parsed.values())[0]
|
|
final = []
|
|
for s in summaries:
|
|
idx = s.get("id")
|
|
if idx is not None and 0 <= idx < len(candidates):
|
|
candidates[idx]["ai_summary"] = s.get("summary")
|
|
final.append(candidates[idx])
|
|
return final
|
|
except Exception as e:
|
|
print(f"[!] Summarize parse error: {e} — using raw summaries")
|
|
for c in candidates:
|
|
c["ai_summary"] = c.get("summary", "")[:200]
|
|
return candidates
|
|
|
|
|
|
# ── Dashboard data ────────────────────────────────────────────────────────────
|
|
def get_zfs_status():
|
|
try:
|
|
cmd = (f"sshpass -p '{REMOTE_PASS}' ssh -o StrictHostKeyChecking=no "
|
|
f"root@{REMOTE_HOST} \"zpool list tank -H -o name,size,alloc,free,cap,health\"")
|
|
res = os.popen(cmd).read().strip().split()
|
|
if len(res) >= 6:
|
|
return {"name": res[0], "size": res[1], "alloc": res[2],
|
|
"free": res[3], "cap": res[4], "health": res[5]}
|
|
except:
|
|
pass
|
|
return None
|
|
|
|
|
|
def get_grafana_metrics():
|
|
headers = {"Authorization": f"Bearer {GRAFANA_TOKEN}"}
|
|
proxmox_hosts = ["192.168.0.173", "192.168.0.112", "192.168.0.197"]
|
|
metrics_data = {}
|
|
queries = {
|
|
"CPU": '100 * (1 - avg by(instance)(irate(node_cpu_seconds_total{mode="idle"}[5m])))',
|
|
"RAM": '100 * (1 - node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes)',
|
|
"Disk": '100 - ((node_filesystem_avail_bytes{mountpoint="/"} * 100) / node_filesystem_size_bytes{mountpoint="/"})',
|
|
}
|
|
try:
|
|
for n, q in queries.items():
|
|
r = requests.get(
|
|
f"{GRAFANA_URL}/api/datasources/proxy/1/api/v1/query",
|
|
params={"query": q}, headers=headers, timeout=10
|
|
).json()
|
|
for res in r.get("data", {}).get("result", []):
|
|
inst = res.get("metric", {}).get("instance", "").split(":")[0]
|
|
if inst in proxmox_hosts:
|
|
metrics_data.setdefault(inst, []).append(
|
|
f"{n}:{float(res.get('value', [0, 0])[1]):.1f}%"
|
|
)
|
|
return [f"H{h.split('.')[-1]}:{''.join(s)}" for h, s in metrics_data.items()]
|
|
except:
|
|
return []
|
|
|
|
|
|
def get_weather():
|
|
try:
|
|
r = requests.get(
|
|
"https://api.open-meteo.com/v1/forecast"
|
|
"?latitude=38.8421&longitude=-77.2683"
|
|
"&hourly=weathercode,temperature_2m,precipitation_probability"
|
|
"&temperature_unit=fahrenheit&timezone=America%2FNew_York&forecast_days=2",
|
|
timeout=15
|
|
).json()
|
|
hourly = r.get("hourly", {})
|
|
times = hourly.get("time", [])
|
|
temps = hourly.get("temperature_2m", [])
|
|
pops = hourly.get("precipitation_probability", [])
|
|
codes = hourly.get("weathercode", [])
|
|
if not times:
|
|
return None
|
|
|
|
now = datetime.now(EASTERN_TZ).replace(tzinfo=None)
|
|
wmo = {
|
|
0: "Sun", 1: "Sun", 2: "PCld", 3: "Ovcst",
|
|
45: "Fog", 48: "Fog",
|
|
51: "Driz", 53: "Driz", 55: "Driz",
|
|
61: "Rain", 63: "Rain", 65: "HvRain",
|
|
71: "Snow", 73: "Snow", 75: "HvSnow", 77: "Sleet",
|
|
80: "Shwr", 81: "Shwr", 82: "HvShwr",
|
|
85: "SnShwr", 86: "HvSnShwr",
|
|
95: "Strm", 96: "Strm", 99: "Strm",
|
|
}
|
|
slots = []
|
|
for i, t in enumerate(times):
|
|
dt = datetime.strptime(t, "%Y-%m-%dT%H:%M")
|
|
if dt >= now:
|
|
slots.append((dt, i))
|
|
if len(slots) >= 24:
|
|
break
|
|
|
|
if not slots:
|
|
return None
|
|
|
|
parts = []
|
|
for idx in range(0, len(slots), 3):
|
|
dt, i = slots[idx]
|
|
cond = wmo.get(codes[i], "Unk")
|
|
temp = temps[i]
|
|
pop = pops[i]
|
|
parts.append(f"{dt.strftime('%H:%M')}:{cond} {temp:.0f}F({pop:.0f}%)")
|
|
|
|
lines = []
|
|
for i in range(0, len(parts), 4):
|
|
lines.append(" | ".join(parts[i:i + 4]))
|
|
return lines
|
|
except Exception as e:
|
|
print(f"[!] Weather error: {e}")
|
|
return None
|
|
|
|
|
|
def get_financial_snapshot():
|
|
tickers = {"SPY": "S&P", "QQQ": "NAS", "BTC-USD": "BTC", "ETH-USD": "ETH", "GOOG": "GOOG"}
|
|
snap = []
|
|
try:
|
|
data = yf.download(list(tickers.keys()), period="5d", progress=False)
|
|
for t, n in tickers.items():
|
|
try:
|
|
snap.append(f"{n}:{data['Close'][t].dropna().iloc[-1]:.0f}")
|
|
except:
|
|
pass
|
|
return "|".join(snap)
|
|
except:
|
|
return None
|
|
|
|
|
|
def get_reddit_top():
|
|
try:
|
|
r = requests.get(
|
|
"https://www.reddit.com/top.json?limit=1",
|
|
headers={"User-Agent": "SethPC/1.0"},
|
|
timeout=10
|
|
).json()
|
|
top = r["data"]["children"][0]["data"]
|
|
return {
|
|
"title": top["title"],
|
|
"subreddit": top["subreddit_name_prefixed"],
|
|
"score": top["score"],
|
|
"url": f"https://www.reddit.com{top['permalink']}",
|
|
}
|
|
except:
|
|
return None
|
|
|
|
|
|
# ── YOURLS ────────────────────────────────────────────────────────────────────
|
|
def shorten_url(long_url):
|
|
if not long_url:
|
|
return ""
|
|
import random, string
|
|
keyword = "r" + "".join(random.choices(string.ascii_lowercase + string.digits, k=3))
|
|
params = {
|
|
"username": YOURLS_USER, "password": YOURLS_PASS,
|
|
"action": "shorturl", "url": long_url,
|
|
"keyword": keyword, "format": "json"
|
|
}
|
|
try:
|
|
data = requests.get(YOURLS_URL, params=params, timeout=5).json()
|
|
if "shorturl" in data:
|
|
return data["shorturl"]
|
|
if "url" in data and "shorturl" in data["url"]:
|
|
return data["url"]["shorturl"]
|
|
if data.get("code") == "error:keyword":
|
|
params.pop("keyword")
|
|
data = requests.get(YOURLS_URL, params=params, timeout=5).json()
|
|
if "shorturl" in data:
|
|
return data["shorturl"]
|
|
if "url" in data and "shorturl" in data["url"]:
|
|
return data["url"]["shorturl"]
|
|
except Exception as e:
|
|
print(f"[!] YOURLS error: {e}")
|
|
return long_url
|
|
|
|
|
|
# ── ESC/POS receipt builder ───────────────────────────────────────────────────
|
|
def build_receipt(articles, zfs, reddit, grafana, weather, finance):
|
|
from escpos.printer import Dummy as EscPosDummy
|
|
p = EscPosDummy(profile="default")
|
|
|
|
def s(**kwargs):
|
|
p.set(font='b', **kwargs)
|
|
|
|
# Header
|
|
s(align='center', bold=True)
|
|
p.text("Daily Briefing\n")
|
|
s(align='center', bold=False)
|
|
p.text(datetime.now(EASTERN_TZ).strftime('%A, %B %d %H:%M ET') + "\n")
|
|
p.text("=" * COLS + "\n")
|
|
|
|
# Dashboard
|
|
s(align='left', bold=True)
|
|
p.text("SYSTEM & WEATHER\n")
|
|
s(align='left', bold=False)
|
|
if weather:
|
|
if isinstance(weather, list):
|
|
p.text("Wx24:\n")
|
|
for line in weather:
|
|
p.text(f" {line}\n")
|
|
else:
|
|
p.text(f"Wx: {weather}\n")
|
|
if zfs:
|
|
p.text(f"ZFS: {zfs['alloc']}/{zfs['size']} ({zfs['cap']}) {zfs['health']}\n")
|
|
if grafana:
|
|
p.text(" | ".join(grafana) + "\n")
|
|
if finance:
|
|
p.text(f"Mkts: {finance}\n")
|
|
p.text("-" * COLS + "\n")
|
|
|
|
# Reddit
|
|
if reddit:
|
|
s(align='left', bold=True)
|
|
p.text(f"REDDIT: {reddit.get('subreddit', '')}\n")
|
|
s(align='left', bold=False)
|
|
p.text(reddit['title'] + "\n")
|
|
url_s = shorten_url(reddit.get('url', ''))
|
|
if url_s:
|
|
s(align='center', bold=False)
|
|
p.qr(url_s, native=True, size=3)
|
|
s(align='left', bold=False)
|
|
p.text(f"{url_s} " + "-" * max(0, COLS - len(url_s) - 1) + "\n")
|
|
else:
|
|
p.text("-" * COLS + "\n")
|
|
|
|
# Articles
|
|
for i, item in enumerate(articles, 1):
|
|
s(align='left', bold=True)
|
|
p.text(f"{i}. [{item.get('source', '')}] {item.get('title', '')}\n")
|
|
s(align='left', bold=False)
|
|
summary = item.get('ai_summary', item.get('summary', ''))
|
|
if summary:
|
|
p.text(summary + "\n")
|
|
url_s = shorten_url(item.get('url', ''))
|
|
if url_s:
|
|
s(align='center', bold=False)
|
|
p.qr(url_s, native=True, size=3)
|
|
s(align='left', bold=False)
|
|
p.text(f"{url_s} " + "-" * max(0, COLS - len(url_s) - 1) + "\n")
|
|
else:
|
|
p.text("-" * COLS + "\n")
|
|
|
|
# Footer
|
|
p.text("\n")
|
|
s(align='center', bold=False)
|
|
p.text("-- end --\n\n")
|
|
p.cut()
|
|
return p.output
|
|
|
|
|
|
def _mac_of(ip):
|
|
subprocess.run(["ping", "-c1", "-W1", ip], capture_output=True)
|
|
out = subprocess.run(["ip", "neigh", "show", ip],
|
|
capture_output=True, text=True).stdout
|
|
m = re.search(r"lladdr ([0-9a-f:]{17})", out)
|
|
return m.group(1).lower() if m else None
|
|
|
|
|
|
def _port_open(ip, port, timeout=3):
|
|
try:
|
|
with socket.create_connection((ip, port), timeout=timeout):
|
|
return True
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def _sweep_arp():
|
|
"""Concurrent ping-prime the /24, then return {mac: ip} from the neigh table."""
|
|
procs = [subprocess.Popen(["ping", "-c1", "-W1", f"{PRINTER_SUBNET}.{i}"],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
for i in range(1, 255)]
|
|
for p in procs:
|
|
p.wait()
|
|
out = subprocess.run(["ip", "neigh", "show"],
|
|
capture_output=True, text=True).stdout
|
|
found = {}
|
|
for line in out.splitlines():
|
|
m = re.match(r"([\d.]+).*lladdr ([0-9a-f:]{17})", line)
|
|
if m:
|
|
found[m.group(2).lower()] = m.group(1)
|
|
return found
|
|
|
|
|
|
def _cache_ip(ip):
|
|
CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
|
CACHE_FILE.write_text(ip)
|
|
|
|
|
|
def discover_printer():
|
|
"""Return the printer's current IP, resolved by MAC. Raise if not found."""
|
|
seeds = []
|
|
if CACHE_FILE.exists():
|
|
seeds.append(CACHE_FILE.read_text().strip())
|
|
if PRINTER_IP:
|
|
seeds.append(PRINTER_IP)
|
|
for ip in seeds:
|
|
if ip and _mac_of(ip) == PRINTER_MAC and _port_open(ip, PRINTER_PORT):
|
|
_cache_ip(ip)
|
|
return ip
|
|
if PRINTER_MAC:
|
|
ip = _sweep_arp().get(PRINTER_MAC)
|
|
if ip and _port_open(ip, PRINTER_PORT):
|
|
_cache_ip(ip)
|
|
return ip
|
|
raise RuntimeError(f"printer-not-found mac={PRINTER_MAC}")
|
|
|
|
|
|
def print_receipt(raw_bytes):
|
|
try:
|
|
with socket.create_connection((PRINTER_IP, PRINTER_PORT), timeout=10) as sock:
|
|
sock.sendall(raw_bytes)
|
|
print(f"✓ Receipt sent to {PRINTER_IP}:{PRINTER_PORT}")
|
|
return True
|
|
except Exception as e:
|
|
print(f"[!] Print failed: {e}")
|
|
return False
|
|
|
|
|
|
# ── Main ──────────────────────────────────────────────────────────────────────
|
|
def main():
|
|
items, token = fetch_news()
|
|
zfs = get_zfs_status()
|
|
reddit = get_reddit_top()
|
|
grafana = get_grafana_metrics()
|
|
weather = get_weather()
|
|
finance = get_financial_snapshot()
|
|
|
|
if not items:
|
|
print("[!] No unread articles — printing dashboard only.")
|
|
|
|
articles = summarize_news(items, count=5, length="30-50 words") if items else []
|
|
raw = build_receipt(articles, zfs, reddit, grafana, weather, finance)
|
|
print_receipt(raw)
|
|
|
|
if articles:
|
|
mark_read([a['id'] for a in articles], token)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|