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
