import asyncio from ipaddress import IPv4Address import mysql.connector from mysql.connector import Error import json import requests import time import calendar import os import logging import traceback from datetime import datetime, timedelta import pytz class LoggerManager: def __init__(self, log_directory, delete_log_days=20, log_name="Analyitcs"): self.log_directory = log_directory self.delete_log_days = delete_log_days self.log_name = log_name self.logger = None self.setup_logging() self.remove_old_logs() def setup_logging(self): if not os.path.exists(self.log_directory): os.makedirs(self.log_directory) log_file_path = os.path.join(self.log_directory, f'{self.log_name}_{datetime.now().strftime("%Y-%m-%d")}.log') log_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') file_handler = logging.FileHandler(log_file_path) file_handler.setFormatter(log_formatter) self.logger = logging.getLogger(self.log_name) self.logger.addHandler(file_handler) self.logger.setLevel(logging.INFO) def remove_old_logs(self): cutoff_date = datetime.now() - timedelta(days=self.delete_log_days) for filename in os.listdir(self.log_directory): if filename.endswith('.log'): file_path = os.path.join(self.log_directory, filename) file_mod_time = datetime.fromtimestamp(os.path.getmtime(file_path)) if file_mod_time < cutoff_date: os.remove(file_path) self.logger.info(f"Removed old log file: {filename}") self.logger.info("Log cleanup complete.") def get_logger(self): return self.logger class Database: def __init__(self, host, user, password, database, port): self.host = host self.user = user self.password = password self.database = database self.port = port self.db_conn = None self.db_cursor = None def __enter__(self): try: self.db_conn = mysql.connector.connect( host=self.host, user=self.user, password=self.password, database=self.database, port=self.port ) if self.db_conn.is_connected(): print("Database connection successful") self.db_cursor = self.db_conn.cursor(buffered=True, dictionary=True) except Error as e: print(f"Error connecting to MySQL: {e}") self.db_conn = None return self def execute_query(self, query, params=None): if self.db_conn is None: print("No database connection.") return None try: self.db_cursor.execute(query, params or ()) return self.db_cursor.fetchall() except Error as e: print(f"Error executing query: {e}") return None def execute_update(self, query, params=None): if self.db_conn is None: print("No database connection.") return False try: self.db_cursor.execute(query, params or ()) self.db_conn.commit() return True except Error as e: traceback.print_exc() print(f"Error executing update: {e}") return False def __exit__(self, exc_type, exc_value, traceback): if self.db_cursor: self.db_cursor.close() if self.db_conn and self.db_conn.is_connected(): self.db_conn.close() print("Database connection closed.") class Internetdata: def __init__(self): self.host = 'ovpn-centrealised-server.cl66yeq0k0b7.us-east-2.rds.amazonaws.com' self.user = 'xeg3ua7ooqu1' self.password = 'ahv3eigee2zo' self.database = 'xeg3ua7ooqu12' self.port = 3306 # self.host = '13.200.64.163' # self.user = 'root' # self.password = 'eternal123' # self.database = 'ovpnnew' # self.port = 5306 self._working_directory = os.path.dirname(os.path.abspath(__file__)) self._log_directory = f'{self._working_directory}/analyitcs' self.logger_manager = LoggerManager(self._log_directory, 10, "Analyitcs") self.logger = self.logger_manager.get_logger() self.total_conncetd_data = 0 def insertdata(self): try: with Database(host=self.host, user=self.user, password=self.password, database=self.database, port=self.port) as db: result = db.execute_query("SELECT username, user_device_id, remote_ip, connected_since, created_at, current_bytes_recv, current_bytes_sent, current_total_bytes FROM ovpn_server_logs WHERE is_latest = 1") if result: insert_query = """ INSERT INTO still_connected_files_usage (filename, user_device_id, remote_ip, connected_since, file_creation_date, current_bytes_recv, current_bytes_sent, current_total_bytes, created_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) """ for row in result: dt = time.gmtime() ts = calendar.timegm(dt) data_with_timestamp = ( row['username'], row['user_device_id'], row['remote_ip'], row['connected_since'], row['created_at'], row['current_bytes_recv'], row['current_bytes_sent'], row['current_total_bytes'], ts ) self.logger.info(f'data_with_timestamp : {data_with_timestamp}') self.total_conncetd_data += int(row['current_total_bytes']) db.execute_update(insert_query, data_with_timestamp) disconnected_data = db.execute_query(""" SELECT SUM(current_bytes_recv) AS total_recv, SUM(current_bytes_sent) AS total_sent, SUM(current_total_bytes) AS total_bytes , COUNT(DISTINCT filename) AS unique_filenames FROM disconnected_files_usage WHERE DATE(FROM_UNIXTIME(created_at)) = CURDATE() """) if disconnected_data: insert_analytics_query = """ INSERT INTO internet_usage_analytics (total_bytes_recv, total_bytes_sent, total_bytes, connected_files_usage, active_user_count, created_at) VALUES (%s, %s, %s, %s, %s, %s) """ print('disconnected_data',disconnected_data) for row in disconnected_data: total_recv = (row['total_recv'] or 0) / (1024**3) total_sent = (row['total_sent'] or 0) / (1024**3) total_bytes = (row['total_bytes'] or 0) / (1024**3) connected_files_usage_gb = self.total_conncetd_data / (1024**3) dt = time.gmtime() ts = calendar.timegm(dt) current_timestamp = int(datetime.now().timestamp()) analytics_data_with_timestamp = ( round(total_recv, 2), round(total_sent, 2), round(total_bytes, 2), round(connected_files_usage_gb, 2), row['unique_filenames'], ts ) self.logger.info(f'analytics_data_with_timestamp : {analytics_data_with_timestamp}') db.execute_update(insert_analytics_query, analytics_data_with_timestamp) db.execute_update("TRUNCATE TABLE disconnected_files_usage") except Exception as e: print(e) self.logger.error(f"Error in insertdata: {e}") traceback.print_exc() if __name__ == "__main__": internet_data = Internetdata() internet_data.insertdata()