ythdd/ythdd_api_v1.py

#!/usr/bin/python3
# API is expected to return:
# - HTTP status code,
# - human-readable status message,
# - json with appropriate data
import flask, json, time
import ythdd_globals, ythdd_extractor
#from flask_sqlalchemy import SQLAlchemy
#import ythdd_api_v1_stats, ythdd_api_v1_user, ythdd_api_v1_info, ythdd_api_v1_query, ythdd_api_v1_meta, ythdd_api_v1_admin

def incrementBadRequests():
	ythdd_globals.apiFailedRequests += 1

def notImplemented(data):
	# TODO: change list to string -> data, not data[0]
	return 501, f"not recognised/implemented: {data[0]}", []

def stub_hello():
	return 200, 'hello from v1!', []

def stats():
	data_to_send = {
		"starttime": ythdd_globals.starttime,
		"uptime": ythdd_globals.getUptime(),
		"total_api_requests": ythdd_globals.apiRequests,
		"failed_api_requests": ythdd_globals.apiFailedRequests,
		"outside_api_requests": ythdd_globals.outsideApiHits,
		"local_api_requests": ythdd_globals.apiRequests - ythdd_globals.outsideApiHits
	}
	return 200, "OK", data_to_send

def hot(data):
	#print(data)
	if len(data) <= 2:
		incrementBadRequests()
		return 400, f'error: bad request. supply required arguments.', []
	match data[1]:
		case "video" | "channel" | "handle" | "playlist":
			url_lookup = {'video': 'https://www.youtube.com/watch?v=', 'channel': 'https://www.youtube.com/channel/', 'handle': 'https://www.youtube.com/@', 'playlist': 'https://www.youtube.com/playlist?list='}
			comment_count = ""
			if len(data) <= 3:
				incrementBadRequests()
				return 400, f'error: bad request. supply required arguments.', []
			if data[2] not in ("c", "nc", "lc"):
				incrementBadRequests()
				return notImplemented(data)
			if data[2] == "lc" and len(data) <= 4:
				incrementBadRequests()
				return 400, f'error: bad request. limited comments (lc) requires an extra argument specifying amount of comments.', []
			elif data[2] == "lc":
				try:
					comment_count = str(int(data[3]))
				except:
					incrementBadRequests()
					return 400, f'error: bad request. {data[3]} is not a number.', []
				videoId = data[4]
			else:
				videoId = data[3]

			if len(videoId) != 11: # videoId sanity check
				incrementBadRequests()
				return 400, f'error: bad request. wrong videoId: {videoId} is {len(videoId)} characters long, but should be 11.', []

			getcomments = True
			if data[2] == "nc":
				getcomments = False

			try:
				started = time.time()
				extracted_dict = ythdd_extractor.extract(url_lookup[data[1]] + videoId, getcomments=getcomments, maxcomments=comment_count)
				extracted_dict["took"] = time.time() - started
				return 200, "OK", extracted_dict
			except Exception as e:
				incrementBadRequests()
				return 400, f'error: failed to get "{videoId}" ({data[2]})', {'error_msg': str(e)}
		case "related":
			videoId = data[2]
			if len(videoId) != 11: # videoId sanity check
				incrementBadRequests()
				return 400, f'error: bad request. wrong videoId: {videoId} is {len(videoId)} characters long, but should be 11.', []

			started = time.time()
			try:
				extracted_related = ythdd_extractor.related('https://www.youtube.com/watch?v=' + videoId)
				extracted_related['took'] = time.time() - started
				return 200, "OK", extracted_related
			except Exception as e:
				incrementBadRequests()
				return 400, f'error: unknown error while parsing {videoId}: {e}', []

		case _:
			incrementBadRequests()
			return notImplemented(data)


	'''
	if data[1] not in ("video", "channel", "handle", "playlist", "related"):
		incrementBadRequests()
		return notImplemented(data)
	if data[2] not in ("c", "nc", "lc"): # comments, no comments, limited comments
		incrementBadRequests()
		return notImplemented(data)
	if data[2] == "lc":
		if len(data) <= 4:
			incrementBadRequests()
			return 400, f'error: bad request. limited comments (lc) requires an extra argument specifying amount of comments.', []
		try:
			comment_count = str(int(data[3]))
		except:
			incrementBadRequests()
			return 400, f'error: bad request. {data[3]} is not a number.', []
		videoId = data[4]
	else:
		videoId = data[3]
	try:
		url_lookup = {'video': 'https://www.youtube.com/watch?v=', 'channel': 'https://www.youtube.com/channel/', 'handle': 'https://www.youtube.com/@', 'playlist': 'https://www.youtube.com/playlist?list='}
		if data[2] == "nc":
			getcomments = False
		else:
			getcomments = True
		started = int(time.time())
		extracted_dict = ythdd_extractor.extract(url_lookup[data[1]] + videoId, getcomments=getcomments, maxcomments=comment_count)
		extracted_dict["took"] = int(time.time()) - started
		return 200, "OK", extracted_dict
	except Exception as e:
		incrementBadRequests()
		return 400, f'error: failed to get "{videoId}" ({data[2]}). {e}', []
	'''

def lookup(data):
	match data[0]:
		case 'stats':
			return stats()
		case 'hot': # retrieve live, uncached data
			#print(data)
			return hot(data)
		case 'user':
			return stub_hello()
			#do_user()
		case 'info':
			return stub_hello()
			#do_info()
		case 'query':
			return stub_hello()
		case 'meta':
			return stub_hello()
		case 'admin':
			# REQUIRE CREDENTIALS!
			return stub_hello()
		case _:
			incrementBadRequests()
			return notImplemented(data)