feat(wallabag): import readwise reader documents to wallabag

README.md

@@ -73,12 +73,18 @@ Run continuously and push new annotations to Readwise every 60 minutes.
 wallabag2readwise daemon --wait-time 60
 ```
 
-#### Import all Wallabag entries to Readwise reader
+#### Import all Wallabag entries to Readwise Reader
 
 ```bash
 wallabag2readwise reader
 ```
 
+#### Import all Readwise Reader Documents to Wallabag
+
+```bash
+wallabag2readwise wallabag
+```
+
 ### Configuration
 
 Get a new Readwise API Token from <https://readwise.io/access_token>.

requirements.txt

@@ -1,4 +1,4 @@
-readwise==1.0.14
+readwise==1.1.0
 requests==2.28.2
 rich==13.3.2
 typer==0.8.0

wallabag2readwise/cli.py

@@ -4,6 +4,7 @@
 import typer
 from readwise import Readwise, ReadwiseReader
 
+from wallabag2readwise.logging import _setup_logging
 from wallabag2readwise.misc import push_annotations
 from wallabag2readwise.output import console
 from wallabag2readwise.wallabag import WallabagConnector
@@ -21,7 +22,10 @@ def push(
     wallabag_client_id: str = typer.Option(..., envvar='WALLABAG_CLIENT_ID'),
     wallabag_client_secret: str = typer.Option(..., envvar='WALLABAG_CLIENT_SECRET'),
     readwise_token: str = typer.Option(..., envvar='READWISE_TOKEN', prompt=True),
+    verbose: bool = typer.Option(False, '--verbose', '-v'),
+    debug: bool = typer.Option(False, '--debug'),
 ):
+    _setup_logging(verbose, debug)
     wallabag = WallabagConnector(
         wallabag_url,
         wallabag_user,
@@ -46,7 +50,10 @@ def daemon(
     wait_time: int = typer.Option(
         60, help='time to wait between runs in minutes', envvar='WAIT_TIME'
     ),
+    verbose: bool = typer.Option(False, '--verbose', '-v'),
+    debug: bool = typer.Option(False, '--debug'),
 ) -> None:
+    _setup_logging(verbose, debug)
     console.print(f'> Starting daemon with {wait_time} minutes wait time')
     while True:
         wallabag = WallabagConnector(
@@ -73,7 +80,10 @@ def reader(
     wallabag_client_id: str = typer.Option(..., envvar='WALLABAG_CLIENT_ID'),
     wallabag_client_secret: str = typer.Option(..., envvar='WALLABAG_CLIENT_SECRET'),
     readwise_token: str = typer.Option(..., envvar='READWISE_TOKEN', prompt=True),
+    verbose: bool = typer.Option(False, '--verbose', '-v'),
+    debug: bool = typer.Option(False, '--debug'),
 ):
+    _setup_logging(verbose, debug)
     console.print('> Starting readwise reader import')
     wallabag = WallabagConnector(
         wallabag_url,
@@ -95,6 +105,52 @@ def reader(
             console.print(f'==> Error: {e}')
 
 
+@app.command(help='Import Readwise Reader Documents to Wallabag')
+def wallabag(
+    wallabag_url: str = typer.Option(
+        ..., envvar='WALLABAG_URL', help='url to your wallabag instance'
+    ),
+    wallabag_user: str = typer.Option(..., envvar='WALLABAG_USER'),
+    wallabag_password: str = typer.Option(..., envvar='WALLABAG_PASSWORD', prompt=True),
+    wallabag_client_id: str = typer.Option(..., envvar='WALLABAG_CLIENT_ID'),
+    wallabag_client_secret: str = typer.Option(..., envvar='WALLABAG_CLIENT_SECRET'),
+    readwise_token: str = typer.Option(..., envvar='READWISE_TOKEN', prompt=True),
+    verbose: bool = typer.Option(False, '--verbose', '-v'),
+    debug: bool = typer.Option(False, '--debug'),
+):
+    _setup_logging(verbose, debug)
+    console.print('> Starting wallabag import')
+    wallabag = WallabagConnector(
+        wallabag_url,
+        wallabag_user,
+        wallabag_password,
+        wallabag_client_id,
+        wallabag_client_secret,
+    )
+    readwise = ReadwiseReader(readwise_token)
+    console.print('=> Getting wallabag entries')
+    wallabag_entries = list(wallabag.get_entries())
+    readwise_documents = readwise.get_documents({'category': 'article'})
+    wallabag_urls = [e.url for e in wallabag_entries]
+    wallabag_titles = [e.title for e in wallabag_entries]
+    for document in readwise_documents:
+        if (
+            document.source_url not in wallabag_urls
+            and document.title not in wallabag_titles
+        ):
+            try:
+                console.print(f'=> Importing {document.title}')
+                wallabag.create_entry(
+                    document.source_url,
+                    {'archive': 1 if document.location == 'archive' else 0},
+                )
+                sleep(2)
+            except Exception as e:
+                console.print(f'==> Error: {e}')
+        else:
+            console.print(f'=> Skipping {document.title}')
+
+
 @app.command()
 def version():
     console.print(importlib.metadata.version('wallabag2readwise'))

wallabag2readwise/logging.py

@@ -1,13 +1,27 @@
 import logging
-from os import getenv
 
-log_level = getenv('LOG_LEVEL', 'WARNING')
 logger = logging.getLogger('wallabag2readwise')
 
-format = '%(levelname)s - %(asctime)s - %(name)s - %(message)s'
-stream_handler = logging.StreamHandler()
-stream_handler.setFormatter(logging.Formatter(format))
-stream_handler.setLevel(log_level)
 
-logger.addHandler(stream_handler)
-logger.setLevel('DEBUG')
+def _setup_logging(verbose: bool, debug: bool):
+    level = 'WARNING'
+    if verbose:
+        level = 'INFO'
+    if debug:
+        level = 'DEBUG'
+
+    format = '%(levelname)s - %(asctime)s - %(name)s - %(message)s'
+
+    stream_handler = logging.StreamHandler()
+    stream_handler.setFormatter(logging.Formatter(format))
+    stream_handler.setLevel(level)
+
+    logger.addHandler(stream_handler)
+
+    for handler in logging.root.handlers:
+        handler.setLevel(level)
+        handler.setFormatter(logging.Formatter(format))
+
+    logging.basicConfig(format=format, level=level)
+
+    logger.setLevel(level)

wallabag2readwise/wallabag.py

@@ -42,6 +42,7 @@ def _get_oauth_token(self):
                 'password': self.password,
             },
         )
+        response.raise_for_status()
         data = response.json()
         return data['access_token']
 
@@ -94,3 +95,8 @@ def get_annotations(
                 created_at=datetime.strptime(item['created_at'], '%y-%m-%dT%H:%M+%S'),
                 ranges=item['ranges'],
             )
+
+    def create_entry(self, url: str, params: dict = {}):
+        params.update({'url': url})
+        result = self.post('/api/entries.json', data=params)
+        result.raise_for_status()