Move all in its own package.

Signed-off-by: Jan Losinski <losinski@wh2.tu-dresden.de>

wallabag_kindle_consumer/consumer.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python3
+import asyncio
+
+from logbook import Logger
+from sqlalchemy.orm import joinedload
+
+from wallabag_kindle_consumer.models import User, Job
+
+logger = Logger(__name__)
+
+
+class Consumer:
+    def __init__(self, wallabag, session, sender, interval=30):
+        self.wallabag = wallabag
+        self.session = session
+        self.interval = interval
+        self.sender = sender
+        self.running = True
+
+    async def fetch_jobs(self, user):
+        logger.info("Fetch entries for user {}", user.name)
+        async for entry in self.wallabag.fetch_entries(user):
+            logger.info("Schedule job to send entry {}", entry.id)
+            job = Job(article=entry.id, title=entry.title, format=entry.tag.format)
+            user.jobs.append(job)
+            await self.wallabag.remove_tag(user, entry)
+
+    async def process_job(self, job):
+        logger.info("Process export for job {id} ({format})", id=job.article, format=job.format)
+        data = await self.wallabag.export_article(job.user, job.article, job.format)
+        await self.sender.send_mail(job, data)
+        self.session.delete(job)
+
+    async def consume(self):
+        while self.running:
+            logger.info("Start consume run")
+            fetches = [self.fetch_jobs(user) for user in self.session.query(User).all()]
+            await asyncio.gather(*fetches)
+            self.session.commit()
+
+            jobs = [self.process_job(job) for job in self.session.query(Job).options(joinedload('user'))]
+            await asyncio.gather(*jobs)
+            self.session.commit()
+
+            asyncio.sleep(self.interval)
+

wallabag_kindle_consumer/models.py

@@ -0,0 +1,46 @@
+from sqlalchemy import create_engine
+from sqlalchemy import Integer, String, DateTime, Column, ForeignKey, Enum
+from sqlalchemy.orm import scoped_session, sessionmaker, relationship
+from sqlalchemy.ext.declarative import declarative_base
+
+Base = declarative_base()
+
+
+class User(Base):
+    __tablename__ = "user"
+
+    name = Column(String, primary_key=True)
+    token = Column(String())
+    client_id = Column(String())
+    client_secret = Column(String())
+    auth_token = Column(String)
+    refresh_token = Column(String)
+    token_valid = Column(DateTime)
+    last_check = Column(DateTime)
+    email = Column(String)
+    kindle_mail = Column(String)
+
+    jobs = relationship('Job', backref='user')
+
+
+class Job(Base):
+    __tablename__ = "job"
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    article = Column(Integer)
+    title = Column(String)
+    user_name = Column(Integer, ForeignKey("user.name"))
+    format = Column(Enum('pdf', 'mobi'))
+
+
+def make_session(uri):
+    Session = sessionmaker(autocommit=False,
+                           autoflush=False,
+                           bind=create_engine(uri))
+    return scoped_session(Session)
+
+
+def re_create_db(uri):
+    engine = create_engine(uri)
+    Base.metadata.drop_all(engine)
+    Base.metadata.create_all(engine)

wallabag_kindle_consumer/refresher.py

@@ -0,0 +1,42 @@
+import asyncio
+from datetime import datetime, timedelta
+
+from logbook import Logger
+from sqlalchemy import func
+
+from wallabag_kindle_consumer.models import User
+
+logger = Logger(__name__)
+
+
+class Refresher:
+    def __init__(self, session, wallabag):
+        self.session = session
+        self.wallabag = wallabag
+        self.grace = 120
+
+    def _wait_time(self):
+        next = self.session.query(func.min(User.token_valid).label("min")).first()
+        if next is None or next.min is None:
+            return 3
+        delta = next.min - datetime.utcnow()
+        if delta < timedelta(seconds=self.grace):
+            return 0
+
+        calculated = delta - timedelta(seconds=self.grace)
+        return calculated.total_seconds()
+
+    async def refresh(self):
+        while True:
+            await asyncio.sleep(self._wait_time())
+
+            ts = datetime.utcnow() + timedelta(seconds=self.grace)
+            refreshes = [self._refresh_user(user) for user
+                         in self.session.query(User).filter(User.token_valid < ts).all()]
+            await asyncio.gather(*refreshes)
+
+            self.session.commit()
+
+    async def _refresh_user(self, user):
+        logger.info("Refresh token for {}", user.name)
+        await self.wallabag.refresh_token(user)

wallabag_kindle_consumer/sender.py

@@ -0,0 +1,39 @@
+import smtplib
+from email.encoders import encode_base64
+from email.mime.application import MIMEApplication
+from email.mime.multipart import MIMEMultipart
+from email.utils import formatdate
+
+
+class Sender:
+    def __init__(self, loop, from_addr, smtp_server, smtp_port, smtp_user=None, smtp_passwd=None):
+        self.from_addr = from_addr
+        self.loop = loop
+        self.host = smtp_server
+        self.port = smtp_port
+        self.user = smtp_user
+        self.passwd = smtp_passwd
+
+    def _send_mail(self, job, data):
+        msg = MIMEMultipart()
+        msg['Subject'] = job.title
+        msg['From'] = self.from_addr
+        msg['To'] = job.user.kindle_mail
+        msg['Date'] = formatdate(localtime=True)
+
+        mobi = MIMEApplication(data)
+        encode_base64(mobi)
+        mobi.add_header('Content-Disposition',
+                        'attachment; filename={id}.{format}'.format(id=job.article, format=job.format))
+
+        msg.attach(mobi)
+
+        smtp = smtplib.SMTP(host=self.host, port=self.port)
+        smtp.starttls()
+        if self.user is not None:
+            smtp.login(self.user, self.passwd)
+        smtp.sendmail(self.from_addr, job.user.kindle_mail, msg.as_string())
+        smtp.quit()
+
+    async def send_mail(self, job, data):
+        return self.loop.run_in_executor(None, self._send_mail, job, data)

wallabag_kindle_consumer/wallabag.py

@@ -0,0 +1,131 @@
+import aiohttp
+from datetime import datetime, timedelta
+from collections import namedtuple
+
+from logbook import Logger
+
+logger = Logger(__name__)
+
+
+class Article:
+    def __init__(self, id, tags, title, tag, **kwargs):
+        self.id = id
+        self.tags = tags
+        self.title = title
+        self.tag = tag
+
+    def tag_id(self):
+        for t in self.tags:
+            if t['label'] == self.tag.tag:
+                return t['id']
+
+        return -1
+
+
+Tag = namedtuple('Tag', ['tag', 'format'])
+
+
+def make_tags(tag):
+    return (Tag(tag='{tag}'.format(tag=tag), format='mobi'),
+            Tag(tag='{tag}-mobi'.format(tag=tag), format='mobi'),
+            Tag(tag='{tag}-pdf'.format(tag=tag), format='pdf'))
+
+
+class Wallabag:
+    def __init__(self, host_url):
+        self.url = host_url
+        self.tag = "kindle"
+        self.tags = make_tags(self.tag)
+
+    async def get_token(self, user, passwd):
+        params = {'grant_type': 'password',
+                  'client_id': user.client_id,
+                  'client_secret': user.client_secret,
+                  'username': user.name,
+                  'password': passwd}
+
+        async with aiohttp.ClientSession() as session:
+            async with session.get(self._url('/oauth/v2/token'), params=params) as resp:
+                if resp.status != 200:
+                    logger.warn("Gannot get token for user {user}", user=user.name)
+                    return False
+                data = await resp.json()
+                user.auth_token = data["access_token"]
+                user.refresh_token = data["refresh_token"]
+                user.token_valid = datetime.utcnow() + timedelta(seconds=data["expires_in"])
+
+                return True
+
+    async def refresh_token(self, user):
+        params = {'grant_type': 'refresh_token',
+                  'client_id': user.client_id,
+                  'client_secret': user.client_secret,
+                  'refresh_token': user.refresh_token,
+                  'username': user.name}
+
+        async with aiohttp.ClientSession() as session:
+            async with session.get(self._url('/oauth/v2/token'), params=params) as resp:
+                if resp.status != 200:
+                    logger.warn("Cannot refresh token for user {user}", user=user.name)
+                    return False
+                data = await resp.json()
+                user.auth_token = data["access_token"]
+                user.refresh_token = data["refresh_token"]
+                user.token_valid = datetime.utcnow() + timedelta(seconds=data["expires_in"])
+
+                return True
+
+    def _api_params(self, user, params=None):
+        if params is None:
+            params = {}
+
+        params['access_token'] = user.auth_token
+        return params
+
+    def _url(self, url):
+        return self.url + url
+
+    async def fetch_entries(self, user):
+        async with aiohttp.ClientSession() as session:
+            for tag in self.tags:
+                params = self._api_params(user, {"tags": tag.tag})
+                async with session.get(self._url('/api/entries.json'), params=params) as resp:
+                    if resp.status != 200:
+                        logger.warn("Could not get entries of tag {tag} for user {user}", tag=tag.tag, user=user.name)
+                        return
+
+                    data = await resp.json()
+                    if data['pages'] == 1:
+                        user.last_check = datetime.utcnow()
+
+                    articles = data['_embedded']['items']
+                    for article in articles:
+                        yield Article(tag=tag, **article)
+
+    async def remove_tag(self, user, article):
+        params = self._api_params(user)
+        url = self._url('/api/entries/{entry}/tags/{tag}.json'.format(entry=article.id,
+                                                                      tag=article.tag_id()))
+
+        async with aiohttp.ClientSession() as session:
+            async with session.delete(url, params=params) as resp:
+                if resp.status != 200:
+                    logger.warn("Cannot remove tag {tag} from entry {entry} of user {user}", user=user.name,
+                                entry=article.id, tag=article.tag.tag)
+                    return
+
+                logger.info("Removed tag {tag} from article {article} of user {user}", user=user.name,
+                            article=article.id, tag=article.tag.tag)
+
+    async def export_article(self, user, article_id, format):
+        params = self._api_params(user)
+        url = self._url("/api/entries/{entry}/export.{format}".format(entry=article_id, format=format))
+
+        async with aiohttp.ClientSession() as session:
+            async with session.get(url, params=params) as resp:
+                if resp.status != 200:
+                    logger.warn("Cannot export article {article} of user {user} in format {format}", user=user.name,
+                                article=article_id, format=format)
+                    return
+
+                return await resp.read()