ShioriFeed 🔖

#!/usr/bin/env python3 # *----------------------------------------------------------------------* # # | [ ShioriFeed 🔖 ] | # # | Simple service for getting an Atom/RSS feed from your Shiori profile | # # | v. 2023-02-13-r2, OctoSpacc | # # *----------------------------------------------------------------------* # # *---------------------------------* # # | Configuration | # # *---------------------------------* # Host = ('localhost', 8176) Debug = True # *---------------------------------* # # External Requirements: urllib3 # *-------------------------------------------------------------------------* # import traceback import json from base64 import urlsafe_b64decode, urlsafe_b64encode from html import escape as HtmlEscape from http.server import HTTPServer, BaseHTTPRequestHandler from socketserver import ThreadingMixIn from urllib.request import urlopen, Request from urllib.error import HTTPError, URLError import threading # Usage: http[s]:///http[s]://// HomeTemplate = '''\ ShioriFeed 🔖

ShioriFeed 🔖

Enter the details of your account on a Shiori server to get an Aotm/RSS feed link.

v. 2023-02-13 ▪️ Source Code

''' def SessionHash(Remote, Username, Password): return f'{hash(Remote)}{hash(Username)}{hash(Password)}' def MkFeed(Data, Remote, Username): Feed = '' Date = Data['bookmarks'][0]['modified'] if Data['bookmarks'] else '' for Mark in Data['bookmarks']: Feed += f''' {HtmlEscape(Mark['title'])} {HtmlEscape(Mark['excerpt'])} {Remote}/bookmark/{Mark['id']}/content {Mark['modified']} {Mark['id']} ''' return f'''\ ShioriFeed ({HtmlEscape(Username)}) 🔖 {Date} {Date} {Feed} ''' def MkUrl(Post): Args = {} #Args = Post.split('&') for Arg in Post.split('&'): Arg = Arg.split('=') Args.update({Arg[0]: Arg[1]}) return f'''\ http[s]://\ /{Args['Remote']}\ /{urlsafe_b64encode(Args['Username'].encode()).decode()}\ /{urlsafe_b64encode(Args['Password'].encode()).decode()}/''' def GetSession(Remote, Username, Password): try: Rq = urlopen(Request(f'{Remote}/api/login', data=json.dumps({'username': Username, 'password': Password, 'remember': True, 'owner': True}).encode(), headers={'User-Agent': f'ShioriFeed at {Host[0]}'})) if Rq.code == 200: Data = {SessionHash(Remote, Username, Password): json.loads(Rq.read().decode())['session']} Sessions.update(Data) return { 'Code': 200, 'Body': Data} else: return { 'Code': Rq.code, 'Body': f'[{Rq.code}] External Server Error\n\n{Rq.read().decode()}'} except Exception: #as Ex: #(HTTPError, URLError) as Ex: #print(traceback.format_exc()) return { 'Code': 500, 'Body': '[500] Internal Server Error' + (f'\n\n{traceback.format_exc()}' if Debug else '')} def RqHandle(Path, Attempt=0): try: Rs = {} Args = Path.strip().removeprefix('/').removesuffix('/').strip().split('/') if Args[0] == '': return { 'Code': 200, 'Body': HomeTemplate, 'Content-Type': 'text/html'} else: Remote = '/'.join(Args[:-2]) Username = urlsafe_b64decode(Args[-2]).decode() Password = urlsafe_b64decode(Args[-1]).decode() if not SessionHash(Remote, Username, Password) in Sessions: TrySession = GetSession(Remote, Username, Password) if TrySession['Code'] != 200: return TrySession Rq = urlopen(Request(f'{Remote}/api/bookmarks', headers={ 'X-Session-Id': Sessions[SessionHash(Remote, Username, Password)], 'User-Agent': f'ShioriFeed at {Host[0]}'})) Rs['Code'] = Rq.code # Shiori got us JSON data, parse it and return our result if Rq.code == 200: Rs['Body'] = MkFeed(json.loads(Rq.read().decode()), Remote, Username) Rs['Content-Type'] = 'application/xml' # We probably got an expired Session-Id, let's try to renew it elif Rq.code == 500 and Attempt < 1: TrySession = GetSession(Remote, Username, Password) if TrySession['Code'] != 200: return TrySession return ReqHandle(Path, Attempt+1) else: Rs['Body'] = f'[{Rq.code}] External Server Error\n\n{Rq.read().decode()}' return Rs except Exception: #as Ex: #(HTTPError, URLError) as Ex: #print(traceback.format_exc()) #Rs['Code'] = 500 #Rs['Body'] = f'[500] Internal Server Error\n\n{traceback.format_exc()}' #Rs['Body'] = f'[500] Internal Server Error' #Rs['Content-Type'] = 'text/plain' return { 'Code': 500, 'Body': '[500] Internal Server Error' + (f'\n\n{traceback.format_exc()}' if Debug else '')} class Handler(BaseHTTPRequestHandler): def do_GET(self): Rs = RqHandle(self.path) self.send_response(Rs['Code']) self.send_header('Content-Type', Rs['Content-Type'] if 'Content-Type' in Rs else 'text/plain') self.end_headers() self.wfile.write(Rs['Body'].encode()) def do_POST(self): try: if self.path == '/': Body = HomeTemplate.replace('', f'''

Here's your Atom feed: {MkUrl(self.rfile.read(int(self.headers['Content-Length'])).decode())}

''').replace('/* {{PostCss}} */', '.PostObscure { opacity: 0.5; }') self.send_response(200) self.send_header('Content-Type', 'text/html') self.end_headers() self.wfile.write(Body.encode()) else: self.send_response(400) self.send_header('Content-Type', 'text/plain') self.end_headers() self.wfile.write(b'[400] Bad Request') except Exception: self.send_response(500) self.send_header('Content-Type', 'text/plain') self.end_headers() self.wfile.write(('[500] Internal Server Error' + (f'\n\n{traceback.format_exc()}' if Debug else '')).encode()) # https://stackoverflow.com/a/3389505 def log_message(self, format, *args): return # https://stackoverflow.com/a/51559006 class ThreadedHTTPServer(ThreadingMixIn, HTTPServer): pass def Serve(): ThreadedHTTPServer(Host, Handler).serve_forever() if __name__ == '__main__': Sessions = {} try: Serve() except KeyboardInterrupt: pass