en2zmax / clear_old_data_in_db_voicetech.py

#!/usr/bin/env python3 import os import sys import logging import mysql.connector as con from datetime import date, timedelta, datetime # Дней хранения логов LOGGING_DAYS = 7 # Сохранение данных за указанное количество дней day_delta = 100 # Сохранение данных с конкретной даты # start_date_string = "2024-09-01" # day_delta = ( # date.today() - datetime.strptime(start_date_string, "%Y-%m-%d").date() # ).days # Данные подключения к БД con_data = { "host": "10.101.60.89", "port": 3306, "user": "root", "password": "MySQLP@ssw0rd1", "database": "voicetech", } logging.basicConfig( level=logging.INFO, filename=f"clear_old_data_in_database_{date.today()}.log", filemode="a", format="[%(asctime)s: %(levelname)s] %(message)s", ) logger = logging.getLogger("logger") try: os.remove( f"clear_old_data_in_database_{date.today() - timedelta(days=LOGGING_DAYS)}.log" ) except: pass class MySQLExecutor: def __init__(self): try: self._connector = con.connect(**con_data) except: logger.exception("Exception") sys.exit() def __del__(self): self._connector.close() def execute(self, query: str): cursor = self._connector.cursor() try: cursor.execute(query) except: logger.exception("Exception") sys.exit() cursor.close() def execute_commit(self, query: str): cursor = self._connector.cursor() try: cursor.execute(query) self._connector.commit() except: logger.exception("Exception") sys.exit() cursor.close() def select(self, query) -> list: cursor = self._connector.cursor() try: cursor.execute(query) result = cursor.fetchall() except: logger.exception("Exception") sys.exit() cursor.close() return result def optimize_table(table_name: str, where_column: str): logger.info(f"Start optimize table {table_name}.") executor = MySQLExecutor() query = f"CREATE TABLE voicetech.{table_name}_new LIKE voicetech.{table_name};" executor.execute(query) logger.info(f"Create table '{table_name}_new'.") query = f"select TRIGGER_NAME from information_schema.TRIGGERS where EVENT_OBJECT_SCHEMA = 'voicetech' and EVENT_OBJECT_TABLE = '{table_name}';" create_trigger_queries_list = [ executor.select(f"show create trigger {trigger};")[0][2] for trigger in [item[0] for item in executor.select(query)] ] logger.info(f"Read triggers from table '{table_name}'.") base_insert_query = ( f"insert into voicetech.{table_name}_new select * from voicetech.{table_name}" ) for i in range(day_delta + 1): begin_date = date.today() - timedelta(days=day_delta - i) end_date = begin_date + timedelta(days=1) query = f"{base_insert_query} where {where_column} >= '{begin_date}' and {where_column} < '{end_date}';" executor.execute_commit(query) logger.info(f"Copy data in table '{table_name}_new' for {begin_date}.") query = f"RENAME TABLE voicetech.{table_name} TO voicetech.{table_name}_old, voicetech.{table_name}_new TO voicetech.{table_name};" executor.execute(query) logger.info( f"Rename table '{table_name}' to '{table_name}_old' and '{table_name}_new' to '{table_name}'." ) query = f"DROP TABLE voicetech.{table_name}_old;" executor.execute(query) logger.info(f"Delete table '{table_name}_old'.") for query in create_trigger_queries_list: executor.execute(query) logger.info(f"Restore triggers from table '{table_name}'.") def main(): logger.info("") logger.info("Start script") logger.info("") optimize_table("interactions", "dtStart") optimize_table("phone_cdr", "start") optimize_table("phone_ivr_journeys", "journeyTime") logger.info("") logger.info("Stop script") logger.info("") if __name__ == "__main__": main()