Update app.py

app.py

@@ -11,6 +11,7 @@ import numpy as np
 from urllib.parse import urlparse
 import logging
 from sklearn.preprocessing import normalize
+from concurrent.futures import ThreadPoolExecutor
 
 # Настройка логирования
 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
@@ -68,6 +69,9 @@ db_lock = threading.Lock()
 # Размер пакета для обработки эмбеддингов
 batch_size = 32
 
+# Количество потоков для параллельной обработки
+num_threads = 5
+
 def get_db_connection():
     """Устанавливает соединение с базой данных."""
     try:
@@ -195,6 +199,35 @@ def insert_embedding(conn, table_name, movie_id, embedding_crc32, string_crc32,
         conn.rollback()
         return False
 
+def process_batch(batch):
+    """Обрабатывает пакет фильмов, создавая для них эмбеддинги."""
+    conn = get_db_connection()
+    if conn is None:
+        return
+
+    try:
+        for movie in batch:
+            embedding_string = f"Название: {movie['name']}\nГод: {movie['year']}\nЖанры: {movie['genresList']}\nОписание: {movie['description']}"
+            string_crc32 = calculate_crc32(embedding_string)
+
+            # Проверяем существующий эмбеддинг
+            existing_embedding = get_embedding_from_db(conn, embeddings_table, "string_crc32", string_crc32, model_name)
+
+            if existing_embedding is None:
+                embedding = encode_string(embedding_string)
+                embedding_crc32 = calculate_crc32(str(embedding.tolist()))
+
+                if insert_embedding(conn, embeddings_table, movie['id'], embedding_crc32, string_crc32, embedding):
+                    logging.info(f"Сохранен эмбеддинг для '{movie['name']}'")
+                else:
+                    logging.error(f"Ошибка сохранения эмбеддинга для '{movie['name']}'")
+            else:
+                logging.info(f"Эмбеддинг для '{movie['name']}' уже существует")
+    except Exception as e:
+        logging.error(f"Ошибка при обработке пакета фильмов: {e}")
+    finally:
+        conn.close()
+
 def process_movies():
     """Обрабатывает фильмы, создавая для них эмбеддинги."""
     global processing_complete
@@ -213,53 +246,31 @@ def process_movies():
     for movie in movies_to_process:
         movies_queue.put(movie)
 
-    conn = get_db_connection()
-    if conn is None:
-        processing_complete = True
-        return
+    with ThreadPoolExecutor(max_workers=num_threads) as executor:
+        try:
+            while not movies_queue.empty():
+                if search_in_progress:
+                    time.sleep(1)
+                    continue
+
+                batch = []
+                while not movies_queue.empty() and len(batch) < batch_size:
+                    try:
+                        movie = movies_queue.get_nowait()
+                        batch.append(movie)
+                    except queue.Empty:
+                        break
 
-    try:
-        while not movies_queue.empty():
-            if search_in_progress:
-                time.sleep(1)
-                continue
-
-            batch = []
-            while not movies_queue.empty() and len(batch) < batch_size:
-                try:
-                    movie = movies_queue.get_nowait()
-                    batch.append(movie)
-                except queue.Empty:
+                if not batch:
                     break
+                
+                executor.submit(process_batch, batch)
+                logging.info(f"Отправлен на обработку пакет из {len(batch)} фильмов.")
+        except Exception as e:
+            logging.error(f"Ошибка при обработке фильмов: {e}")
 
-            if not batch:
-                break
-
-            logging.info(f"Обработка пакета из {len(batch)} фильмов...")
-
-            for movie in batch:
-                embedding_string = f"Название: {movie['name']}\nГод: {movie['year']}\nЖанры: {movie['genresList']}\nОписание: {movie['description']}"
-                string_crc32 = calculate_crc32(embedding_string)
-
-                # Проверяем существующий эмбеддинг
-                existing_embedding = get_embedding_from_db(conn, embeddings_table, "string_crc32", string_crc32, model_name)
-
-                if existing_embedding is None:
-                    embedding = encode_string(embedding_string)
-                    embedding_crc32 = calculate_crc32(str(embedding.tolist()))
-
-                    if insert_embedding(conn, embeddings_table, movie['id'], embedding_crc32, string_crc32, embedding):
-                        logging.info(f"Сохранен эмбеддинг для '{movie['name']}'")
-                    else:
-                        logging.error(f"Ошибка сохранения эмбеддинга для '{movie['name']}'")
-                else:
-                    logging.info(f"Эмбеддинг для '{movie['name']}' уже существует")
-    except Exception as e:
-        logging.error(f"Ошибка при обработке фильмов: {e}")
-    finally:
-        conn.close()
-        processing_complete = True
-        logging.info("Обработка фильмов завершена")
+    processing_complete = True
+    logging.info("Обработка фильмов завершена")
 
 def get_movie_embeddings(conn):
     """Загружает все эмбеддинги фильмов из базы данных."""
@@ -278,7 +289,7 @@ def get_movie_embeddings(conn):
         logging.error(f"Ошибка при загрузке эмбеддингов фильмов: {e}")
     return movie_embeddings
 
-def search_movies(query, top_k=10):
+def search_movies(query, top_k=20):
     """Выполняет поиск фильмов по запросу."""
     global search_in_progress
     search_in_progress = True
@@ -368,4 +379,4 @@ iface = gr.Interface(
 )
 
 # Запускаем интерфейс
-iface.launch()
+iface.launch()