orange.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import shutil
import sys
import time
import urllib.request
from pathlib import Path

HN_BASE = "https://hacker-news.firebaseio.com/v0"
DEFAULT_MAX = 50
SLEEP_SECONDS = 2
STATE_PATH = Path.home() / ".config" / "orange" / "state.json"

_warnings: list[str] = []


def load_state() -> dict:
    """Load state from STATE_PATH. Creates file and parent dirs if missing."""
    if not STATE_PATH.exists():
        STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
        STATE_PATH.write_text("{}")
        return {}
    return json.loads(STATE_PATH.read_text())


def save_state(state: dict) -> None:
    """Write state dict to STATE_PATH atomically via a temp file."""
    import os
    import tempfile

    STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=STATE_PATH.parent, prefix=".state-")
    try:
        with os.fdopen(fd, "w") as f:
            json.dump(state, f, indent=2)
        os.replace(tmp, STATE_PATH)
    except Exception:
        os.unlink(tmp)
        raise


def is_submitted(state: dict, url: str) -> bool:
    """Return True if url is already recorded in state."""
    return url in state


def mark_submitted(state: dict, url: str, hn_id: int) -> None:
    """Record url in state with timestamp. Mutates state in place."""
    import datetime

    state[url] = {
        "hn_id": hn_id,
        "submitted_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
    }


def render_progress_bar(current: int, total: int, width: int = 30) -> str:
    """Return a progress bar string like [████████░░░░░░] 25/50."""
    if total <= 0:
        return f"[{'░' * width}] 0/0"
    filled = int(width * current / total)
    bar = "█" * filled + "░" * (width - filled)
    return f"[{bar}] {current}/{total}"


def display_status(
    current: int,
    total: int,
    title: str,
    url: str,
    status: str,
    skipped: int,
    submitted: int,
) -> None:
    """Write a progress bar and current-item detail to stderr, overwriting in place."""
    terminal_width = shutil.get_terminal_size(fallback=(80, 24)).columns
    bar = render_progress_bar(current, total)
    counts = f"submitted: {submitted}  skipped: {skipped}"
    detail = f"{status}: {title}"[:terminal_width]
    url = url[:terminal_width]
    # Move cursor up 3 lines and clear each line (no-op on first call before any output)
    if current > 1:
        sys.stderr.write("\033[3A")
    sys.stderr.write(f"\033[2K{bar}  {counts}\n")
    sys.stderr.write(f"\033[2K{detail}\n")
    sys.stderr.write(f"\033[2K{url}\n")
    sys.stderr.flush()


def display_complete(total: int, skipped: int, submitted: int) -> None:
    """Print a final summary line to stderr after the main loop."""
    terminal_width = shutil.get_terminal_size(fallback=(80, 24)).columns
    bar = render_progress_bar(total, total)
    summary = f"{bar}  submitted: {submitted}  skipped: {skipped}"[:terminal_width]
    sys.stderr.write("\033[3A")
    sys.stderr.write(f"\033[2K{summary}\n")
    sys.stderr.write("\033[2KDone.\n")
    sys.stderr.write("\033[2K\n")
    sys.stderr.flush()


def handle_interrupt(skipped: int, submitted: int) -> bool:
    """Pause on Ctrl-C. Prompts user to resume or quit. Returns True to resume, False to quit."""
    sys.stderr.write("\033[3A")
    sys.stderr.write("\033[2KInterrupted. Press Enter to resume or Ctrl-C / Ctrl-D to quit.\n")
    sys.stderr.write("\033[2K\n")
    sys.stderr.write("\033[2K")
    sys.stderr.flush()
    try:
        input()
        return True
    except (KeyboardInterrupt, EOFError):
        return False


def load_env(env_path: Path) -> dict[str, str]:
    """Parse a .env file and return key-value pairs."""
    env: dict[str, str] = {}
    if not env_path.exists():
        return env
    for line in env_path.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        if "=" not in line:
            continue
        key, _, value = line.partition("=")
        env[key.strip()] = value.strip()
    return env


def get_config() -> tuple[str, str]:
    """Load and validate required env vars from .env or environment."""
    import os

    env_path = Path(__file__).parent / ".env"
    file_env = load_env(env_path)

    karakeep_url = file_env.get("KARAKEEP_URL") or os.environ.get("KARAKEEP_URL", "")
    api_token = file_env.get("KARAKEEP_API_TOKEN") or os.environ.get(
        "KARAKEEP_API_TOKEN", ""
    )

    if not karakeep_url:
        print("Error: KARAKEEP_URL is required", file=sys.stderr)
        sys.exit(1)
    if not api_token:
        print("Error: KARAKEEP_API_TOKEN is required", file=sys.stderr)
        sys.exit(1)

    return karakeep_url.rstrip("/"), api_token


def fetch_json(url: str) -> object:
    """Fetch a URL and return parsed JSON."""
    with urllib.request.urlopen(url) as response:
        return json.loads(response.read())


def fetch_top_story_ids(max_links: int) -> list[int]:
    """Return the top N story IDs from HN."""
    ids = fetch_json(f"{HN_BASE}/topstories.json")
    return ids[:max_links]  # type: ignore[index]


def fetch_story(story_id: int) -> dict:
    """Fetch a single HN story by ID."""
    return fetch_json(f"{HN_BASE}/item/{story_id}.json")  # type: ignore[return-value]


def create_bookmark(karakeep_url: str, token: str, title: str, url: str) -> str | None:
    """POST a new bookmark to Karakeep and return the bookmark ID, or None on failure."""
    payload = json.dumps({"type": "link", "title": title, "url": url}).encode()
    req = urllib.request.Request(
        f"{karakeep_url}/api/v1/bookmarks",
        data=payload,
        headers={
            "Content-Type": "application/json",
            "Accept": "application/json",
            "Authorization": f"Bearer {token}",
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(req) as response:
            data = json.loads(response.read())
            return data.get("id")
    except urllib.error.URLError as exc:
        _warnings.append(f"Warning: failed to create bookmark: {exc}")
        return None


def tag_bookmark(karakeep_url: str, token: str, bookmark_id: str) -> None:
    """Tag a Karakeep bookmark with 'hackernews'."""
    payload = json.dumps({"tags": [{"tagName": "hackernews"}]}).encode()
    req = urllib.request.Request(
        f"{karakeep_url}/api/v1/bookmarks/{bookmark_id}/tags",
        data=payload,
        headers={
            "Content-Type": "application/json",
            "Accept": "application/json",
            "Authorization": f"Bearer {token}",
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(req):
            pass
    except urllib.error.URLError as exc:
        _warnings.append(f"Warning: failed to tag bookmark {bookmark_id}: {exc}")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Fetch top Hacker News stories and save them as Karakeep bookmarks."
    )
    parser.add_argument(
        "-n",
        "--number",
        type=int,
        default=DEFAULT_MAX,
        metavar="NUMBER",
        help=f"Number of top stories to import (default: {DEFAULT_MAX})",
    )
    parser.add_argument(
        "--override",
        action="store_true",
        help="Submit to Karakeep even if URL is already in local state",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    karakeep_url, api_token = get_config()
    state = load_state()

    sys.stderr.write("Fetching Hacker News top stories...\n")
    sys.stderr.flush()
    story_ids = fetch_top_story_ids(args.number)
    total = len(story_ids)
    skipped = 0
    submitted = 0

    try:
        for count, story_id in enumerate(story_ids, start=1):
            try:
                time.sleep(SLEEP_SECONDS)
            except KeyboardInterrupt:
                if not handle_interrupt(skipped, submitted):
                    break

            story = fetch_story(story_id)
            if story is None:
                _warnings.append(f"Skipped story {story_id}: null response from HN API")
                continue

            title = story.get("title", "")
            url = story.get("url") or f"https://news.ycombinator.com/item?id={story_id}"

            if is_submitted(state, url) and not args.override:
                skipped += 1
                display_status(count, total, title, url, "skip", skipped, submitted)
                continue

            mark_submitted(state, url, story_id)
            save_state(state)

            bookmark_id = create_bookmark(karakeep_url, api_token, title, url)
            if bookmark_id:
                tag_bookmark(karakeep_url, api_token, bookmark_id)

            submitted += 1
            display_status(count, total, title, url, "sent", skipped, submitted)
    finally:
        display_complete(total, skipped, submitted)
        for w in _warnings:
            print(w, file=sys.stderr)


if __name__ == "__main__":
    main()