settings application

.env db params+global settings in admin model
ECVI plugin module

antifroud/data_sync.py

@@ -1,11 +1,15 @@
+import logging
 import pymysql
-from django.db import transaction
-from django.utils import timezone
 from datetime import datetime
-from html import unescape
 from urllib.parse import unquote, parse_qs
-
-from .models import ExternalDBSettings, UserActivityLog, RoomDiscrepancy, ImportedHotel
+import pytz
+from django.utils import timezone
+from django.db import transaction
+from django.conf import settings
+from html import unescape
+import chardet
+import html
+from .models import SyncLog, ExternalDBSettings, UserActivityLog, RoomDiscrepancy, ImportedHotel
 from hotels.models import Reservation, Hotel
 
 class DataSyncManager:
@@ -13,54 +17,120 @@ class DataSyncManager:
     Класс для управления загрузкой, записью и сверкой данных.
     """
 
-    def __init__(self, db_settings_id):
+    def __init__(self, db_settings_id, use_local_db=False):
         self.db_settings_id = db_settings_id
+        self.use_local_db = use_local_db  # Если True, используем локальную БД
         self.db_settings = None
         self.connection = None
         self.table_name = None
 
+        # Настройка логирования
+        self.logger = logging.getLogger(__name__)
+        self.logger.setLevel(logging.DEBUG)
+        handler = logging.FileHandler('data_sync.log')
+        handler.setLevel(logging.DEBUG)
+        formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
+        handler.setFormatter(formatter)
+        self.logger.addHandler(handler)
+
     def connect_to_db(self):
         """
-        Устанавливает соединение с внешней базой данных и получает имя таблицы.
+        Устанавливает соединение с БД в зависимости от настройки.
         """
         try:
-            self.db_settings = ExternalDBSettings.objects.get(id=self.db_settings_id)
-            self.table_name = self.db_settings.table_name
-            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',
-                use_unicode=True
-            )
+            if self.use_local_db:
+                # Подключаемся к локальной базе данных
+                self.db_settings = settings.LocalDataBase.objects.first()  # Получаем настройки первой базы
+                self.table_name = self.db_settings.database
+                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',
+                    use_unicode=True,
+                    cursorclass=pymysql.cursors.DictCursor,
+                )
+            else:
+                # Подключаемся к внешней базе данных
+                self.db_settings = ExternalDBSettings.objects.get(id=self.db_settings_id)
+                self.table_name = self.db_settings.table_name
+                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',
+                    use_unicode=True,
+                    cursorclass=pymysql.cursors.DictCursor,
+                )
+
         except ExternalDBSettings.DoesNotExist:
             raise ValueError("Настройки подключения не найдены.")
         except pymysql.MySQLError as e:
             raise ConnectionError(f"Ошибка подключения к базе данных: {e}")
-
-    def fetch_data(self, limit=100):
+    def get_last_saved_record(self):
         """
-        Загружает данные из указанной таблицы.
+        Получает последнюю запись из таблицы UserActivityLog.
+        """
+        last_record = UserActivityLog.objects.order_by('-id').first()
+        if last_record:
+            self.logger.info(f"Последняя запись в UserActivityLog: ID={last_record.id}")
+            return last_record.id
+        self.logger.info("Таблица UserActivityLog пуста.")
+        return None
+
+    def fetch_new_data(self, last_id=0, limit=100):
+        """
+        Загружает новые записи из указанной таблицы, которые идут после last_id.
         """
         if not self.connection:
             self.connect_to_db()
 
-        cursor = self.connection.cursor()
+        cursor = self.connection.cursor(pymysql.cursors.DictCursor)  # Используем DictCursor для получения словарей
         try:
-            cursor.execute(f"SELECT * FROM `{self.table_name}` LIMIT {limit};")
-            columns = [desc[0] for desc in cursor.description]
+            # Формируем SQL-запрос
+            if last_id:
+                query = f"""
+                    SELECT * FROM `{self.table_name}`
+                    WHERE id > {last_id} AND url_parameters IS NOT NULL AND url_parameters != ''
+                    ORDER BY id ASC
+                    LIMIT {limit};
+                """
+            else:
+                query = f"""
+                    SELECT * FROM `{self.table_name}`
+                    WHERE url_parameters IS NOT NULL AND url_parameters != ''
+                    ORDER BY id ASC
+                    LIMIT {limit};
+                """
+            
+            self.logger.info(f"Выполняется запрос: {query}")
+            cursor.execute(query)
+            
+            # Получаем результаты
             rows = cursor.fetchall()
-            return {"columns": columns, "rows": rows}
+            if not rows:
+                self.logger.info("Нет данных для загрузки.")
+                return {"columns": [], "rows": []}
+
+            # Получаем названия колонок
+            columns = rows[0].keys() if rows else []
+            return {"columns": list(columns), "rows": rows}
+
         except pymysql.MySQLError as e:
-            raise RuntimeError(f"Ошибка выполнения запроса: {e}")
+            self.logger.error(f"Ошибка выполнения запроса: {e}")
+            return {"columns": [], "rows": []}
         finally:
             cursor.close()
 
-    def parse_datetime(self, dt_str):
+
+    def parse_datetime(self, dt_str, hotel_timezone=None):
         """
         Преобразует строку формата 'YYYY-MM-DD HH:MM:SS' или 'YYYY-MM-DDTHH:MM:SS' в aware datetime.
+        Преобразует время в часовой пояс отеля и корректирует на часовой пояс сервера.
         """
         if dt_str is None:
             return None
@@ -72,8 +142,22 @@ class DataSyncManager:
 
         for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"):
             try:
+                # Преобразуем строку в naive datetime
                 naive_dt = datetime.strptime(dt_str, fmt)
+
+                # Если передан часовой пояс отеля
+                if hotel_timezone:
+                    # Переводим время из часового пояса отеля в UTC (если данные из PMS уже UTC+X)
+                    tz = pytz.timezone(hotel_timezone)  # Часовой пояс отеля
+                    aware_dt = tz.localize(naive_dt)  # Локализуем в часовой пояс отеля
+
+                    # Теперь приводим время к серверному часовому поясу (например, Московское время)
+                    server_tz = timezone.get_default_timezone()  # Например, Moscow (UTC+3)
+                    return aware_dt.astimezone(server_tz)  # Переводим в серверное время
+
+                # Если часовой пояс отеля не передан, используем серверный часовой пояс по умолчанию
                 return timezone.make_aware(naive_dt, timezone.get_default_timezone())
+
             except ValueError:
                 continue
 
@@ -81,95 +165,168 @@ class DataSyncManager:
 
     def decode_html_entities(self, text):
         """
-        Раскодирует HTML-сущности в строке.
+        Декодирует URL и HTML-сущности в строке.
+        Пытается автоматически декодировать строку в правильную кодировку.
         """
         if text and isinstance(text, str):
-            return unescape(text)
+            text = unquote(text)  # Декодируем URL
+            text = html.unescape(text)  # Расшифровываем HTML сущности
+
+            # Попробуем определить кодировку и привести строку к utf-8
+            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
         return text
 
     def process_url_parameters(self, url_params):
         """
-        Парсит url_parameters, извлекает utm_content (имя отеля) и utm_term (ID отеля).
+        Парсит url_parameters и возвращает hotel_name и hotel_id.
         """
-        if not url_params:
+        # Проверка корректности данных
+        if not url_params or not isinstance(url_params, str):
+            self.logger.error(f"Ошибка: некорректные url_parameters: {url_params}")
             return {}
+
+        # Декодируем параметры URL
         decoded = unquote(url_params)
         qs = parse_qs(decoded)
-        hotel_name = qs.get('utm_content', [None])[0]
-        hotel_id = qs.get('utm_term', [None])[0]
+
+        # Извлекаем hotel_name и hotel_id_term
+        hotel_name = qs.get('utm_content', [None])[0]  # Умолчание: None, если параметр не найден
+        hotel_id_term = qs.get('utm_term', [None])[0]  # Умолчание: None, если параметр не найден
+
+        # Формируем hotel_id
+        hotel_id = f"{hotel_name}_{hotel_id_term}" if hotel_name and hotel_id_term else None
+
+        # Логирование для отладки
+        self.logger.debug(f"Извлечено из url_parameters: hotel_name={hotel_name}, hotel_id_term={hotel_id_term}, hotel_id={hotel_id}")
+
+        # Возврат результата
         return {
             'hotel_name': hotel_name,
             'hotel_id': hotel_id
         }
 
+
     def check_and_store_imported_hotel(self, hotel_name, hotel_id):
-        """
-        Проверяет, есть ли отель с данным ID в основной БД.
-        Если hotel_id не число или отеля с таким ID нет, добавляет во временную таблицу ImportedHotel.
-        """
-        if not hotel_id or not hotel_name:
-            return
+            """
+            Проверяет, есть ли отель с данным ID в таблице ImportedHotel.
+            Если отеля с таким external_id нет, добавляет новый в таблицы ImportedHotel и Hotel.
+            """
+            if not hotel_id or not hotel_name:
+                return None
 
-        # Проверим, что hotel_id — число
-        if hotel_id.isdigit():
-            hotel_id_int = int(hotel_id)
-            hotel_exists = Hotel.objects.filter(id=hotel_id_int).exists()
-        else:
-            # Если не число, считаем что отеля в основной БД нет
-            hotel_exists = False
+            # Генерация external_id в формате 'hotel_name_hotel_id'
+            external_id = f"{hotel_name}_{hotel_id}"
 
-        if not hotel_exists:
-            ImportedHotel.objects.update_or_create(
-                external_id=str(hotel_id),
-                defaults={
-                    'name': hotel_name
-                }
-            )
+            # Проверяем, существует ли запись с таким external_id в ImportedHotel
+            existing_hotel = ImportedHotel.objects.filter(external_id=external_id).first()
+
+            if existing_hotel:
+                self.logger.info(f"Отель с external_id {external_id} уже существует в ImportedHotel.")
+            else:
+                try:
+                    # Создаем новую запись в ImportedHotel
+                    with transaction.atomic():
+                        imported_hotel = ImportedHotel.objects.create(
+                            external_id=external_id,
+                            name=hotel_name,
+                            display_name=hotel_name,
+                            imported=True  # Отмечаем, что отель импортирован
+                        )
+                        self.logger.info(f"Отель с external_id {external_id} добавлен в ImportedHotel.")
+                        
+                        # Создаем новый отель в основной таблице Hotel
+                        hotel = Hotel.objects.create(
+                            hotel_id=external_id,
+                            name=hotel_name,
+                            phone=None,
+                            email=None,
+                            address=None,
+                            city=None,
+                            timezone="UTC",
+                            description="Автоматически импортированный отель",
+                            
+                        )
+                        self.logger.info(f"Отель с hotel_id {external_id} добавлен в Hotel с флагом is_imported=True.")
+                        return hotel
+                except Exception as e:
+                    self.logger.error(f"Ошибка при добавлении отеля {hotel_name} с external_id {external_id}: {e}")
+                    return None
+
+            return existing_hotel
 
-    @transaction.atomic
     def write_to_db(self, data):
         """
         Записывает данные в UserActivityLog и при необходимости в ImportedHotel.
+        Записывает лог синхронизации в SyncLog.
         """
+        processed_records = 0
+        received_records = len(data["rows"])
+
+        print(f"Received records: {received_records}")
+
         for row in data["rows"]:
-            record = dict(zip(data["columns"], row))
+            print(f'\n------\n row: {row}\n------\n')
+            # record = dict(zip(data["columns"], row))  # Преобразуем строку в словарь
+            # Получаем url_parameters из записи
+            url_parameters = self.decode_html_entities(row.get("url_parameters", ""))
+            print(f'\n------\n url_parameters: {url_parameters}\n------\n')
 
-            external_id = record.get("id", None)
-            if external_id is not None:
-                external_id = str(external_id)
+            # Проверка на пустое значение
+            if not url_parameters:
+                print(f"Error: url_parameters is empty in record {row}")
+                continue  # Пропускаем запись, если url_parameters отсутствует
 
-            created = self.parse_datetime(record.get("created"))
-            date_time = self.parse_datetime(record.get("date_time"))
+            # Пытаемся извлечь информацию о отеле из url_parameters
+            hotel_name = None
+            hotel_id = None
 
-            referred = self.decode_html_entities(record.get("referred", ""))
-            agent = self.decode_html_entities(record.get("agent", ""))
-            platform = self.decode_html_entities(record.get("platform", ""))
-            version = self.decode_html_entities(record.get("version", ""))
-            model = self.decode_html_entities(record.get("model", ""))
-            device = self.decode_html_entities(record.get("device", ""))
-            UAString = self.decode_html_entities(record.get("UAString", ""))
-            location = self.decode_html_entities(record.get("location", ""))
-            page_title = self.decode_html_entities(record.get("page_title", ""))
-            page_url = self.decode_html_entities(record.get("page_url", ""))
+            if url_parameters:
+                hotel_info = self.process_url_parameters(url_parameters)
+                hotel_id = hotel_info.get('hotel_id')
 
-            url_parameters = self.decode_html_entities(record.get("url_parameters", ""))
-            hotel_info = self.process_url_parameters(url_parameters)
+            if not hotel_id:
+                print(f"Error: hotel_id is empty in record {row}")
+                continue  # Пропускаем запись, если hotel_id отсутствует
 
-            if hotel_info.get('hotel_name') and hotel_info.get('hotel_id'):
-                self.check_and_store_imported_hotel(
-                    hotel_name=hotel_info['hotel_name'],
-                    hotel_id=hotel_info['hotel_id']
-                )
+            # Проверяем, существует ли отель с таким hotel_id
+            hotel = self.check_and_store_imported_hotel(hotel_name=hotel_id, hotel_id=hotel_id)
 
-            url_parameters = unquote(url_parameters)
+            if not hotel:
+                print(f"Error: Could not find or create hotel for hotel_id {hotel_id}")
+                continue  # Пропускаем запись, если отель не найден или не создан
 
+            # Преобразуем дату
+            created = self.parse_datetime(row.get("created"))
+            date_time = self.parse_datetime(row.get("date_time"))
+
+            # Декодируем все строки, которые могут содержать HTML-сущности
+            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_title = self.decode_html_entities(row.get("page_title", ""))
+            page_url = self.decode_html_entities(row.get("page_url", ""))
+
+            # Запись в UserActivityLog
             UserActivityLog.objects.update_or_create(
-                external_id=external_id,
+                external_id=row.get("id", None),
                 defaults={
-                    "user_id": record.get("user_id"),
-                    "ip": record.get("ip"),
+                    "user_id": row.get("user_id"),
+                    "ip": row.get("ip"),
                     "created": created,
-                    "timestamp": record.get("timestamp"),
+                    "timestamp": row.get("timestamp"),
                     "date_time": date_time,
                     "referred": referred,
                     "agent": agent,
@@ -179,18 +336,24 @@ class DataSyncManager:
                     "device": device,
                     "UAString": UAString,
                     "location": location,
-                    "page_id": record.get("page_id"),
+                    "page_id": row.get("page_id"),
                     "url_parameters": url_parameters,
                     "page_title": page_title,
-                    "type": record.get("type"),
-                    "last_counter": record.get("last_counter"),
-                    "hits": record.get("hits"),
-                    "honeypot": record.get("honeypot"),
-                    "reply": record.get("reply"),
+                    "type": row.get("type"),
+                    "last_counter": row.get("last_counter"),
+                    "hits": row.get("hits"),
+                    "honeypot": row.get("honeypot"),
+                    "reply": row.get("reply"),
                     "page_url": page_url,
                 }
             )
 
+            processed_records += 1
+
+        # Логируем обработанные записи
+        print(f"Processed records: {processed_records}")
+
+
     def reconcile_data(self):
         """
         Сверяет данные таблицы user_activity_log с таблицей hotels.reservations
@@ -218,21 +381,34 @@ class DataSyncManager:
         RoomDiscrepancy.objects.bulk_create(discrepancies)
 
     def sync(self):
-        """
-        Основной метод для загрузки, записи и сверки данных.
-        """
         try:
-            self.connect_to_db()
-            data = self.fetch_data()
-            self.write_to_db(data)
-            self.reconcile_data()
+            last_id = self.get_last_saved_record()
+            self.logger.info(f"Синхронизация начата. Последний сохранённый ID: {last_id}")
+
+            # Загружаем новые данные
+            data = self.fetch_new_data(last_id=last_id)
+            print(f'\n------\n data: {data}\n------\n')
+            if not data["rows"]:
+                self.logger.info("Нет новых данных для синхронизации.")
+                return
+
+            # Логирование типов данных
+            self.logger.debug(f"Тип первой строки: {type(data['rows'][0])}")
+
+            first_row_id = data["rows"][0]["id"]
+            if last_id is not None and first_row_id <= last_id:
+                self.logger.info(f"Нет новых записей для синхронизации. Последний ID: {last_id}, первый ID из внешней таблицы: {first_row_id}.")
+                return
+
+            processed_records = self.write_to_db(data)
+            self.logger.info(f"Синхронизация завершена. Обработано записей: {processed_records}")
         except Exception as e:
+            self.logger.error(f"Ошибка синхронизации данных: {e}")
             raise RuntimeError(f"Ошибка синхронизации данных: {e}")
         finally:
             if self.connection:
                 self.connection.close()
 
-
 def scheduled_sync():
     """
     Плановая задача для синхронизации данных.