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

Merge pull request #290 from dmunozv04/line-length-100

Browse Source 
Change max line length to 100
This commit is contained in:
David
2025-04-12 19:51:30 +02:00
committed by GitHub
parent 1e10ab4e29 ffdeb4579e
commit e9925b02c3
10 changed files with 60 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.deepsource.toml
View File

@@ -6,3 +6,4 @@ enabled = true
[analyzers.meta]
runtime_version = "3.x.x"
max_line_length = 100

4
pyproject.toml
View File

@@ -29,5 +29,5 @@ files = ["requirements.txt"]
requires = ["hatchling", "hatch-requirements-txt"]
build-backend = "hatchling.build"
[tool.black]
line-length = 88
[tool.ruff]
line-length = 100

16
src/iSponsorBlockTV/api_helpers.py
View File

@@ -101,20 +101,14 @@ class ApiHelper:
if channel_data["items"][0]["statistics"]["hiddenSubscriberCount"]:
sub_count = "Hidden"
else:
sub_count = int(
channel_data["items"][0]["statistics"]["subscriberCount"]
)
sub_count = int(channel_data["items"][0]["statistics"]["subscriberCount"])
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
@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):
if await self.is_whitelisted(vid_id):
return (
@@ -132,9 +126,7 @@ class ApiHelper:
}
headers = {"Accept": "application/json"}
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()
if response.status != 200:
response_text = await response.text()

4
src/iSponsorBlockTV/conditional_ttl_cache.py
View File

@@ -33,9 +33,7 @@ class AsyncConditionalTTL:
def __init__(self, time_to_live, maxsize):
super().__init__(maxsize=maxsize)
self.time_to_live = (
datetime.timedelta(seconds=time_to_live) if time_to_live else None
)
self.time_to_live = datetime.timedelta(seconds=time_to_live) if time_to_live else None
self.maxsize = maxsize

23
src/iSponsorBlockTV/config_setup.py
View File

@@ -6,15 +6,12 @@ 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) "
"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) "
)
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 = (
@@ -22,9 +19,7 @@ ENTER_SKIP_CATEGORIES_PROMPT = (
" 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) "
)
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: "
@@ -121,15 +116,11 @@ def main(config, debug: bool) -> None:
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
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
skip_categories = [x for x in skip_categories if x != ""] # Remove empty strings
config.skip_categories = skip_categories
channel_whitelist = config.channel_whitelist
@@ -148,9 +139,7 @@ def main(config, debug: bool) -> None:
if channel == "/exit":
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)
results = task.result()
if len(results) == 0:

8
src/iSponsorBlockTV/dial_client.py
View File

@@ -83,9 +83,7 @@ class Handler(ssdp.aio.SSDP):
self.devices.append(headers["location"])
def request_received(self, request: ssdp.messages.SSDPRequest, addr):
raise NotImplementedError(
"Request received is not implemented, this is a client"
)
raise NotImplementedError("Request received is not implemented, this is a client")
async def find_youtube_app(web_session, url_location):
@@ -121,9 +119,7 @@ async def discover(web_session):
family, _ = network.get_best_family(bind, network.PORT)
loop = asyncio.get_event_loop()
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, _ = await connect
target = network.MULTICAST_ADDRESS_IPV4, network.PORT

29
src/iSponsorBlockTV/helpers.py
View File

@@ -65,9 +65,7 @@ class Config:
sys.exit()
self.devices = [Device(i) for i in self.devices]
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:
self.skip_categories = ["sponsor"]
print("No categories found, using default: sponsor")
@@ -93,14 +91,8 @@ class Config:
f"{github_wiki_base_url}/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:"
),
("This image has recently been updated to v2, and requires changes."),
("Please read this for more information on how to upgrade to V2:"),
f"{github_wiki_base_url}/Migrate-from-V1-to-V2",
)
print("Exiting in 10 seconds...")
@@ -134,15 +126,12 @@ class Config:
@click.option(
"--data",
"-d",
default=lambda: os.getenv("iSPBTV_data_dir")
or user_data_dir("iSponsorBlockTV", "dmunozv04"),
default=lambda: os.getenv("iSPBTV_data_dir") or user_data_dir("iSponsorBlockTV", "dmunozv04"),
help="data directory",
)
@click.option("--debug", is_flag=True, help="debug mode")
# legacy commands as arguments
@click.option(
"--setup", is_flag=True, help="Setup the program graphically", hidden=True
)
@click.option("--setup", is_flag=True, help="Setup the program graphically", hidden=True)
@click.option(
"--setup-cli",
is_flag=True,
@@ -159,9 +148,7 @@ def cli(ctx, data, debug, setup, setup_cli):
logger = logging.getLogger()
ctx.obj["logger"] = logger
sh = logging.StreamHandler()
sh.setFormatter(
logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
)
sh.setFormatter(logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s"))
logger.addHandler(sh)
if debug:
@@ -211,9 +198,7 @@ pyapp_group.add_command(
click.RichCommand("update", help="Update the package to the latest version")
)
pyapp_group.add_command(
click.Command(
"remove", help="Remove the package, wiping the installation but not the data"
)
click.Command("remove", help="Remove the package, wiping the installation but not the data")
)
pyapp_group.add_command(
click.RichCommand(

12
src/iSponsorBlockTV/main.py
View File

@@ -87,9 +87,7 @@ class DeviceListener:
if state.videoId:
segments = await self.api_helper.get_segments(state.videoId)
if state.state.value == 1: # Playing
self.logger.info(
"Playing video %s with %d segments", state.videoId, len(segments)
)
self.logger.info("Playing video %s with %d segments", state.videoId, len(segments))
if segments: # If there are segments
await self.time_to_segment(segments, state.currentTime, time_start)
@@ -107,9 +105,7 @@ class DeviceListener:
if is_within_start_range or is_beyond_current_position:
next_segment = segment
start_next_segment = (
position if is_within_start_range else segment_start
)
start_next_segment = position if is_within_start_range else segment_start
break
if start_next_segment:
time_to_next = (
@@ -148,9 +144,7 @@ class DeviceListener:
async def finish(devices, web_session, tcp_connector):
await asyncio.gather(
*(device.cancel() for device in devices), return_exceptions=True
)
await asyncio.gather(*(device.cancel() for device in devices), return_exceptions=True)
await web_session.close()
await tcp_connector.close()

97
src/iSponsorBlockTV/setup_wizard.py
View File

@@ -122,9 +122,7 @@ class Channel(Element):
if "name" in self.element_data:
self.element_name = self.element_data["name"]
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):
@@ -204,9 +202,7 @@ class ExitScreen(ModalWithClickExit):
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"),
id="dialog-exit",
)
@@ -255,19 +251,13 @@ class AddDevice(ModalWithClickExit):
id="add-device-dial-button",
classes="button-switcher",
)
with ContentSwitcher(
id="add-device-switcher", initial="add-device-pin-container"
):
with ContentSwitcher(id="add-device-switcher", initial="add-device-pin-container"):
with Container(id="add-device-pin-container"):
yield Input(
placeholder=(
"Pairing Code (found in Settings - Link with TV code)"
),
placeholder=("Pairing Code (found in Settings - Link with TV code)"),
id="pairing-code-input",
validators=[
Function(
_validate_pairing_code, "Invalid pairing code format"
)
Function(_validate_pairing_code, "Invalid pairing code format")
],
)
yield Input(
@@ -332,9 +322,7 @@ class AddDevice(ModalWithClickExit):
@on(Input.Changed, "#pairing-code-input")
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(Button.Pressed, "#add-device-pin-add-button")
@@ -382,9 +370,7 @@ class AddDevice(ModalWithClickExit):
@on(SelectionList.SelectedChanged, "#dial-devices-list")
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):
@@ -422,9 +408,7 @@ class AddChannel(ModalWithClickExit):
classes="button-switcher",
)
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"):
if self.config.apikey:
with Grid(id="add-channel-search-inputs"):
@@ -432,9 +416,7 @@ class AddChannel(ModalWithClickExit):
placeholder="Enter channel name",
id="channel-name-input-search",
)
yield Button(
"Search", id="search-channel-button", variant="success"
)
yield Button("Search", id="search-channel-button", variant="success")
yield RadioSet(
RadioButton(label="Search to see results", disabled=True),
id="channel-search-results",
@@ -457,15 +439,12 @@ class AddChannel(ModalWithClickExit):
)
with Vertical(id="add-channel-id-container"):
yield Input(
placeholder=(
"Enter channel ID (example: UCuAXFkgsw1L7xaCfnd5JJOw)"
),
placeholder=("Enter channel ID (example: UCuAXFkgsw1L7xaCfnd5JJOw)"),
id="channel-id-input",
)
yield Input(
placeholder=(
"Enter channel name (only used to display in the config"
" file)"
"Enter channel name (only used to display in the config file)"
),
id="channel-name-input-id",
)
@@ -491,9 +470,7 @@ class AddChannel(ModalWithClickExit):
async def handle_search_channel(self) -> None:
channel_name = self.query_one("#channel-name-input-search").value
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
self.query_one("#search-channel-button").disabled = True
self.query_one("#add-channel-info").update("Searching...")
@@ -502,9 +479,7 @@ class AddChannel(ModalWithClickExit):
try:
channels_list = await self.api_helper.search_channels(channel_name)
except BaseException:
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
return
for i in channels_list:
@@ -517,9 +492,7 @@ class AddChannel(ModalWithClickExit):
def handle_add_channel_search(self) -> None:
channel = self.query_one("#channel-search-results").pressed_button.channel_data
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
self.query_one("#add-channel-info").update("Adding...")
self.dismiss(channel)
@@ -531,9 +504,7 @@ class AddChannel(ModalWithClickExit):
channel_id = self.query_one("#channel-id-input").value
channel_name = self.query_one("#channel-name-input-id").value
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
if not channel_name:
channel_name = channel_id
@@ -624,9 +595,7 @@ class DevicesManager(Vertical):
def compose(self) -> ComposeResult:
yield Label("Devices", classes="title")
with Horizontal(id="add-device-button-container"):
yield Button(
"Add Device", id="add-device", classes="button-100 button-small"
)
yield Button("Add Device", id="add-device", classes="button-100 button-small")
for device in self.devices:
yield Device(device, tooltip="Click to edit")
@@ -821,16 +790,14 @@ class ChannelWhitelistManager(Vertical):
id="warning-no-key",
)
with Horizontal(id="add-channel-button-container"):
yield Button(
"Add Channel", id="add-channel", classes="button-100 button-small"
)
yield Button("Add Channel", id="add-channel", classes="button-100 button-small")
for channel in self.config.channel_whitelist:
yield Channel(channel)
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:
if channel:
@@ -842,18 +809,18 @@ class ChannelWhitelistManager(Vertical):
channel_widget = Channel(channel_dict)
self.mount(channel_widget)
channel_widget.focus(scroll_visible=True)
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
)
@on(Button.Pressed, "#element-remove")
def remove_channel(self, event: Button.Pressed):
channel_to_remove: Element = event.button.parent
self.config.channel_whitelist.remove(channel_to_remove.element_data)
channel_to_remove.remove()
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
)
@on(Button.Pressed, "#add-channel")
def add_channel(self, event: Button.Pressed):
@@ -902,9 +869,7 @@ class ISponsorBlockTVSetupMainScreen(Screen):
yield Header()
yield Footer()
with ScrollableContainer(id="setup-wizard"):
yield DevicesManager(
config=self.config, id="devices-manager", classes="container"
)
yield DevicesManager(config=self.config, id="devices-manager", classes="container")
yield SkipCategoriesManager(
config=self.config, id="skip-categories-manager", classes="container"
)
@@ -917,12 +882,8 @@ class ISponsorBlockTVSetupMainScreen(Screen):
yield ChannelWhitelistManager(
config=self.config, id="channel-whitelist-manager", classes="container"
)
yield ApiKeyManager(
config=self.config, id="api-key-manager", classes="container"
)
yield AutoPlayManager(
config=self.config, id="autoplay-manager", classes="container"
)
yield ApiKeyManager(config=self.config, id="api-key-manager", classes="container")
yield AutoPlayManager(config=self.config, id="autoplay-manager", classes="container")
def on_mount(self) -> None:
if self.check_for_old_config_entries():

20
src/iSponsorBlockTV/ytlounge.py
View File

@@ -18,9 +18,7 @@ class YtLoungeApi(pyytlounge.YtLoungeApi):
api_helper=None,
logger=None,
):
super().__init__(
config.join_name if config else "iSponsorBlockTV", logger=logger
)
super().__init__(config.join_name if config else "iSponsorBlockTV", logger=logger)
self.auth.screen_id = screen_id
self.auth.lounge_id_token = None
self.api_helper = api_helper
@@ -96,9 +94,7 @@ class YtLoungeApi(pyytlounge.YtLoungeApi):
self.logger.info("Ad can be skipped, skipping")
create_task(self.skip_ad())
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")
create_task(self.mute(True, override=True))
# Manages volume, useful since YouTube wants to know the volume
@@ -107,9 +103,7 @@ class YtLoungeApi(pyytlounge.YtLoungeApi):
self.volume_state = args[0]
# Gets segments for the next video before it starts playing
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
self.logger.info(f"Getting segments for next video: {vid_id}")
create_task(self.api_helper.get_segments(vid_id))
@@ -126,9 +120,7 @@ class YtLoungeApi(pyytlounge.YtLoungeApi):
self.logger.info("Ad can be skipped, skipping")
create_task(self.skip_ad())
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")
create_task(self.mute(True, override=True))
@@ -151,9 +143,7 @@ class YtLoungeApi(pyytlounge.YtLoungeApi):
elif event_type == "loungeScreenDisconnected":
if args: # Sometimes it's empty
data = args[0]
if (
data["reason"] == "disconnectedByUserScreenInitiated"
): # Short playing?
if data["reason"] == "disconnectedByUserScreenInitiated": # Short playing?
self.shorts_disconnected = True
elif event_type == "onAutoplayModeChanged":
create_task(self.set_auto_play_mode(self.auto_play))