Add first working version
This commit is contained in:
commit
352c0d17a1
1
subs.json
Normal file
1
subs.json
Normal file
|
@ -0,0 +1 @@
|
||||||
|
{"app_version":"0.16.1","app_version_int":730,"subscriptions":[{"service_id":0,"url":"https://www.youtube.com/channel/UCl2mFZoRqjw_ELax4Yisf6w","name":"Louis Rossmann"},{"service_id":0,"url":"https://www.youtube.com/channel/UCp68_FLety0O-n9QU6phsgw","name":"colinfurze"},{"service_id":0,"url":"https://www.youtube.com/channel/UCcvs3EMePHVV4MexFPaYekg","name":"BasSistaz"},{"service_id":0,"url":"https://www.youtube.com/channel/UC3XTzVzaHQEd30rQbuvCtTQ","name":"LastWeekTonight"},{"service_id":0,"url":"https://www.youtube.com/channel/UCJ0-OtVpF0wOKEqT2Z1HEtA","name":"ElectroBOOM"},{"service_id":0,"url":"https://www.youtube.com/channel/UChkVOG0PqkXIHHmkBGG15Ig","name":"WALULIS"},{"service_id":0,"url":"https://www.youtube.com/channel/UCIyZiiHXIH7KkqfaDvBmG-Q","name":"EZScape"},{"service_id":0,"url":"https://www.youtube.com/channel/UCtUbO6rBht0daVIOGML3c8w","name":"Summoning Salt"},{"service_id":0,"url":"https://www.youtube.com/channel/UCyQKoLtVor_r1OCOBgeOofQ","name":"Sound Territory"},{"service_id":0,"url":"https://www.youtube.com/channel/UCyHqXCNrWYWq2DksYJzwciQ","name":"bisnap"},{"service_id":0,"url":"https://www.youtube.com/user/styropyro","name":"styropyro"},{"service_id":0,"url":"https://www.youtube.com/channel/UCAepXw94EhaO0CZV9f5D3fQ","name":"The Psychedelic Muse"},{"service_id":0,"url":"https://www.youtube.com/channel/UCXqUK3LHJkiNzIeOvjGHwQQ","name":"Pulsewidth"},{"service_id":0,"url":"https://www.youtube.com/user/CaptainDisillusion","name":"CaptainDisillusion"},{"service_id":0,"url":"https://www.youtube.com/user/EventElevator","name":"EventElevator"},{"service_id":0,"url":"https://www.youtube.com/channel/UCyHDQ5C6z1NDmJ4g6SerW8g","name":"maiLab"},{"service_id":0,"url":"https://www.youtube.com/channel/UCtHaxi4GTYDpJgMSGy7AeSw","name":"Michael Reeves"},{"service_id":0,"url":"https://www.youtube.com/channel/UCfMJ2MchTSW2kWaT0kK94Yw","name":"William Osman"},{"service_id":0,"url":"https://www.youtube.com/user/CGPGrey","name":"CGP Grey"},{"service_id":0,"url":"https://www.youtube.com/user/Nerdwriter1","name":"Nerdwriter1"},{"service_id":0,"url":"https://www.youtube.com/channel/UCHC-rrov7TQcyWcelVpt_3A","name":"the man from del monte"}]}
|
111
yt_sub.py
Normal file
111
yt_sub.py
Normal file
|
@ -0,0 +1,111 @@
|
||||||
|
import requests
|
||||||
|
import json
|
||||||
|
import pdb
|
||||||
|
from urllib.parse import urlparse
|
||||||
|
from getpass import getpass
|
||||||
|
from bs4 import BeautifulSoup
|
||||||
|
|
||||||
|
BASE_FEED_URL = "https://www.youtube.com/feeds/videos.xml?channel_id={channel_id}"
|
||||||
|
TTRSS_API_URL = "https://rss.geekify.de/api/"
|
||||||
|
|
||||||
|
def get_channel_feed_url(sub_url):
|
||||||
|
channel_id = ""
|
||||||
|
feed_url = ""
|
||||||
|
url = urlparse(sub_url)
|
||||||
|
if "/channel" in url.path:
|
||||||
|
channel_id = url.path.split("/")[-1]
|
||||||
|
feed_url = BASE_FEED_URL.format(channel_id=channel_id)
|
||||||
|
elif "/user" in url.path:
|
||||||
|
req = requests.get(url_str)
|
||||||
|
soup = BeautifulSoup(req.text, "html.parser")
|
||||||
|
# we're looking for something like this:
|
||||||
|
# <link rel="canonical" href="https://www.youtube.com/channel/UC2C_jShtL725hvbm1arSV9w">
|
||||||
|
channel_link_extracted = soup.find_all("link", rel="canonical")[0].attrs.get("href")
|
||||||
|
url = urlparse(channel_link_extracted)
|
||||||
|
channel_id = url.path.split("/")[-1]
|
||||||
|
feed_url = BASE_FEED_URL.format(channel_id=channel_id)
|
||||||
|
else:
|
||||||
|
debug_msg = "Unknown url ({url}) for user \"{name}\""
|
||||||
|
print(debug_msg.format(url=url_str))
|
||||||
|
feed_url = ""
|
||||||
|
return feed_url
|
||||||
|
|
||||||
|
class TTRSS():
|
||||||
|
username = ""
|
||||||
|
__password = ""
|
||||||
|
api_endpoint = ""
|
||||||
|
sid = ""
|
||||||
|
|
||||||
|
def __init__(self, api_endpoint="", username="", password=""):
|
||||||
|
self.username = username
|
||||||
|
self.api_endpoint = api_endpoint
|
||||||
|
self.api_level = -1
|
||||||
|
self.__password = password
|
||||||
|
self.login()
|
||||||
|
pass
|
||||||
|
|
||||||
|
def login(self):
|
||||||
|
data = self.__build_request(op="login", user=self.username, password=self.__password)
|
||||||
|
resp = self.__send_request(data)
|
||||||
|
self.sid = resp.get("session_id", "")
|
||||||
|
self.api_level = resp.get("api_level", -1)
|
||||||
|
if len(self.sid) > 0:
|
||||||
|
return True
|
||||||
|
else:
|
||||||
|
return False
|
||||||
|
|
||||||
|
def getCategories(self):
|
||||||
|
data = self.__build_request(sid=self.sid, op="getCategories", unread_only="f", enable_nested="t", include_empty="f")
|
||||||
|
resp = self.__send_request(data)
|
||||||
|
return resp
|
||||||
|
|
||||||
|
def subscribeToFeed(self, feed_url, category_id):
|
||||||
|
data = self.__build_request(sid=self.sid, op="subscribeToFeed", feed_url=feed_url, category_id=category_id)
|
||||||
|
resp = self.__send_request(data)
|
||||||
|
return resp
|
||||||
|
|
||||||
|
def getFeeds(self, cat_id):
|
||||||
|
data = self.__build_request(sid=self.sid, op="getFeeds", cat_id=cat_id, unread_only="f")
|
||||||
|
resp = self.__send_request(data)
|
||||||
|
return resp
|
||||||
|
|
||||||
|
def __build_request(self, *args, **kwargs):
|
||||||
|
request = kwargs
|
||||||
|
return request
|
||||||
|
|
||||||
|
def __send_request(self, data):
|
||||||
|
req = requests.post(self.api_endpoint, data=json.dumps(data))
|
||||||
|
resp = req.json()
|
||||||
|
if resp.get("status", -1) == 0:
|
||||||
|
return resp.get("content", {})
|
||||||
|
else:
|
||||||
|
print("REQUEST FAILED!")
|
||||||
|
return {}
|
||||||
|
|
||||||
|
print("TinyTinyRSS credentials:")
|
||||||
|
ttrss = TTRSS(TTRSS_API_URL, input("Username: "), getpass())
|
||||||
|
cats = ttrss.getCategories()
|
||||||
|
yt_cat_id_obj = next(filter(lambda x: "YouTube" in x.get("title", ""), cats))
|
||||||
|
yt_cat_id = yt_cat_id_obj.get("id")
|
||||||
|
feeds = ttrss.getFeeds(yt_cat_id)
|
||||||
|
already_subed = []
|
||||||
|
for feed in feeds:
|
||||||
|
already_subed.append(feed.get("feed_url"))
|
||||||
|
|
||||||
|
with open("subs.json", "r") as f:
|
||||||
|
newpipe_json = json.loads(f.read())
|
||||||
|
|
||||||
|
subs = newpipe_json.get("subscriptions", {})
|
||||||
|
for sub in subs:
|
||||||
|
if sub.get("service_id", -1) == 0: # this indicates a youtube link
|
||||||
|
url_str = sub.get("url", "")
|
||||||
|
feed_url = get_channel_feed_url(url_str)
|
||||||
|
if feed_url in already_subed:
|
||||||
|
print("Already subscribed to: {}".format(sub.get("name")))
|
||||||
|
else:
|
||||||
|
print("Subscribing to: {}…".format(sub.get("name")), end="")
|
||||||
|
resp = ttrss.subscribeToFeed(feed_url, yt_cat_id)
|
||||||
|
if resp.get("status", {}).get("feed_id", -1) > 0:
|
||||||
|
print("SUCCESS!")
|
||||||
|
else:
|
||||||
|
print("FAILED!")
|
Loading…
Reference in a new issue