Mirror/iSponsorBlockTV
1
0
Fork 0
mirror of https://github.com/dmunozv04/iSponsorBlockTV.git synced 2025-12-06 11:56:45 +03:00
Code Issues Packages Projects Releases Wiki Activity

[pre-commit.ci] auto fixes from pre-commit.com hooks

Browse Source 
for more information, see https://pre-commit.ci
This commit is contained in:
pre-commit-ci[bot]
2024-01-14 13:47:13 +00:00
parent dae27e7aa3
commit a4f0b5fffe
23 changed files with 543 additions and 240 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
src/iSponsorBlockTV/api_helpers.py
View File

@@ -1,9 +1,11 @@
from cache import AsyncLRU
from .conditional_ttl_cache import AsyncConditionalTTL
from . import constants, dial_client
from hashlib import sha256
from aiohttp import ClientSession
import html import html
from hashlib import sha256
from aiohttp import ClientSession
from cache import AsyncLRU
from . import constants, dial_client
from .conditional_ttl_cache import AsyncConditionalTTL
def list_to_tuple(function): def list_to_tuple(function):
@@ -72,7 +74,13 @@ class ApiHelper:
@AsyncLRU(maxsize=10) @AsyncLRU(maxsize=10)
async def search_channels(self, channel): async def search_channels(self, channel):
channels = [] channels = []
params = {"q": channel, "key": self.apikey, "part": "snippet", "type": "channel", "maxResults": "5"} params = {
"q": channel,
"key": self.apikey,
"part": "snippet",
"type": "channel",
"maxResults": "5",
}
url = constants.Youtube_api + "search" url = constants.Youtube_api + "search"
async with self.web_session.get(url, params=params) as resp: async with self.web_session.get(url, params=params) as resp:
data = await resp.json() data = await resp.json()
@@ -81,7 +89,11 @@ class ApiHelper:
for i in data["items"]: for i in data["items"]:
# Get channel subscription number # Get channel subscription number
params = {"id": i["snippet"]["channelId"], "key": self.apikey, "part": "statistics"} params = {
"id": i["snippet"]["channelId"],
"key": self.apikey,
"part": "statistics",
}
url = constants.Youtube_api + "channels" url = constants.Youtube_api + "channels"
async with self.web_session.get(url, params=params) as resp: async with self.web_session.get(url, params=params) as resp:
channel_data = await resp.json() channel_data = await resp.json()
@@ -89,17 +101,26 @@ class ApiHelper:
if channel_data["items"][0]["statistics"]["hiddenSubscriberCount"]: if channel_data["items"][0]["statistics"]["hiddenSubscriberCount"]:
sub_count = "Hidden" sub_count = "Hidden"
else: else:
sub_count = int(channel_data["items"][0]["statistics"]["subscriberCount"]) sub_count = int(
channel_data["items"][0]["statistics"]["subscriberCount"]
)
sub_count = format(sub_count, "_") sub_count = format(sub_count, "_")
channels.append((i["snippet"]["channelId"], i["snippet"]["channelTitle"], sub_count)) channels.append(
(i["snippet"]["channelId"], i["snippet"]["channelTitle"], sub_count)
)
return channels return channels
@list_to_tuple # Convert list to tuple so it can be used as a key in the cache @list_to_tuple # Convert list to tuple so it can be used as a key in the cache
@AsyncConditionalTTL(time_to_live=300, maxsize=10) # 5 minutes for non-locked segments @AsyncConditionalTTL(
time_to_live=300, maxsize=10
) # 5 minutes for non-locked segments
async def get_segments(self, vid_id): async def get_segments(self, vid_id):
if await self.is_whitelisted(vid_id): if await self.is_whitelisted(vid_id):
return [], True # Return empty list and True to indicate that the cache should last forever return (
[],
True,
) # Return empty list and True to indicate that the cache should last forever
vid_id_hashed = sha256(vid_id.encode("utf-8")).hexdigest()[ vid_id_hashed = sha256(vid_id.encode("utf-8")).hexdigest()[
:4 :4
] # Hashes video id and gets the first 4 characters ] # Hashes video id and gets the first 4 characters
@@ -110,13 +131,16 @@ class ApiHelper:
} }
headers = {"Accept": "application/json"} headers = {"Accept": "application/json"}
url = constants.SponsorBlock_api + "skipSegments/" + vid_id_hashed url = constants.SponsorBlock_api + "skipSegments/" + vid_id_hashed
async with self.web_session.get(url, headers=headers, params=params) as response: async with self.web_session.get(
url, headers=headers, params=params
) as response:
response_json = await response.json() response_json = await response.json()
if response.status != 200: if response.status != 200:
response_text = await response.text() response_text = await response.text()
print( print(
f"Error getting segments for video {vid_id}, hashed as {vid_id_hashed}. " f"Error getting segments for video {vid_id}, hashed as {vid_id_hashed}. "
f"Code: {response.status} - {response_text}") f"Code: {response.status} - {response_text}"
)
return [], True return [], True
for i in response_json: for i in response_json:
if str(i["videoID"]) == str(vid_id): if str(i["videoID"]) == str(vid_id):
@@ -130,7 +154,9 @@ class ApiHelper:
ignore_ttl = True ignore_ttl = True
try: try:
for i in response["segments"]: for i in response["segments"]:
ignore_ttl = ignore_ttl and i["locked"] == 1 # If all segments are locked, ignore ttl ignore_ttl = (
ignore_ttl and i["locked"] == 1
) # If all segments are locked, ignore ttl
segment = i["segment"] segment = i["segment"]
UUID = i["UUID"] UUID = i["UUID"]
segment_dict = {"start": segment[0], "end": segment[1], "UUID": [UUID]} segment_dict = {"start": segment[0], "end": segment[1], "UUID": [UUID]}

23
src/iSponsorBlockTV/conditional_ttl_cache.py
View File

@@ -1,6 +1,7 @@
import datetime
from cache.key import KEY from cache.key import KEY
from cache.lru import LRU from cache.lru import LRU
import datetime
"""MIT License """MIT License
@@ -23,7 +24,7 @@ AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.""" SOFTWARE."""
'''Modified code from https://github.com/iamsinghrajat/async-cache''' """Modified code from https://github.com/iamsinghrajat/async-cache"""
class AsyncConditionalTTL: class AsyncConditionalTTL:
@@ -31,9 +32,9 @@ class AsyncConditionalTTL:
def __init__(self, time_to_live, maxsize): def __init__(self, time_to_live, maxsize):
super().__init__(maxsize=maxsize) super().__init__(maxsize=maxsize)
self.time_to_live = datetime.timedelta( self.time_to_live = (
seconds=time_to_live datetime.timedelta(seconds=time_to_live) if time_to_live else None
) if time_to_live else None )
self.maxsize = maxsize self.maxsize = maxsize
@@ -55,13 +56,13 @@ class AsyncConditionalTTL:
def __setitem__(self, key, value): def __setitem__(self, key, value):
value, ignore_ttl = value # unpack tuple value, ignore_ttl = value # unpack tuple
ttl_value = ( ttl_value = (
datetime.datetime.now() + self.time_to_live (datetime.datetime.now() + self.time_to_live)
) if (self.time_to_live and not ignore_ttl) else None # ignore ttl if ignore_ttl is True if (self.time_to_live and not ignore_ttl)
else None
) # ignore ttl if ignore_ttl is True
super().__setitem__(key, (value, ttl_value)) super().__setitem__(key, (value, ttl_value))
def __init__( def __init__(self, time_to_live=60, maxsize=1024, skip_args: int = 0):
self, time_to_live=60, maxsize=1024, skip_args: int = 0
):
""" """
:param time_to_live: Use time_to_live as None for non expiring cache :param time_to_live: Use time_to_live as None for non expiring cache
@@ -73,7 +74,7 @@ class AsyncConditionalTTL:
def __call__(self, func): def __call__(self, func):
async def wrapper(*args, **kwargs): async def wrapper(*args, **kwargs):
key = KEY(args[self.skip_args:], kwargs) key = KEY(args[self.skip_args :], kwargs)
if key in self.ttl: if key in self.ttl:
val = self.ttl[key] val = self.ttl[key]
else: else:

57
src/iSponsorBlockTV/config_setup.py
View File

@@ -1,13 +1,19 @@
import asyncio import asyncio
import aiohttp import aiohttp
from . import api_helpers, ytlounge from . import api_helpers, ytlounge
async def pair_device(): async def pair_device():
try: try:
lounge_controller = ytlounge.YtLoungeApi("iSponsorBlockTV") lounge_controller = ytlounge.YtLoungeApi("iSponsorBlockTV")
pairing_code = input("Enter pairing code (found in Settings - Link with TV code): ") pairing_code = input(
pairing_code = int(pairing_code.replace("-", "").replace(" ", "")) # remove dashes and spaces "Enter pairing code (found in Settings - Link with TV code): "
)
pairing_code = int(
pairing_code.replace("-", "").replace(" ", "")
) # remove dashes and spaces
print("Pairing...") print("Pairing...")
paired = await lounge_controller.pair(pairing_code) paired = await lounge_controller.pair(pairing_code)
if not paired: if not paired:
@@ -33,8 +39,14 @@ def main(config, debug: bool) -> None:
if hasattr(config, "atvs"): if hasattr(config, "atvs"):
print( print(
"The atvs config option is deprecated and has stopped working. Please read this for more information on " "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") "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": )
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"] del config["atvs"]
devices = config.devices devices = config.devices
while not input(f"Paired with {len(devices)} Device(s). Add more? (y/n) ") == "n": while not input(f"Paired with {len(devices)} Device(s). Add more? (y/n) ") == "n":
@@ -50,7 +62,12 @@ def main(config, debug: bool) -> None:
if input("API key already specified. Change it? (y/n) ") == "y": if input("API key already specified. Change it? (y/n) ") == "y":
apikey = input("Enter your API key: ") apikey = input("Enter your API key: ")
else: else:
if input("API key only needed for the channel whitelist function. Add it? (y/n) ") == "y": if (
input(
"API key only needed for the channel whitelist function. Add it? (y/n) "
)
== "y"
):
print( print(
"Get youtube apikey here: https://developers.google.com/youtube/registering_an_application" "Get youtube apikey here: https://developers.google.com/youtube/registering_an_application"
) )
@@ -65,31 +82,41 @@ def main(config, debug: bool) -> None:
"interaction poi_highlight intro outro preview filler music_offtopic]:\n" "interaction poi_highlight intro outro preview filler music_offtopic]:\n"
) )
skip_categories = categories.replace(",", " ").split(" ") skip_categories = categories.replace(",", " ").split(" ")
skip_categories = [x for x in skip_categories if x != ''] # Remove empty strings skip_categories = [
x for x in skip_categories if x != ""
] # Remove empty strings
else: else:
categories = input( categories = input(
"Enter skip categories (space or comma sepparated) Options: [sponsor, selfpromo, exclusive_access, " "Enter skip categories (space or comma sepparated) Options: [sponsor, selfpromo, exclusive_access, "
"interaction, poi_highlight, intro, outro, preview, filler, music_offtopic:\n" "interaction, poi_highlight, intro, outro, preview, filler, music_offtopic:\n"
) )
skip_categories = categories.replace(",", " ").split(" ") skip_categories = categories.replace(",", " ").split(" ")
skip_categories = [x for x in skip_categories if x != ''] # Remove empty strings skip_categories = [
x for x in skip_categories if x != ""
] # Remove empty strings
config.skip_categories = skip_categories config.skip_categories = skip_categories
channel_whitelist = config.channel_whitelist channel_whitelist = config.channel_whitelist
if input("Do you want to whitelist any channels from being ad-blocked? (y/n) ") == "y": if (
input("Do you want to whitelist any channels from being ad-blocked? (y/n) ")
== "y"
):
if not apikey: if not apikey:
print( print(
"WARNING: You need to specify an API key to use this function, otherwise the program will fail to " "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.") "start.\nYou can add one by re-running this setup wizard."
)
web_session = aiohttp.ClientSession() web_session = aiohttp.ClientSession()
api_helper = api_helpers.ApiHelper(config, web_session) api_helper = api_helpers.ApiHelper(config, web_session)
while True: while True:
channel_info = {} channel_info = {}
channel = input("Enter a channel name or \"/exit\" to exit: ") channel = input('Enter a channel name or "/exit" to exit: ')
if channel == "/exit": if channel == "/exit":
break break
task = loop.create_task(api_helper.search_channels(channel, apikey, web_session)) task = loop.create_task(
api_helper.search_channels(channel, apikey, web_session)
)
loop.run_until_complete(task) loop.run_until_complete(task)
results = task.result() results = task.result()
if len(results) == 0: if len(results) == 0:
@@ -123,7 +150,11 @@ def main(config, debug: bool) -> None:
config.channel_whitelist = channel_whitelist config.channel_whitelist = channel_whitelist
config.skip_count_tracking = not input( config.skip_count_tracking = (
"Do you want to report skipped segments to sponsorblock. Only the segment UUID will be sent? (y/n) ") == "n" not input(
"Do you want to report skipped segments to sponsorblock. Only the segment UUID will be sent? (y/n) "
)
== "n"
)
print("Config finished") print("Config finished")
config.save() config.save()

20
src/iSponsorBlockTV/constants.py
View File

@@ -6,16 +6,16 @@ SponsorBlock_api = "https://sponsor.ajay.app/api/"
Youtube_api = "https://www.googleapis.com/youtube/v3/" Youtube_api = "https://www.googleapis.com/youtube/v3/"
skip_categories = ( skip_categories = (
('Sponsor', 'sponsor'), ("Sponsor", "sponsor"),
('Self Promotion', 'selfpromo'), ("Self Promotion", "selfpromo"),
('Intro', 'intro'), ("Intro", "intro"),
('Outro', 'outro'), ("Outro", "outro"),
('Music Offtopic', 'music_offtopic'), ("Music Offtopic", "music_offtopic"),
('Interaction', 'interaction'), ("Interaction", "interaction"),
('Exclusive Access', 'exclusive_access'), ("Exclusive Access", "exclusive_access"),
('POI Highlight', 'poi_highlight'), ("POI Highlight", "poi_highlight"),
('Preview', 'preview'), ("Preview", "preview"),
('Filler', 'filler'), ("Filler", "filler"),
) )
youtube_client_blacklist = ["TVHTML5_FOR_KIDS"] youtube_client_blacklist = ["TVHTML5_FOR_KIDS"]

26
src/iSponsorBlockTV/dial_client.py
View File

@@ -1,11 +1,12 @@
"""Send out a M-SEARCH request and listening for responses.""" """Send out a M-SEARCH request and listening for responses."""
import asyncio import asyncio
import socket import socket
import ssdp
from ssdp import network
import xmltodict
'''Redistribution and use of the DIAL DIscovery And Launch protocol specification (the “DIAL Specification”), import ssdp
import xmltodict
from ssdp import network
"""Redistribution and use of the DIAL DIscovery And Launch protocol specification (the “DIAL Specification”),
with or without modification, are permitted provided that the following conditions are met: ● Redistributions of the with or without modification, are permitted provided that the following conditions are met: ● Redistributions of the
DIAL Specification must retain the above copyright notice, this list of conditions and the following disclaimer. ● DIAL Specification must retain the above copyright notice, this list of conditions and the following disclaimer. ●
Redistributions of implementations of the DIAL Specification in source code form must retain the above copyright Redistributions of implementations of the DIAL Specification in source code form must retain the above copyright
@@ -13,9 +14,9 @@ notice, this list of conditions and the following disclaimer. ● Redistribution
Specification in binary form must include the above copyright notice. ● The DIAL mark, the NETFLIX mark and the names Specification in binary form must include the above copyright notice. ● The DIAL mark, the NETFLIX mark and the names
of contributors to the DIAL Specification may not be used to endorse or promote specifications, software, products, of contributors to the DIAL Specification may not be used to endorse or promote specifications, software, products,
or any other materials derived from the DIAL Specification without specific prior written permission. The DIAL mark or any other materials derived from the DIAL Specification without specific prior written permission. The DIAL mark
is owned by Netflix and information on licensing the DIAL mark is available at www.dial-multiscreen.org.''' is owned by Netflix and information on licensing the DIAL mark is available at www.dial-multiscreen.org."""
''' """
MIT License MIT License
Copyright (c) 2018 Johannes Hoppe Copyright (c) 2018 Johannes Hoppe
@@ -36,8 +37,8 @@ FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.''' SOFTWARE."""
'''Modified code from https://github.com/codingjoe/ssdp/blob/main/ssdp/__main__.py''' """Modified code from https://github.com/codingjoe/ssdp/blob/main/ssdp/__main__.py"""
def get_ip(): def get_ip():
@@ -45,17 +46,16 @@ def get_ip():
s.settimeout(0) s.settimeout(0)
try: try:
# doesn't even have to be reachable # doesn't even have to be reachable
s.connect(('10.254.254.254', 1)) s.connect(("10.254.254.254", 1))
ip = s.getsockname()[0] ip = s.getsockname()[0]
except Exception: except Exception:
ip = '127.0.0.1' ip = "127.0.0.1"
finally: finally:
s.close() s.close()
return ip return ip
class Handler(ssdp.aio.SSDP): class Handler(ssdp.aio.SSDP):
def __init__(self): def __init__(self):
super().__init__() super().__init__()
self.devices = [] self.devices = []
@@ -107,7 +107,9 @@ async def discover(web_session):
family, addr = network.get_best_family(bind, network.PORT) family, addr = network.get_best_family(bind, network.PORT)
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()
ip_address = get_ip() ip_address = get_ip()
connect = loop.create_datagram_endpoint(handler, family=family, local_addr=(ip_address, None)) connect = loop.create_datagram_endpoint(
handler, family=family, local_addr=(ip_address, None)
)
transport, protocol = await connect transport, protocol = await connect
target = network.MULTICAST_ADDRESS_IPV4, network.PORT target = network.MULTICAST_ADDRESS_IPV4, network.PORT

35
src/iSponsorBlockTV/helpers.py
View File

@@ -59,7 +59,9 @@ class Config:
sys.exit() sys.exit()
self.devices = [Device(i) for i in self.devices] self.devices = [Device(i) for i in self.devices]
if not self.apikey and self.channel_whitelist: if not self.apikey and self.channel_whitelist:
raise ValueError("No youtube API key found and channel whitelist is not empty") raise ValueError(
"No youtube API key found and channel whitelist is not empty"
)
if not self.skip_categories: if not self.skip_categories:
self.skip_categories = ["sponsor"] self.skip_categories = ["sponsor"]
print("No categories found, using default: sponsor") print("No categories found, using default: sponsor")
@@ -79,11 +81,15 @@ class Config:
print("Creating data directory") print("Creating data directory")
os.makedirs(self.data_dir) os.makedirs(self.data_dir)
else: # Running in docker without mounting the data dir else: # Running in docker without mounting the data dir
print("Running in docker without mounting the data dir, check the wiki for more information: " print(
"https://github.com/dmunozv04/iSponsorBlockTV/wiki/Installation#Docker") "Running in docker without mounting the data dir, check the wiki for more information: "
print("This image has recently been updated to v2, and requires changes.", "https://github.com/dmunozv04/iSponsorBlockTV/wiki/Installation#Docker"
)
print(
"This image has recently been updated to v2, and requires changes.",
"Please read this for more information on how to upgrade to V2:", "Please read this for more information on how to upgrade to V2:",
"https://github.com/dmunozv04/iSponsorBlockTV/wiki/Migrate-from-V1-to-V2") "https://github.com/dmunozv04/iSponsorBlockTV/wiki/Migrate-from-V1-to-V2",
)
print("Exiting in 10 seconds...") print("Exiting in 10 seconds...")
time.sleep(10) time.sleep(10)
sys.exit() sys.exit()
@@ -110,11 +116,22 @@ class Config:
def app_start(): def app_start():
# If env has a data dir use that, otherwise use the default # If env has a data dir use that, otherwise use the default
default_data_dir = os.getenv("iSPBTV_data_dir") or user_data_dir("iSponsorBlockTV", "dmunozv04") default_data_dir = os.getenv("iSPBTV_data_dir") or user_data_dir(
"iSponsorBlockTV", "dmunozv04"
)
parser = argparse.ArgumentParser(description="iSponsorblockTV") parser = argparse.ArgumentParser(description="iSponsorblockTV")
parser.add_argument("--data-dir", "-d", default=default_data_dir, help="data directory") parser.add_argument(
parser.add_argument("--setup", "-s", action="store_true", help="setup the program graphically") "--data-dir", "-d", default=default_data_dir, help="data directory"
parser.add_argument("--setup-cli", "-sc", action="store_true", help="setup the program in the command line") )
parser.add_argument(
"--setup", "-s", action="store_true", help="setup the program graphically"
)
parser.add_argument(
"--setup-cli",
"-sc",
action="store_true",
help="setup the program in the command line",
)
parser.add_argument("--debug", action="store_true", help="debug mode") parser.add_argument("--debug", action="store_true", help="debug mode")
args = parser.parse_args() args = parser.parse_args()

24
src/iSponsorBlockTV/logging_helpers.py
View File

@@ -1,10 +1,10 @@
import logging import logging
from rich.logging import RichHandler from rich.logging import RichHandler
from rich.text import Text
from rich.style import Style from rich.style import Style
from rich.text import Text
''' """
Copyright (c) 2020 Will McGugan Copyright (c) 2020 Will McGugan
Permission is hereby granted, free of charge, to any person obtaining a copy Permission is hereby granted, free of charge, to any person obtaining a copy
@@ -25,7 +25,7 @@ LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE. SOFTWARE.
Modified code from rich (https://github.com/textualize/rich) Modified code from rich (https://github.com/textualize/rich)
''' """
class LogHandler(RichHandler): class LogHandler(RichHandler):
@@ -40,7 +40,8 @@ class LogHandler(RichHandler):
show_path=False, show_path=False,
time_format="[%x %X]", time_format="[%x %X]",
omit_repeated_times=True, omit_repeated_times=True,
level_width=None) level_width=None,
)
def add_filter_string(self, s): def add_filter_string(self, s):
self.filter_strings.append(s) self.filter_strings.append(s)
@@ -56,14 +57,17 @@ class LogHandler(RichHandler):
class LogRender: class LogRender:
def __init__(self, device_name, def __init__(
self,
device_name,
log_name_len, log_name_len,
show_time=True, show_time=True,
show_level=False, show_level=False,
show_path=True, show_path=True,
time_format="[%x %X]", time_format="[%x %X]",
omit_repeated_times=True, omit_repeated_times=True,
level_width=8): level_width=8,
):
self.filter_strings = [] self.filter_strings = []
self.log_name_len = log_name_len self.log_name_len = log_name_len
self.device_name = device_name self.device_name = device_name
@@ -95,7 +99,9 @@ class LogRender:
output.add_column(style="log.time") output.add_column(style="log.time")
if self.show_level: if self.show_level:
output.add_column(style="log.level", width=self.level_width) output.add_column(style="log.level", width=self.level_width)
output.add_column(width=self.log_name_len, style=Style(color="yellow"), overflow="fold") output.add_column(
width=self.log_name_len, style=Style(color="yellow"), overflow="fold"
)
output.add_column(ratio=1, style="log.message", overflow="fold") output.add_column(ratio=1, style="log.message", overflow="fold")
if self.show_path and path: if self.show_path and path:
output.add_column(style="log.path") output.add_column(style="log.path")
@@ -134,7 +140,9 @@ class LogRender:
class LogFormatter(logging.Formatter): class LogFormatter(logging.Formatter):
def __init__(self, fmt=None, datefmt=None, style="%", validate=True, filter_strings=None): def __init__(
self, fmt=None, datefmt=None, style="%", validate=True, filter_strings=None
):
super().__init__(fmt, datefmt, style, validate) super().__init__(fmt, datefmt, style, validate)
self.filter_strings = filter_strings or [] self.filter_strings = filter_strings or []

8
src/iSponsorBlockTV/macos_install.py
View File

@@ -1,6 +1,8 @@
import plistlib
import os import os
import plistlib
from . import config_setup from . import config_setup
"""Not updated to V2 yet, should still work. Here be dragons""" """Not updated to V2 yet, should still work. Here be dragons"""
default_plist = { default_plist = {
"Label": "com.dmunozv04iSponsorBlockTV", "Label": "com.dmunozv04iSponsorBlockTV",
@@ -40,8 +42,8 @@ def main():
create_plist(correct_path) create_plist(correct_path)
run_setup(correct_path + "/config.json") run_setup(correct_path + "/config.json")
print( print(
'Launch daemon installed. Please restart the computer to enable it or use:\n launchctl load ' "Launch daemon installed. Please restart the computer to enable it or use:\n launchctl load "
'~/Library/LaunchAgents/com.dmunozv04.iSponsorBlockTV.plist' "~/Library/LaunchAgents/com.dmunozv04.iSponsorBlockTV.plist"
) )
else: else:
if not os.path.exists(correct_path): if not os.path.exists(correct_path):

36
src/iSponsorBlockTV/main.py
View File

@@ -1,12 +1,13 @@
import asyncio import asyncio
import aiohttp
import time
import logging import logging
import rich import time
from signal import SIGINT, SIGTERM, signal
from typing import Optional from typing import Optional
from signal import signal, SIGINT, SIGTERM
from . import api_helpers, ytlounge, logging_helpers import aiohttp
import rich
from . import api_helpers, logging_helpers, ytlounge
class DeviceListener: class DeviceListener:
@@ -16,7 +17,7 @@ class DeviceListener:
self.offset = device.offset self.offset = device.offset
self.name = device.name self.name = device.name
self.cancelled = False self.cancelled = False
self.logger = logging.getLogger(f'iSponsorBlockTV-{device.screen_id}') self.logger = logging.getLogger(f"iSponsorBlockTV-{device.screen_id}")
if debug: if debug:
self.logger.setLevel(logging.DEBUG) self.logger.setLevel(logging.DEBUG)
else: else:
@@ -26,7 +27,8 @@ class DeviceListener:
self.logger.addHandler(rh) self.logger.addHandler(rh)
self.logger.info(f"Starting device") self.logger.info(f"Starting device")
self.lounge_controller = ytlounge.YtLoungeApi( self.lounge_controller = ytlounge.YtLoungeApi(
device.screen_id, config, api_helper, self.logger) device.screen_id, config, api_helper, self.logger
)
# Ensures that we have a valid auth token # Ensures that we have a valid auth token
async def refresh_auth_loop(self): async def refresh_auth_loop(self):
@@ -68,7 +70,9 @@ class DeviceListener:
await lounge_controller.connect() await lounge_controller.connect()
except: except:
pass pass
self.logger.info("Connected to device %s (%s)", lounge_controller.screen_name, self.name) self.logger.info(
"Connected to device %s (%s)", lounge_controller.screen_name, self.name
)
try: try:
# print("Subscribing to lounge") # print("Subscribing to lounge")
sub = await lounge_controller.subscribe_monitored(self) sub = await lounge_controller.subscribe_monitored(self)
@@ -91,7 +95,9 @@ class DeviceListener:
if state.videoId: if state.videoId:
segments = await self.api_helper.get_segments(state.videoId) segments = await self.api_helper.get_segments(state.videoId)
if state.state.value == 1: # Playing if state.state.value == 1: # Playing
self.logger.info(f"Playing video {state.videoId} with {len(segments)} segments") self.logger.info(
f"Playing video {state.videoId} with {len(segments)} segments"
)
if segments: # If there are segments if segments: # If there are segments
await self.time_to_segment(segments, state.currentTime, time_start) await self.time_to_segment(segments, state.currentTime, time_start)
@@ -100,18 +106,20 @@ class DeviceListener:
start_next_segment = None start_next_segment = None
next_segment = None next_segment = None
for segment in segments: for segment in segments:
if position < 2 and ( if position < 2 and (segment["start"] <= position < segment["end"]):
segment["start"] <= position < segment["end"]
):
next_segment = segment next_segment = segment
start_next_segment = position # different variable so segment doesn't change start_next_segment = (
position # different variable so segment doesn't change
)
break break
if segment["start"] > position: if segment["start"] > position:
next_segment = segment next_segment = segment
start_next_segment = next_segment["start"] start_next_segment = next_segment["start"]
break break
if start_next_segment: if start_next_segment:
time_to_next = start_next_segment - position - (time.time() - time_start) - self.offset time_to_next = (
start_next_segment - position - (time.time() - time_start) - self.offset
)
await self.skip(time_to_next, next_segment["end"], next_segment["UUID"]) await self.skip(time_to_next, next_segment["end"], next_segment["UUID"])
# Skips to the next segment (waits for the time to pass) # Skips to the next segment (waits for the time to pass)

374
src/iSponsorBlockTV/setup_wizard.py
View File

@@ -2,15 +2,33 @@ import asyncio
import copy import copy
import aiohttp import aiohttp
# Textual imports (Textual is awesome!) # Textual imports (Textual is awesome!)
from textual import on from textual import on
from textual.app import App, ComposeResult from textual.app import App, ComposeResult
from textual.containers import ScrollableContainer, Grid, Container, Vertical, Horizontal from textual.containers import (
Container,
Grid,
Horizontal,
ScrollableContainer,
Vertical,
)
from textual.events import Click from textual.events import Click
from textual.screen import Screen from textual.screen import Screen
from textual.validation import Function from textual.validation import Function
from textual.widgets import Button, Footer, Header, Static, Label, Input, SelectionList, Checkbox, ContentSwitcher, \ from textual.widgets import (
RadioSet, RadioButton Button,
Checkbox,
ContentSwitcher,
Footer,
Header,
Input,
Label,
RadioButton,
RadioSet,
SelectionList,
Static,
)
from textual.widgets.selection_list import Selection from textual.widgets.selection_list import Selection
from textual_slider import Slider from textual_slider import Slider
@@ -30,7 +48,9 @@ def _validate_pairing_code(pairing_code: str) -> bool:
class ModalWithClickExit(Screen): class ModalWithClickExit(Screen):
"""A modal screen that exits when clicked outside its bounds. """A modal screen that exits when clicked outside its bounds.
https://discord.com/channels/1026214085173461072/1033754296224841768/1136015817356611604""" https://discord.com/channels/1026214085173461072/1033754296224841768/1136015817356611604
"""
DEFAULT_CSS = """ DEFAULT_CSS = """
ModalWithClickExit { ModalWithClickExit {
align: center middle; align: center middle;
@@ -62,8 +82,15 @@ class Element(Static):
pass pass
def compose(self) -> ComposeResult: def compose(self) -> ComposeResult:
yield Button(label=self.element_name, classes="element-name", disabled=True, id="element-name") yield Button(
yield Button("Remove", classes="element-remove", variant="error", id="element-remove") label=self.element_name,
classes="element-name",
disabled=True,
id="element-name",
)
yield Button(
"Remove", classes="element-remove", variant="error", id="element-remove"
)
def on_mount(self) -> None: def on_mount(self) -> None:
if self.tooltip: if self.tooltip:
@@ -80,9 +107,11 @@ class Device(Element):
if "name" in self.element_data and self.element_data["name"]: if "name" in self.element_data and self.element_data["name"]:
self.element_name = self.element_data["name"] self.element_name = self.element_data["name"]
else: else:
self.element_name = (f"Unnamed device with id " self.element_name = (
f"Unnamed device with id "
f"{self.element_data['screen_id'][:5]}..." f"{self.element_data['screen_id'][:5]}..."
f"{self.element_data['screen_id'][-5:]}") f"{self.element_data['screen_id'][-5:]}"
)
class Channel(Element): class Channel(Element):
@@ -92,7 +121,9 @@ class Channel(Element):
if "name" in self.element_data: if "name" in self.element_data:
self.element_name = self.element_data["name"] self.element_name = self.element_data["name"]
else: else:
self.element_name = f"Unnamed channel with id {self.element_data['channel_id']}" self.element_name = (
f"Unnamed channel with id {self.element_data['channel_id']}"
)
class ChannelRadio(RadioButton): class ChannelRadio(RadioButton):
@@ -106,9 +137,12 @@ class ChannelRadio(RadioButton):
class MigrationScreen(ModalWithClickExit): class MigrationScreen(ModalWithClickExit):
"""Screen with a dialog to remove old ATVS config.""" """Screen with a dialog to remove old ATVS config."""
BINDINGS = [("escape", "dismiss()", "Cancel"),
BINDINGS = [
("escape", "dismiss()", "Cancel"),
("s", "remove_and_save", "Remove and save"), ("s", "remove_and_save", "Remove and save"),
("q,ctrl+c", "exit", "Exit")] ("q,ctrl+c", "exit", "Exit"),
]
AUTO_FOCUS = "#exit-save" AUTO_FOCUS = "#exit-save"
def compose(self) -> ComposeResult: def compose(self) -> ComposeResult:
@@ -116,9 +150,21 @@ class MigrationScreen(ModalWithClickExit):
Label( Label(
"Welcome to the new configurator! You seem to have the legacy 'atvs' entry on your config file, " "Welcome to the new configurator! You seem to have the legacy 'atvs' entry on your config file, "
"do you want to remove it?\n(The app won't start with it present)", "do you want to remove it?\n(The app won't start with it present)",
id="question", classes="button-100"), id="question",
Button("Remove and save", variant="primary", id="migrate-remove-save", classes="button-100"), classes="button-100",
Button("Don't remove", variant="error", id="migrate-no-change", classes="button-100"), ),
Button(
"Remove and save",
variant="primary",
id="migrate-remove-save",
classes="button-100",
),
Button(
"Don't remove",
variant="error",
id="migrate-no-change",
classes="button-100",
),
id="dialog-migration", id="dialog-migration",
) )
@@ -138,16 +184,25 @@ class MigrationScreen(ModalWithClickExit):
class ExitScreen(ModalWithClickExit): class ExitScreen(ModalWithClickExit):
"""Screen with a dialog to exit.""" """Screen with a dialog to exit."""
BINDINGS = [("escape", "dismiss()", "Cancel"),
BINDINGS = [
("escape", "dismiss()", "Cancel"),
("s", "save", "Save"), ("s", "save", "Save"),
("q,ctrl+c", "exit", "Exit")] ("q,ctrl+c", "exit", "Exit"),
]
AUTO_FOCUS = "#exit-save" AUTO_FOCUS = "#exit-save"
def compose(self) -> ComposeResult: def compose(self) -> ComposeResult:
yield Grid( yield Grid(
Label("Are you sure you want to exit without saving?", id="question", classes="button-100"), Label(
"Are you sure you want to exit without saving?",
id="question",
classes="button-100",
),
Button("Save", variant="success", id="exit-save", classes="button-100"), Button("Save", variant="success", id="exit-save", classes="button-100"),
Button("Don't save", variant="error", id="exit-no-save", classes="button-100"), Button(
"Don't save", variant="error", id="exit-no-save", classes="button-100"
),
Button("Cancel", variant="primary", id="exit-cancel", classes="button-100"), Button("Cancel", variant="primary", id="exit-cancel", classes="button-100"),
id="dialog-exit", id="dialog-exit",
) )
@@ -171,6 +226,7 @@ class ExitScreen(ModalWithClickExit):
class AddDevice(ModalWithClickExit): class AddDevice(ModalWithClickExit):
"""Screen with a dialog to add a device, either with a pairing code or with lan discovery.""" """Screen with a dialog to add a device, either with a pairing code or with lan discovery."""
BINDINGS = [("escape", "dismiss({})", "Return")] BINDINGS = [("escape", "dismiss({})", "Return")]
def __init__(self, config, **kwargs) -> None: def __init__(self, config, **kwargs) -> None:
@@ -184,28 +240,58 @@ class AddDevice(ModalWithClickExit):
with Container(id="add-device-container"): with Container(id="add-device-container"):
yield Label("Add Device", classes="title") yield Label("Add Device", classes="title")
with Grid(id="add-device-switch-buttons"): with Grid(id="add-device-switch-buttons"):
yield Button("Add with pairing code", id="add-device-pin-button", classes="button-switcher") yield Button(
yield Button("Add with lan discovery", id="add-device-dial-button", classes="button-switcher") "Add with pairing code",
with ContentSwitcher(id="add-device-switcher", initial="add-device-pin-container"): id="add-device-pin-button",
classes="button-switcher",
)
yield Button(
"Add with lan discovery",
id="add-device-dial-button",
classes="button-switcher",
)
with ContentSwitcher(
id="add-device-switcher", initial="add-device-pin-container"
):
with Container(id="add-device-pin-container"): with Container(id="add-device-pin-container"):
yield Input(placeholder="Pairing Code (found in Settings - Link with TV code)", yield Input(
placeholder="Pairing Code (found in Settings - Link with TV code)",
id="pairing-code-input", id="pairing-code-input",
validators=[ validators=[
Function(_validate_pairing_code, "Invalid pairing code format") Function(
] _validate_pairing_code, "Invalid pairing code format"
)
],
)
yield Input(
placeholder="Device Name (auto filled if empty/optional)",
id="device-name-input",
)
yield Button(
"Add",
id="add-device-pin-add-button",
variant="success",
disabled=True,
) )
yield Input(placeholder="Device Name (auto filled if empty/optional)", id="device-name-input")
yield Button("Add", id="add-device-pin-add-button", variant="success", disabled=True)
yield Label(id="add-device-info") yield Label(id="add-device-info")
with Container(id="add-device-dial-container"): with Container(id="add-device-dial-container"):
yield Label( yield Label(
"Make sure your device is on the same network as this computer\nIf it isn't showing up, " "Make sure your device is on the same network as this computer\nIf it isn't showing up, "
"try restarting the app.\nIf running in docker, make sure to use `--network=host`\nTo refresh " "try restarting the app.\nIf running in docker, make sure to use `--network=host`\nTo refresh "
"the list, close and open the dialog again", "the list, close and open the dialog again",
classes="subtitle") classes="subtitle",
yield SelectionList(("Searching for devices...", "", False), id="dial-devices-list", disabled=True) )
yield Button("Add selected devices", id="add-device-dial-add-button", variant="success", yield SelectionList(
disabled=True) ("Searching for devices...", "", False),
id="dial-devices-list",
disabled=True,
)
yield Button(
"Add selected devices",
id="add-device-dial-add-button",
variant="success",
disabled=True,
)
async def on_mount(self) -> None: async def on_mount(self) -> None:
self.devices_discovered_dial = [] self.devices_discovered_dial = []
@@ -228,11 +314,15 @@ class AddDevice(ModalWithClickExit):
@on(Button.Pressed, "#add-device-switch-buttons > *") @on(Button.Pressed, "#add-device-switch-buttons > *")
def handle_switch_buttons(self, event: Button.Pressed) -> None: def handle_switch_buttons(self, event: Button.Pressed) -> None:
self.query_one("#add-device-switcher").current = event.button.id.replace("-button", "-container") self.query_one("#add-device-switcher").current = event.button.id.replace(
"-button", "-container"
)
@on(Input.Changed, "#pairing-code-input") @on(Input.Changed, "#pairing-code-input")
def changed_pairing_code(self, event: Input.Changed): def changed_pairing_code(self, event: Input.Changed):
self.query_one("#add-device-pin-add-button").disabled = not event.validation_result.is_valid self.query_one(
"#add-device-pin-add-button"
).disabled = not event.validation_result.is_valid
@on(Input.Submitted, "#pairing-code-input") @on(Input.Submitted, "#pairing-code-input")
@on(Button.Pressed, "#add-device-pin-add-button") @on(Button.Pressed, "#add-device-pin-add-button")
@@ -240,7 +330,9 @@ class AddDevice(ModalWithClickExit):
self.query_one("#add-device-pin-add-button").disabled = True self.query_one("#add-device-pin-add-button").disabled = True
lounge_controller = ytlounge.YtLoungeApi("iSponsorBlockTV") lounge_controller = ytlounge.YtLoungeApi("iSponsorBlockTV")
pairing_code = self.query_one("#pairing-code-input").value pairing_code = self.query_one("#pairing-code-input").value
pairing_code = int(pairing_code.replace("-", "").replace(" ", "")) # remove dashes and spaces pairing_code = int(
pairing_code.replace("-", "").replace(" ", "")
) # remove dashes and spaces
device_name = self.parent.query_one("#device-name-input").value device_name = self.parent.query_one("#device-name-input").value
paired = False paired = False
try: try:
@@ -255,7 +347,9 @@ class AddDevice(ModalWithClickExit):
} }
self.query_one("#pairing-code-input").value = "" self.query_one("#pairing-code-input").value = ""
self.query_one("#device-name-input").value = "" self.query_one("#device-name-input").value = ""
self.query_one("#add-device-info").update(f"[#00ff00][b]Successfully added {device['name']}") self.query_one("#add-device-info").update(
f"[#00ff00][b]Successfully added {device['name']}"
)
self.dismiss([device]) self.dismiss([device])
else: else:
self.query_one("#pairing-code-input").value = "" self.query_one("#pairing-code-input").value = ""
@@ -273,11 +367,14 @@ class AddDevice(ModalWithClickExit):
@on(SelectionList.SelectedChanged, "#dial-devices-list") @on(SelectionList.SelectedChanged, "#dial-devices-list")
def changed_device_list(self, event: SelectionList.SelectedChanged): def changed_device_list(self, event: SelectionList.SelectedChanged):
self.query_one("#add-device-dial-add-button").disabled = not event.selection_list.selected self.query_one(
"#add-device-dial-add-button"
).disabled = not event.selection_list.selected
class AddChannel(ModalWithClickExit): class AddChannel(ModalWithClickExit):
"""Screen with a dialog to add a channel, either using search or with a channel id.""" """Screen with a dialog to add a channel, either using search or with a channel id."""
BINDINGS = [("escape", "dismiss(())", "Return")] BINDINGS = [("escape", "dismiss(())", "Return")]
def __init__(self, config, **kwargs) -> None: def __init__(self, config, **kwargs) -> None:
@@ -291,32 +388,67 @@ class AddChannel(ModalWithClickExit):
yield Label("Add Channel", classes="title") yield Label("Add Channel", classes="title")
yield Label( yield Label(
"Select a method to add a channel. Adding via search only works if a YouTube api key has been set", "Select a method to add a channel. Adding via search only works if a YouTube api key has been set",
id="add-channel-label", classes="subtitle") id="add-channel-label",
classes="subtitle",
)
with Grid(id="add-channel-switch-buttons"): with Grid(id="add-channel-switch-buttons"):
yield Button("Add by channel name", id="add-channel-search-button", classes="button-switcher") yield Button(
yield Button("Add by channel ID", id="add-channel-id-button", classes="button-switcher") "Add by channel name",
id="add-channel-search-button",
classes="button-switcher",
)
yield Button(
"Add by channel ID",
id="add-channel-id-button",
classes="button-switcher",
)
yield Label(id="add-channel-info", classes="subtitle") yield Label(id="add-channel-info", classes="subtitle")
with ContentSwitcher(id="add-channel-switcher", initial="add-channel-search-container"): with ContentSwitcher(
id="add-channel-switcher", initial="add-channel-search-container"
):
with Vertical(id="add-channel-search-container"): with Vertical(id="add-channel-search-container"):
if self.config.apikey: if self.config.apikey:
with Grid(id="add-channel-search-inputs"): with Grid(id="add-channel-search-inputs"):
yield Input(placeholder="Enter channel name", id="channel-name-input-search") yield Input(
yield Button("Search", id="search-channel-button", variant="success") placeholder="Enter channel name",
yield RadioSet(RadioButton(label="Search to see results", disabled=True), id="channel-name-input-search",
id="channel-search-results") )
yield Button("Add", id="add-channel-button-search", variant="success", disabled=True, yield Button(
classes="button-100") "Search", id="search-channel-button", variant="success"
)
yield RadioSet(
RadioButton(label="Search to see results", disabled=True),
id="channel-search-results",
)
yield Button(
"Add",
id="add-channel-button-search",
variant="success",
disabled=True,
classes="button-100",
)
else: else:
yield Label( yield Label(
"[#ff0000]No api key set, cannot search for channels. You can add it the config section " "[#ff0000]No api key set, cannot search for channels. You can add it the config section "
"below", "below",
id="add-channel-search-no-key", classes="subtitle") id="add-channel-search-no-key",
classes="subtitle",
)
with Vertical(id="add-channel-id-container"): with Vertical(id="add-channel-id-container"):
yield Input(placeholder="Enter channel ID (example: UCuAXFkgsw1L7xaCfnd5JJOw)", yield Input(
id="channel-id-input") placeholder="Enter channel ID (example: UCuAXFkgsw1L7xaCfnd5JJOw)",
yield Input(placeholder="Enter channel name (only used to display in the config file)", id="channel-id-input",
id="channel-name-input-id") )
yield Button("Add", id="add-channel-button-id", variant="success", classes="button-100") yield Input(
placeholder="Enter channel name (only used to display in the config file)",
id="channel-name-input-id",
)
yield Button(
"Add",
id="add-channel-button-id",
variant="success",
classes="button-100",
)
@on(RadioSet.Changed, "#channel-search-results") @on(RadioSet.Changed, "#channel-search-results")
def handle_radio_set_changed(self, event: RadioSet.Changed) -> None: def handle_radio_set_changed(self, event: RadioSet.Changed) -> None:
@@ -325,14 +457,18 @@ class AddChannel(ModalWithClickExit):
@on(Button.Pressed, "#add-channel-switch-buttons > *") @on(Button.Pressed, "#add-channel-switch-buttons > *")
def handle_switch_buttons(self, event: Button.Pressed) -> None: def handle_switch_buttons(self, event: Button.Pressed) -> None:
button_ = event.button.id button_ = event.button.id
self.query_one("#add-channel-switcher").current = event.button.id.replace("-button", "-container") self.query_one("#add-channel-switcher").current = event.button.id.replace(
"-button", "-container"
)
@on(Button.Pressed, "#search-channel-button") @on(Button.Pressed, "#search-channel-button")
@on(Input.Submitted, "#channel-name-input-search") @on(Input.Submitted, "#channel-name-input-search")
async def handle_search_channel(self) -> None: async def handle_search_channel(self) -> None:
channel_name = self.query_one("#channel-name-input-search").value channel_name = self.query_one("#channel-name-input-search").value
if not channel_name: if not channel_name:
self.query_one("#add-channel-info").update("[#ff0000]Please enter a channel name") self.query_one("#add-channel-info").update(
"[#ff0000]Please enter a channel name"
)
return return
self.query_one("#search-channel-button").disabled = True self.query_one("#search-channel-button").disabled = True
self.query_one("#add-channel-info").update("Searching...") self.query_one("#add-channel-info").update("Searching...")
@@ -341,7 +477,9 @@ class AddChannel(ModalWithClickExit):
try: try:
channels_list = await self.api_helper.search_channels(channel_name) channels_list = await self.api_helper.search_channels(channel_name)
except: except:
self.query_one("#add-channel-info").update("[#ff0000]Failed to search for channel") self.query_one("#add-channel-info").update(
"[#ff0000]Failed to search for channel"
)
self.query_one("#search-channel-button").disabled = False self.query_one("#search-channel-button").disabled = False
return return
for i in channels_list: for i in channels_list:
@@ -354,7 +492,9 @@ class AddChannel(ModalWithClickExit):
def handle_add_channel_search(self) -> None: def handle_add_channel_search(self) -> None:
channel = self.query_one("#channel-search-results").pressed_button.channel_data channel = self.query_one("#channel-search-results").pressed_button.channel_data
if not channel: if not channel:
self.query_one("#add-channel-info").update("[#ff0000]Please select a channel") self.query_one("#add-channel-info").update(
"[#ff0000]Please select a channel"
)
return return
self.query_one("#add-channel-info").update("Adding...") self.query_one("#add-channel-info").update("Adding...")
self.dismiss(channel) self.dismiss(channel)
@@ -366,7 +506,9 @@ class AddChannel(ModalWithClickExit):
channel_id = self.query_one("#channel-id-input").value channel_id = self.query_one("#channel-id-input").value
channel_name = self.query_one("#channel-name-input-id").value channel_name = self.query_one("#channel-name-input-id").value
if not channel_id: if not channel_id:
self.query_one("#add-channel-info").update("[#ff0000]Please enter a channel ID") self.query_one("#add-channel-info").update(
"[#ff0000]Please enter a channel ID"
)
return return
if not channel_name: if not channel_name:
channel_name = channel_id channel_name = channel_id
@@ -377,6 +519,7 @@ class AddChannel(ModalWithClickExit):
class EditDevice(ModalWithClickExit): class EditDevice(ModalWithClickExit):
"""Screen with a dialog to edit a device. Used by the DevicesManager.""" """Screen with a dialog to edit a device. Used by the DevicesManager."""
BINDINGS = [("escape", "close_screen_saving", "Return")] BINDINGS = [("escape", "close_screen_saving", "Return")]
def __init__(self, device: Element, **kwargs) -> None: def __init__(self, device: Element, **kwargs) -> None:
@@ -402,13 +545,24 @@ class EditDevice(ModalWithClickExit):
yield Input(placeholder="Device name", id="device-name-input", value=name) yield Input(placeholder="Device name", id="device-name-input", value=name)
yield Label("Device screen id") yield Label("Device screen id")
with Grid(id="device-id-container"): with Grid(id="device-id-container"):
yield Input(placeholder="Device id", id="device-id-input", value=self.device_data["screen_id"], yield Input(
password=True) placeholder="Device id",
id="device-id-input",
value=self.device_data["screen_id"],
password=True,
)
yield Button("Show id", id="device-id-view") yield Button("Show id", id="device-id-view")
yield Label("Device offset (in milliseconds)") yield Label("Device offset (in milliseconds)")
with Horizontal(id="device-offset-container"): with Horizontal(id="device-offset-container"):
yield Input(id="device-offset-input", value=str(offset)) yield Input(id="device-offset-input", value=str(offset))
yield Slider(name="Device offset", id="device-offset-slider", min=0, max=2000, step=100, value=offset) yield Slider(
name="Device offset",
id="device-offset-slider",
min=0,
max=2000,
step=100,
value=offset,
)
def on_slider_changed(self, event: Slider.Changed) -> None: def on_slider_changed(self, event: Slider.Changed) -> None:
offset_input = self.query_one("#device-offset-offset_input") offset_input = self.query_one("#device-offset-offset_input")
@@ -491,9 +645,15 @@ class ApiKeyManager(Vertical):
yield Label( yield Label(
"You can get a YouTube Data API v3 Key from the [" "You can get a YouTube Data API v3 Key from the ["
"link=https://console.developers.google.com/apis/credentials]Google Cloud Console[/link]. This key is " "link=https://console.developers.google.com/apis/credentials]Google Cloud Console[/link]. This key is "
"only required if you're whitelisting channels.") "only required if you're whitelisting channels."
)
with Grid(id="api-key-grid"): with Grid(id="api-key-grid"):
yield Input(placeholder="YouTube Api Key", id="api-key-input", password=True, value=self.config.apikey) yield Input(
placeholder="YouTube Api Key",
id="api-key-input",
password=True,
value=self.config.apikey,
)
yield Button("Show key", id="api-key-view") yield Button("Show key", id="api-key-view")
@on(Input.Changed, "#api-key-input") @on(Input.Changed, "#api-key-input")
@@ -549,9 +709,14 @@ class SkipCountTrackingManager(Vertical):
"helped others and used as a metric along with upvotes to ensure that spam doesn't get into the database. " "helped others and used as a metric along with upvotes to ensure that spam doesn't get into the database. "
"The program sends a message to the sponsor block server each time you skip a segment. Hopefully most " "The program sends a message to the sponsor block server each time you skip a segment. Hopefully most "
"people don't change this setting so that the view numbers are accurate. :)", "people don't change this setting so that the view numbers are accurate. :)",
classes="subtitle", id="skip-count-tracking-subtitle") classes="subtitle",
yield Checkbox(value=self.config.skip_count_tracking, id="skip-count-tracking-switch", id="skip-count-tracking-subtitle",
label="Enable skip count tracking") )
yield Checkbox(
value=self.config.skip_count_tracking,
id="skip-count-tracking-switch",
label="Enable skip count tracking",
)
@on(Checkbox.Changed, "#skip-count-tracking-switch") @on(Checkbox.Changed, "#skip-count-tracking-switch")
def changed_skip_tracking(self, event: Checkbox.Changed): def changed_skip_tracking(self, event: Checkbox.Changed):
@@ -570,12 +735,20 @@ class AdSkipMuteManager(Vertical):
yield Label( yield Label(
"This feature allows you to automatically mute and/or skip native YouTube ads. Skipping ads only works if " "This feature allows you to automatically mute and/or skip native YouTube ads. Skipping ads only works if "
"that ad shows the 'Skip Ad' button, if it doesn't then it will only be able to be muted.", "that ad shows the 'Skip Ad' button, if it doesn't then it will only be able to be muted.",
classes="subtitle", id="skip-count-tracking-subtitle") classes="subtitle",
id="skip-count-tracking-subtitle",
)
with Horizontal(id="ad-skip-mute-container"): with Horizontal(id="ad-skip-mute-container"):
yield Checkbox(value=self.config.skip_ads, id="skip-ads-switch", yield Checkbox(
label="Enable skipping ads") value=self.config.skip_ads,
yield Checkbox(value=self.config.mute_ads, id="mute-ads-switch", id="skip-ads-switch",
label="Enable muting ads") label="Enable skipping ads",
)
yield Checkbox(
value=self.config.mute_ads,
id="mute-ads-switch",
label="Enable muting ads",
)
@on(Checkbox.Changed, "#mute-ads-switch") @on(Checkbox.Changed, "#mute-ads-switch")
def changed_mute(self, event: Checkbox.Changed): def changed_mute(self, event: Checkbox.Changed):
@@ -598,16 +771,22 @@ class ChannelWhitelistManager(Vertical):
yield Label( yield Label(
"This feature allows to whitelist channels from being skipped. This feature is automatically disabled " "This feature allows to whitelist channels from being skipped. This feature is automatically disabled "
"when no channels have been specified.", "when no channels have been specified.",
classes="subtitle", id="channel-whitelist-subtitle") classes="subtitle",
yield Label(":warning: [#FF0000]You need to set your YouTube Api Key in order to use this feature", id="channel-whitelist-subtitle",
id="warning-no-key") )
yield Label(
":warning: [#FF0000]You need to set your YouTube Api Key in order to use this feature",
id="warning-no-key",
)
with Horizontal(id="add-channel-button-container"): with Horizontal(id="add-channel-button-container"):
yield Button("Add Channel", id="add-channel", classes="button-100") yield Button("Add Channel", id="add-channel", classes="button-100")
for channel in self.config.channel_whitelist: for channel in self.config.channel_whitelist:
yield Channel(channel) yield Channel(channel)
def on_mount(self) -> None: def on_mount(self) -> None:
self.app.query_one("#warning-no-key").display = (not self.config.apikey) and bool(self.config.channel_whitelist) self.app.query_one("#warning-no-key").display = (
not self.config.apikey
) and bool(self.config.channel_whitelist)
def new_channel(self, channel: tuple) -> None: def new_channel(self, channel: tuple) -> None:
if channel: if channel:
@@ -619,16 +798,18 @@ class ChannelWhitelistManager(Vertical):
channel_widget = Channel(channel_dict) channel_widget = Channel(channel_dict)
self.mount(channel_widget) self.mount(channel_widget)
channel_widget.focus(scroll_visible=True) channel_widget.focus(scroll_visible=True)
self.app.query_one("#warning-no-key").display = (not self.config.apikey) and bool( self.app.query_one("#warning-no-key").display = (
self.config.channel_whitelist) not self.config.apikey
) and bool(self.config.channel_whitelist)
@on(Button.Pressed, "#element-remove") @on(Button.Pressed, "#element-remove")
def remove_channel(self, event: Button.Pressed): def remove_channel(self, event: Button.Pressed):
channel_to_remove: Element = event.button.parent channel_to_remove: Element = event.button.parent
self.config.channel_whitelist.remove(channel_to_remove.element_data) self.config.channel_whitelist.remove(channel_to_remove.element_data)
channel_to_remove.remove() channel_to_remove.remove()
self.app.query_one("#warning-no-key").display = (not self.config.apikey) and bool( self.app.query_one("#warning-no-key").display = (
self.config.channel_whitelist) not self.config.apikey
) and bool(self.config.channel_whitelist)
@on(Button.Pressed, "#add-channel") @on(Button.Pressed, "#add-channel")
def add_channel(self, event: Button.Pressed): def add_channel(self, event: Button.Pressed):
@@ -637,12 +818,10 @@ class ChannelWhitelistManager(Vertical):
class ISponsorBlockTVSetupMainScreen(Screen): class ISponsorBlockTVSetupMainScreen(Screen):
"""Making this a separate screen to avoid a bug: https://github.com/Textualize/textual/issues/3221""" """Making this a separate screen to avoid a bug: https://github.com/Textualize/textual/issues/3221"""
TITLE = "iSponsorBlockTV" TITLE = "iSponsorBlockTV"
SUB_TITLE = "Setup Wizard" SUB_TITLE = "Setup Wizard"
BINDINGS = [ BINDINGS = [("q,ctrl+c", "exit_modal", "Exit"), ("s", "save", "Save")]
("q,ctrl+c", "exit_modal", "Exit"),
("s", "save", "Save")
]
AUTO_FOCUS = None AUTO_FOCUS = None
def __init__(self, config, **kwargs) -> None: def __init__(self, config, **kwargs) -> None:
@@ -655,12 +834,24 @@ class ISponsorBlockTVSetupMainScreen(Screen):
yield Header() yield Header()
yield Footer() yield Footer()
with ScrollableContainer(id="setup-wizard"): with ScrollableContainer(id="setup-wizard"):
yield DevicesManager(config=self.config, id="devices-manager", classes="container") yield DevicesManager(
yield SkipCategoriesManager(config=self.config, id="skip-categories-manager", classes="container") config=self.config, id="devices-manager", classes="container"
yield SkipCountTrackingManager(config=self.config, id="count-segments-manager", classes="container") )
yield AdSkipMuteManager(config=self.config, id="ad-skip-mute-manager", classes="container") yield SkipCategoriesManager(
yield ChannelWhitelistManager(config=self.config, id="channel-whitelist-manager", classes="container") config=self.config, id="skip-categories-manager", classes="container"
yield ApiKeyManager(config=self.config, id="api-key-manager", classes="container") )
yield SkipCountTrackingManager(
config=self.config, id="count-segments-manager", classes="container"
)
yield AdSkipMuteManager(
config=self.config, id="ad-skip-mute-manager", classes="container"
)
yield ChannelWhitelistManager(
config=self.config, id="channel-whitelist-manager", classes="container"
)
yield ApiKeyManager(
config=self.config, id="api-key-manager", classes="container"
)
def on_mount(self) -> None: def on_mount(self) -> None:
if self.check_for_old_config_entries(): if self.check_for_old_config_entries():
@@ -686,7 +877,9 @@ class ISponsorBlockTVSetupMainScreen(Screen):
def changed_api_key(self, event: Input.Changed): def changed_api_key(self, event: Input.Changed):
try: # ChannelWhitelist might not be mounted try: # ChannelWhitelist might not be mounted
# Show if no api key is set and at least one channel is in the whitelist # Show if no api key is set and at least one channel is in the whitelist
self.app.query_one("#warning-no-key").display = (not event.input.value) and self.config.channel_whitelist self.app.query_one("#warning-no-key").display = (
not event.input.value
) and self.config.channel_whitelist
except: except:
pass pass
@@ -694,10 +887,7 @@ class ISponsorBlockTVSetupMainScreen(Screen):
class ISponsorBlockTVSetup(App): class ISponsorBlockTVSetup(App):
CSS_PATH = "setup-wizard-style.tcss" # tcss is the recommended extension for textual css files CSS_PATH = "setup-wizard-style.tcss" # tcss is the recommended extension for textual css files
# Bindings for the whole app here, so they are available in all screens # Bindings for the whole app here, so they are available in all screens
BINDINGS = [ BINDINGS = [("q,ctrl+c", "exit_modal", "Exit"), ("s", "save", "Save")]
("q,ctrl+c", "exit_modal", "Exit"),
("s", "save", "Save")
]
def __init__(self, config, **kwargs) -> None: def __init__(self, config, **kwargs) -> None:
super().__init__(**kwargs) super().__init__(**kwargs)

36
src/iSponsorBlockTV/ytlounge.py
View File

@@ -1,5 +1,6 @@
import asyncio import asyncio
import json import json
import pyytlounge import pyytlounge
from .constants import youtube_client_blacklist from .constants import youtube_client_blacklist
@@ -25,7 +26,9 @@ class YtLoungeApi(pyytlounge.YtLoungeApi):
# Ensures that we still are subscribed to the lounge # Ensures that we still are subscribed to the lounge
async def _watchdog(self): async def _watchdog(self):
await asyncio.sleep(35) # YouTube sends at least a message every 30 seconds (no-op or any other) await asyncio.sleep(
35
) # YouTube sends at least a message every 30 seconds (no-op or any other)
try: try:
self.subscribe_task.cancel() self.subscribe_task.cancel()
except Exception: except Exception:
@@ -68,14 +71,18 @@ class YtLoungeApi(pyytlounge.YtLoungeApi):
create_task(self.mute(False, override=True)) create_task(self.mute(False, override=True))
elif event_type == "onAdStateChange": elif event_type == "onAdStateChange":
data = args[0] data = args[0]
if data["adState"] == '0': # Ad is not playing if data["adState"] == "0": # Ad is not playing
# print("Ad has ended, unmuting") # print("Ad has ended, unmuting")
create_task(self.mute(False, override=True)) create_task(self.mute(False, override=True))
elif self.skip_ads and data["isSkipEnabled"] == "true": # YouTube uses strings for booleans elif (
self.skip_ads and data["isSkipEnabled"] == "true"
): # YouTube uses strings for booleans
self._logger.info("Ad can be skipped, skipping") self._logger.info("Ad can be skipped, skipping")
create_task(self.skip_ad()) create_task(self.skip_ad())
create_task(self.mute(False, override=True)) create_task(self.mute(False, override=True))
elif self.mute_ads: # Seen multiple other adStates, assuming they are all ads elif (
self.mute_ads
): # Seen multiple other adStates, assuming they are all ads
self._logger.info("Ad has started, muting") self._logger.info("Ad has started, muting")
create_task(self.mute(True, override=True)) create_task(self.mute(True, override=True))
# Manages volume, useful since YouTube wants to know the volume when unmuting (even if they already have it) # Manages volume, useful since YouTube wants to know the volume when unmuting (even if they already have it)
@@ -84,7 +91,9 @@ class YtLoungeApi(pyytlounge.YtLoungeApi):
pass pass
# Gets segments for the next video before it starts playing # Gets segments for the next video before it starts playing
elif event_type == "autoplayUpNext": elif event_type == "autoplayUpNext":
if len(args) > 0 and (vid_id := args[0]["videoId"]): # if video id is not empty if len(args) > 0 and (
vid_id := args[0]["videoId"]
): # if video id is not empty
print(f"Getting segments for next video: {vid_id}") print(f"Getting segments for next video: {vid_id}")
create_task(self.api_helper.get_segments(vid_id)) create_task(self.api_helper.get_segments(vid_id))
@@ -95,11 +104,15 @@ class YtLoungeApi(pyytlounge.YtLoungeApi):
if vid_id := data["contentVideoId"]: if vid_id := data["contentVideoId"]:
self._logger.info(f"Getting segments for next video: {vid_id}") self._logger.info(f"Getting segments for next video: {vid_id}")
create_task(self.api_helper.get_segments(vid_id)) create_task(self.api_helper.get_segments(vid_id))
elif self.skip_ads and data["isSkipEnabled"] == "true": # YouTube uses strings for booleans elif (
self.skip_ads and data["isSkipEnabled"] == "true"
): # YouTube uses strings for booleans
self._logger.info("Ad can be skipped, skipping") self._logger.info("Ad can be skipped, skipping")
create_task(self.skip_ad()) create_task(self.skip_ad())
create_task(self.mute(False, override=True)) create_task(self.mute(False, override=True))
elif self.mute_ads: # Seen multiple other adStates, assuming they are all ads elif (
self.mute_ads
): # Seen multiple other adStates, assuming they are all ads
self._logger.info("Ad has started, muting") self._logger.info("Ad has started, muting")
create_task(self.mute(True, override=True)) create_task(self.mute(True, override=True))
@@ -142,10 +155,15 @@ class YtLoungeApi(pyytlounge.YtLoungeApi):
if override or not (self.volume_state.get("muted", "false") == mute_str): if override or not (self.volume_state.get("muted", "false") == mute_str):
self.volume_state["muted"] = mute_str self.volume_state["muted"] = mute_str
# YouTube wants the volume when unmuting, so we send it # YouTube wants the volume when unmuting, so we send it
await super()._command("setVolume", {"volume": self.volume_state.get("volume", 100), "muted": mute_str}) await super()._command(
"setVolume",
{"volume": self.volume_state.get("volume", 100), "muted": mute_str},
)
async def set_auto_play_mode(self, enabled: bool): async def set_auto_play_mode(self, enabled: bool):
await super()._command("setAutoplayMode", {"autoplayMode": "ENABLED" if enabled else "DISABLED"}) await super()._command(
"setAutoplayMode", {"autoplayMode": "ENABLED" if enabled else "DISABLED"}
)
async def play_video(self, video_id: str) -> bool: async def play_video(self, video_id: str) -> bool:
return await self._command("setPlaylist", {"videoId": video_id}) return await self._command("setPlaylist", {"videoId": video_id})