Plex Transcode Notification To Users

#!/usr/bin/env python3
"""
A Flask-based webhook receiver for Tautulli to notify Plex users
when they are transcoding rather than direct playing.

When a transcode event is received:
1. If it's the user's first transcode attempt within the retry window, 
   the script terminates the session and shows a notification.
2. If it happens again within the window and the user has reached 
   the maximum number of blocked retries, the script allows the transcode
   to continue (i.e., does not terminate it).
3. After the retry window expires, the counters reset.
"""

import time
import requests
import logging
from flask import Flask, request

###############################################################################
#                          CONFIGURATION & CONSTANTS
###############################################################################
# Change the values below to match your Tautulli setup.
TAUTULLI_API_URL = "http://<ENDPOINT>:8181/api/v2"
TAUTULLI_API_KEY = "<API KEY>"

# How long a user’s transcode attempts remain tracked (in seconds).
RETRY_WINDOW = 3600  # 1 hour

# How many times to block transcode attempts within the RETRY_WINDOW.
# Once the user has hit this many blocks, they will be allowed to transcode again.
MAX_RETRIES = 1

# List of platforms commonly used in a browser.
BROWSER_CLIENTS = {"plex web", "chrome", "firefox", "edge", "safari"}

# This dictionary will track each user's transcode attempts:
#   user_retry_tracker = {
#       "<user_name_or_id>": {
#           "retries": <int>,
#           "last_attempt": <float_timestamp>
#       },
#       ...
#   }
user_retry_tracker = {}

###############################################################################
#                          APPLICATION SETUP & LOGGING
###############################################################################
app = Flask(__name__)

logging.basicConfig(
    filename='plex_transcode_monitor.log',
    level=logging.DEBUG,
    format='%(asctime)s - %(levelname)s - %(message)s',
)

###############################################################################
#                              HELPER FUNCTIONS
###############################################################################
def notify_and_stop_transcoding(session_id: str, user: str, message: str) -> None:
    """
    Send a command to Tautulli to terminate a transcoding session.
    This also pushes a notification message to the user’s Plex client.
    
    :param session_id: The Plex session ID.
    :param user: The Plex user name or ID.
    :param message: The message that will be displayed to the user.
    """
    try:
        response = requests.get(
            TAUTULLI_API_URL,
            params={
                "apikey": TAUTULLI_API_KEY,
                "cmd": "terminate_session",
                "session_id": session_id,
                "message": message,
            },
            timeout=10  # Set a timeout to avoid hanging.
        )
        response.raise_for_status()
        logging.info(f"Transcode session stopped: Session ID={session_id}, "
                     f"User={user}, Message={message}")
    except requests.RequestException as e:
        logging.error(f"Error stopping stream for user {user}: {e}")


def should_allow_retry(user_id: str) -> bool:
    """
    Check whether a user should be allowed to continue transcoding 
    based on their retry history within the configured window.
    
    The logic is:
      - If the user has no record, block them (first attempt), 
        and initialize their tracking.
      - If the user is still within the retry window:
          * If their total retries is less than MAX_RETRIES, 
            block again and increment their count.
          * Otherwise allow.
      - If the user is outside the retry window, reset and block again.
    
    :param user_id: The Plex user name or ID.
    :return: True if the user should be ALLOWED to transcode, False if blocked.
    """
    current_time = time.time()

    # If the user is not in the tracker, block and create an entry with 0 retries.
    if user_id not in user_retry_tracker:
        user_retry_tracker[user_id] = {
            "last_attempt": current_time,
            "retries": 0
        }
        return False  # Block them on the first attempt.

    user_data = user_retry_tracker[user_id]
    last_attempt_time = user_data["last_attempt"]
    retries = user_data["retries"]

    # Check if we're within the retry window
    if current_time - last_attempt_time <= RETRY_WINDOW:
        # If user still has blocks left, increment and block
        if retries < MAX_RETRIES:
            user_data["retries"] += 1
            user_data["last_attempt"] = current_time
            return False  # Block
        else:
            return True   # Allow
    else:
        # Retry window expired, reset counters and block this new attempt
        user_retry_tracker[user_id] = {
            "last_attempt": current_time,
            "retries": 0
        }
        return False  # Block on "first" attempt again


def get_notification_message(platform: str) -> str:
    """
    Return an appropriate notification message based on the user’s platform.
    
    :param platform: The detected platform name from Tautulli’s webhook data.
    :return: A string message to be displayed to the user.
    """
    if platform.lower() in BROWSER_CLIENTS:
        return (
            "You are using a web browser to watch this video, which causes transcoding. "
            "To avoid slowing down your stream and the server, please use the native Plex app "
            "on your device. If you must watch in a browser, set the playback quality to 'Original' "
            "or 'Maximum' where possible. You can try playing again if you understand the impact."
        )
    else:
        return (
            "You are not playing this video in its original quality. This transcoding slows down "
            "your stream and the server. Please update your Plex app settings to 'Original' or "
            "'Maximum' quality. If you truly need to watch at a lower quality, you can try again."
        )

###############################################################################
#                               FLASK WEBHOOK
###############################################################################
@app.route("/webhook", methods=["POST"])
def tautulli_webhook():
    """
    Handle incoming webhooks from Tautulli.
    Expects JSON data with at least:
      - user
      - session_id
      - transcode_decision
      - platform
    
    If the 'transcode_decision' is 'transcode', this script checks whether the 
    user should be blocked (and notified) or allowed to continue.
    """
    try:
        data = request.json

        # Basic validation of required fields
        user = data.get("user")
        session_id = data.get("session_id")
        transcode_decision = data.get("transcode_decision")
        platform = data.get("platform")

        if not all([user, session_id, transcode_decision, platform]):
            logging.error("Invalid webhook payload: missing required fields.")
            return "Invalid data", 400

        # Only proceed if it's a transcoding session
        if transcode_decision.lower() == "transcode":
            message = get_notification_message(platform)

            # Decide if we should allow or block this transcode attempt
            if should_allow_retry(user):
                logging.info(
                    f"User '{user}' is allowed to continue transcoding "
                    f"after {user_retry_tracker[user]['retries']} blocked attempt(s)."
                )
            else:
                # Stop the transcode session and show the notification
                notify_and_stop_transcoding(session_id, user, message)

        return "OK", 200

    except Exception as e:
        logging.error(f"Error processing the Tautulli webhook: {e}")
        return "Internal Server Error", 500


###############################################################################
#                                 MAIN ENTRY
###############################################################################
if __name__ == "__main__":
    # Adjust host/port as needed for your environment
    app.run(host="0.0.0.0", port=2000)