write to file in search script

search_script.py

@@ -1,5 +1,5 @@
 # <GRACE_MODULE id="search_script" name="search_script.py">  
-#   @SEMANTICS: search, superset, dataset, regex
+#   @SEMANTICS: search, superset, dataset, regex, file_output  
 #   @PURPOSE: Предоставляет утилиты для поиска по текстовым паттернам в метаданных датасетов Superset.  
 #   @DEPENDS_ON: superset_tool.client -> Для взаимодействия с API Superset.  
 #   @DEPENDS_ON: superset_tool.utils -> Для логирования и инициализации клиентов.  
@@ -7,6 +7,7 @@
 # <IMPORTS>  
 import logging  
 import re
+import os
 from typing import Dict, Optional  
 from requests.exceptions import RequestException  
 from superset_tool.client import SupersetClient  
@@ -76,6 +77,29 @@ def search_datasets(
         raise  
 # </ANCHOR id="search_datasets">  
   
+# <ANCHOR id="save_results_to_file" type="Function">  
+#   @PURPOSE: Сохраняет результаты поиска в текстовый файл.  
+#   @PRE: `results` является словарем, возвращенным `search_datasets`, или `None`.  
+#   @PRE: `filename` должен быть допустимым путем к файлу.  
+#   @POST: Записывает отформатированные результаты в указанный файл.  
+#   @PARAM: results: Optional[Dict] - Словарь с результатами поиска.  
+#   @PARAM: filename: str - Имя файла для сохранения результатов.  
+#   @PARAM: logger: Optional[SupersetLogger] - Инстанс логгера.  
+#   @RETURN: bool - Успешно ли выполнено сохранение.  
+def save_results_to_file(results: Optional[Dict], filename: str, logger: Optional[SupersetLogger] = None) -> bool:  
+    logger = logger or SupersetLogger(name="file_writer")  
+    logger.info(f"[save_results_to_file][Enter] Saving results to file: {filename}")  
+    try:  
+        formatted_report = print_search_results(results)  
+        with open(filename, 'w', encoding='utf-8') as f:  
+            f.write(formatted_report)  
+        logger.info(f"[save_results_to_file][Success] Results saved to {filename}")  
+        return True  
+    except Exception as e:  
+        logger.error(f"[save_results_to_file][Failure] Failed to save results to file: {e}", exc_info=True)  
+        return False  
+# </ANCHOR id="save_results_to_file">  
+  
 # <ANCHOR id="print_search_results" type="Function">  
 #   @PURPOSE: Форматирует результаты поиска для читаемого вывода в консоль.  
 #   @PRE: `results` является словарем, возвращенным `search_datasets`, или `None`.  
@@ -89,7 +113,24 @@ def print_search_results(results: Optional[Dict], context_lines: int = 3) -> str
   
     output = []  
     for dataset_id, matches in results.items():  
+        # Получаем информацию о базе данных для текущего датасета
+        database_info = ""
+        # Ищем поле database среди совпадений, чтобы вывести его
+        for match_info in matches:
+            if match_info['field'] == 'database':
+                database_info = match_info['value']
+                break
+        # Если database не найден в совпадениях, пробуем получить из других полей
+        if not database_info:
+            # Предполагаем, что база данных может быть в одном из полей, например sql или table_name
+            # Но для точности лучше использовать специальное поле, которое мы уже получили
+            pass  # Пока не выводим, если не нашли явно
+
         output.append(f"\n--- Dataset ID: {dataset_id} ---")
+        if database_info:
+            output.append(f"  Database: {database_info}")
+        output.append("")  # Пустая строка для читабельности
+        
         for match_info in matches:  
             field, match_text, full_value = match_info['field'], match_info['match'], match_info['value']  
             output.append(f"  - Поле: {field}")  
@@ -125,12 +166,18 @@ def print_search_results(results: Optional[Dict], context_lines: int = 3) -> str
 #   @RELATION: CALLS -> setup_clients  
 #   @RELATION: CALLS -> search_datasets  
 #   @RELATION: CALLS -> print_search_results  
+#   @RELATION: CALLS -> save_results_to_file  
 def main():  
     logger = SupersetLogger(level=logging.INFO, console=True)  
     clients = setup_clients(logger)  
   
     target_client = clients['prod']  
-    search_query = r".account_balance_by_contract"
+    search_query = r"from dm_view.[a-z_]*"  
+
+    # Генерируем имя файла на основе времени
+    import datetime
+    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
+    output_filename = f"search_results_{timestamp}.txt"
     
     results = search_datasets(  
         client=target_client,  
@@ -139,7 +186,16 @@ def main():
     )  
   
     report = print_search_results(results)  
+  
     logger.info(f"[main][Success] Search finished. Report:\n{report}")  
+    
+    # Сохраняем результаты в файл
+    success = save_results_to_file(results, output_filename, logger)
+    if success:
+        logger.info(f"[main][Success] Results also saved to file: {output_filename}")
+    else:
+        logger.error(f"[main][Failure] Failed to save results to file: {output_filename}")
+
 # </ANCHOR id="main">  
   
 if __name__ == "__main__":