sawz/last30days-skill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add animated progress UI with ASCII art

Browse Source 
- New ui.py module with colored output and animations
- Animated spinner during Reddit/X searches
- Progress tracking for enrichment phase [1/N]
- Fun random status messages per phase
- Mini ASCII banner at start
- Completion summary with timing

Makes the research feel more alive while waiting!

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Matt Van Horn
2026-01-23 16:11:45 -08:00
parent 664b1cc52d
commit 9daf671b82
2 changed files with 250 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/last30days.py
+50 -12
View File
@@ -37,6 +37,7 @@ from lib import (
render,
schema,
score,
ui,
xai_x,
)
@@ -59,6 +60,7 @@ def run_research(
to_date: str,
depth: str = "default",
mock: bool = False,
progress: ui.ProgressDisplay = None,
) -> tuple:
"""Run the research pipeline.
@@ -75,6 +77,9 @@ def run_research(
# Reddit search via OpenAI
if sources in ("both", "reddit"):
if progress:
progress.start_reddit()
if mock:
raw_openai = load_fixture("openai_sample.json")
else:
@@ -86,31 +91,47 @@ def run_research(
depth=depth,
)
except http.HTTPError as e:
print(f"[REDDIT ERROR] OpenAI API request failed: {e}", flush=True)
if e.body:
print(f"[REDDIT ERROR] Response body: {e.body[:500]}", flush=True)
if progress:
progress.show_error(f"Reddit API failed: {e}")
raw_openai = {"error": str(e)}
reddit_error = f"API error: {e}"
except Exception as e:
print(f"[REDDIT ERROR] Unexpected error: {type(e).__name__}: {e}", flush=True)
if progress:
progress.show_error(f"Reddit error: {e}")
raw_openai = {"error": str(e)}
reddit_error = f"{type(e).__name__}: {e}"
# Parse response
reddit_items = openai_reddit.parse_reddit_response(raw_openai)
# Enrich with real Reddit data
for i, item in enumerate(reddit_items):
if mock:
mock_thread = load_fixture("reddit_thread_sample.json")
reddit_items[i] = reddit_enrich.enrich_reddit_item(item, mock_thread)
else:
reddit_items[i] = reddit_enrich.enrich_reddit_item(item)
if progress:
progress.end_reddit(len(reddit_items))
raw_reddit_enriched.append(reddit_items[i])
# Enrich with real Reddit data
if reddit_items:
if progress:
progress.start_reddit_enrich(1, len(reddit_items))
for i, item in enumerate(reddit_items):
if progress and i > 0:
progress.update_reddit_enrich(i + 1, len(reddit_items))
if mock:
mock_thread = load_fixture("reddit_thread_sample.json")
reddit_items[i] = reddit_enrich.enrich_reddit_item(item, mock_thread)
else:
reddit_items[i] = reddit_enrich.enrich_reddit_item(item)
raw_reddit_enriched.append(reddit_items[i])
if progress:
progress.end_reddit_enrich()
# X search via xAI
if sources in ("both", "x"):
if progress:
progress.start_x()
if mock:
raw_xai = load_fixture("xai_sample.json")
else:
@@ -126,6 +147,9 @@ def run_research(
# Parse response
x_items = xai_x.parse_x_response(raw_xai)
if progress:
progress.end_x(len(x_items))
return reddit_items, x_items, raw_openai, raw_xai, raw_reddit_enriched, reddit_error, x_error
@@ -207,10 +231,15 @@ def main():
# Check cache (unless refresh or mock)
cache_key = cache.get_cache_key(args.topic, from_date, to_date, f"{sources}:{depth}")
# Initialize progress display
progress = ui.ProgressDisplay(args.topic, show_banner=True)
if not args.refresh and not args.mock:
cached = cache.load_cache(cache_key)
if cached:
# Use cached data
progress.show_cached()
report = schema.Report.from_dict(cached)
output_result(report, args.emit)
return
@@ -250,8 +279,12 @@ def main():
to_date,
depth,
args.mock,
progress,
)
# Processing phase
progress.start_processing()
# Normalize items
normalized_reddit = normalize.normalize_reddit_items(reddit_items, from_date, to_date)
normalized_x = normalize.normalize_x_items(x_items, from_date, to_date)
@@ -268,6 +301,8 @@ def main():
deduped_reddit = dedupe.dedupe_reddit(sorted_reddit)
deduped_x = dedupe.dedupe_x(sorted_x)
progress.end_processing()
# Create report
report = schema.create_report(
args.topic,
@@ -292,6 +327,9 @@ def main():
if not args.mock:
cache.save_cache(cache_key, report.to_dict())
# Show completion
progress.show_complete(len(deduped_reddit), len(deduped_x))
# Output result
output_result(report, args.emit)
scripts/lib/ui.py
+200
View File
@@ -0,0 +1,200 @@
"""Terminal UI utilities for last30days skill."""
import sys
import time
import threading
import random
from typing import Optional
# ANSI color codes
class Colors:
PURPLE = '\033[95m'
BLUE = '\033[94m'
CYAN = '\033[96m'
GREEN = '\033[92m'
YELLOW = '\033[93m'
RED = '\033[91m'
BOLD = '\033[1m'
DIM = '\033[2m'
RESET = '\033[0m'
BANNER = f"""{Colors.PURPLE}{Colors.BOLD}
██╗ █████╗ ███████╗████████╗██████╗ ██████╗ ██████╗ █████╗ ██╗ ██╗███████╗
██║ ██╔══██╗██╔════╝╚══██╔══╝╚════██╗██╔═████╗██╔══██╗██╔══██╗╚██╗ ██╔╝██╔════╝
██║ ███████║███████╗ ██║ █████╔╝██║██╔██║██║ ██║███████║ ╚████╔╝ ███████╗
██║ ██╔══██║╚════██║ ██║ ╚═══██╗████╔╝██║██║ ██║██╔══██║ ╚██╔╝ ╚════██║
███████╗██║ ██║███████║ ██║ ██████╔╝╚██████╔╝██████╔╝██║ ██║ ██║ ███████║
╚══════╝╚═╝ ╚═╝╚══════╝ ╚═╝ ╚═════╝ ╚═════╝ ╚═════╝ ╚═╝ ╚═╝ ╚═╝ ╚══════╝
{Colors.RESET}{Colors.DIM} 30 days of research. 30 seconds of work.{Colors.RESET}
"""
MINI_BANNER = f"""{Colors.PURPLE}{Colors.BOLD}/last30days{Colors.RESET} {Colors.DIM}· researching...{Colors.RESET}"""
# Fun status messages for each phase
REDDIT_MESSAGES = [
"Diving into Reddit threads...",
"Scanning subreddits for gold...",
"Reading what Redditors are saying...",
"Exploring the front page of the internet...",
"Finding the good discussions...",
"Upvoting mentally...",
"Scrolling through comments...",
]
X_MESSAGES = [
"Checking what X is buzzing about...",
"Reading the timeline...",
"Finding the hot takes...",
"Scanning tweets and threads...",
"Discovering trending insights...",
"Following the conversation...",
"Reading between the posts...",
]
ENRICHING_MESSAGES = [
"Getting the juicy details...",
"Fetching engagement metrics...",
"Reading top comments...",
"Extracting insights...",
"Analyzing discussions...",
]
PROCESSING_MESSAGES = [
"Crunching the data...",
"Scoring and ranking...",
"Finding patterns...",
"Removing duplicates...",
"Organizing findings...",
]
# Spinner frames
SPINNER_FRAMES = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏']
DOTS_FRAMES = [' ', '. ', '.. ', '...']
class Spinner:
"""Animated spinner for long-running operations."""
def __init__(self, message: str = "Working", color: str = Colors.CYAN):
self.message = message
self.color = color
self.running = False
self.thread: Optional[threading.Thread] = None
self.frame_idx = 0
def _spin(self):
while self.running:
frame = SPINNER_FRAMES[self.frame_idx % len(SPINNER_FRAMES)]
sys.stderr.write(f"\r{self.color}{frame}{Colors.RESET} {self.message} ")
sys.stderr.flush()
self.frame_idx += 1
time.sleep(0.08)
def start(self):
self.running = True
self.thread = threading.Thread(target=self._spin, daemon=True)
self.thread.start()
def update(self, message: str):
self.message = message
def stop(self, final_message: str = ""):
self.running = False
if self.thread:
self.thread.join(timeout=0.2)
# Clear the line
sys.stderr.write("\r" + " " * 80 + "\r")
if final_message:
sys.stderr.write(f"{Colors.GREEN}✓{Colors.RESET} {final_message}\n")
sys.stderr.flush()
class ProgressDisplay:
"""Progress display for research phases."""
def __init__(self, topic: str, show_banner: bool = True):
self.topic = topic
self.spinner: Optional[Spinner] = None
self.start_time = time.time()
if show_banner:
self._show_banner()
def _show_banner(self):
sys.stderr.write(MINI_BANNER + "\n")
sys.stderr.write(f"{Colors.DIM}Topic: {Colors.RESET}{Colors.BOLD}{self.topic}{Colors.RESET}\n\n")
sys.stderr.flush()
def start_reddit(self):
msg = random.choice(REDDIT_MESSAGES)
self.spinner = Spinner(f"{Colors.YELLOW}Reddit{Colors.RESET} {msg}", Colors.YELLOW)
self.spinner.start()
def end_reddit(self, count: int):
if self.spinner:
self.spinner.stop(f"{Colors.YELLOW}Reddit{Colors.RESET} Found {count} threads")
def start_reddit_enrich(self, current: int, total: int):
if self.spinner:
self.spinner.stop()
msg = random.choice(ENRICHING_MESSAGES)
self.spinner = Spinner(f"{Colors.YELLOW}Reddit{Colors.RESET} [{current}/{total}] {msg}", Colors.YELLOW)
self.spinner.start()
def update_reddit_enrich(self, current: int, total: int):
if self.spinner:
msg = random.choice(ENRICHING_MESSAGES)
self.spinner.update(f"{Colors.YELLOW}Reddit{Colors.RESET} [{current}/{total}] {msg}")
def end_reddit_enrich(self):
if self.spinner:
self.spinner.stop(f"{Colors.YELLOW}Reddit{Colors.RESET} Enriched with engagement data")
def start_x(self):
msg = random.choice(X_MESSAGES)
self.spinner = Spinner(f"{Colors.CYAN}X{Colors.RESET} {msg}", Colors.CYAN)
self.spinner.start()
def end_x(self, count: int):
if self.spinner:
self.spinner.stop(f"{Colors.CYAN}X{Colors.RESET} Found {count} posts")
def start_processing(self):
msg = random.choice(PROCESSING_MESSAGES)
self.spinner = Spinner(f"{Colors.PURPLE}Processing{Colors.RESET} {msg}", Colors.PURPLE)
self.spinner.start()
def end_processing(self):
if self.spinner:
self.spinner.stop()
def show_complete(self, reddit_count: int, x_count: int):
elapsed = time.time() - self.start_time
sys.stderr.write(f"\n{Colors.GREEN}{Colors.BOLD}✓ Research complete{Colors.RESET} ")
sys.stderr.write(f"{Colors.DIM}({elapsed:.1f}s){Colors.RESET}\n")
sys.stderr.write(f" {Colors.YELLOW}Reddit:{Colors.RESET} {reddit_count} threads ")
sys.stderr.write(f"{Colors.CYAN}X:{Colors.RESET} {x_count} posts\n\n")
sys.stderr.flush()
def show_cached(self):
sys.stderr.write(f"{Colors.GREEN}⚡{Colors.RESET} {Colors.DIM}Using cached results{Colors.RESET}\n\n")
sys.stderr.flush()
def show_error(self, message: str):
sys.stderr.write(f"{Colors.RED}✗ Error:{Colors.RESET} {message}\n")
sys.stderr.flush()
def print_phase(phase: str, message: str):
"""Print a phase message."""
colors = {
"reddit": Colors.YELLOW,
"x": Colors.CYAN,
"process": Colors.PURPLE,
"done": Colors.GREEN,
"error": Colors.RED,
}
color = colors.get(phase, Colors.RESET)
sys.stderr.write(f"{color}▸{Colors.RESET} {message}\n")
sys.stderr.flush()