import requests
import os
import functools
import dotenv
import giteapy
import time
import base64
import yaml
from slugify import slugify
from datetime import datetime
from flask import Flask, request, url_for, Response
from requests import api
from flask_micropub import MicropubClient
dotenv.load_dotenv()
PERMITTED_DOMAIN = os.environ.get(
'PERMITTED_DOMAINS', 'https://brainsteam.co.uk/').split(';')
app = Flask(__name__)
app.config['SECRET_KEY'] = 'my super secret key'
micropub = MicropubClient(app, client_id='https://brainsteam.co.uk')
def authed_endpoint(f):
@functools.wraps(f)
def wrapper(*args, **kwargs):
authtok = request.headers.get('Authorization')
if authtok is None:
return {"error": "No token provided"}, 401
auth = requests.get("https://tokens.indieauth.com/token", headers={
"Authorization": authtok, "Accept": "application/json"}).json()
if auth['me'] not in PERMITTED_DOMAIN:
return {"error": f"User {auth['me']} not permitted to post here"}, 403
return f(*args, *kwargs)
return wrapper
_api_client = None
def get_api_client():
global _api_client
if _api_client is None:
config = giteapy.Configuration()
config.host = os.environ.get('GITEA_URL')
config.api_key['access_token'] = os.environ.get('GITEA_API_KEY')
_api_client = giteapy.RepositoryApi(giteapy.ApiClient(config))
return _api_client
@app.route('/', methods=['POST'])
@authed_endpoint
def req():
doc = request.form.to_dict(flat=True)
if 'name' in doc:
entry_type = "posts"
else:
entry_type = "micros"
now = datetime.now()
now_ts = int(time.mktime(now.timetuple()))
url = os.path.join("/",entry_type,now.strftime("%Y/%m/%d"), str(now_ts))
file_path = os.path.join(os.environ.get('CONTENT_PREFIX'), entry_type, now.strftime("%Y/%m/%d"), str(now_ts) + ".md")
frontmatter = {
"url":url,
"type": entry_type,
"date": now.isoformat(sep='T'),
}
categories = request.form.getlist('category[]')
if len(categories) > 0:
frontmatter['categories'] = categories
frontmatter_str = yaml.dump(frontmatter)
content = base64.encodestring(f"---\n{frontmatter_str}\n---\n\n{doc['content']}".encode("utf8")).decode("utf8")
api = get_api_client()
body = giteapy.CreateFileOptions(content=content)
try:
r = api.repo_create_file(os.environ.get('GITEA_REPO_OWNER'), os.environ.get('GITEA_REPO_NAME'), file_path, body)
return Response(status=202, headers={"Location": url})
except Exception as e:
return {"error": str(e)}, 500
# print(r)
# return {"hello": "world"}
@app.route('/', methods=['GET'])
def index():
return """
"""
@app.route('/authenticate')
def authenticate():
return micropub.authenticate(
request.args.get('me'), next_url=url_for('index'))
@app.route('/authorize')
def authorize():
return micropub.authorize(
request.args.get('me'), next_url=url_for('index'),
scope=request.args.get('scope'))
@app.route('/indieauth-callback')
@micropub.authenticated_handler
def indieauth_callback(resp):
return """
Authenticated:
- me: {}
- next: {}
- error: {}
""".format(resp.me, resp.next_url, resp.error)
@app.route('/micropub-callback')
@micropub.authorized_handler
def micropub_callback(resp):
return """
Authorized:
- me: {}
- endpoint: {}
- token: {}
- next: {}
- error: {}
""".format(resp.me, resp.micropub_endpoint, resp.access_token,
resp.next_url, resp.error)
if __name__ == '__main__':
app.run(debug=True)