Source code for platypush.plugins.wallabag

import json
import os
import pathlib
import time
from datetime import datetime, timedelta
from typing import Iterable, List, Optional
from urllib.parse import urljoin

import requests

from platypush.config import Config
from platypush.plugins import Plugin, action
from platypush.schemas.wallabag import WallabagEntrySchema


[docs] class WallabagPlugin(Plugin): """ Plugin to interact with Wallabag (https://wallabag.it), an open-source alternative to Instapaper and Pocket. """ _default_credentials_file = os.path.join( str(Config.get('workdir')), 'wallabag', 'credentials.json' ) _api_timeout = 15
[docs] def __init__( self, client_id: str, client_secret: str, server_url: str = 'https://wallabag.it', username: Optional[str] = None, password: Optional[str] = None, credentials_file: str = _default_credentials_file, **kwargs, ): """ :param client_id: Client ID for your application - you can create one at ``<server_url>/developer``. :param client_secret: Client secret for your application - you can create one at ``<server_url>/developer``. :param server_url: Base URL of the Wallabag server (default: ``https://wallabag.it``). :param username: Wallabag username. This configuration attribute is required for the first login. You can delete afterwards. However, note that, without username and password, the session is valid only as long as the ``refresh_token`` returned by the API is valid (that's usually a couple of months by default). When the ``refresh_token`` expires, you will have to either add again this attribute to the configuration and restart the app (so the service can create a new session), or manually provide your credentials through the :meth:`.login` action. :param password: Wallabag password. This configuration attribute is required for the first login. You can delete afterwards. However, note that, without username and password, the session is valid only as long as the ``refresh_token`` returned by the API is valid (that's usually a couple of months by default). When the ``refresh_token`` expires, you will have to either add again this attribute to the configuration and restart the app (so the service can create a new session), or manually provide your credentials through the :meth:`.login` action. :param credentials_file: Path to the file where the OAuth session parameters will be stored (default: ``<WORKDIR>/wallabag/credentials.json``). """ super().__init__(**kwargs) self._client_id = client_id self._client_secret = client_secret self._server_url = server_url self._username = username self._password = password self._credentials_file = os.path.expanduser(credentials_file) self._session = {}
def _oauth_open_saved_session(self): try: with open(self._credentials_file, 'r') as f: data = json.load(f) except Exception as e: self.logger.warning('Could not load %s: %s', self._credentials_file, e) return self._session = { 'username': data['username'], 'client_id': data.get('client_id', self._client_id), 'client_secret': data.get('client_secret', self._client_secret), 'access_token': data['access_token'], 'refresh_token': data['refresh_token'], } if data.get('expires_at') and time.time() > data['expires_at']: self.logger.info('OAuth token expired, refreshing it') self._oauth_refresh_token() def _oauth_refresh_token(self): url = urljoin(self._server_url, '/oauth/v2/token') payload = { 'grant_type': 'refresh_token', 'client_id': self._client_id, 'client_secret': self._client_secret, 'access_token': self._session['access_token'], 'refresh_token': self._session['refresh_token'], } if self._username and self._password: payload['username'] = self._username payload['password'] = self._password rs = requests.post( url, timeout=self._api_timeout, json=payload, ) try: rs.raise_for_status() except Exception as e: self.logger.warning( 'Failure occurred when refreshing the OAuth token, creating a new session: %s', e, ) assert self._username and self._password, ( "Can't refresh the OAuth session: username and password have not been provided. " "Please either provide them in the configuration and restart " "the service, or call the `wallabag.login` action" ) self._oauth_create_new_session(self._username, self._password) return rs = rs.json() self._session.update( { 'access_token': rs['access_token'], 'refresh_token': rs['refresh_token'], 'expires_at': ( int( ( datetime.now() + timedelta(seconds=rs['expires_in']) ).timestamp() ) if rs.get('expires_in') else None ), } ) self._oauth_flush_session() def _oauth_create_new_session(self, username: str, password: str): url = urljoin(self._server_url, '/oauth/v2/token') rs = requests.post( url, timeout=self._api_timeout, json={ 'grant_type': 'password', 'client_id': self._client_id, 'client_secret': self._client_secret, 'username': username, 'password': password, }, ) rs.raise_for_status() rs = rs.json() self._session = { 'client_id': self._client_id, 'client_secret': self._client_secret, 'username': username, 'access_token': rs['access_token'], 'refresh_token': rs['refresh_token'], 'expires_at': ( int((datetime.now() + timedelta(seconds=rs['expires_in'])).timestamp()) if rs.get('expires_in') else None ), } self._oauth_flush_session() def _oauth_flush_session(self): pathlib.Path(self._credentials_file).parent.mkdir(parents=True, exist_ok=True) pathlib.Path(self._credentials_file).touch(mode=0o600, exist_ok=True) with open(self._credentials_file, 'w') as f: f.write(json.dumps(self._session))
[docs] @action def login(self, username: Optional[str] = None, password: Optional[str] = None): """ Create a new user session if not logged in. :param username: Default ``username`` override. :param password: Default ``password`` override. """ self._oauth_open_saved_session() if self._session: return username = username or self._username password = password or self._password assert ( username and password ), 'No stored user session and no username/password provided' self._oauth_create_new_session(username, password)
def _request(self, url: str, method: str, *args, as_json=True, **kwargs): url = urljoin(self._server_url, f'api/{url}') func = getattr(requests, method.lower()) self.login() kwargs['headers'] = { **kwargs.get('headers', {}), 'Authorization': f'Bearer {self._session["access_token"]}', } rs = func(url, *args, **kwargs) rs.raise_for_status() return rs.json() if as_json else rs.text
[docs] @action def list( self, archived: bool = True, starred: bool = False, sort: str = 'created', descending: bool = False, page: int = 1, limit: int = 30, tags: Optional[Iterable[str]] = None, since: Optional[int] = None, full: bool = True, ) -> List[dict]: """ List saved links. :param archived: Include archived items (default: ``True``). :param starred: Include only starred items (default: ``False``). :param sort: Timestamp sort criteria. Supported: ``created``, ``updated``, ``archived`` (default: ``created``). :param descending: Sort in descending order (default: ``False``). :param page: Results page to be retrieved (default: ``1``). :param limit: Maximum number of entries per page (default: ``30``). :param tags: Filter by a list of tags. :param since: Return entries created after this timestamp (as a UNIX timestamp). :param full: Include the full parsed body of the saved entry. :return: .. schema:: wallabag.WallabagEntrySchema(many=True) """ rs = self._request( '/entries.json', method='get', params={ 'archived': int(archived), 'starred': int(starred), 'sort': sort, 'order': 'desc' if descending else 'asc', 'page': page, 'perPage': limit, 'tags': ','.join(tags or []), 'since': since or 0, 'detail': 'full' if full else 'metadata', }, ) return WallabagEntrySchema().dump( rs.get('_embedded', {}).get('items', []), many=True )
[docs] @action def search( self, term: str, page: int = 1, limit: int = 30, ) -> List[dict]: """ Search links by some text. :param term: Term to be searched. :param page: Results page to be retrieved (default: ``1``). :param limit: Maximum number of entries per page (default: ``30``). :return: .. schema:: wallabag.WallabagEntrySchema(many=True) """ rs = self._request( '/search.json', method='get', params={ 'term': term, 'page': page, 'perPage': limit, }, ) return WallabagEntrySchema().dump( rs.get('_embedded', {}).get('items', []), many=True )
[docs] @action def get(self, id: int) -> Optional[dict]: # pylint: disable=redefined-builtin """ Get the content and metadata of a link by ID. :param id: Entry ID. :return: .. schema:: wallabag.WallabagEntrySchema """ rs = self._request(f'/entries/{id}.json', method='get') if not rs: return None return dict(WallabagEntrySchema().dump(rs))
[docs] @action def export( self, id: int, # pylint: disable=redefined-builtin file: str, format: str = 'txt', # pylint: disable=redefined-builtin ): """ Export a saved entry to a file in the specified format. :param id: Entry ID. :param file: Output filename. :param format: Output format. Supported: ``txt``, ``xml``, ``csv``, ``pdf``, ``epub`` and ``mobi`` (default: ``txt``). """ rs = self._request( f'/entries/{id}/export.{format}', method='get', as_json=False ) if isinstance(rs, str): rs = rs.encode() with open(os.path.expanduser(file), 'wb') as f: f.write(rs)
[docs] @action def save( self, url: str, title: Optional[str] = None, content: Optional[str] = None, tags: Optional[Iterable[str]] = None, authors: Optional[Iterable[str]] = None, archived: bool = False, starred: bool = False, public: bool = False, language: Optional[str] = None, preview_picture: Optional[str] = None, ) -> Optional[dict]: """ Save a link to Wallabag. :param url: URL to be saved. :param title: Entry title (default: parsed from the page content). :param content: Entry content (default: parsed from the entry itself). :param tags: List of tags to attach to the entry. :param authors: List of authors of the entry (default: parsed from the page content). :param archived: Whether the entry should be created in the archive (default: ``False``). :param starred: Whether the entry should be starred (default: ``False``). :param public: Whether the entry should be publicly available. If so, a public URL will be generated (default: ``False``). :param language: Language of the entry. :param preview_picture: URL of a picture to be used for the preview (default: parsed from the page itself). :return: .. schema:: wallabag.WallabagEntrySchema """ rs = self._request( '/entries.json', method='post', json={ 'url': url, 'title': title, 'content': content, 'tags': ','.join(tags or []), 'authors': ','.join(authors or []), 'archive': int(archived), 'starred': int(starred), 'public': int(public), 'language': language, 'preview_picture': preview_picture, }, ) return WallabagEntrySchema().dump(rs) # type: ignore
[docs] @action def update( self, id: int, # pylint: disable=redefined-builtin title: Optional[str] = None, content: Optional[str] = None, tags: Optional[Iterable[str]] = None, authors: Optional[Iterable[str]] = None, archived: bool = False, starred: bool = False, public: bool = False, language: Optional[str] = None, preview_picture: Optional[str] = None, ) -> Optional[dict]: """ Update a link entry saved to Wallabag. :param id: Entry ID. :param title: New entry title. :param content: New entry content. :param tags: List of tags to attach to the entry. :param authors: List of authors of the entry. :param archived: Archive/unarchive the entry. :param starred: Star/unstar the entry. :param public: Mark the entry as public/private. :param language: Change the language of the entry. :param preview_picture: Change the preview picture URL. :return: .. schema:: wallabag.WallabagEntrySchema """ rs = self._request( f'/entries/{id}.json', method='patch', json={ 'title': title, 'content': content, 'tags': ','.join(tags or []), 'authors': ','.join(authors or []), 'archive': int(archived), 'starred': int(starred), 'public': int(public), 'language': language, 'preview_picture': preview_picture, }, ) if not rs: return None return dict(WallabagEntrySchema().dump(rs))
[docs] @action def delete(self, id: int) -> Optional[dict]: # pylint: disable=redefined-builtin """ Delete an entry by ID. :param id: Entry ID. :return: .. schema:: wallabag.WallabagEntrySchema """ rs = self._request( f'/entries/{id}.json', method='delete', ) return WallabagEntrySchema().dump(rs) # type: ignore