Add hourly news briefing system with continuous marquee scroll
- Fetches 12 articles hourly from FreshRSS - Generates 4-6 sentence summaries using Ollama - Creates sequential sethpc.xyz short links (pi0, pi1, etc.) via yourls - Continuous vertical marquee scrolling at reading speed - No wait between rescrolls - immediately loops to next pass - New briefing generated every hour at :00 - Systemd service and timer for auto-start and hourly scheduling
This commit is contained in:
@@ -0,0 +1,19 @@
|
||||
[Unit]
|
||||
Description=piNail Hourly News Briefing Display
|
||||
After=network-online.target
|
||||
Wants=network-online.target
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
User=root
|
||||
WorkingDirectory=/home/pi/piNail2
|
||||
ExecStart=/usr/bin/python3 /home/pi/piNail2/tty_briefing.py
|
||||
Restart=always
|
||||
RestartSec=10
|
||||
StandardOutput=tty
|
||||
StandardError=file:/tmp/briefing.log
|
||||
TTYPath=/dev/tty1
|
||||
SyslogIdentifier=pinail-briefing
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
@@ -0,0 +1,11 @@
|
||||
[Unit]
|
||||
Description=piNail Hourly News Briefing Timer
|
||||
Requires=pinail-briefing.service
|
||||
|
||||
[Timer]
|
||||
OnBootSec=2min
|
||||
OnUnitActiveSec=1h
|
||||
Persistent=true
|
||||
|
||||
[Install]
|
||||
WantedBy=timers.target
|
||||
@@ -0,0 +1,265 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Enhanced hourly news briefing with long summaries, yourls links, and continuous marquee scroll.
|
||||
Fetches from FreshRSS, summarizes with Ollama, generates short URLs, rescrolls continuously.
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import time
|
||||
import requests
|
||||
import re
|
||||
import schedule
|
||||
import textwrap
|
||||
from datetime import datetime
|
||||
|
||||
# Debug logging
|
||||
DEBUG_LOG = "/tmp/briefing_debug.log"
|
||||
def debug_log(msg):
|
||||
with open(DEBUG_LOG, "a") as f:
|
||||
f.write("[{}] {}\n".format(datetime.now().strftime('%H:%M:%S'), msg))
|
||||
f.flush()
|
||||
|
||||
# Load config from local copy
|
||||
CONFIG_PATH = "/home/pi/piNail2/briefing_config.json"
|
||||
with open(CONFIG_PATH, "r") 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.get("ollama_model", "mistral")
|
||||
YOURLS_URL = config.get("yourls_url", "http://192.168.0.152:8080/yourls-api.php")
|
||||
YOURLS_USER = config.get("yourls_user", "admin")
|
||||
YOURLS_PASS = config.get("yourls_pass", "REDACTED_PASSWORD")
|
||||
|
||||
def fetch_news(count=12):
|
||||
"""Fetch top articles from FreshRSS."""
|
||||
login_url = "{}/api/greader.php/accounts/ClientLogin".format(FRESHRSS_URL)
|
||||
login_data = {"Email": FRESHRSS_USER, "Passwd": FRESHRSS_TOKEN}
|
||||
try:
|
||||
login_resp = requests.post(login_url, data=login_data, timeout=15)
|
||||
if login_resp.status_code != 200:
|
||||
return []
|
||||
auth_token = ""
|
||||
for line in login_resp.text.splitlines():
|
||||
if line.startswith("Auth="):
|
||||
auth_token = line.split("=", 1)[1]
|
||||
break
|
||||
if not auth_token:
|
||||
return []
|
||||
except Exception as e:
|
||||
print("[!] Login exception: {}".format(e), file=sys.stderr)
|
||||
return []
|
||||
|
||||
api_url = "{}/api/greader.php/reader/api/0/stream/contents/user/-/state/com.google/reading-list".format(FRESHRSS_URL)
|
||||
params = {"n": count + 5, "xt": "user/-/state/com.google/read", "output": "json"}
|
||||
headers = {"Authorization": "GoogleLogin auth={}".format(auth_token)}
|
||||
try:
|
||||
response = requests.get(api_url, params=params, headers=headers, timeout=15)
|
||||
if response.status_code != 200:
|
||||
return []
|
||||
data = response.json()
|
||||
items = []
|
||||
for item in data.get("items", [])[:count]:
|
||||
title = item.get("title", "")
|
||||
summary = item.get("summary", {}).get("content", "") or item.get("content", {}).get("content", "")
|
||||
source = item.get("origin", {}).get("title", "Unknown Source")
|
||||
link = item.get("alternate", [{}])[0].get("href", "") if item.get("alternate") else ""
|
||||
summary_clean = re.sub('<[^<]+?>', '', summary).strip()
|
||||
items.append({
|
||||
"title": title,
|
||||
"summary": summary_clean,
|
||||
"source": source,
|
||||
"link": link
|
||||
})
|
||||
return items
|
||||
except Exception as e:
|
||||
print("[!] Exception fetching news: {}".format(e), file=sys.stderr)
|
||||
return []
|
||||
|
||||
def summarize_article(text):
|
||||
"""Summarize article with Ollama (longer, 4-6 sentences)."""
|
||||
if not text:
|
||||
return ""
|
||||
payload = {
|
||||
"model": OLLAMA_MODEL,
|
||||
"prompt": "Provide a detailed summary in 4-6 sentences: {}".format(text[:1000]),
|
||||
"stream": False,
|
||||
"options": {"num_ctx": 2048, "temperature": 0.3}
|
||||
}
|
||||
try:
|
||||
response = requests.post("{}/api/generate".format(OLLAMA_URL), json=payload, timeout=60)
|
||||
return response.json().get("response", "").strip()
|
||||
except:
|
||||
return text[:400] + "..."
|
||||
|
||||
def create_yourls_link(original_url, index):
|
||||
"""Generate short URL using yourls with pi____ format."""
|
||||
if not original_url:
|
||||
debug_log("No URL provided")
|
||||
return ""
|
||||
try:
|
||||
# Create sequential alias: pi0, pi1, pi2, etc.
|
||||
alias = "pi{}".format(index)
|
||||
debug_log("Creating yourls link with alias: {}".format(alias))
|
||||
|
||||
payload = {
|
||||
"action": "shorturl",
|
||||
"url": original_url,
|
||||
"keyword": alias,
|
||||
"username": YOURLS_USER,
|
||||
"password": YOURLS_PASS,
|
||||
"format": "json"
|
||||
}
|
||||
response = requests.post(YOURLS_URL, data=payload, timeout=10)
|
||||
data = response.json()
|
||||
|
||||
# Check if we got a shorturl (success or URL already exists)
|
||||
if data.get("shorturl"):
|
||||
short_link = data.get("shorturl", "")
|
||||
debug_log("Created link {}: {}".format(alias, short_link))
|
||||
return short_link
|
||||
|
||||
# Try random alias if keyword alias is taken
|
||||
if "already exists" in data.get("message", "").lower() or "keyword already exists" in data.get("message", "").lower():
|
||||
debug_log("Alias {} taken, trying random".format(alias))
|
||||
payload_random = {
|
||||
"action": "shorturl",
|
||||
"url": original_url,
|
||||
"username": YOURLS_USER,
|
||||
"password": YOURLS_PASS,
|
||||
"format": "json"
|
||||
}
|
||||
response = requests.post(YOURLS_URL, data=payload_random, timeout=10)
|
||||
data = response.json()
|
||||
if data.get("shorturl"):
|
||||
short_link = data.get("shorturl", "")
|
||||
debug_log("Created fallback link: {}".format(short_link))
|
||||
return short_link
|
||||
|
||||
debug_log("yourls failed: {}".format(data.get("message", "unknown error")))
|
||||
return ""
|
||||
except Exception as e:
|
||||
debug_log("yourls exception: {}".format(e))
|
||||
return ""
|
||||
|
||||
def render_marquee(articles, tty):
|
||||
"""Render articles as continuous vertical marquee."""
|
||||
# Build full content once
|
||||
full_content = ""
|
||||
for i, article in enumerate(articles, 1):
|
||||
full_content += "\n{}\n".format("="*70)
|
||||
full_content += "[{}/{}] {}\n".format(i, len(articles), article['source'])
|
||||
full_content += "{}\n\n".format("─" * 70)
|
||||
full_content += " {}\n".format(article['title'])
|
||||
full_content += " {}\n\n".format("─" * 70)
|
||||
|
||||
summary = article.get('ai_summary', article['summary'][:300])
|
||||
wrapped = textwrap.fill(summary, width=66, initial_indent=" ", subsequent_indent=" ")
|
||||
full_content += "{}\n".format(wrapped)
|
||||
|
||||
short_link = article.get('short_link', article.get('link', ''))
|
||||
if short_link:
|
||||
full_content += "\n Link: {}\n".format(short_link)
|
||||
|
||||
full_content += "\n"
|
||||
|
||||
full_content += "\n{}\n".format("="*70)
|
||||
full_content += " --- END OF BRIEFING, RESCROLLING IN 30 SECONDS ---\n"
|
||||
full_content += "{}\n".format("="*70)
|
||||
|
||||
return full_content
|
||||
|
||||
def scroll_content(tty, content):
|
||||
"""Scroll content at reading speed, continuous loop without waiting."""
|
||||
char_delay = 0.025 # 25ms per char = ~40 chars/sec
|
||||
|
||||
while True:
|
||||
for char in content:
|
||||
tty.write(char)
|
||||
tty.flush()
|
||||
if char == '\n':
|
||||
time.sleep(0.1)
|
||||
else:
|
||||
time.sleep(char_delay)
|
||||
# Loop back immediately without pause
|
||||
|
||||
def hourly_briefing():
|
||||
"""Fetch, summarize, and display briefing. Continuously scrolls until next hourly refresh."""
|
||||
try:
|
||||
tty = open('/dev/tty1', 'w')
|
||||
except Exception:
|
||||
tty = sys.stdout
|
||||
|
||||
print("[*] Fetching briefing at {}".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')), file=sys.stderr)
|
||||
|
||||
articles = fetch_news(count=12)
|
||||
if not articles:
|
||||
tty.write(" [!] No articles fetched\n")
|
||||
tty.flush()
|
||||
print("[!] No articles fetched", file=sys.stderr)
|
||||
if tty is not sys.stdout:
|
||||
tty.close()
|
||||
return
|
||||
|
||||
# Summarize and generate yourls links
|
||||
debug_log("Summarizing {} articles...".format(len(articles)))
|
||||
for i, article in enumerate(articles):
|
||||
debug_log("Article {}/{}: summarizing...".format(i+1, len(articles)))
|
||||
article['ai_summary'] = summarize_article(article['summary'])
|
||||
debug_log("Article {}/{}: creating link...".format(i+1, len(articles)))
|
||||
article['short_link'] = create_yourls_link(article['link'], i)
|
||||
debug_log("Article {}/{}: done. Link={}".format(i+1, len(articles), article.get('short_link', '')))
|
||||
|
||||
# Render and display header
|
||||
timestamp = datetime.now().strftime("%A, %B %d, %Y | %I:%M %p")
|
||||
tty.write("\033[2J\033[H\033[?25l") # Clear and hide cursor
|
||||
tty.write("\n{}\n".format("="*70))
|
||||
tty.write(" PINAIL NEWS BRIEFING - {}\n".format(timestamp))
|
||||
tty.write(" {} stories\n".format(len(articles)))
|
||||
tty.write("{}\n\n".format("="*70))
|
||||
tty.flush()
|
||||
time.sleep(1)
|
||||
|
||||
# Build full marquee content
|
||||
full_content = render_marquee(articles, tty)
|
||||
|
||||
# Continuously scroll content until next scheduled run
|
||||
print("[*] Starting continuous scroll", file=sys.stderr)
|
||||
try:
|
||||
scroll_content(tty, full_content)
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
finally:
|
||||
tty.write('\033[?25h') # Restore cursor
|
||||
tty.flush()
|
||||
if tty is not sys.stdout:
|
||||
tty.close()
|
||||
|
||||
def main():
|
||||
"""Schedule hourly briefing."""
|
||||
schedule.every().hour.at(":00").do(hourly_briefing)
|
||||
|
||||
print("[*] piNail News Briefing scheduled for every hour", file=sys.stderr)
|
||||
next_run = datetime.now().replace(minute=0, second=0, microsecond=0)
|
||||
print("[*] Next run at {}".format(next_run.strftime('%H:%M')), file=sys.stderr)
|
||||
|
||||
# Run first briefing immediately
|
||||
hourly_briefing()
|
||||
|
||||
# Keep scheduler alive (will interrupt scroll_content when next hour arrives)
|
||||
while True:
|
||||
schedule.run_pending()
|
||||
time.sleep(1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
main()
|
||||
except KeyboardInterrupt:
|
||||
print("\n[*] Briefing scheduler stopped", file=sys.stderr)
|
||||
sys.exit(0)
|
||||
except Exception as e:
|
||||
print("[!] Fatal error: {}".format(e), file=sys.stderr)
|
||||
raise
|
||||
Reference in New Issue
Block a user