canonical_link

This commit is contained in:
BennyThink 2022-02-04 08:30:27 +08:00
parent 42354b6eed
commit 80afed3a0b
No known key found for this signature in database
GPG key ID: 6CD0DBDA5235D481
2 changed files with 17 additions and 7 deletions

View file

@ -106,12 +106,20 @@ class VIP(Redis, MySQL):
return affected_rows
@staticmethod
def get_channel_info(url: "str"):
def extract_canonical_link(url):
# canonic link works for many websites. It will strip out unnecessary stuff
try:
html_doc = requests.get(url).text
soup = BeautifulSoup(html_doc, "html.parser")
element = soup.find("link", rel="canonical")
return element['href']
except Exception:
return url
def get_channel_info(self, url: "str"):
api_key = os.getenv("GOOGLE_API_KEY")
html_doc = requests.get(url).text
soup = BeautifulSoup(html_doc, "html.parser")
element = soup.find("link", rel="canonical")
channel_id = element['href'].split("https://www.youtube.com/channel/")[1]
canonical_link = self.extract_canonical_link(url)
channel_id = canonical_link.split("https://www.youtube.com/channel/")[1]
channel_api = f"https://www.googleapis.com/youtube/v3/channels?part=snippet,contentDetails&" \
f"id={channel_id}&key={api_key}"
data = requests.get(channel_api).json()
@ -282,4 +290,5 @@ def subscribe_query():
if __name__ == '__main__':
subscribe_query()
a=VIP.extract_canonical_link("https://youtu.be/FUACKXI-1BA?t=71")
print(a)

View file

@ -125,7 +125,7 @@ def unsubscribe_handler(client: "Client", message: "types.Message"):
@app.on_message(filters.command(["hot_patch"]))
def help_handler(client: "Client", message: "types.Message"):
def patch_handler(client: "Client", message: "types.Message"):
username = message.from_user.username
chat_id = message.chat.id
if username == OWNER:
@ -285,6 +285,7 @@ if __name__ == '__main__':
scheduler.add_job(Redis().reset_today, 'cron', hour=0, minute=0)
scheduler.add_job(auto_restart, 'interval', seconds=5)
scheduler.add_job(InfluxDB().collect_data, 'interval', seconds=60)
# default quota allocation of 10,000 units per day,
scheduler.add_job(periodic_sub_check, 'interval', seconds=60)
scheduler.start()
banner = f"""