fix migrate

superset_tool/utils/fileio.py

@@ -281,13 +281,13 @@ def save_and_unpack_dashboard(zip_content: bytes, output_dir: Union[str, Path],
 # @PARAM: regexp_pattern (Optional[LiteralString]) - Паттерн для поиска.
 # @PARAM: replace_string (Optional[LiteralString]) - Строка для замены.
 # @PARAM: logger (Optional[SupersetLogger]) - Экземпляр логгера.
-def update_yamls(db_configs: Optional[List[Dict]] = None, path: str = "dashboards", regexp_pattern: Optional[LiteralString] = None, replace_string: Optional[LiteralString] = None, logger: Optional[SupersetLogger] = None) -> None:
+def update_yamls(db_configs: Optional[List[Dict[str, Any]]] = None, path: str = "dashboards", regexp_pattern: Optional[LiteralString] = None, replace_string: Optional[LiteralString] = None, logger: Optional[SupersetLogger] = None) -> None:
     logger = logger or SupersetLogger(name="fileio")
     logger.info("[update_yamls][Enter] Starting YAML configuration update.")
     dir_path = Path(path)
     assert dir_path.is_dir(), f"Путь {path} не существует или не является директорией"
     
-    configs = [db_configs] if isinstance(db_configs, dict) else db_configs or []
+    configs: List[Dict[str, Any]] = db_configs or []
     
     for file_path in dir_path.rglob("*.yaml"):
         _update_yaml_file(file_path, configs, regexp_pattern, replace_string, logger)
@@ -300,7 +300,7 @@ def update_yamls(db_configs: Optional[List[Dict]] = None, path: str = "dashboard
 # @PARAM: regexp_pattern (Optional[str]) - Паттерн.
 # @PARAM: replace_string (Optional[str]) - Замена.
 # @PARAM: logger (SupersetLogger) - Логгер.
-def _update_yaml_file(file_path: Path, db_configs: List[Dict], regexp_pattern: Optional[str], replace_string: Optional[str], logger: SupersetLogger) -> None:
+def _update_yaml_file(file_path: Path, db_configs: List[Dict[str, Any]], regexp_pattern: Optional[str], replace_string: Optional[str], logger: SupersetLogger) -> None:
     # Читаем содержимое файла
     try:
         with open(file_path, 'r', encoding='utf-8') as f:
@@ -330,10 +330,22 @@ def _update_yaml_file(file_path: Path, db_configs: List[Dict], regexp_pattern: O
                 for key, old_val in old_cfg.items():
                     if key in new_cfg:
                         new_val = new_cfg[key]
-                        # Заменяем только точные совпадения старого значения в тексте YAML
+                        # Заменяем только точные совпадения старого значения в тексте YAML, используя ключ для контекста
                         if isinstance(old_val, str):
-                            escaped_old = re.escape(old_val)
-                            modified_content = re.sub(escaped_old, new_val, modified_content)
+                            # Ищем паттерн: key: "value" или key: value
+                            key_pattern = re.escape(key)
+                            val_pattern = re.escape(old_val)
+                            # Группы: 1=ключ+разделитель, 2=открывающая кавычка (опц), 3=значение, 4=закрывающая кавычка (опц)
+                            pattern = rf'({key_pattern}\s*:\s*)(["\']?)({val_pattern})(["\']?)'
+                            
+                            # Функция замены, сохраняющая кавычки если они были
+                            def replacer(match):
+                                prefix = match.group(1)
+                                quote_open = match.group(2)
+                                quote_close = match.group(4)
+                                return f"{prefix}{quote_open}{new_val}{quote_close}"
+
+                            modified_content = re.sub(pattern, replacer, modified_content)
                             logger.info("[_update_yaml_file][State] Replaced '%s' with '%s' for key %s in %s", old_val, new_val, key, file_path)
             # Записываем обратно изменённый контент без парсинга YAML, сохраняем оригинальное форматирование
             with open(file_path, 'w', encoding='utf-8') as f: