Remove expired inbounds in 3x-ui panel

remove-expired.py

"""
File: remove-expired.py
Author: Reza Esmaeili
Date: Nov 09, 2024
Description: 
    This script helps administrators manage inbounds on a 3x-ui panel by identifying and 
    removing expired inbounds or those that exceed specified data usage limits. Additionally, 
    it can notify administrators of actions taken through Telegram Bot API notifications.
    
Features:
    1. Logs into the 3x-ui panel API and retrieves a list of inbounds.
    2. Checks each inbound for expiration based on the `expiryTime` field or data limit.
    3. Deletes expired or overused inbounds if user confirmation is provided.
    4. Sends a detailed report of actions to a specified Telegram channel.

Dependencies:
    - requests: For HTTP communication with the 3x-ui API and Telegram Bot API.
    - json: For parsing JSON responses from the API.
    - datetime: For handling timestamps and calculating expiration.
    - getpass: For securely capturing the user's password.
    - time: For managing delays during deletion requests.

Requirements:
    - Python 3.x
    - Internet connection for API communication
    - Access credentials for the 3x-ui panel

Usage:
    1. Run the script and provide the 3x-ui panel URL, username, and password.
    2. Optionally, enable Telegram notifications by providing a Telegram bot token and channel ID.
    3. The script will log in, retrieve inbound information, and identify expired or overused inbounds.
    4. After confirmation, expired or overused inbounds will be deleted, and a report will be sent to Telegram if enabled.

Note:
    - Telegram messages are split into chunks to avoid exceeding Telegram's character limit.
    - The script requires confirmation before deletion of inbounds.
    - Ensure the Telegram bot has permission to message the provided channel or user.

Example Run:
    $ python remove-expired.py
"""

import requests
import json
from datetime import datetime
import getpass
import time

def log_message(message):
    print(f"[{datetime.now()}] {message}")

# Function to split and send the report if it exceeds the Telegram character limit
def send_report_to_telegram(report_message, telegram_token, telegram_chat_id):  
    telegram_url = f"https://api.telegram.org/bot{telegram_token}/sendMessage"
    max_length = 4096  # Telegram's message character limit

    # Split the report into chunks of complete lines
    report_lines = report_message.splitlines(keepends=True)
    current_chunk = ""
    
    for line in report_lines:
        if len(current_chunk) + len(line) > max_length:
            payload = {
                "chat_id": telegram_chat_id,
                "text": current_chunk,
                "parse_mode": "markdown"
            }
            response = requests.post(telegram_url, data=payload)
            
            if response.status_code != 200:
                log_message(f"Failed to send part of the report to Telegram channel.")
                log_message(f"Status Code: {response.status_code}, Response: {response.text}")
            
            current_chunk = line  # Start a new chunk
        else:
            current_chunk += line

    # Send any remaining text
    if current_chunk:
        payload = {
            "chat_id": telegram_chat_id,
            "text": current_chunk + "\n--- End of the report ---",
            "parse_mode": "markdown"
        }
        response = requests.post(telegram_url, data=payload)

        if response.status_code != 200:
            log_message(f"Failed to send the last part of the report to Telegram channel.")
            log_message(f"Status Code: {response.status_code}, Response: {response.text}")


print("INBOUND CLEANER: This script will help you to delete expired inbounds in 3x-ui panel!\n\n")

panel_url = input("1. Enter the panel's URL (eg. https://mypanel.theserver.com:8998): ")
username = input("2. Enter your panel's username: ")
password = getpass.getpass("3. Enter your panel's password: ")

# Define API URLs
login_url = f"{panel_url}/login"
inbounds_url = f"{panel_url}/panel/api/inbounds/list"
delete_inbound_url = f"{panel_url}/panel/api/inbounds/del/"

# Define login credentials
login_payload = {
    "username": username,
    "password": password
}

use_telegram = input("4. Do you want Telegram notification (via Telegram's bot API)? (Y/n): ").lower() == 'y'

if use_telegram:
    # Telegram Bot API information
    telegram_token = input("Enter your bot token (from @botfather): ")
    telegram_chat_id = input("Enter your channel's ID or username (preceded by @): ")

# Create a session object to persist the session cookie
session = requests.Session()

# Step 1: Login to the API and capture the session cookie
log_message("Logging in to your panel's API...")
response = session.post(login_url, data=login_payload)

if response.status_code == 200:
    log_message("Login successful!")
else:
    log_message("Login failed!")
    log_message(f"Status Code: {response.status_code}, Response: {response.text}")
    exit()

# Step 2: Get the list of inbounds using the session
log_message("\nRetrieving list of inbounds...")
response = session.get(inbounds_url)

if response.status_code == 200:
    try:
        inbounds = response.json().get("obj")
        if inbounds is None:
            log_message("No 'obj' key found in the response.")
            exit()
    except json.JSONDecodeError as e:
        log_message("Failed to parse JSON response.")
        log_message(f"Error: {e}")
        log_message(f"Response text: {response.text}")
        exit()
else:
    log_message("Failed to retrieve the list of inbounds!")
    log_message(f"Status Code: {response.status_code}, Response: {response.text}")
    exit()

if use_telegram:
    # Prepare report content
    report = []
    report.append(f"*INBOUND CLEANER*\nInbound Deletion Report for {panel_url}\n")
    report.append(f"Script executed on: {datetime.now()}\n\n")

# Step 3: Check for expired inbounds and those that exceed total data usage
log_message("\nChecking for expired inbounds or those that exceed total data usage...")
current_time = int(datetime.now().timestamp() * 1000)
expired_inbounds = []

for inbound in inbounds:
    expiry_time = inbound.get("expiryTime")
    inbound_id = inbound.get("id")
    remark = inbound.get("remark")
    up = inbound.get("up", 0)
    down = inbound.get("down", 0)
    total = inbound.get("total", 0)

    # Check if inbound has expired with a valid expiry time
    if expiry_time < current_time and expiry_time > 0:
        expired_inbounds.append({"id": inbound_id, "remark": remark, "reason": "expired"})
        log_message(f"Inbound ID {inbound_id} (Remark: {remark}) has expired.")
    # Check if data usage exceeds total
    elif total > 0 and (up + down) > total:
        expired_inbounds.append({"id": inbound_id, "remark": remark, "reason": "exceeded total data"})
        log_message(f"Inbound ID {inbound_id} (Remark: {remark}) exceeded total data usage.")
    else:
        # print(f"Inbound ID {inbound_id} (Remark: {remark}) is still valid.")
        pass

# Ask for confirmation before deletion
if input("Do you want to delete the expired or data-exceeded inbounds? (Y/n): ").lower() == "y":
    # Step 4: Delete inbounds based on conditions
    if expired_inbounds:
        log_message("\nDeleting inbounds...")
        for inbound in expired_inbounds:
            inbound_id = inbound["id"]
            remark = inbound["remark"]
            reason = inbound["reason"]
            del_url = f"{delete_inbound_url}{inbound_id}"
            response = session.post(del_url)
            time.sleep(0.2) # NOTE: Do we need this?
            
            if response.status_code == 200:
                log_message(f"Inbound ID {inbound_id} (Remark: `{remark}`) deleted successfully! Reason: {reason}")
                if use_telegram: report.append(f"Inbound ID {inbound_id} (Remark: `{remark}`) deleted successfully! Reason: {reason}\n")
            else:
                log_message(f"Failed to delete Inbound ID {inbound_id} (Remark: `{remark}`)! Reason: {reason}")
                if use_telegram: report.append(f"Failed to delete Inbound ID {inbound_id} (Remark: `{remark}`)! Reason: {reason}\n")
                log_message(f"Status Code: {response.status_code}, Response: {response.text}")
    else:
        log_message("No inbounds found for deletion.")
        if use_telegram: report.append("No inbounds found for deletion.\n")
else:
    if use_telegram: report.append("No deletion performed.\n")

if use_telegram:
    # Send the report to the Telegram channel
    log_message("\nSending report to Telegram channel...")
    report_message = "".join(report)
    send_report_to_telegram(report_message, telegram_token, telegram_chat_id)

log_message("\nScript completed.")