556 lines
19 KiB
Python
Executable File
556 lines
19 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""ParkerBot"""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import datetime
|
|
import glob
|
|
import html
|
|
import json
|
|
import os
|
|
import pickle
|
|
import re
|
|
import sqlite3
|
|
import time
|
|
|
|
from google.auth.transport.requests import Request
|
|
from google_auth_oauthlib.flow import InstalledAppFlow
|
|
from googleapiclient.discovery import build
|
|
from googleapiclient import errors
|
|
from nio import AsyncClient, RoomMessageText, SyncResponse, UploadResponse
|
|
|
|
DATA_DIR = os.getenv("DATA_DIR", "./")
|
|
DB_PATH = os.path.join(DATA_DIR, "parkerbot.sqlite3")
|
|
PICKLE_PATH = os.path.join(DATA_DIR, "token.pickle")
|
|
TOKEN_PATH = os.path.join(DATA_DIR, "sync_token")
|
|
|
|
MATRIX_SERVER = os.getenv("MATRIX_SERVER")
|
|
MATRIX_ROOM = os.getenv("MATRIX_ROOM")
|
|
MATRIX_USER = os.getenv("MATRIX_USER")
|
|
MATRIX_PASSWORD = os.getenv("MATRIX_PASSWORD")
|
|
|
|
YOUTUBE_CLIENT_SECRETS_FILE = os.getenv("YOUTUBE_CLIENT_SECRETS_FILE")
|
|
YOUTUBE_PLAYLIST_TITLE = os.getenv("YOUTUBE_PLAYLIST_TITLE")
|
|
|
|
AZURACAST_API_KEY = os.getenv("AZURACAST_API_KEY")
|
|
AZURACAST_URL = os.getenv("AZURACAST_URL")
|
|
AZURACAST_STATION_ID = os.getenv("AZURACAST_STATION_ID")
|
|
|
|
|
|
def connect_db():
|
|
"""Connect to DB and return connection and cursor."""
|
|
conn = sqlite3.connect(DB_PATH)
|
|
cursor = conn.cursor()
|
|
return conn, cursor
|
|
|
|
|
|
def parse_arguments():
|
|
"""Parse command line arguments."""
|
|
parser = argparse.ArgumentParser(
|
|
description=(
|
|
"Matrix bot to generate YouTube (music) playlists from links sent "
|
|
"to a channel."
|
|
)
|
|
)
|
|
parser.add_argument(
|
|
"--backwards-sync",
|
|
action="store_true",
|
|
help=(
|
|
"Run backwards sync on start. This most probably will cause you to "
|
|
"exceed your YouTube daily API quota, and other hidden YouTube rate"
|
|
" limits."
|
|
),
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def define_tables(conn, cursor):
|
|
"""Define tables for use with program."""
|
|
with conn:
|
|
cursor.execute(
|
|
"""CREATE TABLE IF NOT EXISTS messages (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
sender TEXT,
|
|
message TEXT,
|
|
timestamp DATETIME,
|
|
UNIQUE (sender, message, timestamp))"""
|
|
)
|
|
cursor.execute(
|
|
"""CREATE TABLE IF NOT EXISTS playlists (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
title TEXT,
|
|
playlist_id TEXT UNIQUE,
|
|
creation_date DATE)"""
|
|
)
|
|
cursor.execute( # TODO: Write migration script to add video_id.
|
|
"""CREATE TABLE IF NOT EXISTS playlist_tracks (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
playlist_id INTEGER,
|
|
message_id INTEGER,
|
|
video_id TEXT,
|
|
FOREIGN KEY (playlist_id) REFERENCES playlists(id),
|
|
FOREIGN KEY (message_id) REFERENCES messages(id),
|
|
UNIQUE (playlist_id, message_id))"""
|
|
)
|
|
|
|
|
|
def get_authenticated_service():
|
|
"""Get an authentivated YouTube service."""
|
|
credentials = None
|
|
# Stores the user's access and refresh tokens.
|
|
if os.path.exists(PICKLE_PATH):
|
|
with open(PICKLE_PATH, "rb") as token:
|
|
credentials = pickle.load(token)
|
|
|
|
# If there are no valid credentials available, let the user log in.
|
|
if not credentials or not credentials.valid:
|
|
if credentials and credentials.expired and credentials.refresh_token:
|
|
credentials.refresh(Request())
|
|
else:
|
|
flow = InstalledAppFlow.from_client_secrets_file(
|
|
YOUTUBE_CLIENT_SECRETS_FILE,
|
|
scopes=["https://www.googleapis.com/auth/youtube.force-ssl"],
|
|
)
|
|
credentials = flow.run_local_server(port=8080)
|
|
# Save the credentials for the next run
|
|
with open(PICKLE_PATH, "wb") as token:
|
|
pickle.dump(credentials, token)
|
|
|
|
return build("youtube", "v3", credentials=credentials)
|
|
|
|
|
|
def monday_date(timestamp):
|
|
"""Return Monday of week for given timestamp. Weeks start on Monday."""
|
|
date = datetime.datetime.fromtimestamp(timestamp / 1000, datetime.UTC)
|
|
return date - datetime.timedelta(days=date.weekday())
|
|
|
|
|
|
def make_playlist(youtube, title):
|
|
"""Make a playlist with given title."""
|
|
response = (
|
|
youtube.playlists()
|
|
.insert(
|
|
part="snippet,status",
|
|
body={
|
|
"snippet": {
|
|
"title": title,
|
|
"description": "Weekly playlist generated by ParkerBot",
|
|
},
|
|
"status": {"privacyStatus": "public"},
|
|
},
|
|
)
|
|
.execute()
|
|
)
|
|
|
|
return response["id"]
|
|
|
|
|
|
def get_or_make_playlist(conn, cursor, youtube, playlist_date):
|
|
"""Get ID of playlist with given named suffix, make if doesn't exist."""
|
|
title = f"{YOUTUBE_PLAYLIST_TITLE} {playlist_date.strftime('%Y-%m-%d')}"
|
|
|
|
cursor.execute("SELECT playlist_id FROM playlists WHERE title = ?", (title,))
|
|
row = cursor.fetchone()
|
|
if row:
|
|
return row[0]
|
|
|
|
playlist_id = make_playlist(youtube, title)
|
|
with conn:
|
|
cursor.execute( # TODO: https://docs.python.org/3/library/sqlite3.html#default-adapters-and-converters-deprecated
|
|
"INSERT INTO playlists (title, playlist_id, creation_date) VALUES (?, ?, ?)",
|
|
(title, playlist_id, playlist_date),
|
|
)
|
|
|
|
return playlist_id
|
|
|
|
|
|
def add_video_to_playlist(youtube, playlist_id, video_id, retry_count=6):
|
|
"""Add video to playlist."""
|
|
for attempt in range(retry_count):
|
|
try:
|
|
youtube.playlistItems().insert(
|
|
part="snippet",
|
|
body={
|
|
"snippet": {
|
|
"playlistId": playlist_id,
|
|
"resourceId": {"kind": "youtube#video", "videoId": video_id},
|
|
}
|
|
},
|
|
).execute()
|
|
break
|
|
except errors.HttpError as error:
|
|
if attempt < retry_count - 1:
|
|
time.sleep(2**attempt)
|
|
continue
|
|
raise error
|
|
|
|
|
|
def get_video_info(youtube, video_id):
|
|
"""Check whether a YouTube video is music and return its title and channel."""
|
|
try:
|
|
video_details = youtube.videos().list(id=video_id, part="snippet").execute()
|
|
|
|
# Check if the video actually exists/is accessible
|
|
if not video_details.get("items"):
|
|
return False, "[Video unavailable or private]", ""
|
|
|
|
snippet = video_details["items"][0]["snippet"]
|
|
|
|
# Check if the video category is Music (10) or Entertainment (24)
|
|
is_music = snippet.get("categoryId") in ("10", "24")
|
|
title = snippet.get("title", "[Unknown Title]")
|
|
channel = snippet.get("channelTitle", "[Unknown Channel]")
|
|
|
|
return is_music, title, channel
|
|
|
|
except errors.HttpError as error:
|
|
print(f"YouTube API error fetching info for {video_id}: {error}")
|
|
return False, "[Error fetching title]", ""
|
|
except Exception as e:
|
|
print(f"Unexpected error fetching info for {video_id}: {e}")
|
|
return False, "[Error fetching title]", ""
|
|
|
|
|
|
async def send_intro_message(client, sender, room_id):
|
|
"""Sends introduction message in reply to sender, in room with room_id."""
|
|
intro_message = (
|
|
f"Hi {sender}, I'm ParkerBot! I generate YouTube playlists from links "
|
|
"sent to this channel. You can find my source code here: "
|
|
"https://git.abdulocra.cy/abdulocracy/parkerbot"
|
|
)
|
|
await client.room_send(
|
|
room_id=room_id,
|
|
message_type="m.room.message",
|
|
content={"msgtype": "m.text", "body": intro_message},
|
|
)
|
|
|
|
# TODO: Figure out how to properly send GIF, this is broken as shit.
|
|
with open("./parker.gif", "rb") as gif_file:
|
|
response = await client.upload(gif_file, content_type="image/gif")
|
|
if isinstance(response, UploadResponse):
|
|
print("Image was uploaded successfully to server. ")
|
|
gif_uri = response.content_uri
|
|
await client.room_send(
|
|
room_id=room_id,
|
|
message_type="m.room.message",
|
|
content={
|
|
"msgtype": "m.image",
|
|
"url": gif_uri,
|
|
"body": "parker.gif",
|
|
"info": {"mimetype": "image/gif"},
|
|
},
|
|
)
|
|
else:
|
|
print(f"Failed to upload image. Failure response: {response}")
|
|
|
|
|
|
async def send_playlist_of_week(client, sender, room_id, playlist_id):
|
|
"""Sends playlist of the week in reply to sender, in room with room_id."""
|
|
playlist_link = f"https://www.youtube.com/playlist?list={playlist_id}"
|
|
reply_msg = f"{sender}, here's the playlist of the week: {playlist_link}"
|
|
await client.room_send(
|
|
room_id=room_id,
|
|
message_type="m.room.message",
|
|
content={"msgtype": "m.text", "body": reply_msg},
|
|
)
|
|
|
|
|
|
async def send_playlist_of_all(client, sender, room_id, playlist_id):
|
|
"""Sends playlist of all time in reply to sender, in room with room_id."""
|
|
playlist_link = f"https://www.youtube.com/playlist?list={playlist_id}"
|
|
reply_msg = f"{sender}, here's the playlist of all time: {playlist_link}"
|
|
await client.room_send(
|
|
room_id=room_id,
|
|
message_type="m.room.message",
|
|
content={"msgtype": "m.text", "body": reply_msg},
|
|
)
|
|
|
|
|
|
async def message_callback(conn, cursor, youtube, client, room, event):
|
|
"""Event handler for received messages."""
|
|
sender = event.sender
|
|
if sender != MATRIX_USER:
|
|
body = event.body.strip()
|
|
timestamp = event.server_timestamp
|
|
playlist_id = get_or_make_playlist(
|
|
conn, cursor, youtube, monday_date(timestamp)
|
|
)
|
|
all_playlist_id = get_or_make_playlist(
|
|
conn, cursor, youtube, datetime.datetime.fromtimestamp(0)
|
|
)
|
|
|
|
timestamp_sec = datetime.datetime.fromtimestamp(
|
|
event.server_timestamp / 1000,
|
|
datetime.UTC, # millisec to sec
|
|
)
|
|
current_time = datetime.datetime.now(datetime.UTC)
|
|
|
|
# Account for up to 5 minutes of clock drift
|
|
recent = abs(current_time - timestamp_sec) < datetime.timedelta(minutes=5)
|
|
|
|
if body == "!parkerbot" and recent:
|
|
await send_intro_message(client, sender, room.room_id)
|
|
return
|
|
|
|
if body == "!week" and recent:
|
|
await send_playlist_of_week(client, sender, room.room_id, playlist_id)
|
|
return
|
|
|
|
if body == "!all" and recent:
|
|
await send_playlist_of_all(client, sender, room.room_id, all_playlist_id)
|
|
return
|
|
|
|
youtube_link_pattern = (
|
|
r"(https?://(?:www\.|music\.)?youtube\.com/(?!playlist\?list=)watch"
|
|
r"\?v=[\w-]+|https?://youtu\.be/[\w-]+)"
|
|
)
|
|
youtube_links = re.findall(youtube_link_pattern, body)
|
|
|
|
for link in youtube_links:
|
|
video_id = link.split("v=")[-1].split("&")[0].split("/")[-1]
|
|
|
|
# Safely fetch the category check, title, and channel
|
|
is_music_vid, title, channel = get_video_info(youtube, video_id)
|
|
|
|
# Send the title to the channel so people know what the link is
|
|
# Only do this for recent messages to prevent spam during backwards-sync
|
|
if recent:
|
|
plain_text = f"{title}"
|
|
if channel:
|
|
plain_text += f" - {channel}"
|
|
|
|
# Escape HTML characters to prevent broken rendering in Matrix
|
|
escaped_text = html.escape(plain_text)
|
|
html_text = (
|
|
f"<em><span data-mx-color='#808080'>{escaped_text}</span></em>"
|
|
)
|
|
|
|
await client.room_send(
|
|
room_id=room.room_id,
|
|
message_type="m.room.message",
|
|
content={
|
|
"msgtype": "m.text",
|
|
"body": plain_text,
|
|
"format": "org.matrix.custom.html",
|
|
"formatted_body": html_text,
|
|
},
|
|
)
|
|
|
|
# Only add to the playlist if it's categorized as music/entertainment
|
|
if is_music_vid:
|
|
message_id = record_message(conn, cursor, sender, link, timestamp)
|
|
if in_playlist(cursor, video_id, playlist_id):
|
|
print(f"Track is already in this week's playlist: {link}")
|
|
else:
|
|
# Add video to playlists and record it in the database
|
|
add_video_to_playlist(youtube, playlist_id, video_id)
|
|
add_video_to_playlist(youtube, all_playlist_id, video_id)
|
|
with conn:
|
|
cursor.execute(
|
|
(
|
|
"INSERT INTO playlist_tracks (playlist_id, message_id, video_id) "
|
|
"VALUES (?, ?, ?)"
|
|
),
|
|
(playlist_id, message_id, video_id),
|
|
)
|
|
print(f"Added track to this week's playlist: {link}")
|
|
if recent:
|
|
asyncio.create_task(process_radio_track(link, video_id, title))
|
|
|
|
|
|
def in_playlist(cursor, video_id, playlist_id):
|
|
"""Checks if video is in playlist."""
|
|
cursor.execute(
|
|
"SELECT id FROM playlist_tracks WHERE video_id = ? AND playlist_id = ?",
|
|
(video_id, playlist_id),
|
|
)
|
|
if cursor.fetchone():
|
|
return True
|
|
return False
|
|
|
|
|
|
def record_message(conn, cursor, sender, link, timestamp):
|
|
"""Records message to messages table in DB, returns ID."""
|
|
try:
|
|
with conn:
|
|
cursor.execute(
|
|
"INSERT INTO messages (sender, message, timestamp) VALUES (?, ?, ?)",
|
|
(sender, link, timestamp),
|
|
)
|
|
print(f"Saved message: {sender} {link} {timestamp}")
|
|
except sqlite3.IntegrityError as e:
|
|
if "UNIQUE constraint failed" in str(e):
|
|
print(f"Entry already exists: {sender} {link} {timestamp}")
|
|
else:
|
|
raise e
|
|
|
|
cursor.execute(
|
|
"SELECT id FROM messages WHERE sender = ? AND message = ? AND timestamp = ?",
|
|
(sender, link, timestamp),
|
|
)
|
|
return cursor.fetchone()[0]
|
|
|
|
|
|
async def sync_callback(response):
|
|
"""Saves Matrix sync token."""
|
|
with open(TOKEN_PATH, "w", encoding="utf-8") as f:
|
|
f.write(response.next_batch)
|
|
|
|
|
|
def load_sync_token():
|
|
"""Gets saved Matrix sync token if it exists."""
|
|
try:
|
|
with open(TOKEN_PATH, "r", encoding="utf-8") as file:
|
|
return file.read().strip()
|
|
except FileNotFoundError:
|
|
return None
|
|
|
|
|
|
async def get_client(conn, cursor, youtube):
|
|
"""Returns configured and logged in Matrix client."""
|
|
client = AsyncClient(MATRIX_SERVER, MATRIX_USER)
|
|
client.add_event_callback(
|
|
lambda room, event: message_callback(
|
|
conn, cursor, youtube, client, room, event
|
|
),
|
|
RoomMessageText,
|
|
)
|
|
client.add_response_callback(sync_callback, SyncResponse)
|
|
print(await client.login(MATRIX_PASSWORD))
|
|
return client
|
|
|
|
|
|
async def backwards_sync(conn, cursor, youtube, client, room, start_token):
|
|
"""Fetch and process historical messages from a given room."""
|
|
print("Starting to process channel log...")
|
|
from_token = start_token
|
|
room_id = room.room_id
|
|
while True:
|
|
# Fetch room messages
|
|
response = await client.room_messages(room_id, from_token, direction="b")
|
|
|
|
# Process each message
|
|
for event in response.chunk:
|
|
if isinstance(event, RoomMessageText):
|
|
await message_callback(conn, cursor, youtube, client, room, event)
|
|
|
|
# Break if there are no more messages to fetch
|
|
if not response.end or response.end == from_token:
|
|
break
|
|
|
|
# Update the from_token for the next iteration
|
|
from_token = response.end
|
|
|
|
|
|
async def process_radio_track(video_link, video_id, title):
|
|
"""Downloads native audio and pushes it to AzuraCast API in the background."""
|
|
if not AZURACAST_API_KEY:
|
|
return
|
|
|
|
# Sanitize title for the local filesystem
|
|
safe_title = "".join(c for c in title if c.isalnum() or c in " -_").strip()
|
|
base_filename = f"{safe_title}_{video_id}"
|
|
base_filepath = os.path.join(DATA_DIR, base_filename)
|
|
|
|
print(f"📻 Downloading audio for radio: {title}")
|
|
|
|
# 1. Download with yt-dlp
|
|
dl_cmd = [
|
|
"yt-dlp",
|
|
"--no-cache-dir",
|
|
"--extract-audio",
|
|
"--output", f"{base_filepath}.%(ext)s",
|
|
video_link
|
|
]
|
|
process = await asyncio.create_subprocess_exec(
|
|
*dl_cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE
|
|
)
|
|
stdout, stderr = await process.communicate()
|
|
if process.returncode != 0:
|
|
print(f"❌ yt-dlp crashed! Error:\n{stderr.decode().strip()}")
|
|
return
|
|
|
|
# Check for the file using glob since the extension (.m4a, .webm, .opus) is dynamic
|
|
downloaded_files = glob.glob(f"{base_filepath}.*")
|
|
|
|
if not downloaded_files:
|
|
print(f"❌ Failed to download audio for {title}")
|
|
return
|
|
|
|
filepath = downloaded_files[0]
|
|
|
|
print(f"📻 Uploading to AzuraCast: {title}")
|
|
|
|
# 2. Upload to AzuraCast using curl
|
|
upload_cmd = [
|
|
"curl",
|
|
"-s",
|
|
"-X",
|
|
"POST",
|
|
f"{AZURACAST_URL}/api/station/{AZURACAST_STATION_ID}/files",
|
|
"-H",
|
|
f"X-API-Key: {AZURACAST_API_KEY}",
|
|
"-F",
|
|
f"file=@{filepath}",
|
|
]
|
|
|
|
curl_proc = await asyncio.create_subprocess_exec(
|
|
*upload_cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
|
|
)
|
|
stdout, _ = await curl_proc.communicate()
|
|
|
|
# 3. Extract unique_id and push to the live queue
|
|
try:
|
|
response = json.loads(stdout.decode())
|
|
unique_id = response.get("unique_id")
|
|
|
|
if unique_id:
|
|
req_cmd = [
|
|
"curl",
|
|
"-s",
|
|
"-X",
|
|
"POST",
|
|
f"{AZURACAST_URL}/api/station/{AZURACAST_STATION_ID}/request/{unique_id}",
|
|
"-H",
|
|
f"X-API-Key: {AZURACAST_API_KEY}",
|
|
]
|
|
req_proc = await asyncio.create_subprocess_exec(
|
|
*req_cmd,
|
|
stdout=asyncio.subprocess.DEVNULL,
|
|
stderr=asyncio.subprocess.DEVNULL,
|
|
)
|
|
await req_proc.communicate()
|
|
print(f"✅ Queued on radio: {title}")
|
|
else:
|
|
print(f"⚠️ Uploaded, but no unique_id returned. Response: {stdout.decode()}")
|
|
except json.JSONDecodeError:
|
|
print(f"❌ Failed to parse AzuraCast response: {stdout.decode()}")
|
|
|
|
# Cleanup the local file so your container doesn't bloat
|
|
if os.path.exists(filepath):
|
|
os.remove(filepath)
|
|
|
|
|
|
async def main():
|
|
"""Get DB and Matrix client ready, and start syncing."""
|
|
args = parse_arguments()
|
|
conn, cursor = connect_db()
|
|
define_tables(conn, cursor)
|
|
youtube = get_authenticated_service()
|
|
client = await get_client(conn, cursor, youtube)
|
|
sync_token = load_sync_token()
|
|
|
|
# This is incredibly dumb and most probably will exceed your YouTube API quota.
|
|
if args.backwards_sync:
|
|
init_sync = await client.sync(30000)
|
|
room = await client.room_resolve_alias(MATRIX_ROOM)
|
|
await backwards_sync(conn, cursor, youtube, client, room, init_sync.next_batch)
|
|
|
|
await client.sync_forever(30000, full_state=True, since=sync_token)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|