combined_output.txt

# === bybit_flask.py ===
# bybit_flask.py

import os
from os import sep as os_sep
from flask import Flask, render_template, send_from_directory, abort, request, Response
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from functools import wraps
from urllib.parse import unquote
from datetime import datetime, timezone

# Initialize Flask app
app = Flask(__name__)

# Initialize Flask-Limiter
limiter = Limiter(
    get_remote_address,
    app=app,
    default_limits=["200 per day", "50 per hour"]
)

# === Configuration ===

# Toggle authentication
USE_AUTH = False  # Set to True to enable authentication

# Directories for storing CSV files and logs
CSV_FOLDER = "bybit_stream_data"

# Ensure the CSV folder exists
os.makedirs(CSV_FOLDER, exist_ok=True)

# === Authentication Setup ===

# Define your credentials for Basic Authentication (Used only if USE_AUTH is True)
USERNAME = os.getenv('USERNAME', 'admin')        # Change to your desired username
PASSWORD = os.getenv('PASSWORD', 'password')     # Change to your desired password

def check_auth(username, password):
    """Check if a username/password combination is valid."""
    return username == USERNAME and password == PASSWORD

def authenticate():
    """Sends a 401 response that enables basic auth."""
    return Response(
        'Could not verify your access level for that URL.\n'
        'You have to login with proper credentials', 401,
        {'WWW-Authenticate': 'Basic realm="Login Required"'}
    )

def requires_auth(f):
    """Decorator to handle authentication logic."""
    @wraps(f)
    def decorated(*args, **kwargs):
        if USE_AUTH:
            auth = request.authorization
            if not auth or not check_auth(auth.username, auth.password):
                return authenticate()
        return f(*args, **kwargs)
    return decorated

# Security: Prevent directory traversal by ensuring all paths are within CSV_FOLDER
def secure_path(path):
    # Join the CSV_FOLDER with the requested path
    base_path = os.path.abspath(CSV_FOLDER)
    safe_path = os.path.abspath(os.path.join(base_path, path))
    # Ensure the resulting path is within CSV_FOLDER
    if not safe_path.startswith(base_path):
        abort(403)  # Forbidden
    return safe_path

# === Error Handlers ===

@app.errorhandler(404)
def not_found(error):
    return render_template('error.html', message="Resource Not Found.", description="The requested resource could not be found."), 404

@app.errorhandler(403)
def forbidden(error):
    return render_template('error.html', message="Forbidden", description="You don't have permission to access this resource."), 403

@app.errorhandler(401)
def unauthorized(error):
    return render_template('error.html', message="Unauthorized", description="Please provide valid credentials to access this resource."), 401

@app.errorhandler(429)
def ratelimit_handler(error):
    return render_template('error.html', message="Rate Limit Exceeded", description="You have made too many requests. Please try again later."), 429

@app.errorhandler(500)
def internal_error(error):
    return render_template('error.html', message="Internal Server Error", description="An unexpected error has occurred. Please try again later."), 500

# === Routes ===

# Home route - lists the top-level directories
@app.route('/')
@requires_auth
@limiter.limit("10 per minute")
def index():
    return list_directory('')

# Dynamic route to list directories and files
@app.route('/browse/<path:subpath>')
@requires_auth
@limiter.limit("10 per minute")
def list_directory(subpath):
    # Decode the URL-encoded path
    subpath = unquote(subpath)
    safe_path = secure_path(subpath)

    if not os.path.exists(safe_path):
        abort(404)

    if not os.path.isdir(safe_path):
        abort(404)

    # List directories and files
    items = os.listdir(safe_path)
    dirs = []
    files = []
    for item in sorted(items):
        item_path = os.path.join(safe_path, item)
        if os.path.isdir(item_path):
            dirs.append(item)
        elif os.path.isfile(item_path) and (item.endswith('.csv') or item.endswith('.csv.gz')):
            file_stat = os.stat(item_path)
            files.append({
                'name': item,
                'size': f"{file_stat.st_size / (1024**2):.2f} MB",
                'modified': datetime.fromtimestamp(file_stat.st_mtime).strftime('%Y-%m-%d %H:%M:%S')
            })

    # Calculate the relative path for navigation
    relative_path = os.path.relpath(safe_path, os.path.abspath(CSV_FOLDER))
    if relative_path == '.':
        relative_path = ''

    return render_template(
        'directory.html',
        dirs=dirs,
        files=files,
        current_path=relative_path,
        path_separator=os_sep  # Pass the path separator to the template
    )

# Route to download files
@app.route('/download/<path:filepath>')
@requires_auth
@limiter.limit("10 per minute")
def download_file(filepath):
    # Decode the URL-encoded path
    filepath = unquote(filepath)
    safe_path = secure_path(filepath)

    if not os.path.exists(safe_path):
        abort(404)

    if not os.path.isfile(safe_path):
        abort(404)

    # Extract directory and filename
    directory = os.path.dirname(safe_path)
    filename = os.path.basename(safe_path)

    return send_from_directory(directory, filename, as_attachment=True)

# === Main Application Runner ===

if __name__ == "__main__":
    # For development purposes; in production, use passenger_wsgi.py as the entry point
    app.run(host='127.0.0.1', port=8000, debug=False)


# === bybit_websocket_listener.py ===
# bybit_websocket_listener.py

import asyncio
import json
import os
import websockets
from websockets.exceptions import (
    ConnectionClosedError,
    ConnectionClosedOK,
    WebSocketException,
)
from datetime import datetime, timezone
import gzip
import shutil
import aiofiles
from asyncio import Queue, TaskGroup
from concurrent.futures import ThreadPoolExecutor
import logging
from logging.handlers import RotatingFileHandler

# === Configuration ===

# Load configuration from config.json
CONFIG_FILE = "config.json"
if not os.path.exists(CONFIG_FILE):
    raise FileNotFoundError(f"Configuration file '{CONFIG_FILE}' not found.")

with open(CONFIG_FILE, "r") as f:
    config = json.load(f)

CSV_FOLDER = config.get("CSV_FOLDER", "bybit_stream_data")
LOGS_FOLDER = config.get("LOGS_FOLDER", "logs")
LOG_FILE_NAME = config.get("LOG_FILE_NAME", "app.log")
LOG_FILE_SIZE_LIMIT = config.get("LOG_FILE_SIZE_LIMIT_MB", 10) * 1024 * 1024  # in bytes
LOG_FILE_BACKUP_COUNT = config.get("LOG_FILE_BACKUP_COUNT", 5)
CSV_FOLDER_SIZE_LIMIT = config.get("CSV_FOLDER_SIZE_LIMIT_GB", 15) * 1024 * 1024 * 1024  # in bytes
BUFFER_LIMIT = config.get("BUFFER_LIMIT", 100)
WRITE_INTERVAL = config.get("WRITE_INTERVAL_SEC", 1)  # seconds
PING_INTERVAL = config.get("PING_INTERVAL_SEC", 20)  # seconds
PING_TIMEOUT = config.get("PING_TIMEOUT_SEC", 10)  # seconds
WEBSOCKET_PING_INTERVAL = config.get("WEBSOCKET_PING_INTERVAL_SEC", 20)  # seconds
WEBSOCKET_PING_TIMEOUT = config.get("WEBSOCKET_PING_TIMEOUT_SEC", 10)  # seconds
WEBSOCKET_CLOSE_TIMEOUT = config.get("WEBSOCKET_CLOSE_TIMEOUT_SEC", 5)  # seconds
RECONNECT_INITIAL_BACKOFF = config.get("RECONNECT_INITIAL_BACKOFF_SEC", 1)  # seconds
RECONNECT_MAX_BACKOFF = config.get("RECONNECT_MAX_BACKOFF_SEC", 60)  # seconds
COMPRESSION_SLEEP_INTERVAL = config.get("COMPRESSION_SLEEP_INTERVAL_SEC", 86400)  # 24 hours
FOLDER_SIZE_CHECK_INTERVAL = config.get("FOLDER_SIZE_CHECK_INTERVAL_SEC", 600)  # 10 minutes
LOG_INTERVAL = config.get("LOG_INTERVAL", 100)  # Log every 100 messages

os.makedirs(CSV_FOLDER, exist_ok=True)
os.makedirs(LOGS_FOLDER, exist_ok=True)

LOG_FILE_PATH = os.path.join(LOGS_FOLDER, LOG_FILE_NAME)

# Setup logging with RotatingFileHandler
logger = logging.getLogger("BybitWebSocketListener")
logger.setLevel(logging.INFO)
handler = RotatingFileHandler(
    LOG_FILE_PATH,
    maxBytes=LOG_FILE_SIZE_LIMIT,
    backupCount=LOG_FILE_BACKUP_COUNT,
)
formatter = logging.Formatter(
    "%(asctime)s - %(levelname)s - %(message)s"
)
handler.setFormatter(formatter)
logger.addHandler(handler)

# Initialize ThreadPoolExecutor for blocking I/O
executor = ThreadPoolExecutor()

# Load topics from symbols.json
SYMBOLS_FILE = "symbols.json"
if not os.path.exists(SYMBOLS_FILE):
    raise FileNotFoundError(f"Symbols file '{SYMBOLS_FILE}' not found.")

with open(SYMBOLS_FILE, "r") as f:
    TOPICS = json.load(f)

# WebSocket URLs for each channel_type
CHANNEL_URLS = {
    "linear": "wss://stream.bybit.com/v5/public/linear",
    "spot": "wss://stream.bybit.com/v5/public/spot",
    "options": "wss://stream.bybit.com/v5/public/option",
    "inverse": "wss://stream.bybit.com/v5/public/inverse",
}

# === Functions ===

# Flatten topics for a specific channel type
def generate_subscription_args(topics_json, channel_type):
    args = []
    if channel_type in topics_json:
        for symbol, topic_list in topics_json[channel_type].items():
            for topic in topic_list:
                args.append(f"{topic}.{symbol}")
    return args

# Function to get the current UTC date in YYYY-MM-DD format
def get_current_utc_date():
    return datetime.now(timezone.utc).strftime("%Y-%m-%d")

# Dictionary to hold message queues
message_queues = {}

# Counter for throttled logging
log_counter = 0

# Function to save messages to CSV in structured subdirectories
async def save_message(channel_type, topic, message):
    key = (channel_type, topic)
    if key not in message_queues:
        message_queues[key] = Queue()
        asyncio.create_task(process_queue(channel_type, topic, message_queues[key]))
    await message_queues[key].put(message)

async def process_queue(channel_type, topic, queue):
    global log_counter
    while True:
        try:
            utc_date = get_current_utc_date()

            # Extract the symbol from the topic
            # Example: "orderbook.1.BTCUSDT" -> "BTCUSDT"
            symbol = topic.split(".")[-1]
            topic_base = ".".join(topic.split(".")[:-1])

            channel_folder = os.path.join(CSV_FOLDER, channel_type)
            symbol_folder = os.path.join(channel_folder, symbol)  # New layer for symbol
            topic_folder = os.path.join(symbol_folder, topic_base)  # Topic folder under symbol

            # Ensure the directories exist
            os.makedirs(topic_folder, exist_ok=True)

            # Define the CSV file path
            csv_file = os.path.join(topic_folder, f"{utc_date}_{topic}.csv")

            buffer = []
            write_interval = WRITE_INTERVAL  # Seconds

            while True:
                try:
                    message = await asyncio.wait_for(queue.get(), timeout=write_interval)
                    buffer.append(json.dumps(message))
                    if len(buffer) >= BUFFER_LIMIT:
                        await write_buffer_to_file(csv_file, buffer)
                        buffer.clear()
                        log_counter += len(buffer)
                        if log_counter >= LOG_INTERVAL:
                            logger.info(f"Written {log_counter} messages to {csv_file}")
                            log_counter = 0
                except asyncio.TimeoutError:
                    if buffer:
                        await write_buffer_to_file(csv_file, buffer)
                        log_counter += len(buffer)
                        if log_counter >= LOG_INTERVAL:
                            logger.info(f"Written {log_counter} messages to {csv_file}")
                            log_counter = 0
                        buffer.clear()
        except Exception as e:
            logger.error(f"Error in processing queue for {channel_type}.{topic}: {e}")
            await asyncio.sleep(1)  # Prevent tight loop on error

async def write_buffer_to_file(csv_file, buffer):
    try:
        async with aiofiles.open(csv_file, mode="a") as file:
            await file.write("\n".join(buffer) + "\n")
    except Exception as e:
        logger.error(f"Failed to write buffer to {csv_file}: {e}")

# Function to compress old CSV files to .csv.gz
async def compress_old_csv_files_periodically():
    while True:
        try:
            await compress_old_csv_files()
            await asyncio.sleep(COMPRESSION_SLEEP_INTERVAL)
        except Exception as e:
            logger.error(f"Error in compression task: {e}")
            await asyncio.sleep(60)  # Retry after a minute on error

# Function to compress all old CSV files
async def compress_old_csv_files():
    utc_date = get_current_utc_date()
    csv_files = await get_all_csv_files(CSV_FOLDER)
    old_csv_files = [f for f in csv_files if not os.path.basename(f).startswith(utc_date)]

    if old_csv_files:
        logger.info(f"Compressing {len(old_csv_files)} old CSV files.")
        await asyncio.gather(*[compress_and_remove_file(f) for f in old_csv_files])
    else:
        logger.info("No old CSV files to compress.")

# Helper function to get all CSV files
async def get_all_csv_files(base_folder):
    files = []
    try:
        async for root, dirs, filenames in async_walk(base_folder):
            for filename in filenames:
                if filename.endswith(".csv"):
                    files.append(os.path.join(root, filename))
    except Exception as e:
        logger.error(f"Error scanning directory {base_folder}: {e}")
    return files

# Asynchronous directory walk using asyncio and aiofiles
async def async_walk(top):
    for root, dirs, files in os.walk(top):
        yield root, dirs, files

# Helper function to compress a single file and remove the original
async def compress_and_remove_file(csv_path):
    gz_path = f"{csv_path}.gz"
    try:
        await compress_file_async(csv_path, gz_path)
        os.remove(csv_path)
        logger.info(f"Compressed and removed: {csv_path}")
    except Exception as e:
        logger.error(f"Failed to compress {csv_path}: {e}")

# Asynchronously compress a file using aiofiles and gzip
async def compress_file_async(csv_path, gz_path):
    try:
        async with aiofiles.open(csv_path, "rb") as f_in, aiofiles.open(gz_path, "wb") as f_out:
            # Since gzip module is not asynchronous, use run_in_executor
            await asyncio.get_running_loop().run_in_executor(
                executor, compress_file, csv_path, gz_path
            )
    except Exception as e:
        logger.error(f"Error compressing {csv_path} to {gz_path}: {e}")

def compress_file(csv_path, gz_path):
    try:
        with open(csv_path, "rb") as f_in, gzip.open(gz_path, "wb") as f_out:
            shutil.copyfileobj(f_in, f_out)
    except Exception as e:
        logger.error(f"Error compressing {csv_path}: {e}")

# Function to compress all old CSV files at startup
async def compress_old_csv_files_on_startup():
    try:
        logger.info("Starting initial compression of old CSV files.")
        await compress_old_csv_files()
        logger.info("Initial compression of old CSV files completed.")
    except Exception as e:
        logger.error(f"Error during initial compression of old CSV files: {e}")

# Function to log unknown messages
async def log_unknown_message(message):
    try:
        logger.warning(f"Unknown message received: {json.dumps(message)}")
    except Exception as e:
        logger.error(f"Error logging unknown message: {e}")

# Function to calculate folder size
def calculate_folder_size(folder_path):
    total_size = 0
    for dirpath, _, filenames in os.walk(folder_path):
        for filename in filenames:
            file_path = os.path.join(dirpath, filename)
            if os.path.isfile(file_path):
                try:
                    total_size += os.path.getsize(file_path)
                except OSError as e:
                    logger.error(f"Error accessing {file_path}: {e}")
    return total_size

# Global variable to track folder size
current_folder_size = 0

# Function to initialize folder size
async def initialize_folder_size():
    global current_folder_size
    loop = asyncio.get_running_loop()
    current_folder_size = await loop.run_in_executor(
        executor, calculate_folder_size, CSV_FOLDER
    )
    logger.info(f"Initial folder size: {current_folder_size / (1024**3):.2f} GB")

# Function to delete oldest files until space limit is under the threshold
async def enforce_folder_size_limit():
    while True:
        try:
            global current_folder_size
            if current_folder_size > CSV_FOLDER_SIZE_LIMIT:
                logger.info(
                    f"Folder size {current_folder_size / (1024**3):.2f} GB exceeds limit of {CSV_FOLDER_SIZE_LIMIT / (1024**3):.2f} GB. Initiating cleanup..."
                )

                # Get all files with their modification times
                file_paths = await get_all_files_with_mtime(CSV_FOLDER)

                # Sort files by modification time (oldest first)
                file_paths.sort(key=lambda x: x[1])

                # Delete files until the size is under the limit
                for file_path, _ in file_paths:
                    try:
                        file_size = os.path.getsize(file_path)
                        await asyncio.get_running_loop().run_in_executor(
                            executor, os.remove, file_path
                        )
                        current_folder_size -= file_size
                        logger.info(
                            f"Deleted: {file_path}. Freed {file_size / (1024**2):.2f} MB. Remaining size: {current_folder_size / (1024**3):.2f} GB."
                        )
                        if current_folder_size <= CSV_FOLDER_SIZE_LIMIT:
                            logger.info("Folder size is now within the limit.")
                            break
                    except OSError as e:
                        logger.error(f"Error deleting {file_path}: {e}")

            # Wait before next check
            await asyncio.sleep(FOLDER_SIZE_CHECK_INTERVAL)
        except Exception as e:
            logger.error(f"Error in enforcing folder size limit: {e}")
            await asyncio.sleep(60)  # Retry after 1 minute on error

# Helper function to get all files with their modification times
async def get_all_files_with_mtime(base_folder):
    files = []
    try:
        async for root, dirs, filenames in async_walk(base_folder):
            for filename in filenames:
                file_path = os.path.join(root, filename)
                if os.path.isfile(file_path):
                    try:
                        mtime = os.path.getmtime(file_path)
                        files.append((file_path, mtime))
                    except OSError as e:
                        logger.error(f"Error accessing {file_path}: {e}")
    except Exception as e:
        logger.error(f"Error scanning directory {base_folder}: {e}")
    return files

# Function to manage WebSocket connections with exponential backoff
async def connect_and_listen(channel_type):
    url = CHANNEL_URLS[channel_type]
    subscription_args = generate_subscription_args(TOPICS, channel_type)
    backoff = RECONNECT_INITIAL_BACKOFF

    while True:
        try:
            async with websockets.connect(
                url,
                ping_interval=WEBSOCKET_PING_INTERVAL,
                ping_timeout=WEBSOCKET_PING_TIMEOUT,
                close_timeout=WEBSOCKET_CLOSE_TIMEOUT,
            ) as websocket:
                logger.info(f"Connected to {channel_type} channel WebSocket.")
                backoff = RECONNECT_INITIAL_BACKOFF  # Reset backoff after successful connection

                # Subscribe to topics
                subscribe_message = {
                    "op": "subscribe",
                    "args": subscription_args,
                }
                await websocket.send(json.dumps(subscribe_message))
                logger.info(f"Subscribed to {channel_type} topics: {subscription_args}")

                # Receive subscription confirmation
                try:
                    response = await asyncio.wait_for(websocket.recv(), timeout=10)
                    logger.info(f"Subscription response for {channel_type}: {response}")
                except asyncio.TimeoutError:
                    logger.warning(f"No subscription confirmation received for {channel_type}.")

                # Start heartbeat
                heartbeat_task = asyncio.create_task(send_heartbeat(channel_type, websocket))

                # Listen for messages
                async for message in websocket:
                    try:
                        data = json.loads(message)
                        topic = data.get("topic", "unknown")

                        if topic == "unknown":
                            await log_unknown_message(data)
                        else:
                            await save_message(channel_type, topic, data)
                            # Update folder size based on message processing if needed
                            # This can be implemented as per specific requirements
                    except json.JSONDecodeError:
                        logger.warning(f"Received non-JSON message on {channel_type} channel: {message}")
                    except Exception as e:
                        logger.error(f"Error processing message on {channel_type} channel: {e}")

        except (ConnectionClosedError, ConnectionClosedOK, WebSocketException) as e:
            logger.warning(f"Connection lost on {channel_type} channel: {e}. Reconnecting in {backoff} seconds...")
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, RECONNECT_MAX_BACKOFF)
        except Exception as e:
            logger.error(f"Unexpected error on {channel_type} channel: {e}. Reconnecting in {backoff} seconds...")
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, RECONNECT_MAX_BACKOFF)

async def send_heartbeat(channel_type, websocket):
    try:
        while True:
            await asyncio.sleep(PING_INTERVAL)
            await websocket.send(json.dumps({"op": "ping"}))
            logger.info(f"Sent heartbeat for {channel_type}.")
    except (WebSocketException, asyncio.CancelledError) as e:
        logger.warning(f"Heartbeat failed for {channel_type}: {e}")
    except Exception as e:
        logger.error(f"Unexpected error in heartbeat for {channel_type}: {e}")

# === Main Function ===

async def main():
    # Initialize folder size
    await initialize_folder_size()

    # Compress old CSV files at startup
    await compress_old_csv_files_on_startup()

    # Start compression task and folder size enforcement task using TaskGroup
    async with TaskGroup() as tg:
        tg.create_task(compress_old_csv_files_periodically())
        tg.create_task(enforce_folder_size_limit())
        tg.create_task(connect_and_listen("linear"))
        tg.create_task(connect_and_listen("spot"))
        tg.create_task(connect_and_listen("inverse"))

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("WebSocket listener stopped by user.")
    except Exception as e:
        logger.error(f"Application terminated with error: {e}")


# === combine_files.py ===
import os

def combine_files(base_directory='.', additional_directory='from_cpanel_template', output_file='combined_output.txt', file_extensions=None):
    """
    Combines contents of all files with the specified extensions from:
    1. Base directory (non-recursively).
    2. Additional directory (recursively, including nested files).
    Each file's content is preceded by a comment with its filename.
    
    Parameters:
    - base_directory (str): Path to the base directory containing files to combine.
    - additional_directory (str): Path to the additional directory containing files to combine.
    - output_file (str): Path to the output file.
    - file_extensions (list of str): List of file extensions to include (e.g., ['.py', '.sh']).
    """
    if file_extensions is None:
        file_extensions = ['.py', '.sh', '.yaml', '.txt', '.json', '.log']
    
    # Normalize extensions to lowercase
    file_extensions = [ext.lower() for ext in file_extensions]

    # Helper function to combine files from a specific directory
    def process_directory(directory, outfile, recursive=False):
        directory = os.path.abspath(directory)
        if not os.path.isdir(directory):
            print(f"Error: The directory '{directory}' does not exist.")
            return
        if recursive:
            for root, _, files in os.walk(directory):
                for filename in sorted(files):
                    file_path = os.path.join(root, filename)
                    if os.path.isfile(file_path) and any(filename.lower().endswith(ext) for ext in file_extensions):
                        try:
                            with open(file_path, 'r', encoding='utf-8') as infile:
                                # Write a comment with the relative file path
                                relative_path = os.path.relpath(file_path, directory)
                                outfile.write(f"# === {relative_path} ===\n")
                                # Write the file's content
                                content = infile.read()
                                outfile.write(content + "\n\n")  # Add extra newline for separation
                            print(f"Added: {relative_path}")
                        except Exception as e:
                            print(f"Failed to read {relative_path}: {e}")
        else:
            for filename in sorted(os.listdir(directory)):
                file_path = os.path.join(directory, filename)
                if os.path.isfile(file_path) and any(filename.lower().endswith(ext) for ext in file_extensions):
                    try:
                        with open(file_path, 'r', encoding='utf-8') as infile:
                            # Write a comment with the filename
                            outfile.write(f"# === {filename} ===\n")
                            # Write the file's content
                            content = infile.read()
                            outfile.write(content + "\n\n")  # Add extra newline for separation
                        print(f"Added: {filename}")
                    except Exception as e:
                        print(f"Failed to read {filename}: {e}")

    # Open the output file in write mode
    try:
        with open(output_file, 'w', encoding='utf-8') as outfile:
            # Process files in the base directory (non-recursively)
            print(f"\nProcessing files from base directory: {base_directory}")
            process_directory(base_directory, outfile, recursive=False)

            # Process files in the additional directory (recursively)
            print(f"\nProcessing files from additional directory: {additional_directory}")
            process_directory(additional_directory, outfile, recursive=True)
    except Exception as e:
        print(f"Failed to write to output file '{output_file}': {e}")
        return

    print(f"\nAll specified files have been combined into '{output_file}'.")

if __name__ == "__main__":
    # Specify the directories and file extensions
    combine_files(
        base_directory='.',                     # Base directory (non-recursively)
        additional_directory='from_cpanel_template_not_by_me',  # Additional folder (recursively)
        output_file='combined_output.txt',      # Output file name
        file_extensions=['.py', '.sh', '.yaml', '.txt', '.json', '.log']  # File extensions to include
    )


# === combined_output.txt ===
# === bybit_flask.py ===
# bybit_flask.py

import os
from os import sep as os_sep
from flask import Flask, render_template, send_from_directory, abort, request, Response
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from functools import wraps
from urllib.parse import unquote
from datetime import datetime, timezone

# Initialize Flask app
app = Flask(__name__)

# Initialize Flask-Limiter
limiter = Limiter(
    get_remote_address,
    app=app,
    default_limits=["200 per day", "50 per hour"]
)

# === Configuration ===

# Toggle authentication
USE_AUTH = False  # Set to True to enable authentication

# Directories for storing CSV files and logs
CSV_FOLDER = "bybit_stream_data"

# Ensure the CSV folder exists
os.makedirs(CSV_FOLDER, exist_ok=True)

# === Authentication Setup ===

# Define your credentials for Basic Authentication (Used only if USE_AUTH is True)
USERNAME = os.getenv('USERNAME', 'admin')        # Change to your desired username
PASSWORD = os.getenv('PASSWORD', 'password')     # Change to your desired password

def check_auth(username, password):
    """Check if a username/password combination is valid."""
    return username == USERNAME and password == PASSWORD

def authenticate():
    """Sends a 401 response that enables basic auth."""
    return Response(
        'Could not verify your access level for that URL.\n'
        'You have to login with proper credentials', 401,
        {'WWW-Authenticate': 'Basic realm="Login Required"'}
    )

def requires_auth(f):
    """Decorator to handle authentication logic."""
    @wraps(f)
    def decorated(*args, **kwargs):
        if USE_AUTH:
            auth = request.authorization
            if not auth or not check_auth(auth.username, auth.password):
                return authenticate()
        return f(*args, **kwargs)
    return decorated

# Security: Prevent directory traversal by ensuring all paths are within CSV_FOLDER
def secure_path(path):
    # Join the CSV_FOLDER with the requested path
    base_path = os.path.abspath(CSV_FOLDER)
    safe_path = os.path.abspath(os.path.join(base_path, path))
    # Ensure the resulting path is within CSV_FOLDER
    if not safe_path.startswith(base_path):
        abort(403)  # Forbidden
    return safe_path

# === Error Handlers ===

@app.errorhandler(404)
def not_found(error):
    return render_template('error.html', message="Resource Not Found.", description="The requested resource could not be found."), 404

@app.errorhandler(403)
def forbidden(error):
    return render_template('error.html', message="Forbidden", description="You don't have permission to access this resource."), 403

@app.errorhandler(401)
def unauthorized(error):
    return render_template('error.html', message="Unauthorized", description="Please provide valid credentials to access this resource."), 401

@app.errorhandler(429)
def ratelimit_handler(error):
    return render_template('error.html', message="Rate Limit Exceeded", description="You have made too many requests. Please try again later."), 429

@app.errorhandler(500)
def internal_error(error):
    return render_template('error.html', message="Internal Server Error", description="An unexpected error has occurred. Please try again later."), 500

# === Routes ===

# Home route - lists the top-level directories
@app.route('/')
@requires_auth
@limiter.limit("10 per minute")
def index():
    return list_directory('')

# Dynamic route to list directories and files
@app.route('/browse/<path:subpath>')
@requires_auth
@limiter.limit("10 per minute")
def list_directory(subpath):
    # Decode the URL-encoded path
    subpath = unquote(subpath)
    safe_path = secure_path(subpath)

    if not os.path.exists(safe_path):
        abort(404)

    if not os.path.isdir(safe_path):
        abort(404)

    # List directories and files
    items = os.listdir(safe_path)
    dirs = []
    files = []
    for item in sorted(items):
        item_path = os.path.join(safe_path, item)
        if os.path.isdir(item_path):
            dirs.append(item)
        elif os.path.isfile(item_path) and (item.endswith('.csv') or item.endswith('.csv.gz')):
            file_stat = os.stat(item_path)
            files.append({
                'name': item,
                'size': f"{file_stat.st_size / (1024**2):.2f} MB",
                'modified': datetime.fromtimestamp(file_stat.st_mtime).strftime('%Y-%m-%d %H:%M:%S')
            })

    # Calculate the relative path for navigation
    relative_path = os.path.relpath(safe_path, os.path.abspath(CSV_FOLDER))
    if relative_path == '.':
        relative_path = ''

    return render_template(
        'directory.html',
        dirs=dirs,
        files=files,
        current_path=relative_path,
        path_separator=os_sep  # Pass the path separator to the template
    )

# Route to download files
@app.route('/download/<path:filepath>')
@requires_auth
@limiter.limit("10 per minute")
def download_file(filepath):
    # Decode the URL-encoded path
    filepath = unquote(filepath)
    safe_path = secure_path(filepath)

    if not os.path.exists(safe_path):
        abort(404)

    if not os.path.isfile(safe_path):
        abort(404)

    # Extract directory and filename
    directory = os.path.dirname(safe_path)
    filename = os.path.basename(safe_path)

    return send_from_directory(directory, filename, as_attachment=True)

# === Main Application Runner ===

if __name__ == "__main__":
    # For development purposes; in production, use passenger_wsgi.py as the entry point
    app.run(host='127.0.0.1', port=8000, debug=False)


# === bybit_websocket_listener.py ===
# bybit_websocket_listener.py

import asyncio
import json
import os
import websockets
from websockets.exceptions import (
    ConnectionClosedError,
    ConnectionClosedOK,
    WebSocketException,
)
from datetime import datetime, timezone
import gzip
import shutil
import aiofiles
from asyncio import Queue, TaskGroup
from concurrent.futures import ThreadPoolExecutor
import logging
from logging.handlers import RotatingFileHandler

# === Configuration ===

# Load configuration from config.json
CONFIG_FILE = "config.json"
if not os.path.exists(CONFIG_FILE):
    raise FileNotFoundError(f"Configuration file '{CONFIG_FILE}' not found.")

with open(CONFIG_FILE, "r") as f:
    config = json.load(f)

CSV_FOLDER = config.get("CSV_FOLDER", "bybit_stream_data")
LOGS_FOLDER = config.get("LOGS_FOLDER", "logs")
LOG_FILE_NAME = config.get("LOG_FILE_NAME", "app.log")
LOG_FILE_SIZE_LIMIT = config.get("LOG_FILE_SIZE_LIMIT_MB", 10) * 1024 * 1024  # in bytes
LOG_FILE_BACKUP_COUNT = config.get("LOG_FILE_BACKUP_COUNT", 5)
CSV_FOLDER_SIZE_LIMIT = config.get("CSV_FOLDER_SIZE_LIMIT_GB", 15) * 1024 * 1024 * 1024  # in bytes
BUFFER_LIMIT = config.get("BUFFER_LIMIT", 100)
WRITE_INTERVAL = config.get("WRITE_INTERVAL_SEC", 1)  # seconds
PING_INTERVAL = config.get("PING_INTERVAL_SEC", 20)  # seconds
PING_TIMEOUT = config.get("PING_TIMEOUT_SEC", 10)  # seconds
WEBSOCKET_PING_INTERVAL = config.get("WEBSOCKET_PING_INTERVAL_SEC", 20)  # seconds
WEBSOCKET_PING_TIMEOUT = config.get("WEBSOCKET_PING_TIMEOUT_SEC", 10)  # seconds
WEBSOCKET_CLOSE_TIMEOUT = config.get("WEBSOCKET_CLOSE_TIMEOUT_SEC", 5)  # seconds
RECONNECT_INITIAL_BACKOFF = config.get("RECONNECT_INITIAL_BACKOFF_SEC", 1)  # seconds
RECONNECT_MAX_BACKOFF = config.get("RECONNECT_MAX_BACKOFF_SEC", 60)  # seconds
COMPRESSION_SLEEP_INTERVAL = config.get("COMPRESSION_SLEEP_INTERVAL_SEC", 86400)  # 24 hours
FOLDER_SIZE_CHECK_INTERVAL = config.get("FOLDER_SIZE_CHECK_INTERVAL_SEC", 600)  # 10 minutes
LOG_INTERVAL = config.get("LOG_INTERVAL", 100)  # Log every 100 messages

os.makedirs(CSV_FOLDER, exist_ok=True)
os.makedirs(LOGS_FOLDER, exist_ok=True)

LOG_FILE_PATH = os.path.join(LOGS_FOLDER, LOG_FILE_NAME)

# Setup logging with RotatingFileHandler
logger = logging.getLogger("BybitWebSocketListener")
logger.setLevel(logging.INFO)
handler = RotatingFileHandler(
    LOG_FILE_PATH,
    maxBytes=LOG_FILE_SIZE_LIMIT,
    backupCount=LOG_FILE_BACKUP_COUNT,
)
formatter = logging.Formatter(
    "%(asctime)s - %(levelname)s - %(message)s"
)
handler.setFormatter(formatter)
logger.addHandler(handler)

# Initialize ThreadPoolExecutor for blocking I/O
executor = ThreadPoolExecutor()

# Load topics from symbols.json
SYMBOLS_FILE = "symbols.json"
if not os.path.exists(SYMBOLS_FILE):
    raise FileNotFoundError(f"Symbols file '{SYMBOLS_FILE}' not found.")

with open(SYMBOLS_FILE, "r") as f:
    TOPICS = json.load(f)

# WebSocket URLs for each channel_type
CHANNEL_URLS = {
    "linear": "wss://stream.bybit.com/v5/public/linear",
    "spot": "wss://stream.bybit.com/v5/public/spot",
    "options": "wss://stream.bybit.com/v5/public/option",
    "inverse": "wss://stream.bybit.com/v5/public/inverse",
}

# === Functions ===

# Flatten topics for a specific channel type
def generate_subscription_args(topics_json, channel_type):
    args = []
    if channel_type in topics_json:
        for symbol, topic_list in topics_json[channel_type].items():
            for topic in topic_list:
                args.append(f"{topic}.{symbol}")
    return args

# Function to get the current UTC date in YYYY-MM-DD format
def get_current_utc_date():
    return datetime.now(timezone.utc).strftime("%Y-%m-%d")

# Dictionary to hold message queues
message_queues = {}

# Counter for throttled logging
log_counter = 0

# Function to save messages to CSV in structured subdirectories
async def save_message(channel_type, topic, message):
    key = (channel_type, topic)
    if key not in message_queues:
        message_queues[key] = Queue()
        asyncio.create_task(process_queue(channel_type, topic, message_queues[key]))
    await message_queues[key].put(message)

async def process_queue(channel_type, topic, queue):
    global log_counter
    while True:
        try:
            utc_date = get_current_utc_date()

            # Extract the symbol from the topic
            # Example: "orderbook.1.BTCUSDT" -> "BTCUSDT"
            symbol = topic.split(".")[-1]
            topic_base = ".".join(topic.split(".")[:-1])

            channel_folder = os.path.join(CSV_FOLDER, channel_type)
            symbol_folder = os.path.join(channel_folder, symbol)  # New layer for symbol
            topic_folder = os.path.join(symbol_folder, topic_base)  # Topic folder under symbol

            # Ensure the directories exist
            os.makedirs(topic_folder, exist_ok=True)

            # Define the CSV file path
            csv_file = os.path.join(topic_folder, f"{utc_date}_{topic}.csv")

            buffer = []
            write_interval = WRITE_INTERVAL  # Seconds

            while True:
                try:
                    message = await asyncio.wait_for(queue.get(), timeout=write_interval)
                    buffer.append(json.dumps(message))
                    if len(buffer) >= BUFFER_LIMIT:
                        await write_buffer_to_file(csv_file, buffer)
                        buffer.clear()
                        log_counter += len(buffer)
                        if log_counter >= LOG_INTERVAL:
                            logger.info(f"Written {log_counter} messages to {csv_file}")
                            log_counter = 0
                except asyncio.TimeoutError:
                    if buffer:
                        await write_buffer_to_file(csv_file, buffer)
                        log_counter += len(buffer)
                        if log_counter >= LOG_INTERVAL:
                            logger.info(f"Written {log_counter} messages to {csv_file}")
                            log_counter = 0
                        buffer.clear()
        except Exception as e:
            logger.error(f"Error in processing queue for {channel_type}.{topic}: {e}")
            await asyncio.sleep(1)  # Prevent tight loop on error

async def write_buffer_to_file(csv_file, buffer):
    try:
        async with aiofiles.open(csv_file, mode="a") as file:
            await file.write("\n".join(buffer) + "\n")
    except Exception as e:
        logger.error(f"Failed to write buffer to {csv_file}: {e}")

# Function to compress old CSV files to .csv.gz
async def compress_old_csv_files_periodically():
    while True:
        try:
            await compress_old_csv_files()
            await asyncio.sleep(COMPRESSION_SLEEP_INTERVAL)
        except Exception as e:
            logger.error(f"Error in compression task: {e}")
            await asyncio.sleep(60)  # Retry after a minute on error

# Function to compress all old CSV files
async def compress_old_csv_files():
    utc_date = get_current_utc_date()
    csv_files = await get_all_csv_files(CSV_FOLDER)
    old_csv_files = [f for f in csv_files if not os.path.basename(f).startswith(utc_date)]

    if old_csv_files:
        logger.info(f"Compressing {len(old_csv_files)} old CSV files.")
        await asyncio.gather(*[compress_and_remove_file(f) for f in old_csv_files])
    else:
        logger.info("No old CSV files to compress.")

# Helper function to get all CSV files
async def get_all_csv_files(base_folder):
    files = []
    try:
        async for root, dirs, filenames in async_walk(base_folder):
            for filename in filenames:
                if filename.endswith(".csv"):
                    files.append(os.path.join(root, filename))
    except Exception as e:
        logger.error(f"Error scanning directory {base_folder}: {e}")
    return files

# Asynchronous directory walk using asyncio and aiofiles
async def async_walk(top):
    for root, dirs, files in os.walk(top):
        yield root, dirs, files

# Helper function to compress a single file and remove the original
async def compress_and_remove_file(csv_path):
    gz_path = f"{csv_path}.gz"
    try:
        await compress_file_async(csv_path, gz_path)
        os.remove(csv_path)
        logger.info(f"Compressed and removed: {csv_path}")
    except Exception as e:
        logger.error(f"Failed to compress {csv_path}: {e}")

# Asynchronously compress a file using aiofiles and gzip
async def compress_file_async(csv_path, gz_path):
    try:
        async with aiofiles.open(csv_path, "rb") as f_in, aiofiles.open(gz_path, "wb") as f_out:
            # Since gzip module is not asynchronous, use run_in_executor
            await asyncio.get_running_loop().run_in_executor(
                executor, compress_file, csv_path, gz_path
            )
    except Exception as e:
        logger.error(f"Error compressing {csv_path} to {gz_path}: {e}")

def compress_file(csv_path, gz_path):
    try:
        with open(csv_path, "rb") as f_in, gzip.open(gz_path, "wb") as f_out:
            shutil.copyfileobj(f_in, f_out)
    except Exception as e:
        logger.error(f"Error compressing {csv_path}: {e}")

# Function to compress all old CSV files at startup
async def compress_old_csv_files_on_startup():
    try:
        logger.info("Starting initial compression of old CSV files.")
        await compress_old_csv_files()
        logger.info("Initial compression of old CSV files completed.")
    except Exception as e:
        logger.error(f"Error during initial compression of old CSV files: {e}")

# Function to log unknown messages
async def log_unknown_message(message):
    try:
        logger.warning(f"Unknown message received: {json.dumps(message)}")
    except Exception as e:
        logger.error(f"Error logging unknown message: {e}")

# Function to calculate folder size
def calculate_folder_size(folder_path):
    total_size = 0
    for dirpath, _, filenames in os.walk(folder_path):
        for filename in filenames:
            file_path = os.path.join(dirpath, filename)
            if os.path.isfile(file_path):
                try:
                    total_size += os.path.getsize(file_path)
                except OSError as e:
                    logger.error(f"Error accessing {file_path}: {e}")
    return total_size

# Global variable to track folder size
current_folder_size = 0

# Function to initialize folder size
async def initialize_folder_size():
    global current_folder_size
    loop = asyncio.get_running_loop()
    current_folder_size = await loop.run_in_executor(
        executor, calculate_folder_size, CSV_FOLDER
    )
    logger.info(f"Initial folder size: {current_folder_size / (1024**3):.2f} GB")

# Function to delete oldest files until space limit is under the threshold
async def enforce_folder_size_limit():
    while True:
        try:
            global current_folder_size
            if current_folder_size > CSV_FOLDER_SIZE_LIMIT:
                logger.info(
                    f"Folder size {current_folder_size / (1024**3):.2f} GB exceeds limit of {CSV_FOLDER_SIZE_LIMIT / (1024**3):.2f} GB. Initiating cleanup..."
                )

                # Get all files with their modification times
                file_paths = await get_all_files_with_mtime(CSV_FOLDER)

                # Sort files by modification time (oldest first)
                file_paths.sort(key=lambda x: x[1])

                # Delete files until the size is under the limit
                for file_path, _ in file_paths:
                    try:
                        file_size = os.path.getsize(file_path)
                        await asyncio.get_running_loop().run_in_executor(
                            executor, os.remove, file_path
                        )
                        current_folder_size -= file_size
                        logger.info(
                            f"Deleted: {file_path}. Freed {file_size / (1024**2):.2f} MB. Remaining size: {current_folder_size / (1024**3):.2f} GB."
                        )
                        if current_folder_size <= CSV_FOLDER_SIZE_LIMIT:
                            logger.info("Folder size is now within the limit.")
                            break
                    except OSError as e:
                        logger.error(f"Error deleting {file_path}: {e}")

            # Wait before next check
            await asyncio.sleep(FOLDER_SIZE_CHECK_INTERVAL)
        except Exception as e:
            logger.error(f"Error in enforcing folder size limit: {e}")
            await asyncio.sleep(60)  # Retry after 1 minute on error

# Helper function to get all files with their modification times
async def get_all_files_with_mtime(base_folder):
    files = []
    try:
        async for root, dirs, filenames in async_walk(base_folder):
            for filename in filenames:
                file_path = os.path.join(root, filename)
                if os.path.isfile(file_path):
                    try:
                        mtime = os.path.getmtime(file_path)
                        files.append((file_path, mtime))
                    except OSError as e:
                        logger.error(f"Error accessing {file_path}: {e}")
    except Exception as e:
        logger.error(f"Error scanning directory {base_folder}: {e}")
    return files

# Function to manage WebSocket connections with exponential backoff
async def connect_and_listen(channel_type):
    url = CHANNEL_URLS[channel_type]
    subscription_args = generate_subscription_args(TOPICS, channel_type)
    backoff = RECONNECT_INITIAL_BACKOFF

    while True:
        try:
            async with websockets.connect(
                url,
                ping_interval=WEBSOCKET_PING_INTERVAL,
                ping_timeout=WEBSOCKET_PING_TIMEOUT,
                close_timeout=WEBSOCKET_CLOSE_TIMEOUT,
            ) as websocket:
                logger.info(f"Connected to {channel_type} channel WebSocket.")
                backoff = RECONNECT_INITIAL_BACKOFF  # Reset backoff after successful connection

                # Subscribe to topics
                subscribe_message = {
                    "op": "subscribe",
                    "args": subscription_args,
                }
                await websocket.send(json.dumps(subscribe_message))
                logger.info(f"Subscribed to {channel_type} topics: {subscription_args}")

                # Receive subscription confirmation
                try:
                    response = await asyncio.wait_for(websocket.recv(), timeout=10)
                    logger.info(f"Subscription response for {channel_type}: {response}")
                except asyncio.TimeoutError:
                    logger.warning(f"No subscription confirmation received for {channel_type}.")

                # Start heartbeat
                heartbeat_task = asyncio.create_task(send_heartbeat(channel_type, websocket))

                # Listen for messages
                async for message in websocket:
                    try:
                        data = json.loads(message)
                        topic = data.get("topic", "unknown")

                        if topic == "unknown":
                            await log_unknown_message(data)
                        else:
                            await save_message(channel_type, topic, data)
                            # Update folder size based on message processing if needed
                            # This can be implemented as per specific requirements
                    except json.JSONDecodeError:
                        logger.warning(f"Received non-JSON message on {channel_type} channel: {message}")
                    except Exception as e:
                        logger.error(f"Error processing message on {channel_type} channel: {e}")

        except (ConnectionClosedError, ConnectionClosedOK, WebSocketException) as e:
            logger.warning(f"Connection lost on {channel_type} channel: {e}. Reconnecting in {backoff} seconds...")
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, RECONNECT_MAX_BACKOFF)
        except Exception as e:
            logger.error(f"Unexpected error on {channel_type} channel: {e}. Reconnecting in {backoff} seconds...")
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, RECONNECT_MAX_BACKOFF)

async def send_heartbeat(channel_type, websocket):
    try:
        while True:
            await asyncio.sleep(PING_INTERVAL)
            await websocket.send(json.dumps({"op": "ping"}))
            logger.info(f"Sent heartbeat for {channel_type}.")
    except (WebSocketException, asyncio.CancelledError) as e:
        logger.warning(f"Heartbeat failed for {channel_type}: {e}")
    except Exception as e:
        logger.error(f"Unexpected error in heartbeat for {channel_type}: {e}")

# === Main Function ===

async def main():
    # Initialize folder size
    await initialize_folder_size()

    # Compress old CSV files at startup
    await compress_old_csv_files_on_startup()

    # Start compression task and folder size enforcement task using TaskGroup
    async with TaskGroup() as tg:
        tg.create_task(compress_old_csv_files_periodically())
        tg.create_task(enforce_folder_size_limit())
        tg.create_task(connect_and_listen("linear"))
        tg.create_task(connect_and_listen("spot"))
        tg.create_task(connect_and_listen("inverse"))

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("WebSocket listener stopped by user.")
    except Exception as e:
        logger.error(f"Application terminated with error: {e}")




# === config.json ===
{
    "CSV_FOLDER": "bybit_stream_data",
    "LOGS_FOLDER": "logs",
    "LOG_FILE_NAME": "app.log",
    "LOG_FILE_SIZE_LIMIT_MB": 10,
    "LOG_FILE_BACKUP_COUNT": 5,
    "CSV_FOLDER_SIZE_LIMIT_GB": 15,
    "BUFFER_LIMIT": 100,
    "WRITE_INTERVAL_SEC": 1,
    "PING_INTERVAL_SEC": 20,
    "PING_TIMEOUT_SEC": 10,
    "WEBSOCKET_PING_INTERVAL_SEC": 20,
    "WEBSOCKET_PING_TIMEOUT_SEC": 10,
    "WEBSOCKET_CLOSE_TIMEOUT_SEC": 5,
    "RECONNECT_INITIAL_BACKOFF_SEC": 1,
    "RECONNECT_MAX_BACKOFF_SEC": 60,
    "COMPRESSION_SLEEP_INTERVAL_SEC": 86400,
    "FOLDER_SIZE_CHECK_INTERVAL_SEC": 600,
    "LOG_INTERVAL": 100
}


# === config.yaml ===
# config.yaml

# Logging Settings
log_file: "./logs/listener.log"

# Trigger Settings
trigger_file: "./save_now.trigger"  # Path to the trigger file for on-demand saving

# Scheduling Settings
save_time_utc: "00:00"  # Time to trigger daily save in UTC (HH:MM format)


# === passenger_wsgi.py ===
import importlib.util
import os
import sys

# Add the project directory to the Python path
sys.path.insert(0, os.path.dirname(__file__))

# Specify the path to the Flask application
app_path = os.path.join(os.path.dirname(__file__), 'bybit_flask.py')

# Load the Flask application module
spec = importlib.util.spec_from_file_location('wsgi', app_path)
wsgi = importlib.util.module_from_spec(spec)
spec.loader.exec_module(wsgi)

# Assign the Flask app to 'application' for WSGI
application = wsgi.app


# === requirements.txt ===
websocket-client>=1.6.1
schedule>=1.2.0
pyyaml>=6.0
pytz>=2023.3
apscheduler>=3.11.0
websockets>=11.0
aiofiles>=23.1.0
Flask>=2.0
Flask-Limiter>=2.5.4
python-dotenv>=1.0.0


# === restart_listener.sh ===
#!/bin/bash

# Function to print messages with timestamp
echo_msg() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1"
}

echo_msg "Restarting Flask web application and WebSocket Listener..."

# Stop existing services
./stop_listener.sh

# Start services
./start_listener.sh

echo_msg "All services restarted successfully."


# === setup.sh ===
#!/bin/bash

# Exit immediately if a command exits with a non-zero status
set -e

# Function to print messages with timestamp
echo_msg() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1"
}

echo_msg "Starting setup..."

# Determine the directory where the script is located
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
PROJECT_DIR="$SCRIPT_DIR"

# Define other directories based on PROJECT_DIR
VENV_DIR="$PROJECT_DIR/env"
LOG_DIR="$PROJECT_DIR/logs"
LISTENER_LOG="$LOG_DIR/listener.log"
GUNICORN_LOG="$LOG_DIR/gunicorn.log"
STORAGE_PATH="$PROJECT_DIR/storage"

# Create necessary directories with restrictive permissions
echo_msg "Creating directories..."
mkdir -p "$STORAGE_PATH" && chmod 700 "$STORAGE_PATH"
mkdir -p "$LOG_DIR" && chmod 700 "$LOG_DIR"

# Navigate to the project directory
cd "$PROJECT_DIR"

# Create virtual environment if it doesn't exist
if [ ! -d "$VENV_DIR" ]; then
    echo_msg "Creating Python virtual environment..."
    python3 -m venv "$VENV_DIR"
    echo_msg "Virtual environment created."
else
    echo_msg "Virtual environment already exists."
fi

# Activate virtual environment
echo_msg "Activating virtual environment..."
source "$VENV_DIR/bin/activate"

# Upgrade pip, setuptools, and wheel
echo_msg "Upgrading pip, setuptools, and wheel..."
pip install --upgrade pip setuptools wheel

# Install remaining dependencies
echo_msg "Installing dependencies..."
pip install -r "$PROJECT_DIR/requirements.txt"

# Make start and restart scripts executable
echo_msg "Setting permissions for scripts..."
chmod 700 "$PROJECT_DIR/start_listener.sh"
chmod 700 "$PROJECT_DIR/restart_listener.sh"
chmod 700 "$PROJECT_DIR/stop_listener.sh"

echo_msg "Setup completed successfully."


# === start_listener.sh ===
#!/bin/bash

# Variables
SCRIPT="bybit_websocket_listener.py"  # WebSocket listener script
FLASK_APP_MODULE="bybit_flask:app"    # Gunicorn requires module:app
VENV_DIR="$(pwd)/env"                 # Path to the virtual environment
PROJECT_DIR="$(pwd)"                  # Path to the project directory
LISTENER_LOG="./logs/listener.log"
GUNICORN_LOG="./logs/gunicorn.log"

# Function to print messages with timestamp
echo_msg() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1"
}

if [ "$PRODUCTION" = "true" ]; then
    echo_msg "Production mode detected. Starting Flask app with Passenger (cPanel managed)."
    # In production, Passenger handles the Flask app. Ensure cPanel is configured correctly.
    # Start the WebSocket Listener if required and allowed by cPanel.
    # Note: cPanel might have restrictions on running persistent background processes.
    
    # Example: Starting WebSocket Listener via a background process
    # Adjust as per cPanel's capabilities and policies.
    echo_msg "Starting Bybit WebSocket Listener..."
    nohup python "$PROJECT_DIR/$SCRIPT" > "$LISTENER_LOG" 2>&1 &
    LISTENER_PID=$!
    echo_msg "WebSocket Listener started with PID $LISTENER_PID."
else
    echo_msg "Local mode detected. Starting Flask web application with Gunicorn."

    # Activate virtual environment
    source "$VENV_DIR/bin/activate"

    # Start Gunicorn in the background
    nohup gunicorn -w 4 -b 127.0.0.1:8000 "$FLASK_APP_MODULE" > "$GUNICORN_LOG" 2>&1 &
    GUNICORN_PID=$!
    echo_msg "Gunicorn started with PID $GUNICORN_PID."

    echo_msg "Starting Bybit WebSocket Listener..."

    # Start the WebSocket Listener in the background
    nohup python "$PROJECT_DIR/$SCRIPT" > "$LISTENER_LOG" 2>&1 &
    LISTENER_PID=$!
    echo_msg "WebSocket Listener started with PID $LISTENER_PID."

    echo_msg "All services started successfully."
fi


# === stop_listener.sh ===
#!/bin/bash

# Function to print messages with timestamp
echo_msg() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1"
}

echo_msg "Stopping Flask web application and WebSocket Listener..."

if [ "$PRODUCTION" = "true" ]; then
    echo_msg "Production mode detected. Stopping WebSocket Listener if running."

    # Find and kill WebSocket listener processes
    LISTENER_PIDS=$(pgrep -f "bybit_websocket_listener.py")
    if [ -n "$LISTENER_PIDS" ]; then
        kill $LISTENER_PIDS
        echo_msg "WebSocket Listener processes (PIDs: $LISTENER_PIDS) stopped."
    else
        echo_msg "No WebSocket Listener processes found."
    fi
else
    echo_msg "Local mode detected. Stopping Gunicorn and WebSocket Listener."

    # Find and kill Gunicorn processes
    GUNICORN_PIDS=$(pgrep -f "gunicorn")
    if [ -n "$GUNICORN_PIDS" ]; then
        kill $GUNICORN_PIDS
        echo_msg "Gunicorn processes (PIDs: $GUNICORN_PIDS) stopped."
    else
        echo_msg "No Gunicorn processes found."
    fi

    # Find and kill WebSocket listener processes
    LISTENER_PIDS=$(pgrep -f "bybit_websocket_listener.py")
    if [ -n "$LISTENER_PIDS" ]; then
        kill $LISTENER_PIDS
        echo_msg "WebSocket Listener processes (PIDs: $LISTENER_PIDS) stopped."
    else
        echo_msg "No WebSocket Listener processes found."
    fi
fi

echo_msg "All services stopped successfully."


# === symbols.json ===
{
    "linear": {
        "BTCUSDT": [
            "orderbook.500",
            "publicTrade",
            "liquidation",
            "tickers"
        ],
        "ETHUSDT": [
            "orderbook.500",
            "publicTrade",
            "liquidation",
            "tickers"
        ]
    },
    "spot": {
        "BTCUSDT": [
            "orderbook.200",
            "tickers",
            "publicTrade"
        ]
    },
    "inverse": {
        "BTCUSD": [
            "orderbook.500",
            "publicTrade",
            "kline.1",
            "liquidation"
        ]
    }
}


# === bybit_listener.py ===
import os
import sys


sys.path.insert(0, os.path.dirname(__file__))


def app(environ, start_response):
    start_response('200 OK', [('Content-Type', 'text/plain')])
    message = 'It works!\n'
    version = 'Python v' + sys.version.split()[0] + '\n'
    response = '\n'.join([message, version])
    return [response.encode()]


# === passenger_wsgi.py ===
import importlib.util
import os
import sys

# Add the project directory to the Python path
sys.path.insert(0, os.path.dirname(__file__))

# Specify the path to the Flask application
app_path = os.path.join(os.path.dirname(__file__), 'bybit_flask.py')

# Load the Flask application module
spec = importlib.util.spec_from_file_location('wsgi', app_path)
wsgi = importlib.util.module_from_spec(spec)
spec.loader.exec_module(wsgi)

# Assign the Flask app to 'application' for WSGI
application = wsgi.app


# === logs/passenger.log ===
App 1235087 output: Traceback (most recent call last):
App 1235087 output:   File "/opt/cpanel/ea-ruby27/root/usr/share/passenger/helper-scripts/wsgi-loader.py", line 384, in <module>
App 1235087 output:     app_module = load_app()
App 1235087 output:                  ^
App 1235087 output: ^^^^^^^^^
App 1235087 output:   File "/opt/cpanel/ea-ruby27/root/usr/share/passenger/helper-scripts/wsgi-loader.py", line 88, in load_app
App 1235087 output:     spec.loader.exec_module(app_module)
App 1235087 output:   File "<frozen importlib._bootstrap_external>", line 995, in exec_module
App 1235087 output:   File "<frozen importlib._bootstrap>", line 488, in _call_with_frames_removed
App 1235087 output:   File "/home/osgawcom/ichhaberecht.com/bybit_stream_data/passenger_wsgi.py", line 1, in <module>
App 1235087 output:     import imp
App 1235087 output: ModuleNotFoundError: No module named 'imp'


# === tmp/restart.txt ===