Added main.py

main.py

@@ -0,0 +1,292 @@
+#!/usr/bin/env python3
+
+import argparse
+import atexit
+from getpass import getpass
+import http.client
+import json
+import os
+import re
+import sys
+import xdg.BaseDirectory
+
+# Helpers
+def jsonToObj(cls, dataDict):
+    transDict = cls.TRANS
+    return cls(**{transDict.get(key, key): val for key, val in dataDict.items() if transDict.get(key, key) is not None})
+
+# Class Invalid
+class TtRssCounters:
+    def __init__(feeds=None, labels=None, categories=None, tags=None):
+        self.feeds = feeds
+        self.labels = labels
+        self.categories = categories
+        self.tags = tags
+
+class Category:
+    TRANS = {
+        "id": "catId",
+        "order_id": "orderId",
+    }
+    def __init__(self, catId, title=None, unread=None, orderId=None):
+        self.catId = catId
+        self.title = title
+        self.unread = unread
+        self.orderId = orderId
+    @classmethod
+    def fromJson(cls, jsonData):
+        #return cls(**jsonToInput(cls.TRANS, jsonData))
+        return jsonToObj(cls, jsonData)
+
+class Feed:
+    TRANS = {
+        "id": "feedId",
+        "feed_url": "url",
+        "cat_id": "catId",
+        "last_updated": "lastUpdated",
+        "order_id": "orderId",
+        "has_icon": None,
+    }
+    def __init__(self, feedId, title=None, url=None, catId=None, unread=None, lastUpdated=None, orderId=None):
+        self.feedId = feedId
+        self.title = title
+        self.url = url
+        self.catId = catId
+        self.unread = unread
+        self.lastUpdated = lastUpdated
+        self.orderId = orderId
+    @classmethod
+    def fromJson(cls, jsonData):
+        #return cls(**jsonToInput(cls.TRANS, jsonData))
+        return jsonToObj(cls, jsonData)
+    def toJson(self):
+        #return objToJson(self)
+        pass
+
+class Headline:
+    TRANS = {
+        "id": "headlineId",
+        "guid": None,
+        "is_updated": "isUpdated",
+        "link": "url",
+        "feed_id": "feedId",
+        "feed_title": "feedTitle",
+        "comments_count": None,
+        "comments_link": None,
+        "always_display_attachments": None,
+        "note": None,
+        "lang": None,
+        "content": None,
+        "flavor_image": None,
+        "flavor_stream": None,
+    }
+    def __init__(self, headlineId, unread, marked, published, updated, isUpdated, title, url, feedId, tags, labels, feedTitle, author, score):
+        self.headlineId = headlineId
+        self.unread = unread
+        self.marked = marked
+        self.published = published
+        self.updated = updated
+        self.isUpdated = isUpdated
+        self.title = title
+        self.url = url
+        self.feedId = feedId
+        self.tags = tags
+        self.labels = labels
+        self.feedTitle = feedTitle
+        self.author = author
+        self.score = score
+    @classmethod
+    def fromJson(cls, jsonData):
+        #return cls(**jsonToInput(cls.TRANS, jsonData))
+        return jsonToObj(cls, jsonData)
+
+class TtRssInstance:
+
+    def __init__(self, host, endpoint="/api/"):
+        self._host = host
+        self._endpoint = endpoint
+        self._sid = None
+        self.__conn = http.client.HTTPSConnection(host)
+        atexit.register(lambda: self.__conn.close())
+
+    def __raiseError(self, op, info):
+        if type(info) == 'dict':
+            if 'error' not in info:
+                if 'content' not in info:
+                    raise Exception(f"Invalid info for error: {info}")
+                info = info['content']
+            err = info['error']
+        else:
+            err = info
+        raise Exception(f"API Call {op} failed: {info['error']}")
+
+    def _get(self, op, sendSid=True, **parameters):
+        headers = {
+                'Content-Type': 'application/json',
+        }
+        post_body = {k: v for k, v in parameters.items() if v is not None}
+        post_body['op'] = op
+        if sendSid:
+            if self.__sid is None:
+                raise Exception(f"Login required before API Call {op}")
+            post_body['sid'] = self.__sid
+        self.__conn.request('POST', self._endpoint, json.dumps(post_body), headers)
+        res = self.__conn.getresponse()
+        if res.status != 200:
+            raise Exception(f"Return Code is {res.status} {res.reason}!")
+        ret = json.loads(res.read().decode('utf8'))
+        return ret
+
+    def _getSafe(self, op, sendSid=True, **parameters):
+        r = self._get(op, sendSid=sendSid, **parameters)
+        if r['status'] != 0:
+            if r['content']['error'] == 'NOT_LOGGED_IN':
+                raise Exception(f"Authentication failed!")
+            self.__raiseError(op, r)
+        return r['content']
+
+    def getApiLevel(self):
+        r = self._get("getApiLevel")
+        if r['status'] != 0:
+            return 0 # Asume api level 0 on fail
+        return r['content']['level']
+
+    def getVersion(self):
+        return self._getSafe("getVersion")['version']
+
+    def login(self, username, password):
+        r = self._get("login", sendSid=False, user=username, password=password)
+        if r['status'] != 0:
+            err = r['content']['error']
+            if err == 'API_DISABLED':
+                raise Exception(f"API disabled for user {username}")
+            elif err == 'LOGIN_ERROR':
+                raise Exception(f"Login failed for user {username}")
+            else:
+                self.__raiseError('login', err)
+        self.__sid = r['content']['session_id']
+        return True
+
+    def logout(self):
+        r = self._get("logout")
+        if r['status'] != 0:
+            err = r['content']['error']
+            if err != 'NOT_LOGGED_IN':
+                self.__raiseError('logout', err)
+        self._sid = None
+        return True
+
+    def isLoggedIn(self):
+        if self.__sid is None:
+            return False
+        r = self._get("isLoggedIn")
+        if r['status'] != 0:
+            err = r['content']['error']
+            if err == 'NOT_LOGGED_IN':
+                return False
+            else:
+                self.__raiseError('logout', err)
+        return r['status']['content']['status']
+
+    def getUnread(self):
+        return int(self._getSafe("getUnread")['unread'])
+
+    # TODO Broken
+    def getCounters(self, feeds=True, labels=True, categories=True, tags=False):
+        mode = ''
+        if feeds:
+            mode += 'f'
+        if labels:
+            mode += 'l'
+        if categories:
+            mode += 'c'
+        if tags:
+            mode += 't'
+        r = self._getSafe("getCounters", output_mode=mode)
+        #return TtRssCounters(r.get('feeds', None), r.get('labels', None), r.get('categories', None), r.get('tags', None))
+        return r
+
+    def getFeeds(self, cat_id=-3, unread_only=False, limit=None, offset=None, include_nested=False):
+        r = self._getSafe("getFeeds", cat_id=cat_id, unread_only=unread_only, limit=limit, offset=offset, include_nested=include_nested)
+        return [Feed.fromJson(data) for data in r]
+
+    def getCategories(self, unread=False, nested=False, empty=True):
+        r = self._getSafe('getCategories', unread_only=unread, enable_nested=nested, include_empty=empty)
+        return [Category.fromJson(data) for data in r]
+
+    def getHeadlines(self, feed_id=None, cat_id=None, limit=None, skip=None, show_excerpt=False, show_content=False, view_mode='all_articles', include_attachments=False, since_id=None, nested=None, order_by=None, sanitize=True, force_update=False, has_sandbox=False):
+        is_cat = False
+        send_feed_id = feed_id
+        if cat_id is not None:
+            if feed_id is not None:
+                raise Exception("cat_id and feed_id cannot be set both!")
+            is_cat = True
+            send_feed_id = cat_id
+        r = self._getSafe('getHeadlines', feed_id=send_feed_id, limit=limit, skip=skip, show_excerpt=show_excerpt, show_content=show_content, view_mode=view_mode, include_attachments=include_attachments, since_id=since_id, include_nested=nested, order_by=order_by, sanitize=sanitize, force_update=force_update, has_sandbox=has_sandbox)
+        return [Headline.fromJson(data) for data in r]
+
+# Main Code
+
+def func_articles(server, args):
+    filtered = {key: value for key, value in args.__dict__.items() if key in ["feed_id", "cat_id", "limit", "skip", "view_mode", "since_id", "include_nested", "order_by"]}
+    headlines = server.getHeadlines(**filtered)
+    if args.only_url:
+        for h in headlines:
+            print(h.url)
+    else:
+        print(json.dumps([h.objToJson() for h in headlines]))
+
+def parser_articles(sub):
+    p = sub.add_parser('articles', help="Get articles")
+    p.set_defaults(func=func_articles)
+    p.add_argument('--feed-id', dest='feed_id', help="Only output articles for this feed, cannot be used with --cat-id")
+    p.add_argument('--cat-id', dest='cat_id', help="Only output articles for feeds of this category, cannot be used with --feed-id")
+    p.add_argument('--limit', dest='limit', type=int, help="Maximum count of articles")
+    p.add_argument('--skip', '--offset', dest='skip', type=int, help="Skip this amount of feeds first")
+    p.add_argument('--view-mode', '--filter', dest='view_mode', choices=['all_articles', 'unread', 'adaptive', 'marked', 'updated'], default='all_articles', help='Only show articles of certain type')
+    p.add_argument('--only-url', dest='only_url', action='store_true', help="Only output urls of articles instead of full article info")
+
+def configure_parser():
+    parser = argparse.ArgumentParser(description="Allows access to feeds and articles of a Tiny Tiny RSS instance using the API.")
+    #parser.add_argument('--proto', '--protocol', dest='proto', default='https', choices=['http', 'https'], help="The protocol used to access the api, defaults to https")
+    parser.add_argument('-H', '--host', '--url', dest='url', required=True, help="URL of the TT-RSS instace to access, examples: rss.example.com, example.com/tt-rss")
+    parser.add_argument('-u', '--user', '--username', dest='user', required=True, help="Name of the user used to access the instance")
+    parser.add_argument('-p', '--pass', '--password', dest='passwd', help="Password of the user used to access the instance")
+    parser.add_argument('-P', '--pass-stdin', '--password-stdin', action='store_true', dest='passStdin', help="Read the password for the user used to access the instance from stdin")
+    sub = parser.add_subparsers(help="Operations", dest='op', description="Operations available to send to server")
+    parser_articles(sub)
+    return parser
+
+def parse(args):
+    return configure_parser().parse_args(args=args)
+
+def main():
+    APP_NAME = "ttrss-cli"
+    URL_REGEX = re.compile(r'^(?P<proto>(https?://)?)(?P<host>[^:/ ]+)(:(?P<port>\d+))?(?P<path>.*)$')
+    # Retrieve args from config and shell
+    storedArgs = sys.argv.copy()
+    storedArgs.pop(0)
+    for d in xdg.BaseDirectory.load_config_paths(APP_NAME):
+        with open(os.path.join(d, "config.txt")) as f:
+            storedArgs = [line.strip() for line in f if line.strip()[0] != '#'] + storedArgs
+    # Parse args
+    args = parse(storedArgs)
+    # Read Pass if stdin
+    passwd = args.passwd
+    if args.passStdin:
+        passwd = getpass()
+    # Check for pass
+    if passwd is None:
+        raise ValueError("Password is missing!") # TODO make more beautiful for user
+    # Parse url
+    urlM = URL_REGEX.match(args.url).groupdict()
+    host = urlM['host']
+    # Call api
+    # TODO Support scheme (http / https)
+    # TODO Support port
+    server = TtRssInstance(host)
+    server.login(args.user, passwd)
+    args.func(server, args)
+
+if __name__ == "__main__":
+    main()