# import logging # import pymysql # from datetime import datetime # from urllib.parse import unquote, parse_qs # import pytz # from django.utils import timezone # from django.db import transaction # from django.conf import settings # import chardet # import html # from .models import SyncLog, ExternalDBSettings, UserActivityLog, RoomDiscrepancy # from hotels.models import Room, Hotel, Reservation # class DatabaseConnector: # """ # Класс для подключения к внешней или локальной базе данных. # """ # def __init__(self, db_settings_id, use_local_db=False): # self.db_settings_id = db_settings_id # self.use_local_db = use_local_db # self.connection = None # self.db_settings = None # self.table_name = None # self.logger = self.setup_logger() # def setup_logger(self): # logger = logging.getLogger(__name__) # logger.setLevel(logging.DEBUG) # handler = logging.FileHandler("data_sync.log") # handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")) # logger.addHandler(handler) # return logger # def connect(self): # """Подключение к базе данных.""" # try: # if self.use_local_db: # self.db_settings = settings.LocalDataBase.objects.first() # else: # self.db_settings = ExternalDBSettings.objects.get(id=self.db_settings_id) # self.connection = pymysql.connect( # host=self.db_settings.host, # port=self.db_settings.port, # user=self.db_settings.user, # password=self.db_settings.password, # database=self.db_settings.database, # charset="utf8mb4", # cursorclass=pymysql.cursors.DictCursor, # ) # self.table_name = self.db_settings.table_name # self.logger.info("Подключение к базе данных успешно установлено.") # except Exception as e: # self.logger.error(f"Ошибка подключения к БД: {e}") # raise ConnectionError(e) # def close(self): # """Закрывает соединение с базой данных.""" # if self.connection: # self.connection.close() # self.logger.info("Соединение с базой данных закрыто.") # def execute_query(self, query): # """Выполнение запроса и возврат результатов.""" # with self.connection.cursor() as cursor: # cursor.execute(query) # return cursor.fetchall() # class DataProcessor: # """ # Обрабатывает и сохраняет данные. # """ # def __init__(self, logger): # self.logger = logger # def decode_html_entities(self, text): # """Декодирует URL и HTML-сущности.""" # if text and isinstance(text, str): # text = unquote(text) # text = html.unescape(text) # try: # detected = chardet.detect(text.encode()) # encoding = detected['encoding'] # if encoding and encoding != 'utf-8': # text = text.encode(encoding).decode('utf-8', errors='ignore') # except Exception as e: # self.logger.error(f"Ошибка кодировки: {e}") # return text # def process_url_parameters(self, url_params): # """Парсит параметры URL и возвращает информацию о отеле и номере.""" # decoded = unquote(url_params) # qs = parse_qs(decoded) # hotel_name = qs.get('utm_content', [None])[0] # room_number = qs.get('utm_term', [None])[0] # return {"hotel_name": hotel_name, "room_number": room_number} # def parse_datetime(self, dt_str): # """Преобразует строку даты в aware datetime.""" # try: # if isinstance(dt_str, datetime): # return timezone.make_aware(dt_str) if timezone.is_naive(dt_str) else dt_str # if dt_str: # return timezone.make_aware(datetime.strptime(dt_str, "%Y-%m-%d %H:%M:%S")) # except Exception as e: # self.logger.error(f"Ошибка парсинга даты: {e}") # return None # class HotelRoomManager: # """ # Управляет созданием отелей и номеров. # """ # def __init__(self, logger): # self.logger = logger # def get_or_create_hotel(self, hotel_name): # """Создает или получает отель.""" # if not hotel_name: # return None # hotel, created = Hotel.objects.get_or_create( # name=hotel_name, # defaults={"description": "Автоматически добавленный отель"} # ) # if created: # self.logger.info(f"Создан отель: {hotel_name}") # return hotel # def get_or_create_room(self, hotel, room_number): # """Создает или получает номер отеля.""" # if not hotel or not room_number: # return None # room, created = Room.objects.get_or_create( # hotel=hotel, # number=room_number, # defaults={"external_id": f"{hotel.name}_{room_number}"} # ) # if created: # self.logger.info(f"Добавлен номер: {room_number} в отель {hotel.name}") # return room # class DataSyncManager: # """ # Главный класс для синхронизации данных. # """ # def __init__(self, db_settings_id, use_local_db=False): # self.logger = self.setup_logger() # self.db_connector = DatabaseConnector(db_settings_id, use_local_db) # self.data_processor = DataProcessor(self.logger) # self.hotel_manager = HotelRoomManager(self.logger) # def setup_logger(self): # logger = logging.getLogger(__name__) # logger.setLevel(logging.DEBUG) # handler = logging.FileHandler("data_sync.log") # handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")) # logger.addHandler(handler) # return logger # def get_last_saved_record(self): # """Получает ID последней записи.""" # record = UserActivityLog.objects.order_by("-id").first() # return record.id if record else 0 # def fetch_new_data(self, last_id): # """Получает новые данные из БД для переноса 1-в-1.""" # query = f""" # SELECT id AS external_id, user_id, ip, created, timestamp, date_time, # referred, agent, platform, version, model, device, UAString, location, # page_id, url_parameters, page_title, type, last_counter, hits, # honeypot, reply, page_url # FROM `{self.db_connector.table_name}` # WHERE id > {last_id} # ORDER BY id ASC LIMIT 100; # """ # return self.db_connector.execute_query(query) # def process_and_save_data(self, rows): # """Обрабатывает данные и сохраняет их 1-в-1 в UserActivityLog.""" # for row in rows: # try: # # Парсинг и сохранение всех полей # UserActivityLog.objects.update_or_create( # external_id=row["external_id"], # defaults={ # "user_id": row.get("user_id"), # "ip": row.get("ip"), # "created": self.data_processor.parse_datetime(row.get("created")), # "timestamp": row.get("timestamp"), # "date_time": self.data_processor.parse_datetime(row.get("date_time")), # "referred": row.get("referred"), # "agent": row.get("agent"), # "platform": row.get("platform"), # "version": row.get("version"), # "model": row.get("model"), # "device": row.get("device"), # "UAString": row.get("UAString"), # "location": row.get("location"), # "page_id": row.get("page_id"), # "url_parameters": row.get("url_parameters"), # "page_title": row.get("page_title"), # "type": row.get("type"), # "last_counter": row.get("last_counter"), # "hits": row.get("hits"), # "honeypot": row.get("honeypot"), # "reply": row.get("reply"), # "page_url": row.get("page_url"), # } # ) # self.logger.info(f"Запись external_id={row['external_id']} успешно сохранена.") # except Exception as e: # self.logger.error(f"Ошибка обработки записи {row['external_id']}: {e}") # def check_and_store_room(self, hotel, room_number): # """ # Проверяет и создает номер в отеле, если он еще не существует. # """ # try: # Room.objects.get_or_create( # hotel=hotel, # number=room_number, # defaults={ # "external_id": f"{hotel.hotel_id}_{room_number}", # "description": "Автоматически добавленный номер", # } # ) # self.logger.info(f"Добавлен номер: {room_number} в отель {hotel.name}") # except Exception as e: # self.logger.error(f"Ошибка при добавлении номера {room_number} для отеля {hotel.name}: {e}") # def reconcile_data(self): # """ # Сверяет данные UserActivityLog с Reservation и фиксирует несоответствия. # """ # discrepancies = [] # reservations = Reservation.objects.values("hotel_id", "room_number", "check_in", "check_out") # for log in UserActivityLog.objects.all(): # try: # # Преобразование page_id в число # page_id = int(float(log.page_id)) # except (ValueError, TypeError): # self.logger.warning(f"Некорректное значение page_id: {log.page_id} - пропущено.") # continue # # Сверка с бронированиями # matching_reservation = reservations.filter( # hotel_id=log.url_parameters, room_number=str(page_id) # ).first() # if not matching_reservation: # discrepancy = RoomDiscrepancy( # hotel_id=log.hotel_id, # room_number=page_id, # booking_id=f"Log-{log.id}", # check_in_date_expected=None, # check_in_date_actual=log.created.date() if log.created else None, # discrepancy_type="Mismatch", # ) # discrepancies.append(discrepancy) # RoomDiscrepancy.objects.bulk_create(discrepancies) # self.logger.info(f"Обнаружено несоответствий: {len(discrepancies)}") # def write_to_db(self, data): # """ # Записывает данные в UserActivityLog и при необходимости в ImportedHotel. # """ # processed_records = 0 # received_records = len(data["rows"]) # self.logger.info(f"Начата обработка {received_records} записей.") # for row in data["rows"]: # # Обработка page_id с валидацией # raw_page_id = row.get("page_id") # try: # page_id = int(float(raw_page_id)) # Пытаемся привести к числу # except (ValueError, TypeError): # self.logger.warning(f"Некорректное значение page_id: {raw_page_id} - установлено 0") # page_id = 0 # Используем значение по умолчанию # # Получаем url_parameters # url_parameters = self.encode_html_entities(self.process_url_parameters(self.decode_html_entities(row.get("url_parameters", "")))) # # Проверка на пустое значение url_parameters # if not url_parameters: # self.logger.warning("Пропущена запись из-за отсутствующих url_parameters.") # continue # # Извлечение информации об отеле из url_parameters # hotel_info = self.process_url_parameters(url_parameters) # hotel_name = hotel_info.get("hotel_name") # hotel_id = hotel_info.get("hotel_id") # if not hotel_id: # self.logger.warning("Пропущена запись из-за отсутствующего hotel_id.") # continue # # Проверяем или создаем отель # hotel = self.check_and_store_imported_hotel(hotel_name=hotel_name, hotel_id=hotel_id) # if not hotel: # self.logger.error(f"Не удалось найти или создать отель: {hotel_id}") # continue # # Преобразование дат # created = self.parse_datetime(row.get("created")) # date_time = self.parse_datetime(row.get("date_time")) # # Запись данных в UserActivityLog # UserActivityLog.objects.update_or_create( # external_id=row.get("id"), # defaults={ # "user_id": row.get("user_id"), # "ip": row.get("ip") or "0.0.0.0", # "created": created, # "timestamp": row.get("timestamp") or timezone.now().timestamp(), # "date_time": date_time, # "referred": self.decode_html_entities(row.get("referred", "")), # "agent": self.decode_html_entities(row.get("agent", "")), # "platform": self.decode_html_entities(row.get("platform", "")), # "version": self.decode_html_entities(row.get("version", "")), # "model": self.decode_html_entities(row.get("model", "")), # "device": self.decode_html_entities(row.get("device", "")), # "UAString": self.decode_html_entities(row.get("UAString", "")), # "location": self.decode_html_entities(row.get("location", "")), # "page_id": page_id, # Обновленный page_id # "url_parameters": url_parameters, # "page_title": self.decode_html_entities(row.get("page_title", "")), # "type": row.get("type"), # "last_counter": row.get("last_counter") or 0, # "hits": row.get("hits") or 0, # "honeypot": row.get("honeypot") or False, # "reply": row.get("reply") or False, # "page_url": self.decode_html_entities(row.get("page_url", "")), # }, # ) # processed_records += 1 # self.logger.info(f"Запись успешно обработана: external_id={row.get('id')}") # self.logger.info(f"Обработано записей: {processed_records} из {received_records}.") # def sync(self): # """Запускает процесс синхронизации.""" # self.db_connector.connect() # try: # last_id = self.get_last_saved_record() # rows = self.fetch_new_data(last_id) # self.process_and_save_data(rows) # self.logger.info("Синхронизация завершена.") # finally: # self.db_connector.close() # import logging # from concurrent.futures import ThreadPoolExecutor # from .models import ExternalDBSettings # logger = logging.getLogger(__name__) # logger.setLevel(logging.DEBUG) # def scheduled_sync(): # """ # Планировщик синхронизации для всех активных подключений. # Каждое подключение обрабатывается отдельно. # """ # logger.info("Запуск планировщика синхронизации.") # # Получаем все активные настройки подключения # active_db_settings = ExternalDBSettings.objects.filter(is_active=True) # if not active_db_settings.exists(): # logger.warning("Не найдено активных подключений для синхронизации.") # return # logger.info(f"Найдено активных подключений: {len(active_db_settings)}") # def sync_task(db_settings): # """ # Выполняет синхронизацию для одного подключения. # """ # try: # logger.info(f"Начало синхронизации для подключения: {db_settings.name} (ID={db_settings.id})") # sync_manager = DataSyncManager(db_settings.id) # sync_manager.sync() # logger.info(f"Синхронизация успешно завершена для подключения: {db_settings.name}") # except Exception as e: # logger.error(f"Ошибка синхронизации для подключения {db_settings.name}: {e}") # # Параллельное выполнение задач синхронизации # with ThreadPoolExecutor(max_workers=5) as executor: # Максимальное количество потоков = 5 # for db_settings in active_db_settings: # executor.submit(sync_task, db_settings) # logger.info("Планировщик синхронизации завершил работу.") import logging import pymysql from datetime import datetime from urllib.parse import unquote, parse_qs import pytz from django.utils import timezone from django.conf import settings import chardet import html from hotels.models import Room, Hotel from .models import UserActivityLog, ExternalDBSettings class DatabaseConnector: """ Класс для подключения к внешней базе данных. """ def __init__(self, db_settings_id): self.db_settings_id = db_settings_id self.connection = None self.logger = self.setup_logger() self.db_settings = None def setup_logger(self): logger = logging.getLogger(__name__) logger.setLevel(logging.DEBUG) handler = logging.FileHandler("data_sync.log") handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")) logger.addHandler(handler) return logger def connect(self): """Подключение к базе данных.""" try: self.db_settings = ExternalDBSettings.objects.get(id=self.db_settings_id) self.connection = pymysql.connect( host=self.db_settings.host, port=self.db_settings.port, user=self.db_settings.user, password=self.db_settings.password, database=self.db_settings.database, charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, ) self.logger.info("Подключение к базе данных успешно установлено.") except Exception as e: self.logger.error(f"Ошибка подключения к БД: {e}") raise ConnectionError(e) def close(self): """Закрывает соединение с базой данных.""" if self.connection: self.connection.close() self.logger.info("Соединение с базой данных закрыто.") def execute_query(self, query): """Выполнение запроса и возврат результатов.""" with self.connection.cursor() as cursor: cursor.execute(query) return cursor.fetchall() class DataProcessor: """ Обрабатывает и сохраняет данные. """ def __init__(self, logger): self.logger = logger def decode_html_entities(self, text): """Декодирует URL и HTML-сущности.""" if text and isinstance(text, str): text = unquote(text) text = html.unescape(text) try: detected = chardet.detect(text.encode()) encoding = detected['encoding'] if encoding and encoding != 'utf-8': text = text.encode(encoding).decode('utf-8', errors='ignore') except Exception as e: self.logger.error(f"Ошибка кодировки: {e}") return text def parse_datetime(self, dt_str): """Преобразует строку даты в aware datetime.""" try: if isinstance(dt_str, datetime): return timezone.make_aware(dt_str) if timezone.is_naive(dt_str) else dt_str if dt_str: return timezone.make_aware(datetime.strptime(dt_str, "%Y-%m-%d %H:%M:%S")) except Exception as e: self.logger.error(f"Ошибка парсинга даты: {e}") return None def url_parameters_parser(self, url_parameters): """ Парсит строку URL-параметров и возвращает словарь с ключами и значениями. Пример входа: url_parameters = "utm_medium=qr-3&utm_content=chisto-pozhit&utm_term=101" Возвращает: { "utm_medium": "qr-3", "utm_content": "chisto-pozhit", "utm_term": "101" } """ try: if not url_parameters: return {} # Декодируем параметры URL decoded_params = unquote(url_parameters) parsed_params = parse_qs(decoded_params) # Преобразуем список значений в строку result = {key: value[0] for key, value in parsed_params.items()} return result except Exception as e: self.logger.error(f"Ошибка парсинга URL-параметров: {e}") return {} class HotelRoomManager: """ Управляет созданием отелей и номеров. """ def __init__(self, logger): self.logger = logger def get_or_create_hotel(self, hotel_name, hotel_id): """Создает или получает отель.""" if not hotel_name: return None hotel, created = Hotel.objects.get_or_create( name=hotel_name, hotel_id = hotel_id, defaults={"description": "Автоматически добавленный отель"} ) if created: self.logger.info(f"Создан отель: {hotel_name}") return hotel def get_or_create_room(self, hotel, room_number): """Создает или получает номер отеля.""" if not hotel or not room_number: return None room, created = Room.objects.get_or_create( hotel=hotel, number=room_number, defaults={"external_id": f"{hotel.name}_{room_number}"} ) if created: self.logger.info(f"Добавлен номер: {room_number} в отель {hotel.name}") return room class DataSyncManager: """ Главный класс для синхронизации данных. """ def __init__(self, db_settings_id): self.logger = self.setup_logger() self.db_connector = DatabaseConnector(db_settings_id) self.data_processor = DataProcessor(self.logger) self.hotel_manager = HotelRoomManager(self.logger) def setup_logger(self): logger = logging.getLogger(__name__) logger.setLevel(logging.DEBUG) handler = logging.FileHandler("data_sync.log") handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")) logger.addHandler(handler) return logger def get_last_saved_record(self): """Получает ID последней записи.""" record = UserActivityLog.objects.order_by("-id").first() return record.id if record else 0 def fetch_new_data(self, last_id): """Получает новые данные из БД.""" query = f""" SELECT * FROM `{self.db_connector.db_settings.table_name}` WHERE id > {last_id} AND url_parameters IS NOT NULL AND url_parameters LIKE '%utm_medium%' ORDER BY id ASC LIMIT 1000; """ self.logger.info(f"Запрос на получение новых данных отправлен. \n Содержание запроса: {query}") return self.db_connector.execute_query(query) def process_and_save_data(self, rows): """Обрабатывает и сохраняет данные, включая парсинг отсутствующих значений.""" for row in rows: try: # Декодирование параметров URL url_params = self.data_processor.decode_html_entities(row.get("url_parameters", "")) params = parse_qs(url_params) param_dict = DataProcessor.url_parameters_parser(self.data_processor, url_params) self.logger.info(f"Параметры URL успешно декодированы: {param_dict}") # Извлечение и обработка данных external_id = row.get("id") created = self.data_processor.parse_datetime(row.get("created")) or timezone.now() hotel_name = params.get("utm_content", [None])[0] or "Неизвестный отель" room_number = params.get("utm_term", [None])[0] or "000" page_title = row.get("page_title") or f"Информация отсутствует для ID {external_id}" # Создание отеля и номера, если они отсутствуют hotel = self.hotel_manager.get_or_create_hotel(hotel_name, hotel_id=row.get("hotel_id")) room = self.hotel_manager.get_or_create_room(hotel, room_number) # Заполнение отсутствующих данных user_id = row.get("user_id") or 0 ip = row.get("ip") or "0.0.0.0" hits = row.get("hits") or 0 date_time = row.get("date_time") or timezone.now().strftime("%Y-%m-%d %H:%M:%S") timeatamp = row.get("timestamp") or timezone.now().timestamp() # Создание или обновление записи в UserActivityLog UserActivityLog.objects.update_or_create( external_id=external_id, defaults={ "user_id": user_id, "ip": ip, "timestamp": timeatamp, "date_time": date_time, "created": created, "page_id": room.id if room else None, "url_parameters": url_params, "page_title": html.unescape(page_title), "hits": hits, } ) self.logger.info(f"Запись ID {external_id} успешно обработана и дополнена.") except Exception as e: self.logger.error(f"Ошибка при обработке записи ID {row.get('id')}: {e}") def sync(self): """Запускает процесс синхронизации.""" self.db_connector.connect() try: last_id = self.get_last_saved_record() rows = self.fetch_new_data(last_id) self.process_and_save_data(rows) self.logger.info("Синхронизация завершена.") finally: self.db_connector.close() import logging from concurrent.futures import ThreadPoolExecutor from .models import ExternalDBSettings logger = logging.getLogger(__name__) logger.setLevel(logging.DEBUG) def scheduled_sync(): """ Планировщик синхронизации для всех активных подключений. Каждое подключение обрабатывается отдельно. """ logger.info("Запуск планировщика синхронизации.") # Получаем все активные настройки подключения active_db_settings = ExternalDBSettings.objects.filter(is_active=True) if not active_db_settings.exists(): logger.warning("Не найдено активных подключений для синхронизации.") return logger.info(f"Найдено активных подключений: {len(active_db_settings)}") def sync_task(db_settings): """ Выполняет синхронизацию для одного подключения. """ try: logger.info(f"Начало синхронизации для подключения: {db_settings.name} (ID={db_settings.id})") sync_manager = DataSyncManager(db_settings.id) sync_manager.sync() logger.info(f"Синхронизация успешно завершена для подключения: {db_settings.name}") except Exception as e: logger.error(f"Ошибка синхронизации для подключения {db_settings.name}: {e}") # Параллельное выполнение задач синхронизации with ThreadPoolExecutor(max_workers=5) as executor: # Максимальное количество потоков = 5 for db_settings in active_db_settings: executor.submit(sync_task, db_settings) logger.info("Планировщик синхронизации завершил работу.")