150 lines
4.7 KiB
Python
150 lines
4.7 KiB
Python
from authlib.integrations.flask_client import OAuth, token_update, OAuthError
|
|
from flask import Flask, request, url_for, render_template, redirect, session, flash
|
|
from config import SETTINGS
|
|
import urllib
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import os
|
|
|
|
def fetch_token(name):
|
|
return session['token']
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = 'super secret key'
|
|
app.config.from_object(SETTINGS)
|
|
oauth = OAuth(app, fetch_token=fetch_token)
|
|
oauth.register(
|
|
name='gitea',
|
|
access_token_url='https://git.seidlm.at/login/oauth/access_token',
|
|
access_token_params=None,
|
|
authorize_url='https://git.seidlm.at/login/oauth/authorize',
|
|
authorize_params=None,
|
|
api_base_url='https://git.seidlm.at/api/v1/',
|
|
client_kwargs=None,
|
|
)
|
|
|
|
@token_update.connect_via(app)
|
|
def on_token_update(sender, name, token, refresh_token=None, access_token=None):
|
|
if refresh_token:
|
|
item = OAuth2Token.find(name=name, refresh_token=refresh_token)
|
|
elif access_token:
|
|
item = OAuth2Token.find(name=name, access_token=access_token)
|
|
else:
|
|
return
|
|
|
|
# update old token
|
|
item.access_token = token['access_token']
|
|
item.refresh_token = token.get('refresh_token')
|
|
item.expires_at = token['expires_at']
|
|
item.save()
|
|
|
|
@app.errorhandler(OAuthError)
|
|
def handle_error(error):
|
|
return render_template('error.html', error=error)
|
|
|
|
@app.route('/login')
|
|
def login():
|
|
redirect_uri = url_for('authorize', _external=True)
|
|
return oauth.gitea.authorize_redirect(redirect_uri)
|
|
|
|
@app.route('/authorize')
|
|
def authorize():
|
|
token = oauth.gitea.authorize_access_token()
|
|
resp = oauth.gitea.get('user')
|
|
user = resp.json()
|
|
session['token'] = token
|
|
session['user'] = user
|
|
return redirect('/')
|
|
|
|
@app.route('/logout')
|
|
def logout():
|
|
session.pop('token', None)
|
|
session.pop('user', None)
|
|
return redirect('/')
|
|
|
|
@app.route('/')
|
|
def home():
|
|
user = session.get('user')
|
|
files = {}
|
|
path = request.args.get('path')
|
|
if not path:
|
|
path = '/'
|
|
if user:
|
|
resp = oauth.gitea.get(urllib.parse.quote('{}/contents/{}'.format(app.config.get('REPO_URL'), path)))
|
|
files = resp.json()
|
|
|
|
return render_template('home.html', user=user, files=files, path=path)
|
|
|
|
@app.route('/create', methods=['POST'])
|
|
def create():
|
|
path = request.form.get('path')
|
|
filename = request.form.get('filename')
|
|
if not path:
|
|
flash('No path given')
|
|
redirect(request.referrer)
|
|
if not filename:
|
|
flash('no filename given')
|
|
redirect(request.referrer)
|
|
|
|
payload = {"content": "", "message": "WEB API file created"}
|
|
path = path+"/"+filename
|
|
resp = oauth.gitea.request('POST', urllib.parse.quote('{}/contents/{}'.format(app.config.get('REPO_URL'), path)), data=json.dumps(payload), headers={'Content-type': 'application/json'})
|
|
return redirect('/edit?path='+path)
|
|
|
|
|
|
|
|
@app.route('/edit')
|
|
def edit():
|
|
path = request.args.get('path')
|
|
if not path:
|
|
flash('no path given')
|
|
redirect(request.referrer)
|
|
|
|
resp = oauth.gitea.get(urllib.parse.quote('{}/contents/{}'.format(app.config.get('REPO_URL'), path)))
|
|
jresp = resp.json()
|
|
if 'errors' in jresp:
|
|
error = {}
|
|
error['description'] = jresp['errors']
|
|
return render_template('error.html', error=error)
|
|
content = base64.b64decode(jresp['content']).decode('UTF-8')
|
|
sha = jresp['sha']
|
|
user = session.get('user')
|
|
return render_template('edit.html',filecontent=content,sha=sha,path=path, user=user)
|
|
|
|
@app.route('/update', methods=['POST'])
|
|
def update():
|
|
sample = {
|
|
"branch": "master",
|
|
"content": "",
|
|
"sha": "c7ef6a89fe8b888ad897479ac592f1e4841e4f18",
|
|
"message": "Web API Update"}
|
|
content = request.form.get('content')
|
|
content = base64.b64encode(content.encode('UTF-8'))
|
|
sha = request.form.get('sha')
|
|
path = request.form.get('path')
|
|
sample['content'] = content.decode('UTF-8')
|
|
sample['sha'] = sha
|
|
|
|
resp = oauth.gitea.request('PUT', urllib.parse.quote('{}/contents/{}'.format(app.config.get('REPO_URL'), path)), data=json.dumps(sample), headers={'Content-type': 'application/json'})
|
|
print(resp)
|
|
|
|
return str(resp)
|
|
|
|
@app.route('/upload', methods=['POST'])
|
|
def upload():
|
|
file = request.files
|
|
path = request.form.get('path')
|
|
head, tail = os.path.split(path)
|
|
path = head + '/' + file['file'].filename
|
|
image_string = base64.b64encode(file['file'].read())
|
|
payload = {"content": "", "message": "WEB API add file"}
|
|
payload['content'] = image_string.decode('UTF-8')
|
|
resp = oauth.gitea.request('POST', urllib.parse.quote('{}/contents/{}'.format(app.config.get('REPO_URL'), path)), data=json.dumps(payload), headers={'Content-type': 'application/json'})
|
|
|
|
return 'OK'
|
|
|
|
if __name__ == "__main__":
|
|
gitea = oauth.create_client('gitea')
|
|
app.run(debug=True)
|