xevion/v1.xevion.dev
1
0
Fork 0
mirror of https://github.com/Xevion/v1.xevion.dev.git synced 2025-12-06 03:16:57 -06:00
Code Issues Packages Projects Releases Wiki Activity

PyCharm grand repo wide reformat

Browse Source
This commit is contained in:
Xevion
2020-03-08 20:21:18 -05:00
parent 6aeb41249c
commit 66c2ff228c
17 changed files with 277 additions and 187 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
app/__init__.py
View File

@@ -1,12 +1,13 @@
# Main Flask and Flask configs
from flask import Flask
from config import Config
# Flask Extensions
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_login import LoginManager
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from flask_login import LoginManager
from flask_migrate import Migrate
# Flask Extensions
from flask_sqlalchemy import SQLAlchemy
from config import Config
# App & App config setup
app = Flask(__name__)
@@ -21,4 +22,4 @@ limiter = Limiter(app, key_func=get_remote_address, default_limits=["10 per seco
from app import models
from app import routes, simple_routes, hidden, dashboard
from app import ftbhot, custom, spotify, panzer, sound
from app import ftbhot, custom, spotify, panzer, sound

8
app/custom.py
View File

@@ -1,6 +1,8 @@
from functools import wraps
from flask import abort
from flask_login import current_user
from functools import wraps
def require_role(roles=["User"]):
def wrap(func):
@@ -10,5 +12,7 @@ def require_role(roles=["User"]):
if current_user.has_roles(roles):
return func(*args, **kwargs)
return abort(401)
return decorated_view
return wrap
return wrap

21
app/dashboard.py
View File

@@ -1,15 +1,17 @@
from app import app, db, login
from app.forms import ProfileSettingsForm, ProfilePictureForm
from app.models import User, Search
from flask import render_template, request, jsonify
from flask_login import login_required
from app import app
from app.custom import require_role
from flask import render_template, redirect, url_for, request, jsonify
from flask_login import current_user, login_required
from app.forms import ProfileSettingsForm, ProfilePictureForm
@app.route('/dashboard')
@login_required
def dashboard():
return render_template('/dashboard/dashboard.html')
@app.route('/dashboard/profile_settings', methods=['GET'])
@login_required
def profile_settings():
@@ -17,25 +19,28 @@ def profile_settings():
ppform = ProfilePictureForm()
return render_template('/dashboard/profile_settings.html', psform=psform, ppform=ppform)
@app.route('/dashboard/profile_settings/submit', methods=['POST'])
@login_required
def profile_settings_submit():
form = ProfileSettingsForm()
if form.validate_on_submit():
data = {
'show_email' : form.show_email.data or None,
'profile_picture_file' : request.files
'show_email': form.show_email.data or None,
'profile_picture_file': request.files
}
return jsonify(data=data)
return '{}'
@app.route('/dashboard/constants')
@login_required
@require_role(roles=['Admin'])
def constants():
return render_template('/dashboard/constants.html')
@app.route('/dashboard/rbac')
@login_required
def rbac():
return render_template('/dashboard/rbac.html')
return render_template('/dashboard/rbac.html')

13
app/forms.py
View File

@@ -1,14 +1,17 @@
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, BooleanField, SubmitField, RadioField, FileField
from wtforms.validators import ValidationError, DataRequired, EqualTo, Email, URL
from app.models import User
class LoginForm(FlaskForm):
username = StringField('Username', validators=[DataRequired()])
password = PasswordField('Password', validators=[DataRequired()])
remember_me = BooleanField('Remember Me')
submit = SubmitField('Sign in')
class RegistrationForm(FlaskForm):
username = StringField('Username', validators=[DataRequired()])
email = StringField('Email', validators=[DataRequired(), Email()])
@@ -20,17 +23,21 @@ class RegistrationForm(FlaskForm):
user = User.query.filter_by(username=username.data).first()
if user is not None:
raise ValidationError('That username is not available.')
def validate_email(self, email):
user = User.query.filter_by(email=email.data).first()
if user is not None:
raise ValidationError('That email address is not available.')
class ProfileSettingsForm(FlaskForm):
show_email = RadioField('Show Email', default='registered', choices=[('public', 'Public'), ('registered', 'Registered Users Only'), ('hidden', 'Hidden')])
show_email = RadioField('Show Email', default='registered',
choices=[('public', 'Public'), ('registered', 'Registered Users Only'),
('hidden', 'Hidden')])
submit = SubmitField('Save Profile Settings')
class ProfilePictureForm(FlaskForm):
profile_picture_file = FileField('Upload Profile Picture')
profile_picture_url = StringField('Use URL for Profile Picture', validators=[URL()])
submit = SubmitField('Submit Profile Picture')
submit = SubmitField('Submit Profile Picture')

9
app/ftbhot.py
View File

@@ -1,22 +1,27 @@
from app import app
import flask
from app import app
@app.route('/ftbhot/about')
@app.route('/ftbhot/about/')
def ftbhot_about():
return flask.render_template('/ftbhot/about.html')
@app.route('/ftbhot/auth')
@app.route('/ftbhot/auth/')
def ftbhot_auth():
return 'WIP'
@app.route('/ftbhot')
@app.route('/ftbhot/')
def ftbhot():
return flask.render_template('/ftbhot/embed.html')
@app.route('/ftbhot/json')
@app.route('/ftbhot/json/')
def ftbhot_embed():
return flask.render_template('/ftbhot/current.json')
return flask.render_template('/ftbhot/current.json')

48
app/hidden.py
View File

@@ -1,13 +1,16 @@
from app import app, db, login
from app.models import Search
from app.custom import require_role
from flask_login import current_user, login_user, logout_user, login_required
from flask import request, render_template
import requests
import xmltodict
import base64
import json
import requests
import xmltodict
from flask import request, render_template
from flask_login import current_user, login_required
from app import app, db
from app.custom import require_role
from app.models import Search
@app.route('/hidden/history')
@login_required
@require_role(roles=['Hidden', 'Admin'])
@@ -21,6 +24,7 @@ def hidden_history():
def hidden_help():
return render_template('hidden_help.html')
# Parses strings to test for "boolean-ness"
def boolparse(string, default=False):
trues = ['true', '1']
@@ -30,6 +34,7 @@ def boolparse(string, default=False):
return True
return False
@app.route('/hidden/')
@login_required
@require_role(roles=['Hidden'])
@@ -39,7 +44,8 @@ def hidden():
try:
page = int(request.args.get('page') or 1)
except (TypeError, ValueError):
return '\"page\" parameter must be Integer.<br>Invalid \"page\" parameter: \"{}\"'.format(request.args.get('page'))
return '\"page\" parameter must be Integer.<br>Invalid \"page\" parameter: \"{}\"'.format(
request.args.get('page'))
# Handled within building
try:
count = int(request.args.get('count') or 50)
@@ -50,7 +56,7 @@ def hidden():
showfull = boolparse(request.args.get('showfull'))
showtags = boolparse(request.args.get('showtags'))
# Request, Parse & Build Data
data = build_data(tags, page-1, count, base64, showfull)
data = build_data(tags, page - 1, count, base64, showfull)
# Handling for limiters
if base64:
if showfull:
@@ -60,14 +66,18 @@ def hidden():
search = Search(user_id=current_user.id, exact_url=str(request.url), query_args=json.dumps(request.args.to_dict()))
db.session.add(search)
db.session.commit()
return render_template('hidden.html', title='Gelbooru Browser', data=data, tags=tags, page=page, count=count, base64=base64, showfull=showfull, showtags=showtags)
return render_template('hidden.html', title='Gelbooru Browser', data=data, tags=tags, page=page, count=count,
base64=base64, showfull=showfull, showtags=showtags)
def base64ify(url):
return base64.b64encode(requests.get(url).content).decode()
gelbooru_api_url = "https://gelbooru.com/index.php?page=dapi&s=post&q=index&tags={}&pid={}&limit={}"
gelbooru_view_url = "https://gelbooru.com/index.php?page=post&s=view&id={}"
def build_data(tags, page, count, base64, showfull):
# URL Building & Request
temp = gelbooru_api_url.format(tags, page, count)
@@ -75,7 +85,7 @@ def build_data(tags, page, count, base64, showfull):
# XML Parsing & Data Building
parse = xmltodict.parse(response)
build = []
try:
parse['posts']['post']
except KeyError:
@@ -83,13 +93,13 @@ def build_data(tags, page, count, base64, showfull):
for index, element in enumerate(parse['posts']['post'][:count]):
temp = {
'index' : str(index + 1),
'real_url' : element['@file_url'],
'sample_url' : element['@preview_url'],
# strips tags, ensures no empty tags (may be unnecessary)
'tags' : list(filter(lambda tag : tag != '', [tag.strip() for tag in element['@tags'].split(' ')])),
'view' : gelbooru_view_url.format(element['@id'])
}
'index': str(index + 1),
'real_url': element['@file_url'],
'sample_url': element['@preview_url'],
# strips tags, ensures no empty tags (may be unnecessary)
'tags': list(filter(lambda tag: tag != '', [tag.strip() for tag in element['@tags'].split(' ')])),
'view': gelbooru_view_url.format(element['@id'])
}
if base64:
if not showfull:
temp['base64'] = base64ify(temp['sample_url'])
@@ -97,4 +107,4 @@ def build_data(tags, page, count, base64, showfull):
temp['base64'] = base64ify(temp['real_url'])
build.append(temp)
return build
return build

21
app/models.py
View File

@@ -1,9 +1,11 @@
from flask import abort
from flask_login import UserMixin
from datetime import datetime
from app import db, login
from flask_login import UserMixin
from werkzeug.security import generate_password_hash, check_password_hash
from app import db, login
# Just a note, my role system is really quite terrible, but I've implemented as good as a system as I can for a simple RBAC without Hierarchy.
# Once could create a complex system, but it would be better to properly work with SQLAlchemy to create proper permissions, hierarchy, parent/child etc. rather than to work with simple strings.
# One should look into perhaps Pickled Python objects if they were interested in simplifying interactions while opening a lot more data storage.
@@ -20,7 +22,7 @@ class User(UserMixin, db.Model):
about_me = db.Column(db.String(320))
last_seen = db.Column(db.DateTime, default=datetime.utcnow)
show_email = db.Column(db.Boolean, default=False)
def set_password(self, password):
self.password_hash = generate_password_hash(password)
@@ -28,7 +30,7 @@ class User(UserMixin, db.Model):
if self.password_hash is None:
raise "{} has no password_hash set!".format(self.__repr__())
return check_password_hash(self.password_hash, password)
# Retains order while making sure that there are no duplicate role values and they are capitalized
def post_role_processing(self):
user_roles = self.get_roles()
@@ -51,7 +53,7 @@ class User(UserMixin, db.Model):
raise e
success = False
return success
def get_roles(self):
return self.uroles.split(' ')
@@ -70,7 +72,7 @@ class User(UserMixin, db.Model):
self.uroles = user_roles
if postprocess:
self.post_role_processing()
def has_role(self, role):
return self.has_roles([role])
@@ -92,6 +94,7 @@ class User(UserMixin, db.Model):
def __repr__(self):
return '<User {}>'.format(self.username)
class Search(db.Model):
id = db.Column(db.Integer, primary_key=True)
exact_url = db.Column(db.String(160))
@@ -102,6 +105,7 @@ class Search(db.Model):
def __repr__(self):
return '<Search by {} @ {}>'.format(User.query.filter_by(id=self.user_id).first().username, self.timestamp)
class Post(db.Model):
id = db.Column(db.Integer, primary_key=True)
body = db.Column(db.String(140))
@@ -111,6 +115,7 @@ class Post(db.Model):
def __repr__(self):
return '<Post {}>'.format(self.body)
@login.user_loader
def load_user(id):
return User.query.get(int(id))
return User.query.get(int(id))

13
app/panzer.py
View File

@@ -1,8 +1,11 @@
from app import app
from textwrap import wrap
from PIL import Image, ImageDraw, ImageFont
from io import BytesIO
from textwrap import wrap
import flask
from PIL import Image, ImageDraw, ImageFont
from app import app
@app.route('/panzer/')
@app.route('/panzer')
@@ -14,6 +17,7 @@ def panzer(string='bionicles are cooler than sex'):
image = create_panzer(string)
return serve_pil_image(image)
def create_panzer(string):
img = Image.open("./app/static/panzer.jpeg")
draw = ImageDraw.Draw(img)
@@ -27,8 +31,9 @@ def create_panzer(string):
draw.text((topleft[0], topleft[1] + (y * 33)), text, font=font2)
return img
def serve_pil_image(pil_img):
img_io = BytesIO()
pil_img.save(img_io, 'JPEG', quality=50)
img_io.seek(0)
return flask.send_file(img_io, mimetype='image/jpeg')
return flask.send_file(img_io, mimetype='image/jpeg')

79
app/routes.py
View File

@@ -1,35 +1,37 @@
from app import app, db, login
from app.models import User, Search
from app.forms import LoginForm, RegistrationForm
from app.custom import require_role
from werkzeug.urls import url_parse
from flask import render_template, redirect, url_for, flash, request, jsonify, abort, send_file
from flask_login import current_user, login_user, logout_user, login_required
from multiprocessing import Value
import flask
import requests
import xmltodict
import random
import string
import faker
import json
import pprint
import os
import sys
import random
import string
import faker
import requests
from flask import render_template, redirect, url_for, flash, request, jsonify
from flask_login import current_user, login_user, logout_user, login_required
from werkzeug.urls import url_parse
from app import app, db
from app.custom import require_role
from app.forms import LoginForm, RegistrationForm
from app.models import User
print = pprint.PrettyPrinter().pprint
fake = faker.Faker()
strgen = lambda length, charset=string.ascii_letters, weights=None : ''.join(random.choices(list(charset), k=length, weights=weights))
strgen = lambda length, charset=string.ascii_letters, weights=None: ''.join(
random.choices(list(charset), k=length, weights=weights))
@app.route('/', subdomain='api')
def api_index():
return "api"
@app.route('/time/')
def time():
value = request.args.get('value')
if not value:
return '<br>'.join(['[int] value', '[int list] lengths', '[string list] strings', '[boolean] reverse', '[string] pluralappend', '[boolean] synonym'])
return '<br>'.join(
['[int] value', '[int list] lengths', '[string list] strings', '[boolean] reverse', '[string] pluralappend',
'[boolean] synonym'])
value = int(value)
lengths = request.args.get('lengths')
if lengths: lengths = lengths.split(',')
@@ -40,9 +42,13 @@ def time():
if lengths: lengths = list(map(int, lengths))
reverse = request.args.get('reverse')
if reverse: reverse = bool(reverse)
return timeformat(value=value, lengths=lengths or [60, 60, 24, 365], strings=strings or ['second', 'minute', 'hour', 'day', 'year'], reverse=True if reverse is None else reverse)
return timeformat(value=value, lengths=lengths or [60, 60, 24, 365],
strings=strings or ['second', 'minute', 'hour', 'day', 'year'],
reverse=True if reverse is None else reverse)
def timeformat(value, lengths=[60, 60, 24, 365], strings=['second', 'minute', 'hour', 'day', 'year'], reverse=True, pluralappend='s', synonym=False):
def timeformat(value, lengths=[60, 60, 24, 365], strings=['second', 'minute', 'hour', 'day', 'year'], reverse=True,
pluralappend='s', synonym=False):
converted = [value]
for index, length in enumerate(lengths):
temp = converted[-1] // length
@@ -51,18 +57,20 @@ def timeformat(value, lengths=[60, 60, 24, 365], strings=['second', 'minute', 'h
if temp != 0:
converted.append(temp)
else:
break
break
strings = strings[:len(converted)]
build = ['{} {}'.format(value, strings[i] + pluralappend if value > 1 or value == 0 else strings[i]) for i, value in enumerate(converted)][::-1]
build = ['{} {}'.format(value, strings[i] + pluralappend if value > 1 or value == 0 else strings[i]) for i, value in
enumerate(converted)][::-1]
build = ', '.join(build)
return build
@app.route('/avatar/')
@app.route('/avatar/<id>/')
@app.route('/avatar/<id>')
def getAvatar(id=''):
# Constants
headers = {'Authorization' : f'Bot {app.config["DISCORD_TOKEN"]}'}
headers = {'Authorization': f'Bot {app.config["DISCORD_TOKEN"]}'}
api = "https://discordapp.com/api/v6/users/{}"
cdn = "https://cdn.discordapp.com/avatars/{}/{}.png"
# Get User Data which contains Avatar Hash
@@ -73,23 +81,25 @@ def getAvatar(id=''):
url = cdn.format(id, user['avatar'])
return "<img src=\"{}\">".format(url)
@app.route('/userinfo/')
@login_required
@require_role(roles=['Admin'])
def user_info():
prepare = {
'id' : current_user.get_id(),
'email' : current_user.email,
'username' : current_user.username,
'password_hash' : current_user.password_hash,
'is_active' : current_user.is_active,
'is_anonymous' : current_user.is_anonymous,
'is_authenticated' : current_user.is_authenticated,
'metadata' : current_user.metadata.info,
'uroles' : current_user.get_roles()
'id': current_user.get_id(),
'email': current_user.email,
'username': current_user.username,
'password_hash': current_user.password_hash,
'is_active': current_user.is_active,
'is_anonymous': current_user.is_anonymous,
'is_authenticated': current_user.is_authenticated,
'metadata': current_user.metadata.info,
'uroles': current_user.get_roles()
}
return jsonify(prepare)
@app.route('/')
def index():
jobs = [
@@ -101,6 +111,7 @@ def index():
]
return render_template('index.html', job=random.choice(jobs))
@app.route('/register/', methods=['GET', 'POST'])
def register():
if current_user.is_authenticated:
@@ -115,6 +126,7 @@ def register():
return redirect(url_for('login'))
return render_template('register.html', title='Register', form=form, hideRegister=True)
@app.route('/login/', methods=['GET', 'POST'])
def login():
if current_user.is_authenticated:
@@ -132,7 +144,8 @@ def login():
return redirect(next_page)
return render_template('login.html', title='Login', form=form, hideLogin=True)
@app.route('/logout/')
def logout():
logout_user()
return redirect(url_for('index'))
return redirect(url_for('index'))

18
app/simple_routes.py
View File

@@ -1,27 +1,35 @@
from app import app
from flask import send_from_directory, redirect, url_for, render_template
import mistune
import os
import mistune
from flask import send_from_directory, redirect, url_for, render_template
from app import app
markdown = mistune.Markdown()
@app.route('/keybase.txt')
def keybase():
return app.send_static_file('keybase.txt')
@app.route('/modpacks')
def modpacks():
return markdown(open(os.path.join(app.root_path, 'static', 'MODPACKS.MD'), 'r').read())
@app.route('/favicon.ico')
def favicon():
return send_from_directory(os.path.join(app.root_path, 'static'), 'favicon.ico', mimetype='image/vnd.microsoft.icon')
return send_from_directory(os.path.join(app.root_path, 'static'), 'favicon.ico',
mimetype='image/vnd.microsoft.icon')
@app.errorhandler(401)
def unauthorized(e):
return redirect(url_for('login'))
@app.errorhandler(404)
def page_not_found(e):
# note that we set the 404 status explicitly
return render_template('error.html', code=404, message='Content not found...'), 404
return render_template('error.html', code=404, message='Content not found...'), 404

58
app/sound.py
View File

@@ -1,20 +1,16 @@
from app import app, db, limiter
from app.sound_models import YouTubeAudio, SoundcloudAudio, CouldNotDecode, CouldNotDownload, CouldNotProcess
from flask import Response, send_file, redirect, url_for, render_template, request, jsonify
from flask import Response, send_file, request, jsonify
from flask_login import current_user
from multiprocessing import Value
from mutagen.mp3 import MP3
import os
import re
import json
import subprocess
from app import app, db, limiter
from app.sound_models import YouTubeAudio, CouldNotDecode, CouldNotDownload, CouldNotProcess
# Selection of Lambdas for creating new responses
# Not sure if Responses change based on Request Context, but it doesn't hurt.
getBadRequest = lambda : Response('Bad request', status=400, mimetype='text/plain')
getNotImplemented = lambda : Response('Not implemented', status=501, mimetype='text/plain')
getInvalidID = lambda : Response('Invalid ID', status=400, mimetype='text/plain')
getNotDownloaded = lambda : Response('Media not yet downloaded', status=400, mimetype='text/plain')
getBadRequest = lambda: Response('Bad request', status=400, mimetype='text/plain')
getNotImplemented = lambda: Response('Not implemented', status=501, mimetype='text/plain')
getInvalidID = lambda: Response('Invalid ID', status=400, mimetype='text/plain')
getNotDownloaded = lambda: Response('Media not yet downloaded', status=400, mimetype='text/plain')
# Retrieves the YouTubeAudio object relevant to the mediaid if available. If not, it facilitates the creation and writing of one.
# Also helps with access times.
@@ -22,22 +18,24 @@ def get_youtube(mediaid):
audio = YouTubeAudio.query.get(mediaid)
if audio is not None:
audio.access()
return audio # sets the access time to now
return audio # sets the access time to now
else:
audio = YouTubeAudio(id=mediaid)
audio.fill_metadata()
audio.download()
# Commit and save new audio object into the database
db.session.add(audio)
db.session.commit()
return audio
audio = YouTubeAudio(id=mediaid)
audio.fill_metadata()
audio.download()
# Commit and save new audio object into the database
db.session.add(audio)
db.session.commit()
return audio
basic_responses = {
CouldNotDecode : 'Could not decode process response.',
CouldNotDownload : 'Could not download video.',
CouldNotProcess : 'Could not process.'
CouldNotDecode: 'Could not decode process response.',
CouldNotDownload: 'Could not download video.',
CouldNotProcess: 'Could not process.'
}
# A simple function among the routes to determine what should be returned.
# Not particularly sure how request context is passed, but it seems that either it passed or can access current_user's authenitcation/role's properly, so no problem.
# Shows error in full context IF authenticated + admin, otherwise basic error description, OTHERWISE a basic error message.
@@ -48,7 +46,8 @@ def errorCheck(e):
raise e
if current_user.is_authenticated and current_user.has_role('Admin'): response = str(e) + '\n' + response
return Response(response, status=200, mimetype='text/plain')
# Under the request context, it grabs the same args needed to decide whether the stream has been downloaded previously
# It applies rate limiting differently based on service, and whether the stream has been accessed previously
def downloadLimiter():
@@ -60,9 +59,10 @@ def downloadLimiter():
else:
return '10/minute'
# Streams back the specified media back to the client
@app.route('/stream/<service>/<mediaid>')
@limiter.limit(downloadLimiter, lambda : 'global', error_message='429 Too Many Requests')
@limiter.limit(downloadLimiter, lambda: 'global', error_message='429 Too Many Requests')
def stream(service, mediaid):
if service == 'youtube':
if YouTubeAudio.isValid(mediaid):
@@ -80,6 +80,7 @@ def stream(service, mediaid):
else:
return getBadRequest()
# Returns the duration of a specific media
@app.route('/duration/<service>/<mediaid>')
def duration(service, mediaid):
@@ -93,6 +94,7 @@ def duration(service, mediaid):
else:
return getBadRequest()
# Returns a detailed JSON export of a specific database entry.
# Will not create a new database entry where one didn't exist before.
@app.route('/status/<service>/<mediaid>')
@@ -113,6 +115,7 @@ def status(service, mediaid):
else:
return getBadRequest()
@app.route('/list/<service>')
def list(service):
if service == 'youtube':
@@ -125,6 +128,7 @@ def list(service):
else:
return getBadRequest()
@app.route('/all/<service>')
def all(service):
if service == 'youtube':
@@ -136,4 +140,4 @@ def all(service):
elif service == 'spotify':
return getNotImplemented()
else:
return getBadRequest()
return getBadRequest()

59
app/sound_models.py
View File

@@ -1,9 +1,11 @@
from datetime import datetime
from app import db
import subprocess
import json
import os
import re
import subprocess
from datetime import datetime
from app import db
# Returned when a erroring status code is returned. May end up hitting false positives, where the file was still produced properly
# yet a erroring status code was returned. May be a good measure to always disconnect when a error code is found.
@@ -11,27 +13,31 @@ import re
class CouldNotProcess(Exception):
pass
# Shouldn't happen in most cases. When a file isn't found, yet the status code for the process returned positive.
class CouldNotDownload(Exception):
pass
# When a JSON returning command returns undecodable JSON
# This shouldn't occur and will only be available when a unforseen error occurs where JSON cannot be read,
# yet a non-erroring status code was returned!
class CouldNotDecode(Exception):
pass
# A Database Object describing a Audio File originating from YouTube
# Stores basic information like Title/Uploader/URL etc. as well as holds methods useful
# for manipulating, deleting, downloading, updating, and accessing the relevant information or file.
class YouTubeAudio(db.Model):
id = db.Column(db.String(11), primary_key=True) # 11 char id, presumed to stay the same for the long haul. Should be able to change to 12 chars.
url = db.Column(db.String(64)) # 43 -> 64
title = db.Column(db.String(128)) # 120 > 128
creator = db.Column(db.String(128)) # Seems to be Uploader set, so be careful with this
uploader = db.Column(db.String(32)) # 20 -> 32
filename = db.Column(db.String(156)) # 128 + 11 + 1 -> 156
duration = db.Column(db.Integer)
id = db.Column(db.String(11),
primary_key=True) # 11 char id, presumed to stay the same for the long haul. Should be able to change to 12 chars.
url = db.Column(db.String(64)) # 43 -> 64
title = db.Column(db.String(128)) # 120 > 128
creator = db.Column(db.String(128)) # Seems to be Uploader set, so be careful with this
uploader = db.Column(db.String(32)) # 20 -> 32
filename = db.Column(db.String(156)) # 128 + 11 + 1 -> 156
duration = db.Column(db.Integer)
access_count = db.Column(db.Integer, default=0)
download_timestamp = db.Column(db.DateTime, index=True, default=datetime.utcnow)
last_access_timestamp = db.Column(db.DateTime, index=True, default=datetime.utcnow)
@@ -61,20 +67,23 @@ class YouTubeAudio(db.Model):
self.filename = self.id + '.mp3'
command = f'youtube-dl -4 -x --audio-format mp3 --restrict-filenames --dump-json {self.id}'
process = subprocess.Popen(command.split(' '),
encoding='utf-8', stdout=subprocess.PIPE, stderr=subprocess.PIPE)
encoding='utf-8', stdout=subprocess.PIPE, stderr=subprocess.PIPE)
data = process.communicate()
if process.returncode != 0:
raise CouldNotProcess(f'Command: {command}\n{data[1]}Exit Code: {process.returncode}') # process ends with a newline, not needed between
raise CouldNotProcess(
f'Command: {command}\n{data[1]}Exit Code: {process.returncode}') # process ends with a newline, not needed between
try:
data = json.loads(data[0])
except json.JSONDecodeError:
raise CouldNotDecode(data) # We'll return the process data, figure out what to do with it higher up in stack (return/diagnose etc.)
raise CouldNotDecode(
data) # We'll return the process data, figure out what to do with it higher up in stack (return/diagnose etc.)
print(f'JSON acquired for {self.id}, beginning to fill.')
self.duration = data['duration']
self.url = data['webpage_url'] # Could be created, but we'll just infer from JSON response
self.url = data['webpage_url'] # Could be created, but we'll just infer from JSON response
self.creator = data['creator'] or data['uploader']
self.uploader = data['uploader'] or data['creator']
self.title = data['title'] or data['alt_title'] # Do not trust alt-title ; it is volatile and uploader set, e.x. https://i.imgur.com/Tgff4rI.png
self.title = data['title'] or data[
'alt_title'] # Do not trust alt-title ; it is volatile and uploader set, e.x. https://i.imgur.com/Tgff4rI.png
print(f'Metadata filled for {self.id}')
db.session.commit()
@@ -83,26 +92,27 @@ class YouTubeAudio(db.Model):
print(f'Attempting download of {self.id}')
command = f'youtube-dl -x -4 --restrict-filenames --audio-quality 64K --audio-format mp3 -o ./app/sounds/youtube/%(id)s.%(ext)s {self.id}'
process = subprocess.Popen(command.split(' '), encoding='utf-8', stdout=subprocess.PIPE, stderr=subprocess.PIPE)
data = process.communicate() # Not the data for the mp3, just the output. We have to separate this in order to 'wait' for the process to complete fully.
data = process.communicate() # Not the data for the mp3, just the output. We have to separate this in order to 'wait' for the process to complete fully.
print('Checking process return code...')
if process.returncode != 0:
raise CouldNotProcess(f'Command: {command}\n{data[1] or data[0]}Exit Code: {process.returncode}')
print('Checking for expected file...')
if not os.path.exists(self.getPath()):
raise CouldNotDownload(data[1] or data[0])
print(f'Download attempt for {self.id} finished successfully.')
print(f'Download attempt for {self.id} finished successfully.')
# Validates whether the specified ID could be a valid YouTube video ID
# Validates whether the specified ID could be a valid YouTube video ID
@staticmethod
def isValid(id):
return re.match(r'^[A-Za-z0-9_-]{11}$', id) is not None
# Returns a JSON serialization of the database entry
def toJSON(self, noConvert=False):
data = {'id' : self.id, 'url' : self.url, 'title' : self.title, 'creator' : self.creator,
'uploader' : self.uploader, 'filename' : self.filename, 'duration' : self.duration,
'access_count' : self.access_count, 'download_timestamp' : self.download_timestamp.isoformat(),
'last_access_timestamp' : self.last_access_timestamp.isoformat()}
data = {'id': self.id, 'url': self.url, 'title': self.title, 'creator': self.creator,
'uploader': self.uploader, 'filename': self.filename, 'duration': self.duration,
'access_count': self.access_count, 'download_timestamp': self.download_timestamp.isoformat(),
'last_access_timestamp': self.last_access_timestamp.isoformat()}
return data if noConvert else json.dumps(data)
def delete(self):
@@ -114,10 +124,11 @@ class YouTubeAudio(db.Model):
db.session.delete(self)
db.session.commit()
class SoundcloudAudio(db.Model):
id = db.Column(db.Integer, primary_key=True) # hidden API-accessible only ID
id = db.Column(db.Integer, primary_key=True) # hidden API-accessible only ID
url = db.Column(db.String(256))
title = db.Column(db.String(128))
creator = db.Column(db.String(64))
filename = db.Column(db.String(156))
duration = db.Column(db.Integer)
duration = db.Column(db.Integer)

24
app/spotify.py
View File

@@ -1,22 +1,23 @@
import json
import os
import time
from flask import send_file
from app import app
from config import Config
from flask import send_from_directory, redirect, url_for, render_template, send_file
import json
import subprocess
import time
import os
from .spotify_explicit import main
path = os.path.join('app/spotify_explicit/recent.json')
def check_and_update():
try:
with open(path) as file:
file = json.load(file)
except (FileNotFoundError, json.JSONDecodeError):
file = {'last_generated' : -1}
file = {'last_generated': -1}
if file['last_generated'] == -1:
return True
else:
@@ -28,12 +29,13 @@ def check_and_update():
ideal = file['last_generated'] + Config.SPOTIFY_CACHE_TIME
# print(f'Waiting another {int(ideal - time.time())} seconds')
return False
@app.route('/spotify/')
def spotify():
if check_and_update():
print('Graph out of date - running update command')
with open(path, 'w+') as file:
file = json.dump({'last_generated' : int(time.time())}, file)
file = json.dump({'last_generated': int(time.time())}, file)
main.main()
return send_file('spotify_explicit/export/export.png')
return send_file('spotify_explicit/export/export.png')

17
app/spotify_explicit/auth.py
View File

@@ -1,4 +1,7 @@
import logging, sys, os, json
import json
import logging
import os
import sys
# Path to API Credentials file
PATH = os.path.join(sys.path[0], 'auth.json')
@@ -9,11 +12,11 @@ if not os.path.exists(PATH):
# Dump a pretty-printed dictionary with default values
json.dump(
{
'USERNAME' : 'Your Username Here',
'CLIENT_ID' : 'Your Client ID Here',
'CLIENT_SECRET' : 'Your Client Secret Here',
'REDIRECT_URI' : 'Your Redirect URI Callback Here',
'SCOPE' : ['Your Scopes Here']
'USERNAME': 'Your Username Here',
'CLIENT_ID': 'Your Client ID Here',
'CLIENT_SECRET': 'Your Client Secret Here',
'REDIRECT_URI': 'Your Redirect URI Callback Here',
'SCOPE': ['Your Scopes Here']
},
file,
indent=3
@@ -31,4 +34,4 @@ USERNAME = FILE['USERNAME']
CLIENT_ID = FILE['CLIENT_ID']
CLIENT_SECRET = FILE['CLIENT_SECRET']
REDIRECT_URI = FILE['REDIRECT_URI']
SCOPE = ' '.join(FILE['SCOPE'])
SCOPE = ' '.join(FILE['SCOPE'])

11
app/spotify_explicit/main.py
View File

@@ -1,11 +1,12 @@
import json
import logging
import os
import sys
import time
import json
from . import auth
from . import pull
from . import process
import logging
from . import pull
def main():
@@ -14,6 +15,7 @@ def main():
refresh()
process.main()
# Refreshes tracks from files if the token from Spotipy has expired,
# thus keeping us up to date in most cases while keeping rate limits
def refresh():
@@ -28,5 +30,6 @@ def refresh():
else:
pull.main()
if __name__ == "__main__":
main()
main()

31
app/spotify_explicit/process.py
View File

@@ -1,13 +1,11 @@
import os
import sys
import json
import logging
import datetime
import collections
import numpy as np
import os
import dateutil.parser
import PIL.Image as Image
import matplotlib.pyplot as plt
import numpy as np
# Gets all files in tracks folder, returns them in parsed JSON
def get_files():
@@ -20,6 +18,7 @@ def get_files():
)
return files
# Simple function to combine a bunch of items from different files
def combine_files(files):
items = []
@@ -27,6 +26,7 @@ def combine_files(files):
items.extend(file['items'])
return items
# Prints the data in a interesting format
def print_data(data):
for i, item in enumerate(data):
@@ -36,6 +36,7 @@ def print_data(data):
artists = ' & '.join(artist['name'] for artist in item['track']['artists'])
print('[{}] {} "{}" by {}'.format(date, explicit, track_name, artists))
def process_data(data):
# Process the data by Month/Year, then by Clean/Explicit
scores = {}
@@ -44,7 +45,7 @@ def process_data(data):
if date not in scores.keys():
scores[date] = [0, 0]
scores[date][1 if item['track']['explicit'] else 0] += 1
# Create simplified arrays for each piece of data
months = list(scores.keys())[::-1]
clean, explicit = [], []
@@ -59,17 +60,17 @@ def process_data(data):
ind = np.arange(n)
width = 0.55
# Resizer figuresize to be 2.0 wider
plt.figure(figsize=(10.0, 6.0))
plt.figure(figsize=(10.0, 6.0))
# Stacked Bars
p1 = plt.bar(ind, explicit, width)
p2 = plt.bar(ind, clean, width, bottom=explicit) # bottom= just has the bar sit on top of the explicit
p2 = plt.bar(ind, clean, width, bottom=explicit) # bottom= just has the bar sit on top of the explicit
# Plot labeling
plt.title('Song Count by Clean/Explicit')
plt.ylabel('Song Count')
plt.xlabel('Month')
plt.xticks(ind, months, rotation=270) # Rotation 90 will have the
plt.xticks(ind, months, rotation=270) # Rotation 90 will have the
plt.legend((p1[0], p2[0]), ('Explicit', 'Clean'))
fig = plt.gcf() # Magic to save to image and then show
fig = plt.gcf() # Magic to save to image and then show
# Save the figure, overwriting anything in your way
logging.info('Saving the figure to the \'export\' folder')
@@ -86,7 +87,7 @@ def process_data(data):
dpi=100,
quality=95
)
# Finally show the figure to
logging.info('Showing plot to User')
# plt.show()
@@ -95,6 +96,7 @@ def process_data(data):
# logging.info('Copying the plot data to clipboard')
# copy(months, clean, explicit)
# Simple method for exporting data to a table like format
# Will paste into Excel very easily
def copy(months, clean, explicit):
@@ -104,6 +106,7 @@ def copy(months, clean, explicit):
f'{item[0]}\t{item[1]}\t{item[2]}' for item in zip(months, clean, explicit)
]))
def main():
# logging.basicConfig(level=logging.INFO)
logging.info("Reading track files")
@@ -111,7 +114,7 @@ def main():
logging.info(f"Read and parse {len(files)} track files")
logging.info("Combining into single track file for ease of access")
data = combine_files(files)
data.sort(key=lambda item : dateutil.parser.parse(item['added_at']).timestamp(), reverse=True)
data.sort(key=lambda item: dateutil.parser.parse(item['added_at']).timestamp(), reverse=True)
logging.info(f'File combined with {len(data)} items')
logging.info('Processing file...')
process_data(data)
process_data(data)

21
app/spotify_explicit/pull.py
View File

@@ -1,13 +1,14 @@
import os
import sys
from . import auth
import json
import shutil
import pprint
import spotipy
import logging
from hurry.filesize import size, alternative
import os
import shutil
import spotipy
import spotipy.util as util
from hurry.filesize import size, alternative
from . import auth
def main():
# Get Authorization
@@ -27,8 +28,8 @@ def main():
tracks_folder = os.path.join(os.path.dirname(__file__), 'tracks')
logging.warning('Clearing all files in tracks folder for new files')
if os.path.exists(tracks_folder):
shutil.rmtree(tracks_folder) # Delete folder and all contents (old track files)
os.makedirs(tracks_folder) # Recreate the folder just deleted
shutil.rmtree(tracks_folder) # Delete folder and all contents (old track files)
os.makedirs(tracks_folder) # Recreate the folder just deleted
logging.info('Cleared folder, ready to download new track files')
curoffset, curlimit = 0, 50
@@ -62,4 +63,4 @@ def main():
))
break
# Continuing, so increment offset
curoffset += curlimit
curoffset += curlimit