Skip to content
Extraits de code Groupes Projets

Kagari token

Fusionnées Kubat a demandé de fusionner kagari-token vers master
Affichage du commit 03240907
Suivant
Afficher la dernière version
1 file
+ 99
36
Comparer les modifications
  • Côte à côte
  • En ligne
+ 99
36
@@ -10,6 +10,8 @@ import argparse
from urllib.error import URLError
from urllib.parse import urlencode
from urllib.request import (urlopen, Request)
from configparser import ConfigParser
from pathlib import Path
from textwrap import wrap
@@ -20,6 +22,68 @@ class CustomFormatter(
pass
class Config:
def __init__(self):
"Create the config file if it doesn't exist, or parse it"
config_folder = Path.home() / ".config" / "kagari"
config_folder.mkdir(mode=0o700, parents=True, exist_ok=True)
config_file = config_folder / "kagari.ini"
config = ConfigParser()
# Read the config
if config_file.is_file():
with open(config_file, 'r') as f:
config.read_file(f)
if (repo := config.get('DEFAULT', 'repo')) not in config:
raise Exception(f"the repo {repo} is not in the config file, available repos are: {config.sections()}")
ok = config.has_option(repo, 'api') \
and config.has_option(repo, 'url') \
and config.get(repo, 'token', fallback='NO_TOKEN') != 'NO_TOKEN'
if not ok:
raise Exception(f"the repo {repo} is not valid...")
# Write the default config
else:
config['DEFAULT'] = {
'repo': 'kurisu',
'comments': False,
}
config['kurisu'] = {
'api': 'v1',
'url': 'https://kurisu.iiens.net',
'token': 'NO_TOKEN',
}
with open(config_file, 'w') as f:
config.write(f)
config_file = config_file.as_posix()
raise Exception(f"you need to set the repo token in the config file {config_file}")
self.config = config
def build_url(self):
"Builds the complete url to access the API"
api = self.get_repo('api')
url = self.get_repo('url')
if api == 'v1':
return f"{url}/api"
if api == 'v2':
return f"{url}/api/v2"
else:
raise Exception(f'Unsuported api version {api}')
def show_comments(self):
"Returns whever we must show the comments on karas or not"
return self.config.getboolean('DEFAULT', 'comments', fallback=True)
def get_repo(self, key):
"Returns the default repo of the configuration"
repo = self.config.get('DEFAULT', 'repo')
return self.config.get(repo, key)
def create_parser():
"Creates and returns a parser for the CLI arguments"
usage = "kagari [-h] (dl [dl options] | search [search options])"
@@ -125,28 +189,23 @@ def create_parser():
return parser
def make_request(url):
def make_request(url, config):
"Executes an HTTP request with basic error handling"
try:
req = Request(url)
req.add_header("X-Token-Authorization", "49duf300b8nmm469uzvz6dxwfarb05x3")
req.add_header("X-Token-Authorization", config.get_repo('token'))
response = urlopen(req)
except URLError as err:
if hasattr(err, "reason"):
emsg = f"Failed to reach target URL. Reason: {err.reason}"
print(emsg)
sys.exit(1)
elif hasattr(err, "code"):
emsg = f"The server could not fulfill the request. Code {err.code}"
print(emsg)
sys.exit(1)
if hasattr(err, "reason"): print(f"Failed to reach target URL. Reason: {err.reason}")
elif hasattr(err, "code"): print(f"The server could not fulfill the request. Code {err.code}")
sys.exit(1)
return response
def search_query(args):
def search_query(args, config):
"Searches the database with ID or filters, and returns the response json"
url = "https://kurisu.iiens.net/api"
url_values = dict()
url = config.build_url()
# If ID, it's the only parameter needed
if args.id:
@@ -163,7 +222,7 @@ def search_query(args):
if url_args:
url += "?" + url_args
response = make_request(url)
response = make_request(url, config)
try:
kara_json = json.loads(response.read().decode("utf-8"))
@@ -182,16 +241,16 @@ def search_query(args):
return kara_json
def build_kara_name(data):
def build_kara_name(data, config):
"Builds the name of the file from the response dictionary"
kara_id = data["id"]
src_name = data["source_name"]
kara_id = data["id"]
src_name = data["source_name"]
song_name = data["song_name"]
song_type = data["song_type"]
song_nbr = data["song_number"]
category = data["category"]
language = data["language"]
author = data["author_name"]
song_nbr = data["song_number"]
category = data["category"]
language = data["language"]
author = data["author_name"]
kara_name = f"{kara_id} - {src_name} - {song_type}{song_nbr} - {song_name}"
kara_additional = f" ({category}/{language}) by {author}"
@@ -224,15 +283,16 @@ def kara_prompt(kara_string="", kara_nbr=0):
return answers[ans]
def dl_mode(args, kara_json):
def dl_mode(args, kara_json, config):
"Downloads the targeted files"
kara_id = None
kara_string = ""
kara_nbr = 0
base_url = config.build_url()
if type(kara_json) is not list:
kara_name, kara_additional = build_kara_name(kara_json)
kara_name, kara_additional = build_kara_name(kara_json, config)
kara_string = '"' + kara_name + '"' + kara_additional
kara_id = kara_json["id"]
else:
@@ -246,12 +306,12 @@ def dl_mode(args, kara_json):
return
if kara_id:
url = f"https://kurisu.iiens.net/api/download/{kara_id}"
url = f"{base_url}/download/{kara_id}"
path = args.path if args.path else os.getcwd()
filename = path + "/" + kara_name + ".mkv"
filename = f"{path}/{kara_name}.mkv"
print(f'Downloading and writing video as "{filename}"')
response = make_request(url)
response = make_request(url, config)
# Writing the video file
with open(filename, "wb") as out_file:
@@ -261,12 +321,12 @@ def dl_mode(args, kara_json):
sys.stdout.flush()
for curr_nbr, kara in enumerate(kara_json):
kara_id = kara["id"]
kara_name, _ = build_kara_name(kara)
url = f"https://kurisu.iiens.net/api/download/{kara_id}"
kara_name, _ = build_kara_name(kara, config)
url = f"{base_url}/download/{kara_id}"
path = args.path if args.path else os.getcwd()
filename = path + "/" + kara_name + ".mkv"
filename = f"{path}/{kara_name}.mkv"
response = make_request(url)
response = make_request(url, config)
# Writing the video file
with open(filename, "wb") as out_file:
@@ -277,16 +337,18 @@ def dl_mode(args, kara_json):
print("All done !")
def search_mode(args, kara_json):
def search_mode(args, kara_json, config):
"Performs a search request on the database and returns the answer"
if args.json:
pretty_output = json.dumps(kara_json, indent=4, ensure_ascii=False)
pretty_output += "\n"
else:
pretty_output = ""
show_comments = config.show_comments()
if type(kara_json) is not list: kara_json = [kara_json]
for res in kara_json:
new_kara = " NEW!" if res["is_new"] == "1" else ""
output_info = build_kara_name(res)
output_info = build_kara_name(res, config)
output_string = (
f"[{res['id']}] {output_info[0]}\n"
+ f"{output_info[1]} in {res['author_year']}{new_kara}\n"
@@ -295,7 +357,7 @@ def search_mode(args, kara_json):
pretty_output += output_string
if res["popularity"]:
pretty_output += " Popularity: " + res["popularity"] + "\n"
if res["upload_comment"]:
if res["upload_comment"] and show_comments:
for comment_line in wrap(
res["upload_comment"], fix_sentence_endings=True
):
@@ -318,18 +380,19 @@ def search_mode(args, kara_json):
def run():
"Runs the command"
config = Config()
parser = create_parser()
args = parser.parse_args()
args = parser.parse_args()
kara_json = search_query(args)
kara_json = search_query(args, config)
if args.mode == "dl":
try:
dl_mode(args, kara_json)
dl_mode(args, kara_json, config)
except KeyboardInterrupt:
print("\nDownload interrupted by user input.")
elif args.mode == "search":
search_mode(args, kara_json)
search_mode(args, kara_json, config)
else:
print(f'Unknown mode "{args.mode}". Use either "dl" or "search".')
sys.exit(1)
Chargement en cours