Last active
January 23, 2025 20:55
-
-
Save bcrpntr/d203fbe50362f41323d06ad252fc441a to your computer and use it in GitHub Desktop.
Archives Readwise Reader feed articles older than two weeks.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import requests | |
| import datetime | |
| import time | |
| # Replace 'XXX' with your Readwise API token | |
| token = 'XXX' | |
| # API URLs | |
| list_url = "https://readwise.io/api/v3/list/" | |
| update_url_template = "https://readwise.io/api/v3/update/{document_id}/" | |
| rate_limit = 20 # Requests per minute | |
| def fetch_documents(updated_after=None, location=None): | |
| """Fetch documents from Readwise API.""" | |
| full_data = [] | |
| next_page_cursor = None | |
| while True: | |
| params = {} | |
| if next_page_cursor: | |
| params['pageCursor'] = next_page_cursor | |
| if updated_after: | |
| params['updatedAfter'] = updated_after | |
| if location: | |
| params['location'] = location | |
| response = requests.get( | |
| url=list_url, | |
| params=params, | |
| headers={"Authorization": f"Token {token}"}, | |
| ) | |
| if response.status_code == 429: | |
| retry_after = int(response.headers.get("Retry-After", 60)) | |
| print(f"Rate limit hit. Retrying after {retry_after} seconds...") | |
| time.sleep(retry_after) | |
| continue | |
| if response.status_code != 200: | |
| print("Error fetching documents:", response.text) | |
| break | |
| response_json = response.json() | |
| full_data.extend(response_json.get('results', [])) | |
| next_page_cursor = response_json.get('nextPageCursor') | |
| if not next_page_cursor: | |
| break | |
| # Manage rate limiting | |
| time.sleep(60 / rate_limit) | |
| return full_data | |
| def archive_old_documents(): | |
| """Archive documents older than 2 weeks.""" | |
| today = datetime.datetime.now(datetime.timezone.utc) | |
| two_weeks_ago = today - datetime.timedelta(days=14) | |
| # Fetch all documents from the "new" location | |
| documents = fetch_documents(location="feed") | |
| for doc in documents: | |
| # Parse the document's saved date | |
| saved_date = datetime.datetime.fromisoformat(doc['saved_at'].replace('Z', '+00:00')) | |
| # Check if the document is older than 2 weeks | |
| if saved_date < two_weeks_ago: | |
| doc_id = doc['id'] | |
| update_url = update_url_template.format(document_id=doc_id) | |
| # Archive the document | |
| response = requests.patch( | |
| url=update_url, | |
| headers={"Authorization": f"Token {token}"}, | |
| json={"location": "archive"}, | |
| ) | |
| if response.status_code == 429: | |
| retry_after = int(response.headers.get("Retry-After", 60)) | |
| print(f"Rate limit hit. Retrying after {retry_after} seconds...") | |
| time.sleep(retry_after) | |
| continue | |
| if response.status_code == 200: | |
| print(f"Archived document: {doc['title']}") | |
| else: | |
| print(f"Failed to archive document {doc['title']}: {response.text}") | |
| # Manage rate limiting | |
| time.sleep(60 / rate_limit) | |
| if __name__ == "__main__": | |
| archive_old_documents() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment