micropub-flask-gitea/example.py

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": "unauthorized",
                "error_description": "An auth token was not 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": "forbidden", "error_description": 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 = "post"
    else:
        entry_type = "note"

    now = datetime.now()
    now_ts = int(time.mktime(now.timetuple()))

    url = os.path.join("/", entry_type + "s",
                       now.strftime("%Y/%m/%d"), str(now_ts))

    if 'name' in doc:
        slug = slugify(doc['name']) + str(now_ts)
    else:
        slug = str(now_ts)

    file_path = os.path.join(os.environ.get(
        'CONTENT_PREFIX'), entry_type + "s",  now.strftime("%Y/%m/%d"), slug + ".md")

    frontmatter = {
        "url": url,
        "type": entry_type,
        "date": now.isoformat(sep='T'),
    }

    if 'category' in doc:
        categories = [doc['category']]
    else:
        categories = request.form.getlist('category[]')

    if 'name' in doc:
        frontmatter['title'] = doc['name']

    if len(categories) > 0:
        frontmatter['tags'] = categories

    frontmatter_str = yaml.dump(frontmatter)

    if 'photo' in doc:

        if os.environ.get('MICROPUB_IMAGE_STRATEGY') == 'copy':
          # download the photo
          r = requests.get(doc['photo'])

          # generate local filename
          filename = os.path.join(os.environ.get('MICROPUB_MEDIA_PATH'), now.strftime("%Y/%m/%d"), str(now_ts) + ".jpg")
          photo_url = os.path.join(os.environ.get('MICROPUB_MEDIA_URL_PREFIX'), now.strftime("%Y/%m/%d"), str(now_ts) + ".jpg")

          # make directory if needed
          if not os.path.exists(os.path.dirname(filename)):
            os.makedirs(os.path.dirname(filename))

          with open(filename, 'wb') as f:
            f.write(r.content)

        # elif os.environ.get('MICROPUB_IMAGE_STRATEGY') == 'gitea':

        else:
          photo_url = doc['photo']
        
        
        docstr = f"![image]({photo_url}) \n\n {doc['content']}"
    else:
        docstr = doc['content']

    content = base64.encodestring(
        f"---\n{frontmatter_str}\n---\n\n{docstr}".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'])
@authed_endpoint
def index():

    if request.args.get('q') == 'config':
        return {
            "media-endpoint": "/micropub/media",
            "syndicate-to": [
                {
                    "uid": "mastodon",
                    "name": "Mastodon"
                }
            ],
            "post-types": [
                {
                    "type": "note",
                    "name": "Note"
                },
                {
                    "type": "article",
                    "name": "Blog Post"
                },
                {
                    "type": "photo",
                    "name": "Photo"
                },
                {
                    "type": "reply",
                    "name": "Reply"
                },
                {
                    "type": "bookmark",
                    "name": "Bookmark"
                }
            ]
        }


@app.route('/form', methods=['GET'])
def authform():
    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>
      <body>
        Authenticated:
        <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):

    return """
    <!DOCTYPE html>
    <html>
      <body>
        Authorized:
        <ul>
          <li>me: {}</li>
          <li>endpoint: {}</li>
          <li>token: {}</li>
          <li>next: {}</li>
          <li>error: {}</li>
        </ul>
      </body>
    </html>
    """.format(resp.me, resp.micropub_endpoint, resp.access_token,
               resp.next_url, resp.error)


if __name__ == '__main__':
    app.run(debug=True)