-
-
Notifications
You must be signed in to change notification settings - Fork 120
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #150 from ryankupk/update-cli
Refactor CLI setup script, add prompts for muting and skipping native youtube ads
- Loading branch information
Showing
1 changed file
with
195 additions
and
171 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,171 +1,195 @@ | ||
import asyncio | ||
|
||
import aiohttp | ||
|
||
from . import api_helpers, ytlounge | ||
|
||
|
||
async def pair_device(web_session): | ||
try: | ||
lounge_controller = ytlounge.YtLoungeApi( | ||
"iSponsorBlockTV", web_session=web_session | ||
) | ||
pairing_code = input( | ||
"Enter pairing code (found in Settings - Link with TV code): " | ||
) | ||
pairing_code = int( | ||
pairing_code.replace("-", "").replace(" ", "") | ||
) # remove dashes and spaces | ||
print("Pairing...") | ||
paired = await lounge_controller.pair(pairing_code) | ||
if not paired: | ||
print("Failed to pair device") | ||
return | ||
device = { | ||
"screen_id": lounge_controller.auth.screen_id, | ||
"name": lounge_controller.screen_name, | ||
} | ||
print(f"Paired device: {device['name']}") | ||
return device | ||
except Exception as e: | ||
print(f"Failed to pair device: {e}") | ||
return | ||
|
||
|
||
def main(config, debug: bool) -> None: | ||
print("Welcome to the iSponsorBlockTV cli setup wizard") | ||
loop = asyncio.get_event_loop_policy().get_event_loop() | ||
web_session = aiohttp.ClientSession() | ||
if debug: | ||
loop.set_debug(True) | ||
asyncio.set_event_loop(loop) | ||
if hasattr(config, "atvs"): | ||
print( | ||
"The atvs config option is deprecated and has stopped working. Please read" | ||
" this for more information on how to upgrade to V2:" | ||
" \nhttps://github.com/dmunozv04/iSponsorBlockTV/wiki/Migrate-from-V1-to-V2" | ||
) | ||
if ( | ||
input( | ||
"Do you want to remove the legacy 'atvs' entry (the app won't start" | ||
" with it present)? (y/n) " | ||
) | ||
== "y" | ||
): | ||
del config["atvs"] | ||
devices = config.devices | ||
while not input(f"Paired with {len(devices)} Device(s). Add more? (y/n) ") == "n": | ||
task = loop.create_task(pair_device(web_session)) | ||
loop.run_until_complete(task) | ||
device = task.result() | ||
if device: | ||
devices.append(device) | ||
config.devices = devices | ||
|
||
apikey = config.apikey | ||
if apikey: | ||
if input("API key already specified. Change it? (y/n) ") == "y": | ||
apikey = input("Enter your API key: ") | ||
else: | ||
if ( | ||
input( | ||
"API key only needed for the channel whitelist function. Add it? (y/n) " | ||
) | ||
== "y" | ||
): | ||
print( | ||
"Get youtube apikey here:" | ||
" https://developers.google.com/youtube/registering_an_application" | ||
) | ||
apikey = input("Enter your API key: ") | ||
config.apikey = apikey | ||
|
||
skip_categories = config.skip_categories | ||
if skip_categories: | ||
if input("Skip categories already specified. Change them? (y/n) ") == "y": | ||
categories = input( | ||
"Enter skip categories (space or comma sepparated) Options: [sponsor" | ||
" selfpromo exclusive_access interaction poi_highlight intro outro" | ||
" preview filler music_offtopic]:\n" | ||
) | ||
skip_categories = categories.replace(",", " ").split(" ") | ||
skip_categories = [ | ||
x for x in skip_categories if x != "" | ||
] # Remove empty strings | ||
else: | ||
categories = input( | ||
"Enter skip categories (space or comma sepparated) Options: [sponsor," | ||
" selfpromo, exclusive_access, interaction, poi_highlight, intro, outro," | ||
" preview, filler, music_offtopic:\n" | ||
) | ||
skip_categories = categories.replace(",", " ").split(" ") | ||
skip_categories = [ | ||
x for x in skip_categories if x != "" | ||
] # Remove empty strings | ||
config.skip_categories = skip_categories | ||
|
||
channel_whitelist = config.channel_whitelist | ||
if ( | ||
input("Do you want to whitelist any channels from being ad-blocked? (y/n) ") | ||
== "y" | ||
): | ||
if not apikey: | ||
print( | ||
"WARNING: You need to specify an API key to use this function," | ||
" otherwise the program will fail to start.\nYou can add one by" | ||
" re-running this setup wizard." | ||
) | ||
api_helper = api_helpers.ApiHelper(config, web_session) | ||
while True: | ||
channel_info = {} | ||
channel = input('Enter a channel name or "/exit" to exit: ') | ||
if channel == "/exit": | ||
break | ||
|
||
task = loop.create_task( | ||
api_helper.search_channels(channel, apikey, web_session) | ||
) | ||
loop.run_until_complete(task) | ||
results = task.result() | ||
if len(results) == 0: | ||
print("No channels found") | ||
continue | ||
|
||
for i, item in enumerate(results): | ||
print(f"{i}: {item[1]} - Subs: {item[2]}") | ||
print("5: Enter a custom channel ID") | ||
print("6: Go back") | ||
|
||
choice = -1 | ||
choice = input("Select one option of the above [0-6]: ") | ||
while choice not in [str(x) for x in range(7)]: | ||
print("Invalid choice") | ||
choice = input("Select one option of the above [0-6]: ") | ||
|
||
if choice == "5": | ||
channel_info["id"] = input("Enter a channel ID: ") | ||
channel_info["name"] = input("Enter the channel name: ") | ||
channel_whitelist.append(channel_info) | ||
continue | ||
if choice == "6": | ||
continue | ||
|
||
channel_info["id"] = results[int(choice)][0] | ||
channel_info["name"] = results[int(choice)][1] | ||
channel_whitelist.append(channel_info) | ||
# Close web session asynchronously | ||
|
||
config.channel_whitelist = channel_whitelist | ||
|
||
config.skip_count_tracking = ( | ||
not input( | ||
"Do you want to report skipped segments to sponsorblock. Only the segment" | ||
" UUID will be sent? (y/n) " | ||
) | ||
== "n" | ||
) | ||
|
||
config.auto_play = not input("Do you want to enable autoplay? (y/n) ") == "n" | ||
print("Config finished") | ||
config.save() | ||
loop.run_until_complete(web_session.close()) | ||
import asyncio | ||
|
||
import aiohttp | ||
|
||
from . import api_helpers, ytlounge | ||
|
||
# Constants for user input prompts | ||
ATVS_REMOVAL_PROMPT = ( | ||
"Do you want to remove the legacy 'atvs' entry (the app won't start" | ||
" with it present)? (y/N) " | ||
) | ||
PAIRING_CODE_PROMPT = "Enter pairing code (found in Settings - Link with TV code): " | ||
ADD_MORE_DEVICES_PROMPT = "Paired with {num_devices} Device(s). Add more? (y/N) " | ||
CHANGE_API_KEY_PROMPT = "API key already specified. Change it? (y/N) " | ||
ADD_API_KEY_PROMPT = ( | ||
"API key only needed for the channel whitelist function. Add it? (y/N) " | ||
) | ||
ENTER_API_KEY_PROMPT = "Enter your API key: " | ||
CHANGE_SKIP_CATEGORIES_PROMPT = "Skip categories already specified. Change them? (y/N) " | ||
ENTER_SKIP_CATEGORIES_PROMPT = ( | ||
"Enter skip categories (space or comma sepparated) Options: [sponsor," | ||
" selfpromo, exclusive_access, interaction, poi_highlight, intro, outro," | ||
" preview, filler, music_offtopic]:\n" | ||
) | ||
WHITELIST_CHANNELS_PROMPT = ( | ||
"Do you want to whitelist any channels from being ad-blocked? (y/N) " | ||
) | ||
SEARCH_CHANNEL_PROMPT = 'Enter a channel name or "/exit" to exit: ' | ||
SELECT_CHANNEL_PROMPT = "Select one option of the above [0-6]: " | ||
ENTER_CHANNEL_ID_PROMPT = "Enter a channel ID: " | ||
ENTER_CUSTOM_CHANNEL_NAME_PROMPT = "Enter the channel name: " | ||
REPORT_SKIPPED_SEGMENTS_PROMPT = ( | ||
"Do you want to report skipped segments to sponsorblock. Only the segment" | ||
" UUID will be sent? (Y/n) " | ||
) | ||
MUTE_ADS_PROMPT = "Do you want to mute native YouTube ads automatically? (y/N) " | ||
SKIP_ADS_PROMPT = "Do you want to skip native YouTube ads automatically? (y/N) " | ||
AUTOPLAY_PROMPT = "Do you want to enable autoplay? (Y/n) " | ||
|
||
|
||
def get_yn_input(prompt): | ||
while choice := input(prompt): | ||
if choice.lower() in ["y", "n"]: | ||
return choice.lower() | ||
print("Invalid input. Please enter 'y' or 'n'.") | ||
|
||
|
||
async def pair_device(): | ||
try: | ||
lounge_controller = ytlounge.YtLoungeApi("iSponsorBlockTV") | ||
pairing_code = input(PAIRING_CODE_PROMPT) | ||
pairing_code = int( | ||
pairing_code.replace("-", "").replace(" ", "") | ||
) # remove dashes and spaces | ||
print("Pairing...") | ||
paired = await lounge_controller.pair(pairing_code) | ||
if not paired: | ||
print("Failed to pair device") | ||
return | ||
device = { | ||
"screen_id": lounge_controller.auth.screen_id, | ||
"name": lounge_controller.screen_name, | ||
} | ||
print(f"Paired device: {device['name']}") | ||
return device | ||
except Exception as e: | ||
print(f"Failed to pair device: {e}") | ||
return | ||
|
||
|
||
def main(config, debug: bool) -> None: | ||
print("Welcome to the iSponsorBlockTV cli setup wizard") | ||
loop = asyncio.get_event_loop_policy().get_event_loop() | ||
web_session = aiohttp.ClientSession() | ||
if debug: | ||
loop.set_debug(True) | ||
asyncio.set_event_loop(loop) | ||
if hasattr(config, "atvs"): | ||
print( | ||
"The atvs config option is deprecated and has stopped working. Please read" | ||
" this for more information on how to upgrade to V2:" | ||
" \nhttps://github.com/dmunozv04/iSponsorBlockTV/wiki/Migrate-from-V1-to-V2" | ||
) | ||
choice = get_yn_input(ATVS_REMOVAL_PROMPT) | ||
if choice == "y": | ||
del config["atvs"] | ||
|
||
devices = config.devices | ||
choice = get_yn_input(ADD_MORE_DEVICES_PROMPT.format(num_devices=len(devices))) | ||
while choice == "y": | ||
task = loop.create_task(pair_device()) | ||
loop.run_until_complete(task) | ||
device = task.result() | ||
if device: | ||
devices.append(device) | ||
choice = get_yn_input(ADD_MORE_DEVICES_PROMPT.format(num_devices=len(devices))) | ||
config.devices = devices | ||
|
||
apikey = config.apikey | ||
if apikey: | ||
choice = get_yn_input(CHANGE_API_KEY_PROMPT) | ||
if choice == "y": | ||
apikey = input(ENTER_API_KEY_PROMPT) | ||
else: | ||
choice = get_yn_input(ADD_API_KEY_PROMPT) | ||
if choice == "y": | ||
print( | ||
"Get youtube apikey here:" | ||
" https://developers.google.com/youtube/registering_an_application" | ||
) | ||
apikey = input(ENTER_API_KEY_PROMPT) | ||
config.apikey = apikey | ||
|
||
skip_categories = config.skip_categories | ||
if skip_categories: | ||
choice = get_yn_input(CHANGE_SKIP_CATEGORIES_PROMPT) | ||
if choice == "y": | ||
categories = input(ENTER_SKIP_CATEGORIES_PROMPT) | ||
skip_categories = categories.replace(",", " ").split(" ") | ||
skip_categories = [ | ||
x for x in skip_categories if x != "" | ||
] # Remove empty strings | ||
else: | ||
categories = input(ENTER_SKIP_CATEGORIES_PROMPT) | ||
skip_categories = categories.replace(",", " ").split(" ") | ||
skip_categories = [ | ||
x for x in skip_categories if x != "" | ||
] # Remove empty strings | ||
config.skip_categories = skip_categories | ||
|
||
channel_whitelist = config.channel_whitelist | ||
choice = get_yn_input(WHITELIST_CHANNELS_PROMPT) | ||
if choice == "y": | ||
if not apikey: | ||
print( | ||
"WARNING: You need to specify an API key to use this function," | ||
" otherwise the program will fail to start.\nYou can add one by" | ||
" re-running this setup wizard." | ||
) | ||
api_helper = api_helpers.ApiHelper(config, web_session) | ||
while True: | ||
channel_info = {} | ||
channel = input(SEARCH_CHANNEL_PROMPT) | ||
if channel == "/exit": | ||
break | ||
|
||
task = loop.create_task( | ||
api_helper.search_channels(channel, apikey, web_session) | ||
) | ||
loop.run_until_complete(task) | ||
results = task.result() | ||
if len(results) == 0: | ||
print("No channels found") | ||
continue | ||
|
||
for i, item in enumerate(results): | ||
print(f"{i}: {item[1]} - Subs: {item[2]}") | ||
print("5: Enter a custom channel ID") | ||
print("6: Go back") | ||
|
||
while choice := input(SELECT_CHANNEL_PROMPT): | ||
if choice in [str(x) for x in range(7)]: | ||
break | ||
print("Invalid choice") | ||
|
||
if choice == "5": | ||
channel_info["id"] = input(ENTER_CHANNEL_ID_PROMPT) | ||
channel_info["name"] = input(ENTER_CUSTOM_CHANNEL_NAME_PROMPT) | ||
channel_whitelist.append(channel_info) | ||
continue | ||
if choice == "6": | ||
continue | ||
|
||
channel_info["id"] = results[int(choice)][0] | ||
channel_info["name"] = results[int(choice)][1] | ||
channel_whitelist.append(channel_info) | ||
# Close web session asynchronously | ||
|
||
config.channel_whitelist = channel_whitelist | ||
|
||
choice = get_yn_input(REPORT_SKIPPED_SEGMENTS_PROMPT) | ||
config.skip_count_tracking = choice != "n" | ||
|
||
choice = get_yn_input(MUTE_ADS_PROMPT) | ||
config.mute_ads = choice == "y" | ||
|
||
choice = get_yn_input(SKIP_ADS_PROMPT) | ||
config.skip_ads = choice == "y" | ||
|
||
choice = get_yn_input(AUTOPLAY_PROMPT) | ||
config.auto_play = choice != "n" | ||
|
||
print("Config finished") | ||
config.save() | ||
loop.run_until_complete(web_session.close()) |