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

Added channels whitelist

Browse Source
This commit is contained in:
oxixes
2023-04-22 13:20:10 +02:00
parent e0c0548496
commit 73d1a024f3
5 changed files with 103 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
config.json.template
View File

@@ -3,5 +3,6 @@
{"identifier": "", "airplay_credentials": ""}
],
"apikey":"",
"skip_categories": ["sponsor"]
"skip_categories": ["sponsor"],
"channel_whitelist": []
}

34
iSponsorBlockTV/api_helpers.py
View File

@@ -21,13 +21,45 @@ async def get_vid_id(title, artist, api_key, web_session):
url = constants.Youtube_api + "search"
async with web_session.get(url, params=params) as resp:
data = await resp.json()
if "error" in data:
return
for i in data["items"]:
if (i["id"]["kind"] != "youtube#video"):
continue
title_api = html.unescape(i["snippet"]["title"])
artist_api = html.unescape(i["snippet"]["channelTitle"])
if title_api == title and artist_api == artist:
return i["id"]["videoId"]
return (i["id"]["videoId"], i["snippet"]["channelId"])
return
@AsyncLRU(maxsize=10)
async def search_channels(channel, api_key, web_session):
channels = []
params = {"q": channel, "key": api_key, "part": "snippet", "type": "channel", "maxResults": "5"}
url = constants.Youtube_api + "search"
async with web_session.get(url, params=params) as resp:
data = await resp.json()
if "error" in data:
return channels
for i in data["items"]:
# Get channel subcription number
params = {"id": i["snippet"]["channelId"], "key": api_key, "part": "statistics"}
url = constants.Youtube_api + "channels"
async with web_session.get(url, params=params) as resp:
channelData = await resp.json()
if channelData["items"][0]["statistics"]["hiddenSubscriberCount"]:
subCount = "Hidden"
else:
subCount = channelData["items"][0]["statistics"]["subscriberCount"]
channel.append((i["snippet"]["channelId"], i["snippet"]["channelTitle"], subCount))
return channels
@listToTuple
@AsyncTTL(time_to_live=300, maxsize=5)

63
iSponsorBlockTV/config_setup.py
View File

@@ -3,7 +3,9 @@ import json
import asyncio
from pyatv.const import DeviceModel
import sys
import aiohttp
import asyncio
from . import api_helpers
def save_config(config, config_file):
with open(config_file, "w") as f:
@@ -74,9 +76,9 @@ def main(config, config_file, debug):
try:
for i in atvs:
config["atvs"].append(i)
print("done adding")
print("Done adding")
except:
print("rewriting atvs (don't worry if none were saved before)")
print("Rewriting atvs (don't worry if none were saved before)")
config["atvs"] = atvs
try:
@@ -84,12 +86,12 @@ def main(config, config_file, debug):
except:
apikey = ""
if apikey != "":
if input("Apikey already specified. Change it? (y/n) ") == "y":
if input("API key already specified. Change it? (y/n) ") == "y":
apikey = input("Enter your API key: ")
config["apikey"] = apikey
else:
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"
)
apikey = input("Enter your API key: ")
config["apikey"] = apikey
@@ -108,10 +110,55 @@ def main(config, config_file, debug):
skip_categories = [x for x in skip_categories if x != ''] # Remove empty strings
else:
categories = input(
"Enter skip categories (space sepparated) Options: [sponsor, selfpromo, exclusive_access, interaction, poi_highlight, intro, outro, preview, filler, music_offtopic:\n"
"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.split(" ")
skip_categories = categories.replace(",", " ").split(" ")
skip_categories = [x for x in skip_categories if x != ''] # Remove empty strings
config["skip_categories"] = skip_categories
print("config finished")
try:
channel_whitelist = config["channel_whitelist"]
except:
channel_whitelist = []
if channel_whitelist != []:
if input("Do you want to whitelist any channels from being ad-blocked? (y/n) ") == "y":
web_session = aiohttp.ClientSession()
while True:
channel_info = {}
channel = input("Enter a channel name or \"/exit to exit\": ")
if channel == "/exit":
break
results = asyncio.run(api_helpers.search_channels(channel, apikey, web_session))
if len(results) == 0:
print("No channels found")
continue
for i in range(len(results)):
print(f"{i}: {results[i][1]} - Subs: {results[i][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
elif choice == "6":
continue
channel_info["id"] = results[int(choice)][0]
channel_info["name"] = results[int(choice)][1]
channel_whitelist.append(channel_info)
config["channel_whitelist"] = channel_whitelist
print("Config finished")
save_config(config, config_file)

4
iSponsorBlockTV/helpers.py
View File

@@ -46,10 +46,10 @@ def app_start():
else:
try: # Check if config file has the correct structure
config["atvs"], config["apikey"], config["skip_categories"]
config["atvs"], config["apikey"], config["skip_categories"], config["channel_whitelist"]
except: # If not, ask to setup the program
print("invalid config file, please run with --setup")
sys.exit()
main.main(
config["atvs"], config["apikey"], config["skip_categories"], args.debug
config["atvs"], config["apikey"], config["skip_categories"], config["channel_whitelist"], args.debug
)

19
iSponsorBlockTV/main.py
View File

@@ -13,12 +13,14 @@ class MyPushListener(pyatv.interface.PushListener):
web_session = None
categories = ["sponsor"]
whitelist = []
def __init__(self, apikey, atv, categories, web_session):
def __init__(self, apikey, atv, categories, whitelist, web_session):
self.apikey = apikey
self.rc = atv.remote_control
self.web_session = web_session
self.categories = categories
self.whitelist = whitelist
self.atv = atv
def playstatus_update(self, updater, playstatus):
@@ -37,6 +39,7 @@ class MyPushListener(pyatv.interface.PushListener):
self.categories,
self.atv,
time_start,
self.whitelist
)
)
@@ -46,7 +49,7 @@ class MyPushListener(pyatv.interface.PushListener):
async def process_playstatus(
playstatus, apikey, rc, web_session, categories, atv, time_start
playstatus, apikey, rc, web_session, categories, atv, time_start, whitelist
):
logging.debug("App playing is:" + str(atv.metadata.app.identifier))
if (
@@ -57,7 +60,7 @@ async def process_playstatus(
playstatus.title, playstatus.artist, apikey, web_session
)
if vid_id:
print(vid_id)
print(f"ID: {vid_id[0]}, Channel ID: {vid_id[1]}")
segments = await api_helpers.get_segments(vid_id, web_session, categories)
print(segments)
await time_to_segment(
@@ -104,12 +107,12 @@ async def connect_atv(loop, identifier, airplay_credentials):
return await pyatv.connect(config, loop)
async def loop_atv(event_loop, atv_config, apikey, categories, web_session):
async def loop_atv(event_loop, atv_config, apikey, categories, whitelist, web_session):
identifier = atv_config["identifier"]
airplay_credentials = atv_config["airplay_credentials"]
atv = await connect_atv(event_loop, identifier, airplay_credentials)
if atv:
listener = MyPushListener(apikey, atv, categories, web_session)
listener = MyPushListener(apikey, atv, categories, whitelist, web_session)
atv.push_updater.listener = listener
atv.push_updater.start()
@@ -123,19 +126,19 @@ async def loop_atv(event_loop, atv_config, apikey, categories, web_session):
# reconnect to apple tv
atv = await connect_atv(event_loop, identifier, airplay_credentials)
if atv:
listener = MyPushListener(apikey, atv, categories, web_session)
listener = MyPushListener(apikey, atv, categories, whitelist, web_session)
atv.push_updater.listener = listener
atv.push_updater.start()
print("Push updater started")
def main(atv_configs, apikey, categories, debug):
def main(atv_configs, apikey, categories, whitelist, debug):
loop = asyncio.get_event_loop_policy().get_event_loop()
if debug:
loop.set_debug(True)
asyncio.set_event_loop(loop)
web_session = aiohttp.ClientSession()
for i in atv_configs:
loop.create_task(loop_atv(loop, i, apikey, categories, web_session))
loop.create_task(loop_atv(loop, i, apikey, categories, whitelist, web_session))
loop.run_forever()