from calendar import Calendar, calendar import csv import base64 from itertools import chain, count from random import Random, randint, random import time import requests import os import calendar import datetime import runpy import time from collections import Counter import os.path from datetime import datetime, timedelta, date from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart import traceback import logging from config import get_database_config, get_config, DatabaseConnection from sendmail import sendExceptionMail from typing import List, Tuple, Set class LoggerManager: def __init__(self): self.working_dir = os.path.dirname(os.path.abspath(__file__)) self.logs_dir = os.path.join(self.working_dir, 'kr_temp_server_logs') os.makedirs(self.logs_dir, exist_ok=True) def init_logging(self): """Initialize logging to a daily rotating log file.""" log_file = os.path.join(self.logs_dir, f"kr_temp_{datetime.now():%Y-%m-%d}.log") logging.basicConfig( filename=log_file, format='%(asctime)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S', level=logging.INFO ) print("Log file initialized:", log_file) class TemporaryServer: def __init__(self): self.CSV_URL = "http://www.vpngate.net/api/iphone/" self.environment_selected, self.config = get_config() self.db_config = get_database_config(self.config) self.db = DatabaseConnection(self.db_config) # tables name self.server_ip = self.config["server_ip"] self.ovpn_path = self.config['ovpn_main_file_path'] self.live_ovpn_table_name = self.config["ovpn_table_name"] self.live_country_table_name = self.config["country_table_name"] self.decode_vpn_path = "decodeOvpn/" # Create logs dir if it doesn't exist #script details self.abs_path = os.path.abspath(__file__) self.dir_path = os.path.dirname(os.path.abspath(__file__)) print('abs_path',self.abs_path) self.full_script_name = self.environment_selected.upper() + " " + os.path.basename(self.abs_path) #logger logger = LoggerManager() logger.init_logging() logging.info("Logging is now ready!") self.start_time = datetime.now() logging.info(f"Script start time is : {self.start_time}") self.end_time = None self.processed_countries = set() self.processed_servers = set() self.delete_condition = "ping_count < -2" print(self.environment_selected) #print(self.db_config) def set_last_script_run_time(self): filepath = os.path.join(self.dir_path, "kr_temporary_last_run.txt") try: timestamp = calendar.timegm(time.gmtime()) with open(filepath, "w") as f: f.write(str(timestamp)) print("Script time updated!") except Exception as e: print(f"Error updating script time: {e}") def fetch_csv_data(self,csv_url): try: response = requests.get(csv_url) response.raise_for_status() # Ensure the request was successful decoded_content = response.content.decode("utf-8") reader = csv.reader(decoded_content.splitlines(), delimiter=",") rows = list(reader) except requests.exceptions.RequestException as e: print(f"Error fetching CSV data: {e}") logging.error(f"Error fetching CSV data: {e}") return [] except Exception as e: print(f"Error processing CSV data: {e}") logging.error(f"Error processing CSV data: {e}") return [] return rows @staticmethod def get_current_utc_timestamp(): """Return current UTC timestamp as an integer.""" return calendar.timegm(time.gmtime()) def get_all_countries_servers(self): try: query_country = f"SELECT country_id,country_code FROM {self.live_country_table_name}" query_servers = f"SELECT ip_address FROM {self.live_ovpn_table_name}" # Fetch servers self.db.execute_query(query_servers) servers = {row['ip_address'] for row in self.db.fetch_all()} # Fetch countries self.db.execute_query(query_country) # countries = {row['country_id'] for row in self.db.fetch_all()} rows = self.db.fetch_all() countries_dict = {row['country_code']: row['country_id'] for row in rows} countries = {row['country_code'] for row in rows} return countries_dict, servers, countries except Exception as e: print(f"Error fetching countries or servers: {e}") logging.error(f"Error fetching countries or servers: {e}") return set(), set() def bulk_insert_if_not_exists(self, table_name: str, columns: List[str], values: List[Tuple],all_country_codes: Set): try: # Filter out countries that already exist in the database filtered_values = [val for val in values if val[1] not in all_country_codes] # print('filtered_values',filtered_values) if not filtered_values: print("No new records to insert.") return placeholders = ', '.join(['%s'] * len(columns)) columns_str = ', '.join(columns) query = f"INSERT INTO {table_name} ({columns_str}) VALUES ({placeholders})" self.db.cursor.executemany(query, filtered_values) self.db.connection.commit() inserted_count = self.db.cursor.rowcount print(f"Inserted {inserted_count} new records into `{table_name}`.") except Exception as e: if "Duplicate entry" in str(e): print(f"Some entries already exist, skipping") else: print(f"Error in bulk insert into `{table_name}`: {e}") def clean_ovpn_file(self, contrydata): ovpn_file = base64.b64decode(contrydata) mydata = ovpn_file.decode('utf-8').splitlines() cleaned_ovpn_lines = [] for data in mydata: if data.startswith("#") or data.startswith(" ") or data.strip() == "": continue cleaned_ovpn_lines.append(data) cleaned_ovpn = "\n".join(cleaned_ovpn_lines).strip() return cleaned_ovpn def delete_low_ping_count_file(self): try: select_ping_query = ( f"SELECT * FROM {self.live_ovpn_table_name} WHERE {self.delete_condition}" ) self.db.cursor.execute(select_ping_query) selected_data = self.db.fetch_all() for data in selected_data: server_id = data["server_id"] ip_address = data["ip_address"] # file_path = self.ovpn_path + str(server_id) + "_" + ip_address + ".ovpn" file_path = self.ovpn_path + ip_address + ".ovpn" if os.path.isfile(file_path): os.remove(file_path) print(f"Removed: {file_path}") self.db.connection.commit() delete_query = f"DELETE FROM {self.live_ovpn_table_name} WHERE {self.delete_condition}" self.db.execute_query(delete_query) self.db.connection.commit() except Exception as e: print(f"Error deleting low ping count files: {e}") # logging.error(f"Error deleting low ping count files: {e}") def insert_country_server(self): self.db.connect() try: # Fetch CSV data csv_data = self.fetch_csv_data(self.CSV_URL) all_country,all_servers,all_country_code = self.get_all_countries_servers() print('all_country',type(all_country)) print('all_servers',len(all_servers)) bulk_countries = [] bulk_servers = [] for contrydata in csv_data[2:-1]: # print('contrydata',contrydata[1]) print(f"Processing country data: {contrydata[1]}") # host = [contrydata[0], contrydata[1]] ip_address = contrydata[1] countryname = contrydata[5] countrycode = contrydata[6] ts = self.get_current_utc_timestamp() ovpn_file = self.clean_ovpn_file(contrydata[14]) # print('ovpn_file',ovpn_file) if countryname not in self.processed_countries: bulk_countries.append((countryname, countrycode, ts, ts, 1, 0)) self.processed_countries.add(countryname) if ip_address not in self.processed_servers: country_id = all_country.get(countrycode) bulk_servers.append((country_id, ip_address, ovpn_file, ts, ts, countrycode)) self.processed_servers.add(ip_address) os.makedirs(self.ovpn_path, exist_ok=True) file_name_bbb = os.path.join(self.ovpn_path, f"{ip_address}.ovpn") with open(file_name_bbb, "w") as f: f.write(ovpn_file) # print('bulk_countries',bulk_countries) # print('bulk_servers',bulk_servers) if bulk_countries: self.bulk_insert_if_not_exists( table_name=self.live_country_table_name, columns=['country', 'country_code', 'created_at', 'updated_at', 'premium', 'is_active'], values=bulk_countries, all_country_codes=all_country_code) if bulk_servers: self.bulk_insert_if_not_exists( table_name=self.live_ovpn_table_name, columns=['country_id', 'ip_address', 'ovpn', 'created_at', 'updated_at', 'country_code'], values = bulk_servers, all_country_codes=all_servers) # self.delete_low_ping_count_file() self.delete_low_ping_count_file() self.end_time = datetime.now() logging.info(f"Script completed time is : {self.end_time}") except Exception as e: traceback.print_exc() print(e) logging.error(f"Error in add_server: {e}") excp = traceback.format_exc() excp_time = time.time() excp_time_formatted = datetime.fromtimestamp(excp_time).strftime('%Y-%m-%d %H:%M:%S') sendExceptionMail(str(excp), self.full_script_name, self.server_ip, str(excp_time_formatted)) def main(self): try: self.insert_country_server() except Exception as e: excp = traceback.format_exc() logging.error(f"Error in main: {e}") excp_time = time.time() excp_time_formatted = datetime.fromtimestamp(excp_time).strftime('%Y-%m-%d %H:%M:%S') sendExceptionMail(str(excp), self.full_script_name, self.server_ip, str(excp_time_formatted)) # sendExceptionMail(self.full_script_name, str(e), self.environment_selected) temp_obj = TemporaryServer() temp_obj.main()