#!/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): 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): if len(data) <= 3: incrementBadRequests() return 400, f'error: bad request. supply required arguments.', [] comment_count = "" if data[1] not in ("video", "channel", "handle", "playlist"): return notImplemented(data) if data[2] not in ("c", "nc", "lc"): # comments, no comments, limited comments 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)