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 = ["urllib3>=2.0.0", "tqdm>=4.66.0"]
5# ///
6"""
7Linkwarden API Import Script
8===========================
9
10Reads JSON objects (one per line) from stdin (as output by `just rss-stars linkwarden`),
11creates links in Linkwarden via its API, waits for archive status, and reports progress.
12
13Usage:
14 just rss-stars linkwarden | ./linkwarden_import.py --days 6
15
16- Input: JSON objects, one per line, with fields: title, url, externalURL, datePublished, dateArrived, uniqueID, tags, etc.
17- Output: Progress bar and stats to stderr, details for each link.
18
19AGENT-NOTE: CRITICAL FEATURES TO MAINTAIN
201. IDEMPOTENT: Multiple runs must produce identical results (skip duplicates)
212. ERROR HANDLING: Graceful degradation for missing data or API errors
223. RATE LIMITING: Pause after every N links
234. DETERMINISTIC: Sorted processing ensures consistent output
24
25AGENT-NOTE: HTTP/2 COMPATIBILITY SOLUTION
26Uses urllib3 directly instead of requests to avoid HTTP/2 compatibility issues with Caddy.
27urllib3 defaults to HTTP/1.1 which works reliably with Caddy's reverse proxy setup.
28See .agents/docs/caddy_ssl_analysis.md for detailed technical analysis.
29"""
30
31import sys
32import json
33import time
34import argparse
35import os
36import urllib3
37from tqdm import tqdm
38from datetime import datetime, timedelta, timezone
39
40# ANSI color codes for terminal output
41class Colors:
42 GREEN = '\033[92m'
43 BLUE = '\033[94m'
44 YELLOW = '\033[93m'
45 RED = '\033[91m'
46 PURPLE = '\033[95m'
47 CYAN = '\033[96m'
48 WHITE = '\033[97m'
49 BOLD = '\033[1m'
50 END = '\033[0m'
51
52def format_link_detail(entry, link_id=None, action="", details="", archived=False, error=None):
53 """Format link details with discussion action emojis for each platform"""
54 title = (entry.get("title") or "No title")[:60]
55 url = entry.get("externalURL") or entry.get("url", "No URL")
56
57 # Create title: url format with bold title
58 main_link = f"{Colors.BOLD}{title}{Colors.END}: {url}"
59
60 # Action emoji and color mapping
61 action_formats = {
62 "created": f"{Colors.GREEN}✨ CREATED{Colors.END}",
63 "updated": f"{Colors.BLUE}🔄 UPDATED{Colors.END}",
64 "exists": f"{Colors.YELLOW}📋 EXISTS{Colors.END}",
65 "duplicate": f"{Colors.PURPLE}🔗 DUPLICATE{Colors.END}",
66 "failed": f"{Colors.RED}❌ FAILED{Colors.END}"
67 }
68
69 # Get formatted action
70 action_display = action_formats.get(action, f"{Colors.WHITE}❓ {action.upper()}{Colors.END}")
71
72 # Build the main line
73 if error:
74 return f"{action_display} {main_link} | {Colors.RED}{error}{Colors.END}"
75
76 # Build details string
77 detail_parts = []
78 if archived:
79 detail_parts.append(f"{Colors.GREEN}📦 Archived{Colors.END}")
80
81 # Extract and format discussion links with individual action emojis
82 discussion_links = []
83 if details:
84 # Handle collection updates
85 if "Moved to" in details and "collection" in details:
86 collection_name = details.split("Moved to ")[1].split(" collection")[0]
87 detail_parts.append(f"📁 {Colors.BLUE}→{Colors.END} {collection_name}")
88
89 # Handle newly added discussion links
90 if "Added Lobsters link" in details:
91 discussion_links.append(f"💬 {Colors.GREEN}✨{Colors.END} [Lobsters](https://lobste.rs/...)")
92 elif "Added Hacker News link" in details:
93 discussion_links.append(f"💬 {Colors.GREEN}✨{Colors.END} [Hacker News](https://news.ycombinator.com/...)")
94 elif "Added Reddit link" in details:
95 discussion_links.append(f"💬 {Colors.GREEN}✨{Colors.END} [Reddit](https://reddit.com/...)")
96 elif details.startswith("Updated:") and "link" in details:
97 # Extract the aggregator name from the details
98 clean_details = details.replace("Updated: Added ", "").replace(" link to description", "")
99 discussion_links.append(f"💬 {Colors.GREEN}✨{Colors.END} {clean_details}")
100
101 # Handle existing discussion links
102 elif "Aggregator link already exists" in details:
103 # For existing links, we know they have discussion but don't know which platform
104 # We could enhance this by checking the original entry's aggregator URL
105 aggregator_url = entry.get("url", "")
106 if "lobste.rs" in aggregator_url:
107 discussion_links.append(f"💬 {Colors.YELLOW}📋{Colors.END} [Lobsters]({aggregator_url})")
108 elif "news.ycombinator.com" in aggregator_url:
109 discussion_links.append(f"💬 {Colors.YELLOW}📋{Colors.END} [Hacker News]({aggregator_url})")
110 elif "reddit.com" in aggregator_url:
111 discussion_links.append(f"💬 {Colors.YELLOW}📋{Colors.END} [Reddit]({aggregator_url})")
112 else:
113 discussion_links.append(f"💬 {Colors.YELLOW}📋{Colors.END} Available")
114
115 # Handle collection status messages
116 elif "Already in" in details and "collection" in details:
117 collection_name = details.split("Already in ")[1].split(" collection")[0]
118 detail_parts.append(f"📁 {Colors.YELLOW}✓{Colors.END} {collection_name}")
119
120 # Handle generic details that don't match specific patterns
121 elif details and not any(pattern in details for pattern in ["Added", "Aggregator", "Already in", "Moved to", "No updates needed"]):
122 detail_parts.append(f"{Colors.CYAN}ℹ️{Colors.END} {details}")
123
124
125 # Add discussion links to detail parts
126 detail_parts.extend(discussion_links)
127
128 detail_str = " | ".join(detail_parts) if detail_parts else ""
129
130 return f"{action_display} {main_link}" + (f" | {detail_str}" if detail_str else "")
131
132# Disable SSL warnings for self-signed certificates
133urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
134
135API_BASE = os.environ.get("LINKWARDEN_API_BASE", "https://linkwarden.homelab.local/api/v1")
136API_TOKEN = os.environ.get("LINKWARDEN_API_TOKEN")
137
138# Only require API token if not in dry-run mode
139if "--dry-run" not in sys.argv and not API_TOKEN:
140 print("ERROR: LINKWARDEN_API_TOKEN environment variable must be set", file=sys.stderr)
141 sys.exit(1)
142
143HEADERS = {"Authorization": f"Bearer {API_TOKEN}", "Content-Type": "application/json"} if API_TOKEN else {}
144
145RATE_LIMIT = 3 # links per batch
146PAUSE_SECS = 5 # seconds to pause after each batch
147ARCHIVE_POLL_SECS = 2 # seconds between archive status polls
148ARCHIVE_TIMEOUT_SECS = 30 # max seconds to wait for archive
149
150# Create urllib3 pool manager with SSL verification disabled for Caddy's internal TLS
151http = urllib3.PoolManager(cert_reqs='CERT_NONE')
152
153def get_or_create_collection(collection_name: str):
154 """Get collection ID by name, creating it if it doesn't exist"""
155 if not collection_name:
156 collection_name = "rss" # Default fallback
157
158 try:
159 # First, try to find existing collection
160 resp = http.request('GET', f"{API_BASE}/collections",
161 headers=HEADERS, timeout=10)
162
163 if resp.status == 200:
164 data = json.loads(resp.data.decode())
165 collections = data.get("response", [])
166
167 # Look for existing collection by name
168 for collection in collections:
169 if collection.get("name", "").lower() == collection_name.lower():
170 return collection.get("id"), f"Found existing {collection_name} collection (ID: {collection.get('id')})"
171
172 # Collection doesn't exist, create it
173 create_data = {
174 "name": collection_name,
175 "description": f"{collection_name} imported links",
176 "color": "#22c55e" # Green color
177 }
178
179 resp = http.request('POST', f"{API_BASE}/collections",
180 headers=HEADERS,
181 body=json.dumps(create_data).encode('utf-8'),
182 timeout=10)
183
184 if resp.status == 200:
185 data = json.loads(resp.data.decode())
186 collection_id = data.get("response", {}).get("id")
187 return collection_id, f"Created new {collection_name} collection (ID: {collection_id})"
188 else:
189 return None, f"Failed to create {collection_name} collection: HTTP {resp.status}"
190 else:
191 return None, f"Failed to fetch collections: HTTP {resp.status}"
192 except Exception as e:
193 return None, f"Error managing {collection_name} collection: {e}"
194
195# Global variable for collection cache
196COLLECTION_CACHE = {}
197
198def test_api_connection():
199 """Test API connection and token validity"""
200 try:
201 resp = http.request('GET', f"{API_BASE}/links", headers=HEADERS, timeout=10)
202 if resp.status == 200:
203 return True, "API connection successful"
204 elif resp.status == 401 or b"session has expired" in resp.data.lower():
205 return False, "API token expired - please refresh token in Linkwarden UI"
206 else:
207 return False, f"API error: {resp.status} {resp.data.decode()[:100]}"
208 except Exception as e:
209 return False, f"Connection error: {e}"
210
211def search_existing_link(external_url):
212 """Search for existing link by URL using the proper search API"""
213 if not external_url:
214 return None
215
216 try:
217 # Use the proper /api/v1/search endpoint with searchQueryString
218 import urllib.parse
219 encoded_url = urllib.parse.quote(external_url, safe='')
220
221 # Search using the URL as the search query string
222 resp = http.request('GET', f"{API_BASE}/search?searchQueryString={encoded_url}",
223 headers=HEADERS, timeout=15)
224
225 if resp.status == 200:
226 try:
227 data = json.loads(resp.data.decode())
228 # The search API returns data.links instead of response
229 links = data.get("data", {}).get("links", [])
230
231 # Look for exact URL match in search results
232 for link in links:
233 link_url = link.get("url", "")
234 if link_url == external_url:
235 return link
236
237 return None
238 except (json.JSONDecodeError, KeyError):
239 return None
240 else:
241 return None
242 except Exception:
243 return None
244
245def extract_aggregator_info(entry):
246 """Extract aggregator information from RSS entry"""
247 external_url = entry.get("externalURL")
248 aggregator_url = entry.get("url")
249
250 aggregator_name = None
251 if aggregator_url:
252 if "lobste.rs" in aggregator_url:
253 aggregator_name = "Lobsters"
254 elif "news.ycombinator.com" in aggregator_url:
255 aggregator_name = "Hacker News"
256 elif "reddit.com" in aggregator_url:
257 aggregator_name = "Reddit"
258
259 return external_url, aggregator_url, aggregator_name
260
261def update_link_fields(existing_link, entry, target_collection_id, collection_name, force_fields=None):
262 """Update link fields following links.md conflict resolution rules"""
263 if force_fields is None:
264 force_fields = set()
265
266 link_id = existing_link.get("id")
267 collection_data = existing_link.get("collection", {})
268 updates = []
269
270 # Prepare update data with existing values
271 update_data = {
272 "id": link_id,
273 "name": existing_link.get("name"),
274 "url": existing_link.get("url"),
275 "description": existing_link.get("description", ""),
276 "collection": {
277 "id": existing_link.get("collectionId"),
278 "ownerId": collection_data.get("ownerId")
279 },
280 "tags": existing_link.get("tags", [])
281 }
282
283 if existing_link.get("textContent"):
284 update_data["textContent"] = existing_link.get("textContent")
285
286 # Check collection update
287 if existing_link.get("collectionId") != target_collection_id:
288 update_data["collection"]["id"] = target_collection_id
289 updates.append(f"Moved to {collection_name} collection")
290
291 # Check title update (preserve existing unless forced)
292 new_title = entry.get("title", "")
293 current_title = existing_link.get("name", "")
294 if not current_title and new_title:
295 update_data["name"] = new_title
296 updates.append(f"Added missing title '{new_title}'")
297 elif current_title and new_title and current_title != new_title:
298 if "title" in force_fields:
299 update_data["name"] = new_title
300 updates.append(f"Force updated title to '{new_title}'")
301 else:
302 updates.append(f"Title conflict: '{current_title}' vs '{new_title}' (use --force title)")
303
304 # Check aggregator updates (merge approach)
305 external_url, aggregator_url, aggregator_name = extract_aggregator_info(entry)
306 if aggregator_url and aggregator_name and aggregator_url != (external_url or aggregator_url):
307 current_description = update_data["description"]
308 aggregator_link = f"[{aggregator_name}]({aggregator_url})"
309
310 if aggregator_link not in (current_description or ""):
311 if current_description:
312 update_data["description"] = f"{current_description}\n\n**Discussion:** {aggregator_link}"
313 else:
314 update_data["description"] = f"**Discussion:** {aggregator_link}"
315 updates.append(f"Added {aggregator_name} link")
316
317 # Apply updates if any changes needed
318 if updates:
319 # Check if we have actual data changes vs just warnings
320 has_data_changes = any(not msg.startswith("Title conflict:") for msg in updates)
321
322 if has_data_changes:
323 try:
324 resp = http.request('PUT', f"{API_BASE}/links/{link_id}",
325 headers=HEADERS,
326 body=json.dumps(update_data).encode('utf-8'),
327 timeout=10)
328
329 if resp.status == 200:
330 # Try to update timestamp after other updates
331 timestamp = entry.get("datePublished")
332 if timestamp:
333 ts_success, ts_msg = update_link_timestamp(link_id, timestamp)
334 if ts_success:
335 updates.append(ts_msg)
336
337 return True, " + ".join(updates)
338 else:
339 return False, f"Update failed: HTTP {resp.status}"
340 except Exception as e:
341 return False, f"Update error: {e}"
342 else:
343 # Only warnings, no actual updates - but try timestamp update
344 timestamp = entry.get("datePublished")
345 if timestamp:
346 ts_success, ts_msg = update_link_timestamp(link_id, timestamp)
347 if ts_success:
348 return True, ts_msg
349 return False, " + ".join(updates)
350
351 # No updates needed, but try timestamp update
352 timestamp = entry.get("datePublished")
353 if timestamp:
354 ts_success, ts_msg = update_link_timestamp(link_id, timestamp)
355 if ts_success:
356 return True, ts_msg
357
358 return False, "No updates needed"
359
360def update_link_timestamp(link_id, timestamp):
361 """Try to update link timestamp after creation"""
362 if not timestamp:
363 return False, "No timestamp provided"
364
365 try:
366 # Convert Unix timestamp to ISO string
367 import datetime
368 iso_date = datetime.datetime.fromtimestamp(float(timestamp), datetime.timezone.utc).isoformat()
369
370 # Try updating with just the ID and importDate field
371 update_data = {
372 "id": link_id,
373 "importDate": iso_date
374 }
375
376 resp = http.request('PUT', f"{API_BASE}/links/{link_id}",
377 headers=HEADERS,
378 body=json.dumps(update_data).encode('utf-8'),
379 timeout=10)
380
381 if resp.status == 200:
382 return True, f"Updated timestamp to {iso_date[:10]}"
383 else:
384 return False, f"Timestamp update failed: HTTP {resp.status}"
385 except Exception as e:
386 return False, f"Timestamp update error: {e}"
387
388def create_link_data(entry, collection_id):
389 """Create link data structure for new links"""
390 external_url, aggregator_url, aggregator_name = extract_aggregator_info(entry)
391 primary_url = external_url or aggregator_url
392
393 data = {
394 "name": entry.get("title") or primary_url,
395 "url": primary_url,
396 "description": entry.get("content") or "",
397 "collection": {"id": collection_id},
398 }
399
400 # Add textContent if available
401 if entry.get("textContent"):
402 data["textContent"] = entry.get("textContent")
403
404 # Add aggregator info to description if available
405 if aggregator_url and aggregator_name and aggregator_url != primary_url:
406 if data["description"]:
407 data["description"] += f"\n\n**Discussion:** [{aggregator_name}]({aggregator_url})"
408 else:
409 data["description"] = f"**Discussion:** [{aggregator_name}]({aggregator_url})"
410
411 # Add tags
412 tags = entry.get("tags")
413 if tags:
414 if isinstance(tags, str):
415 tags = [t.strip() for t in tags.split("|") if t.strip()]
416 data["tags"] = [{"name": t} for t in tags if t]
417
418 return data
419 """Create link data structure for new links"""
420 external_url, aggregator_url, aggregator_name = extract_aggregator_info(entry)
421 primary_url = external_url or aggregator_url
422
423 data = {
424 "name": entry.get("title") or primary_url,
425 "url": primary_url,
426 "description": entry.get("content") or "",
427 "collection": {"id": collection_id},
428 }
429
430 # Add textContent if available
431 if entry.get("textContent"):
432 data["textContent"] = entry.get("textContent")
433
434 # Add aggregator info to description if available
435 if aggregator_url and aggregator_name and aggregator_url != primary_url:
436 if data["description"]:
437 data["description"] += f"\n\n**Discussion:** [{aggregator_name}]({aggregator_url})"
438 else:
439 data["description"] = f"**Discussion:** [{aggregator_name}]({aggregator_url})"
440
441 # Add tags
442 tags = entry.get("tags")
443 if tags:
444 if isinstance(tags, str):
445 tags = [t.strip() for t in tags.split("|") if t.strip()]
446 data["tags"] = [{"name": t} for t in tags if t]
447
448 return data
449
450def update_link_description(existing_link, new_aggregator_url, aggregator_name):
451 """Update link description with new aggregator link"""
452 if not new_aggregator_url or not aggregator_name:
453 return False, "No aggregator info to add"
454
455 link_id = existing_link.get("id")
456 current_description = existing_link.get("description", "")
457
458 # Create markdown link for aggregator
459 aggregator_link = f"[{aggregator_name}]({new_aggregator_url})"
460
461 # Check if this aggregator link already exists in description
462 if aggregator_link in (current_description or ""):
463 return False, "Aggregator link already exists"
464
465 # Prepare updated description
466 if current_description:
467 # Add aggregator link at the end, separated by newlines
468 updated_description = f"{current_description}\n\n**Discussion:** {aggregator_link}"
469 else:
470 updated_description = f"**Discussion:** {aggregator_link}"
471
472 # Update the link - use correct OpenAPI format
473 collection_data = existing_link.get("collection", {})
474 update_data = {
475 "id": link_id,
476 "name": existing_link.get("name"),
477 "url": existing_link.get("url"),
478 "description": updated_description,
479 "collection": {
480 "id": existing_link.get("collectionId"),
481 "ownerId": collection_data.get("ownerId")
482 }
483 }
484
485 # Add tags - always include tags array (required by API)
486 existing_tags = existing_link.get("tags", [])
487 update_data["tags"] = existing_tags # Include even if empty
488
489 try:
490 resp = http.request('PUT', f"{API_BASE}/links/{link_id}",
491 headers=HEADERS,
492 body=json.dumps(update_data).encode('utf-8'),
493 timeout=10)
494
495 if resp.status == 200:
496 return True, f"Added {aggregator_name} link to description"
497 else:
498 return False, f"Update failed: HTTP {resp.status} - {resp.data.decode()[:100]}"
499 except Exception as e:
500 return False, f"Update error: {e}"
501
502def create_or_update_link(entry, force_fields=None):
503 """Create new link or update existing one with aggregator info"""
504 if force_fields is None:
505 force_fields = set()
506
507 external_url, aggregator_url, aggregator_name = extract_aggregator_info(entry)
508
509 # Use externalURL as primary, fall back to url
510 primary_url = external_url or aggregator_url
511 if not primary_url:
512 return None, "No URL found"
513
514 # Search for existing link
515 existing_link = search_existing_link(primary_url)
516
517 if existing_link:
518 # Link exists - check if we should update it
519 link_id = existing_link.get("id")
520
521 # Get target collection for this entry
522 collection_name = entry.get("folder", "rss")
523 if collection_name not in COLLECTION_CACHE:
524 collection_id, collection_msg = get_or_create_collection(collection_name)
525 if collection_id is None:
526 return None, f"Collection error: {collection_msg}"
527 COLLECTION_CACHE[collection_name] = collection_id
528 print(f"✓ {collection_msg}", file=sys.stderr)
529
530 target_collection_id = COLLECTION_CACHE[collection_name]
531
532 # Update all fields as needed
533 success, message = update_link_fields(existing_link, entry, target_collection_id, collection_name, force_fields)
534 if success:
535 return link_id, f"Updated: {message}"
536 else:
537 return link_id, f"Exists: {message}"
538
539 # Get collection for new link
540 collection_name = entry.get("folder", "rss")
541 if collection_name not in COLLECTION_CACHE:
542 collection_id, collection_msg = get_or_create_collection(collection_name)
543 if collection_id is None:
544 return None, f"Collection error: {collection_msg}"
545 COLLECTION_CACHE[collection_name] = collection_id
546 print(f"✓ {collection_msg}", file=sys.stderr)
547
548 collection_id = COLLECTION_CACHE[collection_name]
549
550 # Create new link
551 data = create_link_data(entry, collection_id)
552
553 try:
554 resp = http.request('POST', f"{API_BASE}/links",
555 headers=HEADERS,
556 body=json.dumps(data).encode('utf-8'),
557 timeout=10)
558
559 if resp.status == 200:
560 try:
561 response_data = json.loads(resp.data.decode())
562 link_id = response_data["response"]["id"]
563
564 # Try to update timestamp after creation
565 timestamp = entry.get("datePublished")
566 if timestamp:
567 success, msg = update_link_timestamp(link_id, timestamp)
568 if success:
569 return link_id, f"Created + {msg}"
570 # If timestamp update fails, still return success for the creation
571
572 return link_id, "Created"
573 except (json.JSONDecodeError, KeyError) as e:
574 return None, f"Response parsing error: {e}"
575 elif resp.status == 400 and b"already exists" in resp.data:
576 return None, "Duplicate"
577 elif resp.status == 409 and b"already exists" in resp.data:
578 return None, "Duplicate"
579 elif resp.status == 401 or b"session has expired" in resp.data.lower():
580 return None, "API token expired - please refresh token in Linkwarden UI"
581 else:
582 return None, f"API error: {resp.status} {resp.data.decode()[:100]}"
583 except Exception as e:
584 return None, f"Request error: {e}"
585
586def wait_for_archive(link_id):
587 start = time.time()
588 while time.time() - start < ARCHIVE_TIMEOUT_SECS:
589 try:
590 resp = http.request('GET', f"{API_BASE}/archives/{link_id}",
591 headers=HEADERS, timeout=10)
592 if resp.status == 200:
593 return True
594 elif resp.status == 404:
595 time.sleep(ARCHIVE_POLL_SECS)
596 else:
597 return False
598 except Exception:
599 time.sleep(ARCHIVE_POLL_SECS)
600 return False
601
602def wait_for_archive_with_progress(link_id, pbar):
603 """Enhanced archive waiting with progress reporting"""
604 start = time.time()
605 attempts = 0
606 max_attempts = ARCHIVE_TIMEOUT_SECS // ARCHIVE_POLL_SECS
607
608 while time.time() - start < ARCHIVE_TIMEOUT_SECS:
609 attempts += 1
610 elapsed = time.time() - start
611
612 # Update progress bar with archive status
613 pbar.set_description(f"Archiving link {link_id} ({attempts}/{max_attempts}, {elapsed:.1f}s)")
614
615 try:
616 resp = http.request('GET', f"{API_BASE}/archives/{link_id}",
617 headers=HEADERS, timeout=10)
618 if resp.status == 200:
619 pbar.set_description(f"✅ Archived link {link_id} in {elapsed:.1f}s")
620 time.sleep(0.5) # Brief pause to show success message
621 return True
622 elif resp.status == 404:
623 time.sleep(ARCHIVE_POLL_SECS)
624 else:
625 pbar.set_description(f"❌ Archive failed for link {link_id} (HTTP {resp.status})")
626 return False
627 except Exception as e:
628 pbar.set_description(f"⚠️ Archive check error for link {link_id}: {str(e)[:30]}")
629 time.sleep(ARCHIVE_POLL_SECS)
630
631 pbar.set_description(f"⏰ Archive timeout for link {link_id} after {ARCHIVE_TIMEOUT_SECS}s")
632 return False
633
634def main():
635 argp = argparse.ArgumentParser(description="Import links to Linkwarden via API")
636 argp.add_argument("--days", type=int, default=6, help="Only include entries from the last N days (by dateArrived or datePublished, -1 for all)")
637 argp.add_argument("--dry-run", action="store_true", help="Count and preview entries without importing")
638 argp.add_argument("--force", action="append", choices=["title"], help="Force update specified fields even if they differ (can be used multiple times)")
639 args = argp.parse_args()
640
641 # Convert force list to set for easier checking
642 if args.force:
643 args.force = set(args.force)
644 else:
645 args.force = set()
646
647 # Test API connection first (skip for dry-run)
648 if not args.dry_run:
649 api_ok, api_msg = test_api_connection()
650 if not api_ok:
651 print(f"API connection failed: {api_msg}", file=sys.stderr)
652 sys.exit(1)
653 print(f"✓ {api_msg}", file=sys.stderr)
654
655 now = datetime.now(timezone.utc)
656 cutoff_timestamp = None
657 if args.days != -1:
658 cutoff_date = now - timedelta(days=args.days)
659 cutoff_timestamp = cutoff_date.timestamp()
660 print(f"📅 Filtering entries from last {args.days} days (since {cutoff_date.strftime('%Y-%m-%d %H:%M:%S')} UTC)", file=sys.stderr)
661
662 entries = []
663 total_processed = 0
664 for line in sys.stdin:
665 line = line.strip()
666 if not line:
667 continue
668 total_processed += 1
669 try:
670 entry = json.loads(line)
671 timestamp = entry.get("dateArrived") or entry.get("datePublished")
672 if cutoff_timestamp is not None:
673 if timestamp is None or float(timestamp) < cutoff_timestamp:
674 continue
675 entries.append(entry)
676 except Exception as e:
677 print(f"Warning: Skipping invalid line {total_processed}: {e}", file=sys.stderr)
678
679 # Sort for deterministic order
680 entries.sort(key=lambda x: x.get("dateArrived") or x.get("datePublished") or 0)
681
682 print(f"📊 Found {len(entries)} entries to import (from {total_processed} total RSS entries)", file=sys.stderr)
683
684 if args.dry_run:
685 print("🔍 DRY RUN MODE - Preview of entries to be imported:", file=sys.stderr)
686 for i, entry in enumerate(entries[:10], 1): # Show first 10
687 title = entry.get("title", "No title")[:60]
688 url = entry.get("externalURL") or entry.get("url", "No URL")
689 timestamp = entry.get("dateArrived") or entry.get("datePublished")
690 date_str = datetime.fromtimestamp(timestamp, timezone.utc).strftime('%Y-%m-%d') if timestamp else "No date"
691 print(f" {i:3d}. [{date_str}] {title} - {url}", file=sys.stderr)
692
693 if len(entries) > 10:
694 print(f" ... and {len(entries) - 10} more entries", file=sys.stderr)
695
696 estimated_time = (len(entries) / RATE_LIMIT) * PAUSE_SECS + len(entries) * 2 # rough estimate
697 print(f"⏱️ Estimated import time: ~{estimated_time/60:.1f} minutes", file=sys.stderr)
698 print(f"🔄 Smart rate limiting: {RATE_LIMIT} created links per batch, {PAUSE_SECS}s pause between batches", file=sys.stderr)
699 print(f"ℹ️ Note: Only new link creation triggers rate limiting (existing links are processed quickly)", file=sys.stderr)
700 return
701
702 if len(entries) == 0:
703 print("ℹ️ No entries to import", file=sys.stderr)
704 return
705
706 # Warn for large datasets
707 if len(entries) > 50:
708 print(f"⚠️ Large dataset detected ({len(entries)} entries). This may take a while.", file=sys.stderr)
709 print(f"⏱️ Estimated time: ~{((len(entries) / RATE_LIMIT) * PAUSE_SECS + len(entries) * 2)/60:.1f} minutes", file=sys.stderr)
710 print(f"ℹ️ Note: Time depends on how many new links need to be created (existing links are fast)", file=sys.stderr)
711
712 stats = {"created": 0, "updated": 0, "failed": 0, "archived": 0, "duplicates": 0, "exists": 0}
713 details = []
714 created_count = 0 # Track only resource-intensive operations for rate limiting
715
716 with tqdm(total=len(entries), file=sys.stderr, desc="Importing links",
717 bar_format='{l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}, {rate_fmt}]') as pbar:
718
719 for i, entry in enumerate(entries):
720 # Update progress bar description with current batch info
721 batch_num = (created_count // RATE_LIMIT) + 1
722 pbar.set_description(f"Processing ({created_count} created)")
723
724 # Extract aggregator info for display
725 external_url, aggregator_url, aggregator_name = extract_aggregator_info(entry)
726
727 link_id, result = create_or_update_link(entry, args.force)
728
729 if link_id and result:
730 if result.startswith("Created"):
731 stats["created"] += 1
732 created_count += 1 # Count created links for rate limiting
733
734 # Enhanced archive waiting with progress
735 pbar.set_description(f"Processing ({created_count} created) - Archiving link {link_id}")
736 archived = wait_for_archive_with_progress(link_id, pbar)
737 if archived:
738 stats["archived"] += 1
739
740 details.append({
741 "title": entry.get("title"),
742 "url": entry.get("externalURL") or entry.get("url"),
743 "aggregator_url": aggregator_url,
744 "id": link_id,
745 "archived": archived,
746 "action": "created"
747 })
748 elif result.startswith("Updated"):
749 stats["updated"] += 1
750 # Updates might also be resource-intensive if they trigger re-archiving
751 # But typically they're just description updates, so don't count for rate limiting
752 details.append({
753 "title": entry.get("title"),
754 "url": entry.get("externalURL") or entry.get("url"),
755 "aggregator_url": aggregator_url,
756 "id": link_id,
757 "archived": False, # Don't re-archive updated links
758 "action": "updated",
759 "details": result
760 })
761 elif result.startswith("Exists"):
762 stats["exists"] += 1
763 # Existing links don't consume resources, no rate limiting needed
764 details.append({
765 "title": entry.get("title"),
766 "url": entry.get("externalURL") or entry.get("url"),
767 "aggregator_url": aggregator_url,
768 "id": link_id,
769 "archived": False,
770 "action": "exists",
771 "details": result
772 })
773 elif result == "Duplicate":
774 stats["duplicates"] += 1
775 # Duplicates don't consume resources, no rate limiting needed
776 details.append({
777 "title": entry.get("title"),
778 "url": entry.get("externalURL") or entry.get("url"),
779 "aggregator_url": aggregator_url,
780 "error": "Duplicate"
781 })
782 else:
783 stats["failed"] += 1
784 details.append({
785 "title": entry.get("title"),
786 "url": entry.get("externalURL") or entry.get("url"),
787 "aggregator_url": aggregator_url,
788 "error": result
789 })
790
791 pbar.update(1)
792
793 # Smart rate limiting: only pause after creating RATE_LIMIT new links
794 if created_count > 0 and created_count % RATE_LIMIT == 0 and (i + 1) < len(entries):
795 pbar.set_description(f"Rate limiting pause ({PAUSE_SECS}s) - {created_count} links created")
796 time.sleep(PAUSE_SECS)
797
798 # Print summary
799 print(f"\n✅ Import complete! Processed {len(entries)} entries", file=sys.stderr)
800 print(json.dumps(stats, indent=2), file=sys.stderr)
801
802 # Print detailed results with colored markdown formatting
803 print(f"\n{Colors.BOLD}📋 Detailed Results:{Colors.END}", file=sys.stderr)
804 for detail in details:
805 # Pass the full entry data including aggregator URL
806 entry_data = {
807 "title": detail.get("title"),
808 "externalURL": detail.get("url"),
809 "url": detail.get("aggregator_url", "") # Include aggregator URL for platform detection
810 }
811 formatted_line = format_link_detail(
812 entry_data,
813 action=detail.get("action", "failed" if detail.get("error") else "unknown"),
814 details=detail.get("details", ""),
815 archived=detail.get("archived", False),
816 error=detail.get("error")
817 )
818 print(formatted_line, file=sys.stderr)
819
820if __name__ == "__main__":
821 main()