122 lines
3.3 KiB
Python
122 lines
3.3 KiB
Python
import json
|
|
import dateutil
|
|
from datetime import datetime, timezone
|
|
|
|
from flask import Flask, request, abort
|
|
from feedgen.feed import FeedGenerator
|
|
|
|
from pony import orm
|
|
from pony.orm import Required, Optional, Set
|
|
|
|
app = Flask(__name__)
|
|
|
|
SITE_URL = 'http://localhost:5000/feeds/'
|
|
|
|
db = orm.Database()
|
|
|
|
class Feed(db.Entity):
|
|
slug = Required(str, unique=True)
|
|
admin_token = Required(str)
|
|
title = Required(str)
|
|
atom_id = Required(str) # also link
|
|
description = Required(str)
|
|
login = Optional(str)
|
|
password = Optional(str)
|
|
items = Set('Item')
|
|
|
|
class Item(db.Entity):
|
|
feed = Required(Feed)
|
|
atom_id = Required(str)
|
|
title = Required(str)
|
|
content = Required(str)
|
|
date = Required(datetime)
|
|
|
|
db.bind(provider='sqlite', filename='db.sqlite', create_db=True)
|
|
db.generate_mapping(create_tables=True)
|
|
orm.set_sql_debug(True)
|
|
|
|
def does_slug_exist(slug):
|
|
with orm.db_session:
|
|
return len(orm.select(f for f in Feed if f.slug == slug)[:1])
|
|
|
|
@app.route('/feeds', methods=['POST'])
|
|
def add_feed():
|
|
data = request.get_json(silent=True)
|
|
|
|
if 'slug' not in data:
|
|
abort(400, 'missing slug')
|
|
|
|
slug = data['slug']
|
|
|
|
if does_slug_exist(slug):
|
|
abort(400, "feed with this slug already exists")
|
|
|
|
if 'title' not in data or 'description' not in data or 'admin_token' not in data:
|
|
abort(400, "missing title or description or admin_token")
|
|
|
|
url = SITE_URL + slug
|
|
|
|
with orm.db_session:
|
|
feed = Feed(slug=slug,
|
|
admin_token=data['admin_token'],
|
|
title=data['title'],
|
|
atom_id=url,
|
|
description=data['description'])
|
|
|
|
if 'login' in data and 'password' in data:
|
|
feed.login = data['login']
|
|
feed.password = data['password']
|
|
|
|
return "OK"
|
|
|
|
@app.route('/feeds/<slug>', methods=['GET', 'POST'])
|
|
@orm.db_session
|
|
def by_feed_name(slug):
|
|
if not does_slug_exist(slug):
|
|
abort(404, "unknown slug")
|
|
|
|
feed = orm.select(f for f in Feed if f.slug == slug)[:1][0]
|
|
|
|
if feed.password is not '':
|
|
auth = request.authorization
|
|
if not auth or auth.username != feed.login or auth.password != feed.password:
|
|
abort(401, "missing or incorrect password")
|
|
|
|
if request.method == 'POST':
|
|
data = request.get_json(silent=True)
|
|
|
|
if 'title' not in data or 'content' not in data or 'admin_token' not in data:
|
|
abort(400, "missing title or content")
|
|
|
|
if data['admin_token'] != feed.admin_token:
|
|
abort(401, "incorrect admin_token")
|
|
|
|
item_id = feed.atom_id + '/' + str(len(feed.items))
|
|
date = datetime.now(dateutil.tz.tzutc())
|
|
|
|
with orm.db_session:
|
|
item = Item(feed=feed,
|
|
atom_id=item_id,
|
|
title=data['title'],
|
|
content=data['content'],
|
|
date=date)
|
|
|
|
return "OK"
|
|
|
|
gen = FeedGenerator()
|
|
gen.title(feed.title)
|
|
gen.id(feed.atom_id)
|
|
gen.link(href=feed.atom_id)
|
|
gen.description(feed.description)
|
|
|
|
for item in feed.items:
|
|
fe = gen.add_entry()
|
|
fe.id(item.atom_id)
|
|
fe.title(item.title)
|
|
fe.content(item.content)
|
|
|
|
date = item.date.replace(tzinfo=timezone.utc)
|
|
fe.updated(date)
|
|
|
|
return gen.atom_str(pretty=True).decode('utf8')
|