turbotraktor
/
ladylilia.com
1
0
Fork 0
Code Issues Pull Requests Releases Activity
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from '7a04d51d61'
${ noResults }
ladylilia.com/twitch-cache-api/app.py

412 lines
16 KiB
Raw Blame History

import logging
import os
import flask
import flask_apscheduler
import flask_restful
import flask_restful.fields
import flask_restful.reqparse
import sqlalchemy
import sqlalchemy.engine
from db import db, Video, Comment, Association, Emote, Clip, Event
app = flask.Flask(__name__)
app.logger.setLevel(logging.INFO)
app.config.update(
ERROR_404_HELP=False,
SQLALCHEMY_TRACK_MODIFICATIONS=False,
SQLALCHEMY_DATABASE_URI=os.getenv('SQLALCHEMY_DATABASE_URI'),
SCHEDULER_TIMEZONE='UTC',
SCHEDULER_JOBS=[
dict(id='sync_videos',
func='sync:Sync.sync_videos',
args=(app, db),
max_instances=1,
trigger='interval',
seconds=600),
dict(id='sync_emotes',
func='sync:Sync.sync_emotes',
args=(app, db),
max_instances=1,
trigger='interval',
seconds=1800),
dict(id='sync_clips',
func='sync:Sync.sync_clips',
args=(app, db),
max_instances=1,
trigger='interval',
seconds=120),
dict(id='sync_events',
func='sync:Sync.sync_events',
args=(app, db),
max_instances=1,
trigger='interval',
seconds=300)])
if app.config.get('SQLALCHEMY_DATABASE_URI', '').startswith('sqlite://'):
@sqlalchemy.event.listens_for(sqlalchemy.engine.Engine, 'connect')
def set_sqlite_pragma(dbapi_connection, connection_record):
dbapi_connection.execute('PRAGMA journal_mode=WAL')
dbapi_connection.execute('PRAGMA synchronous=NORMAL')
db.init_app(app)
db.create_all(app=app)
scheduler = flask_apscheduler.APScheduler()
scheduler.init_app(app)
api = flask_restful.Api(app)
video_fields = {
'id': flask_restful.fields.Integer(),
'broadcast_type': flask_restful.fields.String(),
'title': flask_restful.fields.String(),
'description': flask_restful.fields.String(),
'game': flask_restful.fields.String(),
'length': flask_restful.fields.Integer(),
'thumbnail_small': flask_restful.fields.String(),
'thumbnail_medium': flask_restful.fields.String(),
'thumbnail_large': flask_restful.fields.String(),
'created_at': flask_restful.fields.DateTime(dt_format='iso8601'),
'updated_at': flask_restful.fields.DateTime(dt_format='iso8601'),
'recorded_at': flask_restful.fields.DateTime(dt_format='iso8601'),
'published_at': flask_restful.fields.DateTime(dt_format='iso8601'),
}
comment_fields = {
'id': flask_restful.fields.String(attribute='comment.id'),
'video_id': flask_restful.fields.Integer(),
'video_recorded_at': flask_restful.fields.DateTime(dt_format='iso8601', attribute='video.recorded_at'),
'offset': flask_restful.fields.Float(),
'commenter_id': flask_restful.fields.Integer(attribute='comment.commenter_id'),
'commenter_name': flask_restful.fields.String(attribute='comment.commenter_name'),
'commenter_display_name': flask_restful.fields.String(attribute='comment.commenter_display_name'),
'commenter_logo': flask_restful.fields.String(attribute='comment.commenter_logo'),
'source': flask_restful.fields.String(attribute='comment.source'),
'message_body': flask_restful.fields.String(attribute='comment.message_body'),
'message_user_color': flask_restful.fields.String(attribute='comment.message_user_color'),
'message_user_badges': flask_restful.fields.String(attribute='comment.message_user_badges'),
'created_at': flask_restful.fields.DateTime(dt_format='iso8601', attribute='comment.created_at'),
'updated_at': flask_restful.fields.DateTime(dt_format='iso8601', attribute='comment.updated_at'),
}
search_fields = {
**comment_fields,
'video_title': flask_restful.fields.String(attribute='video.title'),
'video_game': flask_restful.fields.String(attribute='video.game'),
'video_length': flask_restful.fields.Integer(attribute='video.length'),
'video_thumbnail_small': flask_restful.fields.String(attribute='video.thumbnail_small'),
'video_thumbnail_medium': flask_restful.fields.String(attribute='video.thumbnail_medium'),
'video_thumbnail_large': flask_restful.fields.String(attribute='video.thumbnail_large'),
}
emote_fields = {
'id': flask_restful.fields.Integer(),
'code': flask_restful.fields.String(),
}
clip_fields = {
'slug': flask_restful.fields.String(),
'video_id': flask_restful.fields.Integer(),
'video_offset': flask_restful.fields.Float(),
'title': flask_restful.fields.String(),
'game': flask_restful.fields.String(),
'duration': flask_restful.fields.Float(),
'curator_id': flask_restful.fields.Integer(),
'curator_name': flask_restful.fields.String(),
'curator_display_name': flask_restful.fields.String(),
'curator_logo': flask_restful.fields.String(),
'thumbnail_tiny': flask_restful.fields.String(),
'thumbnail_small': flask_restful.fields.String(),
'thumbnail_medium': flask_restful.fields.String(),
'created_at': flask_restful.fields.DateTime(dt_format='iso8601'),
}
event_fields = {
'id': flask_restful.fields.String(),
'start': flask_restful.fields.DateTime(dt_format='iso8601'),
'end': flask_restful.fields.DateTime(dt_format='iso8601'),
'title': flask_restful.fields.String(),
'description': flask_restful.fields.String(),
'cover_image_id': flask_restful.fields.String(),
'cover_image_url': flask_restful.fields.String(),
'game_id': flask_restful.fields.Integer(),
'game_name': flask_restful.fields.String(),
'game_box_small': flask_restful.fields.String(),
'game_box_medium': flask_restful.fields.String(),
'game_box_large': flask_restful.fields.String(),
}
filter_parser = flask_restful.reqparse.RequestParser()
filter_parser.add_argument('filter', type=str)
filter_parser.add_argument('older_than', type=flask_restful.inputs.datetime_from_iso8601)
filter_parser.add_argument('newer_than', type=flask_restful.inputs.datetime_from_iso8601)
filter_parser.add_argument('sort_by', type=str)
filter_parser.add_argument('sort_order', type=str)
filter_parser.add_argument('page_number', type=int)
filter_parser.add_argument('page_size', type=int)
search_parser = flask_restful.reqparse.RequestParser()
search_parser.add_argument('commenter', type=str)
search_parser.add_argument('term', type=str)
search_parser.add_argument('sort_by', type=str)
search_parser.add_argument('sort_order', type=str)
search_parser.add_argument('page_number', type=int)
search_parser.add_argument('page_size', type=int)
class VideoResource(flask_restful.Resource):
@flask_restful.marshal_with(video_fields)
def get(self, id):
q = db.session.query(Video).filter(Video.id == id)
video = q.first()
if not video:
flask_restful.abort(404, message='Video {0} does not exist'.format(id))
return video, 200
class VideosResource(flask_restful.Resource):
@flask_restful.marshal_with(video_fields)
def get(self):
args = filter_parser.parse_args()
q = db.session.query(Video)
if args['filter']:
q = q.filter(Video.title.ilike('%{}%'.format(args['filter'])))
if args['older_than']:
q = q.filter(Video.recorded_at < args['older_than'])
if args['newer_than']:
q = q.filter(Video.recorded_at > args['newer_than'])
count = q.count()
if args['sort_order'] == 'random':
q = q.order_by(sqlalchemy.func.random())
elif args['sort_by']:
col = getattr(Video, args['sort_by'], None)
if col:
if args['sort_order']:
order_by = getattr(col, args['sort_order'], None)
if order_by:
q = q.order_by(order_by())
else:
q = q.order_by(col)
if args['page_size']:
q = q.limit(args['page_size'])
if args['page_number'] and args['page_size']:
q = q.offset(args['page_number'] * args['page_size'])
videos = q.all()
return videos, 200, {'X-Total-Count': count}
class CommentResource(flask_restful.Resource):
@flask_restful.marshal_with(comment_fields)
def get(self, video_id, comment_id):
q = db.session.query(Video).filter(Video.id == video_id)
video = q.first()
if not video:
flask_restful.abort(404, message='Video {0} does not exist'.format(video_id))
q = db.session.query(Association).join(Comment)
q = q.filter((Association.video_id == video_id) &\
(Association.comment_id == comment_id))
assoc = q.first()
if not assoc:
flask_restful.abort(404,
message='Comment {0} does not exist'.format(comment_id))
return assoc, 200
class CommentsResource(flask_restful.Resource):
@flask_restful.marshal_with(comment_fields)
def get(self, id):
args = filter_parser.parse_args()
q = db.session.query(Video).filter(Video.id == id)
video = q.first()
if not video:
flask_restful.abort(404, message='Video {0} does not exist'.format(id))
q = db.session.query(Association).join(Comment).join(Video)
q = q.filter(Association.video_id == id)
if args['filter']:
q = q.filter(Comment.message_body.ilike('%{}%'.format(args['filter'])))
count = q.count()
if args['sort_order'] == 'random':
q = q.order_by(sqlalchemy.func.random())
elif args['sort_by']:
col = getattr(Association, args['sort_by'], None)
if not col:
col = getattr(Comment, args['sort_by'], None)
if not col:
col = getattr(Video, args['sort_by'].split('video_', 1).pop(), None)
if col:
if args['sort_order']:
order_by = getattr(col, args['sort_order'], None)
if order_by:
q = q.order_by(order_by())
else:
q = q.order_by(col)
if args['page_size']:
q = q.limit(args['page_size'])
if args['page_number'] and args['page_size']:
q = q.offset(args['page_number'] * args['page_size'])
assocs = q.all()
return assocs, 200, {'X-Total-Count': count}
class SearchResource(flask_restful.Resource):
@flask_restful.marshal_with(search_fields)
def get(self):
args = search_parser.parse_args()
if not args['commenter'] and not args['term']:
return [], 200, {'X-Total-Count': 0}
q = db.session.query(Association).join(Comment).join(Video)
if args['commenter']:
q = q.filter(Comment.commenter_name.ilike('%{}%'.format(args['commenter'])) |\
Comment.commenter_display_name.ilike('%{}%'.format(args['commenter'])))
if args['term']:
q = q.filter(Comment.message_body.ilike('%{}%'.format(args['term'])))
if args['older_than']:
q = q.filter(Comment.created_at < args['older_than'])
if args['newer_than']:
q = q.filter(Comment.created_at > args['newer_than'])
count = q.count()
if args['sort_order'] == 'random':
q = q.order_by(sqlalchemy.func.random())
elif args['sort_by']:
col = getattr(Association, args['sort_by'], None)
if not col:
col = getattr(Comment, args['sort_by'], None)
if not col:
col = getattr(Video, args['sort_by'].split('video_', 1).pop(), None)
if col:
if args['sort_order']:
order_by = getattr(col, args['sort_order'], None)
if order_by:
q = q.order_by(order_by())
else:
q = q.order_by(col)
if args['page_size']:
q = q.limit(args['page_size'])
if args['page_number'] and args['page_size']:
q = q.offset(args['page_number'] * args['page_size'])
assocs = q.all()
return assocs, 200, {'X-Total-Count': count}
class EmoteResource(flask_restful.Resource):
@flask_restful.marshal_with(emote_fields)
def get(self, id):
q = db.session.query(Emote).filter(Emote.id == id)
emote = q.first()
if not emote:
flask_restful.abort(404, message='Emote {0} does not exist'.format(id))
return emote, 200
class EmotesResource(flask_restful.Resource):
@flask_restful.marshal_with(emote_fields)
def get(self):
q = db.session.query(Emote)
emotes = q.all()
return emotes, 200
class ClipResource(flask_restful.Resource):
@flask_restful.marshal_with(clip_fields)
def get(self, slug):
q = db.session.query(Clip).filter(Clip.slug == slug)
clip = q.first()
if not clip:
flask_restful.abort(404, message='Clip {0} does not exist'.format(slug))
return clip, 200
class ClipsResource(flask_restful.Resource):
@flask_restful.marshal_with(clip_fields)
def get(self):
args = filter_parser.parse_args()
q = db.session.query(Clip)
if args['filter']:
q = q.filter(Clip.title.ilike('%{}%'.format(args['filter'])))
if args['older_than']:
q = q.filter(Clip.created_at < args['older_than'])
if args['newer_than']:
q = q.filter(Clip.created_at > args['newer_than'])
count = q.count()
if args['sort_order'] == 'random':
q = q.order_by(sqlalchemy.func.random())
elif args['sort_by']:
col = getattr(Clip, args['sort_by'], None)
if col:
if args['sort_order']:
order_by = getattr(col, args['sort_order'], None)
if order_by:
q = q.order_by(order_by())
else:
q = q.order_by(col)
if args['page_size']:
q = q.limit(args['page_size'])
if args['page_number'] and args['page_size']:
q = q.offset(args['page_number'] * args['page_size'])
clips = q.all()
return clips, 200, {'X-Total-Count': count}
class EventResource(flask_restful.Resource):
@flask_restful.marshal_with(event_fields)
def get(self, id):
q = db.session.query(Event).filter(Event.id == id)
event = q.first()
if not event:
flask_restful.abort(404, message='Event {0} does not exist'.format(id))
return event, 200
class EventsResource(flask_restful.Resource):
@flask_restful.marshal_with(event_fields)
def get(self):
args = filter_parser.parse_args()
q = db.session.query(Event)
if args['filter']:
q = q.filter(Event.title.ilike('%{}%'.format(args['filter'])))
if args['older_than']:
q = q.filter(Event.start < args['older_than'])
if args['newer_than']:
q = q.filter(Event.end > args['newer_than'])
count = q.count()
if args['sort_order'] == 'random':
q = q.order_by(sqlalchemy.func.random())
elif args['sort_by']:
col = getattr(Event, args['sort_by'], None)
if col:
if args['sort_order']:
order_by = getattr(col, args['sort_order'], None)
if order_by:
q = q.order_by(order_by())
else:
q = q.order_by(col)
if args['page_size']:
q = q.limit(args['page_size'])
if args['page_number'] and args['page_size']:
q = q.offset(args['page_number'] * args['page_size'])
events = q.all()
return events, 200, {'X-Total-Count': count}
api.add_resource(VideoResource, '/videos/<int:id>')
api.add_resource(VideosResource, '/videos')
api.add_resource(CommentResource, '/videos/<int:video_id>/comments/<string:comment_id>')
api.add_resource(CommentsResource, '/videos/<int:id>/comments')
api.add_resource(SearchResource, '/search')
api.add_resource(EmoteResource, '/emotes/<int:id>')
api.add_resource(EmotesResource, '/emotes')
api.add_resource(ClipResource, '/clips/<string:slug>')
api.add_resource(ClipsResource, '/clips')
api.add_resource(EventResource, '/events/<int:id>')
api.add_resource(EventsResource, '/events')
if __name__ == '__main__':
scheduler.start()
app.run(host='0.0.0.0', threaded=True, debug=False)
Reference in new issue View Git Blame Copy Permalink