#!/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 = 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}', [] 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 = int(time.time()) try: extracted_related = ythdd_extractor.related('https://www.youtube.com/watch?v=' + videoId) extracted_related['took'] = int(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)