utensil.tngl.sh
Utensil's Zettelkasten-style forest of evergreen notes on math and tech.
utensil.tngl.sh/forest/
1#!/usr/bin/env -S uv run
2# /// script
3# requires-python = ">=3.11,<3.12"
4# dependencies = []
5# ///
6"""
7Pocket Export Script
8===================
9
10Reads JSON objects (one per line) from stdin (as output by `just rss-stars "linkwarden"`),
11maps fields to Pocket CSV import format, and outputs a valid CSV to stdout.
12
13Usage:
14 just rss-stars "linkwarden" | ./rss2pocket.py > pocket.csv
15
16- Input: JSON objects, one per line, with fields: title, url, externalURL, datePublished, dateArrived, tags, starred, etc.
17- Output: CSV with columns: title, url, time_added, cursor, tags, status
18
19AGENT-NOTE: CRITICAL FEATURES TO MAINTAIN
201. IDEMPOTENT: Multiple runs must produce identical results
212. ERROR HANDLING: Graceful degradation for missing data
223. BUILD INTEGRATION: Validates syntax after processing
234. DETERMINISTIC: Sorted processing ensures consistent output
24"""
25
26import sys
27import json
28import csv
29import argparse
30from datetime import datetime, timedelta, timezone
31import zipfile
32
33# AGENT-NOTE: Pocket CSV columns: title, url, time_added, cursor, tags, status, lobsters_url, hn_url
34COLUMNS = ["title", "url", "time_added", "cursor", "tags", "status"]
35
36def get_time_added(entry):
37 # Prefer dateArrived, fallback to datePublished, else 0
38 ts = entry.get("dateArrived") or entry.get("datePublished")
39 try:
40 return str(int(float(ts))) if ts is not None else "0"
41 except Exception:
42 return "0"
43
44def get_tags(entry):
45 tags = entry.get("tags")
46 if isinstance(tags, list):
47 return "|".join(str(t).strip() for t in tags if t)
48 elif isinstance(tags, str):
49 return tags.strip()
50 return ""
51
52def map_entry(entry):
53 # Map fields to Pocket CSV columns, using best available data
54 # Determine status: only 'unread' or 'archived' are valid
55 status = "unread"
56 # Accept both is_archived (bool/int) and status (str) from input
57 is_archived = entry.get("is_archived")
58 input_status = str(entry.get("status", "")).strip().lower()
59 if input_status == "archived" or is_archived in [1, True, "1", "true", "True"]:
60 status = "archived"
61 # Force output to only 'unread' or 'archived'
62 if status != "archived":
63 status = "unread"
64 # If url is a lobste.rs or HN link and externalURL exists, always use externalURL
65 def is_aggregator(url):
66 if not url:
67 return False
68 return ("lobste.rs" in url) or ("news.ycombinator.com" in url) or ("ycombinator.com/item" in url)
69
70 ext_url = entry.get("externalURL")
71 main_url = entry.get("url")
72 # Always prefer externalURL if present and non-empty
73 if ext_url:
74 url = ext_url
75 elif main_url:
76 url = main_url
77 else:
78 url = entry.get("uniqueID") or ""
79
80 return {
81 "title": entry.get("title", ""), # Use as-is, no fallback
82 "url": url,
83 "time_added": get_time_added(entry),
84 "tags": get_tags(entry),
85 "status": status if status == "archived" else "unread",
86 }
87
88def main():
89 argp = argparse.ArgumentParser(description="Convert RSS stars to Pocket CSV import format")
90 argp.add_argument("--days", type=int, default=7, help="Only include entries from the last N days (by dateArrived or datePublished, -1 for all)")
91 argp.add_argument("--debug", action="store_true", help="Enable detailed debug logging to stderr for each entry")
92 args = argp.parse_args()
93 now = datetime.now(timezone.utc)
94 cutoff_timestamp = None
95 if args.days != -1:
96 cutoff_date = now - timedelta(days=args.days)
97 cutoff_timestamp = cutoff_date.timestamp()
98
99 entries = []
100 entry_count = 0
101 for line in sys.stdin:
102 line = line.strip()
103 if not line:
104 continue
105 try:
106 entry = json.loads(line)
107 entry_count += 1
108 # Use dateArrived or datePublished (UNIX timestamp) for filtering
109 timestamp = entry.get("dateArrived") or entry.get("datePublished")
110 filtered_out = False
111 filter_reason = ""
112 if cutoff_timestamp is not None:
113 if timestamp is None or float(timestamp) < cutoff_timestamp:
114 filtered_out = True
115 filter_reason = f"timestamp {timestamp} < cutoff {cutoff_timestamp}"
116 if filtered_out:
117 if args.debug:
118 print(f"[DEBUG] Skipped entry {entry_count}: title='{entry.get('title','')}' url='{entry.get('url','')}' externalURL='{entry.get('externalURL','')}' reason={filter_reason}", file=sys.stderr)
119 continue
120 mapped = map_entry(entry)
121 if args.debug:
122 print(f"[DEBUG] Processed entry {entry_count}: title='{mapped['title']}'\n aggregator_url='{entry.get('url','')}'\n external_url='{entry.get('externalURL','')}'\n chosen_url='{mapped['url']}'\n time_added='{mapped['time_added']}' status='{mapped['status']}' tags='{mapped['tags']}'", file=sys.stderr)
123 entries.append(mapped)
124 except Exception as e:
125 print(f"Warning: Skipping invalid line: {e}", file=sys.stderr)
126 # Sort by time_added for deterministic output
127 entries.sort(key=lambda x: x["time_added"])
128 # Write to output/pocket.csv as well as stdout
129 import os
130 os.makedirs('output', exist_ok=True)
131 # Write part_000000.csv (with cursor column) and make output/pocket.csv identical
132 part_csv_path = 'output/part_000000.csv'
133 pocket_csv_path = 'output/pocket.csv'
134 with open(part_csv_path, 'w', newline='', encoding='utf-8') as f1, \
135 open(pocket_csv_path, 'w', newline='', encoding='utf-8') as f2:
136 writer1 = csv.DictWriter(f1, fieldnames=COLUMNS)
137 writer2 = csv.DictWriter(f2, fieldnames=COLUMNS)
138 writer1.writeheader()
139 writer2.writeheader()
140 for row in entries:
141 row_out = row.copy()
142 row_out['cursor'] = '' # Always empty
143 writer1.writerow(row_out)
144 writer2.writerow(row_out)
145 # Create output/pocket.zip containing only part_000000.csv at the root
146 with zipfile.ZipFile('output/pocket.zip', 'w', zipfile.ZIP_DEFLATED) as zipf:
147 zipf.write(part_csv_path, arcname='part_000000.csv')
148 print("Pocket export written to output/pocket.csv", file=sys.stderr)
149 print("Pocket export written to output/pocket.zip", file=sys.stderr)
150 # Also write to stdout in the same format
151 writer = csv.DictWriter(sys.stdout, fieldnames=COLUMNS)
152 writer.writeheader()
153 for row in entries:
154 row_out = row.copy()
155 row_out['cursor'] = ''
156 writer.writerow(row_out)
157
158 # Output summary to stderr: count and date range
159 timestamps = [int(e["time_added"]) for e in entries if e["time_added"].isdigit() and int(e["time_added"]) > 0]
160 if timestamps:
161 min_ts = min(timestamps)
162 max_ts = max(timestamps)
163 min_dt = datetime.fromtimestamp(min_ts, tz=timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')
164 max_dt = datetime.fromtimestamp(max_ts, tz=timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')
165 print(f"Exported {len(entries)} entries. Date range: {min_dt} — {max_dt}", file=sys.stderr)
166 else:
167 print(f"Exported {len(entries)} entries. No valid date range.", file=sys.stderr)
168
169if __name__ == "__main__":
170 main()