micropub-flask-gitea/example.py

181 lines
4.3 KiB
Python
Raw Normal View History

2021-12-23 21:39:21 +00:00
import requests
import os
import functools
import dotenv
import giteapy
import time
import base64
import yaml
2021-12-23 21:39:21 +00:00
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__)
2015-01-19 17:01:56 +00:00
app.config['SECRET_KEY'] = 'my super secret key'
2021-12-23 21:39:21 +00:00
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))
2021-12-23 21:39:21 +00:00
return _api_client
2021-12-23 21:39:21 +00:00
@app.route('/', methods=['POST'])
@authed_endpoint
def req():
doc = request.form.to_dict()
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 = yaml.dump({"url":url, "type": entry_type, "date": now.isoformat(sep='T')})
content = base64.encodestring(f"---\n{frontmatter}\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 """
<!DOCTYPE html>
<html>
<body>
<form action="/authenticate" method="GET">
<input type="text" name="me" placeholder="your domain.com"/>
<button type="submit">Authenticate</button>
</form>
<form action="/authorize" method="GET">
<input type="text" name="me" placeholder="your domain.com"/>
<select name="scope">
<option>read</option>
<option>post</option>
<option>comment</option>
</select>
<button type="submit">Authorize</button>
</form>
</body>
</html>
"""
@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 """
<!DOCTYPE html>
<html>
2015-01-19 17:01:56 +00:00
<body>
Authenticated:
2015-01-19 17:01:56 +00:00
<ul>
<li>me: {}</li>
<li>next: {}</li>
<li>error: {}</li>
</ul>
</body>
</html>
""".format(resp.me, resp.next_url, resp.error)
@app.route('/micropub-callback')
@micropub.authorized_handler
def micropub_callback(resp):
2021-12-23 21:39:21 +00:00
return """
<!DOCTYPE html>
<html>
2015-01-19 17:01:56 +00:00
<body>
Authorized:
<ul>
<li>me: {}</li>
<li>endpoint: {}</li>
<li>token: {}</li>
<li>next: {}</li>
<li>error: {}</li>
</ul>
2015-01-19 17:01:56 +00:00
</body>
</html>
""".format(resp.me, resp.micropub_endpoint, resp.access_token,
resp.next_url, resp.error)
if __name__ == '__main__':
app.run(debug=True)