parkerbot/main.py

#!/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(
            """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
    if os.path.exists(PICKLE_PATH):
        with open(PICKLE_PATH, "rb") as token:
            credentials = pickle.load(token)

    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)
            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(
            "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()

        if not video_details.get("items"):
            return False, "[Video unavailable or private]", ""

        snippet = video_details["items"][0]["snippet"]

        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},
    )

    with open("./parker.gif", "rb") as gif_file:
        response = await client.upload(gif_file, content_type="image/gif")
    if isinstance(response, UploadResponse):
        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"},
            },
        )


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,
        )
        current_time = datetime.datetime.now(datetime.UTC)

        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]

            is_music_vid, title, channel = get_video_info(youtube, video_id)

            if recent:
                plain_text = f"{title}"
                if channel:
                    plain_text += f" - {channel}"

                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,
                    },
                )

            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_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:
        response = await client.room_messages(room_id, from_token, direction="b")

        for event in response.chunk:
            if isinstance(event, RoomMessageText):
                await message_callback(conn, cursor, youtube, client, room, event)

        if not response.end or response.end == from_token:
            break

        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

    base_url = AZURACAST_URL.rstrip("/")
    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}")

    dl_cmd = [
        "yt-dlp",
        "--cache-dir", "/usr/src/app/.cache",
        "--extract-audio",
        "--embed-metadata",
        "--embed-thumbnail",
        "--convert-thumbnails", "jpg",
        "--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

    downloaded_files = glob.glob(f"{base_filepath}.*")

    if not downloaded_files:
        print(f"❌ Failed to download audio for {title}")
        return

    filepath = downloaded_files[0]
    filename = os.path.basename(filepath)

    print(f"📻 Uploading to AzuraCast: {title}")

    upload_cmd = [
        "curl",
        "-s",
        "-X",
        "POST",
        f"{base_url}/api/station/{AZURACAST_STATION_ID}/files/upload",
        "-H",
        f"X-API-Key: {AZURACAST_API_KEY}",
        "-F",
        f"path={filename}",
        "-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()

    unique_id = None
    azura_path = None

    try:
        response = json.loads(stdout.decode())
        unique_id = response.get("unique_id")
        azura_path = response.get("path") or response.get("file")

        if not unique_id or not azura_path:
            search_cmd = [
                "curl",
                "-s",
                "-G",
                f"{base_url}/api/station/{AZURACAST_STATION_ID}/files",
                "--data-urlencode",
                f"search={filename}",
                "-H",
                f"X-API-Key: {AZURACAST_API_KEY}",
            ]
            search_proc = await asyncio.create_subprocess_exec(
                *search_cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.DEVNULL,
            )
            search_out, _ = await search_proc.communicate()

            search_data = json.loads(search_out.decode())
            if (
                isinstance(search_data, dict)
                and "rows" in search_data
                and len(search_data["rows"]) > 0
            ):
                unique_id = unique_id or search_data["rows"][0].get("unique_id")
                azura_path = azura_path or search_data["rows"][0].get("path")
            elif isinstance(search_data, list) and len(search_data) > 0:
                unique_id = unique_id or search_data[0].get("unique_id")
                azura_path = azura_path or search_data[0].get("path")

    except json.JSONDecodeError:
        print(f"❌ Failed to parse AzuraCast upload response")

    # 1. Add the track to the "Currents" playlist FIRST
    # AzuraCast rejects requests for files that aren't assigned to any active playlist.
    target_path = azura_path if azura_path else filename

    try:
        pl_cmd = [
            "curl",
            "-s",
            "-X",
            "GET",
            f"{base_url}/api/station/{AZURACAST_STATION_ID}/playlists",
            "-H",
            f"X-API-Key: {AZURACAST_API_KEY}",
        ]
        pl_proc = await asyncio.create_subprocess_exec(
            *pl_cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL
        )
        pl_stdout, _ = await pl_proc.communicate()

        playlists = json.loads(pl_stdout.decode())
        currents_id = next(
            (p.get("id") for p in playlists if p.get("name", "").lower() == "currents"),
            None,
        )

        if currents_id:
            payload = {
                "do": "playlist",
                "playlists": [currents_id],
                "files": [target_path],
            }
            json_path = os.path.join(DATA_DIR, f"assign_{filename}.json")
            with open(json_path, "w") as f:
                json.dump(payload, f)

            batch_cmd = [
                "curl",
                "-s",
                "-X",
                "PUT",
                f"{base_url}/api/station/{AZURACAST_STATION_ID}/files/batch",
                "-H",
                f"X-API-Key: {AZURACAST_API_KEY}",
                "-H",
                "Content-Type: application/json",
                "-d",
                f"@{json_path}",
            ]
            b_proc = await asyncio.create_subprocess_exec(
                *batch_cmd, stdout=asyncio.subprocess.DEVNULL
            )
            await b_proc.communicate()

            if os.path.exists(json_path):
                os.remove(json_path)
            print(f"✅ Auto-assigned to 'Currents' (Path: {target_path})")
    except Exception as e:
        print(f"❌ Failed to auto-assign playlist for {title}: {e}")

    # 2. Issue the Play Next Request SECOND
    # Now that the file is legally in a playlist, AzuraCast will accept the request.
    if unique_id:
        req_cmd = [
            "curl",
            "-s",
            "-X",
            "POST",
            f"{base_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.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        req_stdout, _ = await req_proc.communicate()

        try:
            req_resp = json.loads(req_stdout.decode())
            # Check the JSON response directly so we don't lie about success
            if req_resp.get("success") or req_resp.get("code") == 200:
                print(f"✅ Queued on radio to play next: {title}")
            else:
                print(
                    f"⚠️ AzuraCast rejected the request for {title}: {req_stdout.decode()}"
                )
        except json.JSONDecodeError:
            print(f"⚠️ Failed to queue {title}. API returned: {req_stdout.decode()}")
    else:
        print(f"⚠️ Uploaded, but no unique_id returned to make the request.")

    if os.path.exists(filepath):
        os.remove(filepath)


async def automated_playlist_rotation():
    """Runs continuously, sorting AzuraCast files by age once a day."""
    while True:
        try:
            if not AZURACAST_API_KEY:
                await asyncio.sleep(86400)
                continue

            base_url = AZURACAST_URL.rstrip("/")

            pl_cmd = [
                "curl",
                "-s",
                "-X",
                "GET",
                f"{base_url}/api/station/{AZURACAST_STATION_ID}/playlists",
                "-H",
                f"X-API-Key: {AZURACAST_API_KEY}",
            ]
            pl_proc = await asyncio.create_subprocess_exec(
                *pl_cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.DEVNULL,
            )
            pl_stdout, _ = await pl_proc.communicate()

            try:
                playlists = json.loads(pl_stdout.decode())
                pl_map = {p.get("name", "").lower(): p.get("id") for p in playlists}
            except json.JSONDecodeError:
                await asyncio.sleep(86400)
                continue

            req_playlists = ["currents", "recurrents", "gold"]
            if not all(p in pl_map for p in req_playlists):
                print(
                    "⚠️ Missing required playlists. Please create 'Currents', 'Recurrents', and 'Gold'."
                )
                await asyncio.sleep(86400)
                continue

            currents_id = pl_map["currents"]
            recurrents_id = pl_map["recurrents"]
            gold_id = pl_map["gold"]

            files_cmd = [
                "curl",
                "-s",
                "-X",
                "GET",
                f"{base_url}/api/station/{AZURACAST_STATION_ID}/files",
                "-H",
                f"X-API-Key: {AZURACAST_API_KEY}",
            ]
            f_proc = await asyncio.create_subprocess_exec(
                *files_cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.DEVNULL,
            )
            f_stdout, _ = await f_proc.communicate()

            try:
                files_data = json.loads(f_stdout.decode())
            except json.JSONDecodeError:
                files_data = []

            file_list = (
                files_data.get("rows", [])
                if isinstance(files_data, dict)
                else files_data
            )

            currents = []
            recurrents = []
            gold = []
            current_time = time.time()

            for item in file_list:
                if item.get("type") == "dir":
                    continue

                path = item.get("path")
                mtime = item.get("mtime", current_time)

                age_days = (current_time - mtime) / 86400

                if age_days <= 7:
                    currents.append(path)
                elif age_days <= 14:
                    recurrents.append(path)
                else:
                    gold.append(path)

            async def batch_assign(files, pl_id):
                if not files:
                    return
                for i in range(0, len(files), 50):
                    chunk = files[i : i + 50]
                    payload = {"do": "playlist", "playlists": [pl_id], "files": chunk}
                    json_path = os.path.join(DATA_DIR, f"batch_{pl_id}_{i}.json")
                    with open(json_path, "w") as f:
                        json.dump(payload, f)

                    batch_cmd = [
                        "curl",
                        "-s",
                        "-X",
                        "PUT",
                        f"{base_url}/api/station/{AZURACAST_STATION_ID}/files/batch",
                        "-H",
                        f"X-API-Key: {AZURACAST_API_KEY}",
                        "-H",
                        "Content-Type: application/json",
                        "-d",
                        f"@{json_path}",
                    ]
                    b_proc = await asyncio.create_subprocess_exec(
                        *batch_cmd, stdout=asyncio.subprocess.DEVNULL
                    )
                    await b_proc.communicate()

                    if os.path.exists(json_path):
                        os.remove(json_path)

            await batch_assign(currents, currents_id)
            await batch_assign(recurrents, recurrents_id)
            await batch_assign(gold, gold_id)

            print(
                f"✅ Playlist rotation: {len(currents)} Currents, {len(recurrents)} Recurrents, {len(gold)} Gold."
            )

        except Exception as e:
            print(f"💥 Error in automated_playlist_rotation: {e}")

        await asyncio.sleep(86400)


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()

    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)

    asyncio.create_task(automated_playlist_rotation())

    await client.sync_forever(30000, full_state=True, since=sync_token)


if __name__ == "__main__":
    asyncio.run(main())