My Notes

Using the Pocket API with a Custom Script

6 min read

Detailed Guide: Using the Pocket API with a Custom Script

Let me elaborate on Method 2, which gives you the most control over the synchronization process between Firefox Pocket and ArchiveBox.

Step 1: Set Up Pocket API Access

  1. Create a Pocket API application:

    • Go to https://getpocket.com/developer/apps/new
    • Fill in the application details:
      • Application name: “PocketToArchiveBox”
      • Application description: “Sync Pocket items to ArchiveBox”
      • Permissions: Check “Retrieve” (we only need to read your list)
      • Platforms: Check “Web”
    • Click “Create Application”
    • You’ll receive a Consumer Key - save this for later use

  2. Obtain an Access Token:

    • This is a bit tricky as it requires a three-step OAuth process
    • Let’s create a simple script to handle this authentication:
import requests
import webbrowser
import time
 
CONSUMER_KEY = "114230-e320d22731d275a71b062c1"
REDIRECT_URI = "http://localhost"  # This can be any URL, we'll intercept the redirect
 
# Step 1: Get a request token
request_data = {
    "consumer_key": CONSUMER_KEY,
    "redirect_uri": REDIRECT_URI
}
response = requests.post("https://getpocket.com/v3/oauth/request", json=request_data)
request_token = response.json()["code"]
print(f"Request token: {request_token}")
 
# Step 2: Redirect user to authorization page
auth_url = f"https://getpocket.com/auth/authorize?request_token={request_token}&redirect_uri={REDIRECT_URI}"
print(f"Opening browser to authorize application: {auth_url}")
webbrowser.open(auth_url)
 
# Wait for user to authorize
input("Please authorize the app in your browser, then press Enter to continue...")
 
# Step 3: Convert request token to access token
auth_data = {
    "consumer_key": CONSUMER_KEY,
    "code": request_token
}
response = requests.post("https://getpocket.com/v3/oauth/authorize", json=auth_data)
access_token = response.json()["access_token"]
print(f"Your access token is: {access_token}")
print("Save this token for future use with your script.")

Run this script once to get your access token. Save both the Consumer Key and Access Token securely.

Step 2: Create the Synchronization Script

Now, let’s create a more comprehensive script that:

  1. Retrieves items from Pocket
  2. Filters them if needed (e.g., only unarchived items)
  3. Keeps track of already processed items to avoid duplicates
  4. Adds new URLs to ArchiveBox
#!/usr/bin/env python3
import requests
import subprocess
import json
import os
import logging
from datetime import datetime
 
# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    filename='pocket_to_archivebox.log'
)
 
# Configuration
CONSUMER_KEY = "your_consumer_key"
ACCESS_TOKEN = "your_access_token"
ARCHIVEBOX_PATH = "/path/to/your/archivebox"  # Change this to your ArchiveBox directory
STATE_FILE = os.path.expanduser("~/.pocket_archivebox_state.json")
# Options: 'unread', 'archive', or 'all'
POCKET_STATE = "all"  # Change to 'unread' if you only want unread items
# Options: newest, oldest, title, site
POCKET_SORT = "newest"
# Maximum number of items to retrieve (0 for all)
MAX_ITEMS = 50
 
def load_state():
    """Load the state file containing previously processed items"""
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE, 'r') as f:
            return json.load(f)
    return {"processed_items": [], "last_run": None}
 
def save_state(state):
    """Save the state to keep track of processed items"""
    with open(STATE_FILE, 'w') as f:
        json.dump(state, f, indent=2)
 
def get_pocket_items():
    """Retrieve items from Pocket API"""
    logging.info("Retrieving items from Pocket")
    
    payload = {
        "consumer_key": CONSUMER_KEY,
        "access_token": ACCESS_TOKEN,
        "state": POCKET_STATE,
        "sort": POCKET_SORT,
        "detailType": "simple",
        "count": MAX_ITEMS if MAX_ITEMS > 0 else None
    }
    
    try:
        response = requests.post("https://getpocket.com/v3/get", json=payload)
        response.raise_for_status()
        data = response.json()
        
        # Extract URLs and item IDs
        items = []
        for item_id, item in data.get("list", {}).items():
            url = item.get("resolved_url") or item.get("given_url")
            if url:
                items.append({
                    "id": item_id,
                    "url": url,
                    "title": item.get("given_title", ""),
                    "time_added": item.get("time_added", "")
                })
        
        logging.info(f"Retrieved {len(items)} items from Pocket")
        return items
    
    except Exception as e:
        logging.error(f"Error retrieving Pocket items: {e}")
        return []
 
def add_to_archivebox(urls):
    """Add URLs to ArchiveBox"""
    if not urls:
        logging.info("No new URLs to add to ArchiveBox")
        return True
    
    try:
        # Change to ArchiveBox directory
        os.chdir(ARCHIVEBOX_PATH)
        
        # Create a temporary file with URLs
        temp_file = os.path.join(ARCHIVEBOX_PATH, "pocket_urls.txt")
        with open(temp_file, 'w') as f:
            for url in urls:
                f.write(f"{url}\n")
        
        # Add URLs to ArchiveBox using the file
        cmd = f"archivebox add --depth=1 < {temp_file}"
        logging.info(f"Running: {cmd}")
        result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
        
        # Clean up
        os.remove(temp_file)
        
        if result.returncode != 0:
            logging.error(f"ArchiveBox error: {result.stderr}")
            return False
        
        logging.info(f"Successfully added {len(urls)} URLs to ArchiveBox")
        return True
    
    except Exception as e:
        logging.error(f"Error adding to ArchiveBox: {e}")
        return False
 
def main():
    """Main function to sync Pocket to ArchiveBox"""
    logging.info("Starting Pocket to ArchiveBox sync")
    
    # Load state
    state = load_state()
    processed_ids = set(state["processed_items"])
    
    # Get items from Pocket
    items = get_pocket_items()
    
    # Filter out already processed items
    new_items = [item for item in items if item["id"] not in processed_ids]
    logging.info(f"Found {len(new_items)} new items to process")
    
    if new_items:
        # Extract URLs from new items
        urls = [item["url"] for item in new_items]
        
        # Add to ArchiveBox
        if add_to_archivebox(urls):
            # Update state with successfully processed items
            for item in new_items:
                processed_ids.add(item["id"])
    
    # Update and save state
    state["processed_items"] = list(processed_ids)
    state["last_run"] = datetime.now().isoformat()
    save_state(state)
    
    logging.info("Sync completed")
 
if __name__ == "__main__":
    main()

Step 3: Make the Script Executable and Test It

  1. Save the script to a file named pocket_to_archivebox.py

  2. Make it executable:

    chmod +x pocket_to_archivebox.py

  3. Run it manually to test:

    ./pocket_to_archivebox.py

  4. Check the logs to verify it’s working correctly:

    cat pocket_to_archivebox.log

Step 4: Set Up a Cron Job for Automation

  1. Edit your crontab:

    crontab -e

  2. Add a scheduled task (this example runs daily at 3 AM):

    0 3 * * * /full/path/to/pocket_to_archivebox.py

  3. For more frequent syncing (every 4 hours):

    0 */4 * * * /full/path/to/pocket_to_archivebox.py

Additional Customization Options

Filtering Options

You can modify the script to filter items based on specific criteria:

# Add this inside get_pocket_items() function after retrieving the data
filtered_items = []
for item_id, item in data.get("list", {}).items():
    # Only include items with specific tags
    if 'tags' in item and 'archive-this' in item['tags']:
        # Process item...
        
    # Only include items from specific sites
    if 'given_url' in item and 'github.com' in item['given_url']:
        # Process item...

Adding Tags in ArchiveBox

You can pass additional metadata to ArchiveBox:

# Modify the add_to_archivebox function
def add_to_archivebox(items):
    for item in items:
        url = item["url"]
        title = item["title"]
        tag = "pocket"
        
        cmd = f'archivebox add "{url}" --tag "{tag}" --title "{title}"'
        # Run command...

Archiving After Syncing

You might want to archive items in Pocket after they’ve been added to ArchiveBox:

def archive_pocket_items(item_ids):
    """Archive items in Pocket after they've been added to ArchiveBox"""
    payload = {
        "consumer_key": CONSUMER_KEY,
        "access_token": ACCESS_TOKEN,
        "actions": [{"action": "archive", "item_id": item_id} for item_id in item_ids]
    }
    
    try:
        response = requests.post("https://getpocket.com/v3/send", json=payload)
        response.raise_for_status()
        return True
    except Exception as e:
        logging.error(f"Error archiving Pocket items: {e}")
        return False

This script provides a solid foundation that you can customize based on your specific needs. Let me know if you need any clarification or have specific requirements you’d like to incorporate!

Graph View