import os import time import json import requests import logging DATA_STORE_PATH = "/data/last_update_time.txt" logging.basicConfig(level=logging.INFO) class APIHandler: def __init__(self, base_url, headers={}): self.base_url = base_url self.headers = headers def get(self, endpoint, params=None): response = requests.get(self.base_url + endpoint, params=params, headers=self.headers) response.raise_for_status() return response.json() def post(self, endpoint, data=None): response = requests.post(self.base_url + endpoint, data=json.dumps(data), headers=self.headers) response.raise_for_status() return response.status_code def get_last_update_time(): with open(DATA_STORE_PATH, 'r') as file: return file.read().strip() def update_last_update_time(new_time): with open(DATA_STORE_PATH, 'w') as file: file.write(new_time) def get_new_annotations(last_annotation_time): inoreader = APIHandler( "https://www.inoreader.com/reader/api/0/stream/contents", headers = { 'Authorization': 'Bearer ' + os.getenv("INOREADER_ACCESS_TOKEN") } ) inoreader_response = inoreader.get( "/user/-/state/com.google/annotated", params={ "annotations": 1, "n": 100, } ) data = json.loads(inoreader_response) all_annotations = [] for item in data["items"]: annotations = item.get("annotations", []) for annotation in annotations: annotation['title'] = item['title'] annotation['author'] = item['author'] all_annotations.append(annotation) return [annotation for annotation in all_annotations if annotation['added_on'] > last_annotation_time] def push_annotations_to_readwise(annotations): readwise = APIHandler( "https://readwise.io", headers = { 'Authorization': 'Token ' + os.getenv("READWISE_ACCESS_TOKEN"), 'Content-Type': 'application/json' } ) readwise.post( "/api/v2/highlights/", data={ 'highlights': [ { 'text': annotation['text'], 'title': annotation['title'], 'author': annotation['author'], 'note': annotation['note'], } for annotation in annotations ] } ) def main(): while True: try: last_annotation_time = get_last_update_time() new_annotations = get_new_annotations(last_annotation_time) if new_annotations: latest_added_on = max(annotation['added_on'] for annotation in new_annotations) push_annotations_to_readwise(new_annotations) update_last_update_time(latest_added_on) else: logging.info("No new annotations found") time.sleep(3600) except Exception as e: logging.error(f"An error occurred: {e}") time.sleep(60) if __name__ == "__main__": main()