danvics/ytptube
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

fix: dont reject non-unique names during migration.

Browse source
This commit is contained in:
arabcoders 2026-01-21 18:39:47 +03:00
parent 5b7d9ddbde
commit 48feca72d3
4 changed files with 70 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
app/features/conditions/migration.py
View file

@ -3,7 +3,7 @@ from __future__ import annotations
import json
import logging
from pathlib import Path
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Any, cast
from app.features.conditions.models import ConditionModel
from app.features.core.migration import Migration as FeatureMigration
@ -46,8 +46,9 @@ class Migration(FeatureMigration):
return
inserted = 0
seen_names: dict[str, int] = {}
for index, item in enumerate(items):
if not (normalized := await self._normalize(item, index)):
if not (normalized := await self._normalize(item, index, seen_names)):
continue
try:
await self._repo.create(normalized)
@ -58,17 +59,26 @@ class Migration(FeatureMigration):
LOG.info("Migrated %s condition(s) from %s.", inserted, self._source_file)
await self._move_file(self._source_file)
async def _normalize(self, item: Any, index: int) -> ConditionModel | None:
async def _normalize(self, item: Any, index: int, seen_names: dict[str, int]) -> ConditionModel | None:
if not isinstance(item, dict):
LOG.warning("Skipping condition at index %s due to invalid type.", index)
return None
assert isinstance(item, dict)
name: str | None = item.get("name")
if not name or not isinstance(name, str):
LOG.warning("Skipping condition at index %s due to missing name.", index)
return None
extras: dict[str, Any] = item.get("extras") if isinstance(item.get("extras"), dict) else {}
normalized_name = name.strip()
if not normalized_name:
LOG.warning("Skipping condition at index %s due to empty name.", index)
return None
name = self._unique_name(normalized_name, seen_names)
extras = item.get("extras") if isinstance(item.get("extras"), dict) else {}
extras = cast("dict[str, Any]", extras)
filter_value: str | None = item.get("filter")
if (not filter_value or not isinstance(filter_value, str)) and len(extras) == 0:
LOG.warning("Skipping condition '%s' due to missing filter.", name)
@ -78,12 +88,15 @@ class Migration(FeatureMigration):
if not isinstance(cli, str):
cli = ""
enabled: bool = item.get("enabled") if isinstance(item.get("enabled"), bool) else True
priority: int = item.get("priority") if isinstance(item.get("priority"), int) else 0
enabled_value = item.get("enabled")
enabled: bool = enabled_value if isinstance(enabled_value, bool) else True
priority_value = item.get("priority")
priority: int = priority_value if isinstance(priority_value, int) else 0
if isinstance(priority, bool) or 0 > priority:
priority = 0
description: str = item.get("description") if isinstance(item.get("description"), str) else ""
description_value = item.get("description")
description: str = description_value if isinstance(description_value, str) else ""
return ConditionModel(
id=index,

17
app/features/core/migration.py
View file

@ -48,3 +48,20 @@ class Migration(abc.ABC):
shutil.move(str(source), str(destination))
return destination
def _unique_name(self, name: str, seen_names: dict[str, int]) -> str:
base = name
count = seen_names.get(base, 0)
if count == 0:
seen_names[base] = 1
return base
suffix = count + 1
new_name = f"{base} ({suffix})"
while new_name in seen_names:
suffix += 1
new_name = f"{base} ({suffix})"
seen_names[base] = suffix
seen_names[new_name] = 1
return new_name

18
app/features/dl_fields/migration.py
View file

@ -3,7 +3,7 @@ from __future__ import annotations
import json
import logging
from pathlib import Path
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Any, cast
from app.features.core.migration import Migration as FeatureMigration
from app.features.dl_fields.schemas import DLField
@ -46,8 +46,9 @@ class Migration(FeatureMigration):
return
inserted = 0
seen_names: dict[str, int] = {}
for index, item in enumerate(items):
if not (normalized := self._normalize(item, index)):
if not (normalized := self._normalize(item, index, seen_names)):
continue
try:
await self._repo.create(normalized)
@ -58,16 +59,24 @@ class Migration(FeatureMigration):
LOG.info("Migrated %s dl field(s) from %s.", inserted, self._source_file)
await self._move_file(self._source_file)
def _normalize(self, item: Any, index: int) -> dict[str, Any] | None:
def _normalize(self, item: Any, index: int, seen_names: dict[str, int]) -> dict[str, Any] | None:
if not isinstance(item, dict):
LOG.warning("Skipping dl field at index %s due to invalid type.", index)
return None
assert isinstance(item, dict)
name: str | None = item.get("name")
if not name or not isinstance(name, str):
LOG.warning("Skipping dl field at index %s due to missing name.", index)
return None
normalized_name = name.strip()
if not normalized_name:
LOG.warning("Skipping dl field at index %s due to empty name.", index)
return None
name = self._unique_name(normalized_name, seen_names)
description: str = item.get("description") if isinstance(item.get("description"), str) else ""
field_value: str | None = item.get("field")
if not field_value or not isinstance(field_value, str):
@ -77,7 +86,8 @@ class Migration(FeatureMigration):
kind: str = item.get("kind") if isinstance(item.get("kind"), str) else "text"
icon: str = item.get("icon") if isinstance(item.get("icon"), str) else ""
value: str = item.get("value") if isinstance(item.get("value"), str) else ""
extras: dict[str, Any] = item.get("extras") if isinstance(item.get("extras"), dict) else {}
extras = item.get("extras") if isinstance(item.get("extras"), dict) else {}
extras = cast("dict[str, Any]", extras)
order_value: int = 0
if isinstance(item.get("order"), int) and not isinstance(item.get("order"), bool):

27
app/features/notifications/migration.py
View file

@ -3,7 +3,7 @@ from __future__ import annotations
import json
import logging
from pathlib import Path
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Any, cast
from app.features.core.migration import Migration as FeatureMigration
from app.features.notifications.schemas import NotificationEvents
@ -47,8 +47,9 @@ class Migration(FeatureMigration):
return
inserted = 0
seen_names: dict[str, int] = {}
for index, item in enumerate(items):
normalized = self._normalize(item, index)
normalized = self._normalize(item, index, seen_names)
if not normalized:
continue
try:
@ -60,30 +61,39 @@ class Migration(FeatureMigration):
LOG.info("Migrated %s notification target(s) from %s.", inserted, self._source_file)
await self._move_file(self._source_file)
def _normalize(self, item: Any, index: int) -> dict[str, Any] | None:
def _normalize(self, item: Any, index: int, seen_names: dict[str, int]) -> dict[str, Any] | None:
if not isinstance(item, dict):
LOG.warning("Skipping notification at index %s due to invalid type.", index)
return None
assert isinstance(item, dict)
name: str | None = item.get("name")
if not name or not isinstance(name, str):
LOG.warning("Skipping notification at index %s due to missing name.", index)
return None
normalized_name = name.strip()
if not normalized_name:
LOG.warning("Skipping notification at index %s due to empty name.", index)
return None
name = self._unique_name(normalized_name, seen_names)
request = item.get("request") if isinstance(item.get("request"), dict) else {}
request = cast("dict[str, Any]", request)
url: str | None = request.get("url")
if not url or not isinstance(url, str):
LOG.warning("Skipping notification '%s' due to missing request url.", name)
return None
method = request.get("method") if isinstance(request.get("method"), str) else "POST"
method = method.upper()
raw_method = request.get("method")
method = raw_method.upper() if isinstance(raw_method, str) else "POST"
if method not in {"POST", "PUT"}:
LOG.warning("Skipping notification '%s' due to invalid method '%s'.", name, method)
return None
req_type = request.get("type") if isinstance(request.get("type"), str) else "json"
req_type = req_type.lower()
raw_type = request.get("type")
req_type = raw_type.lower() if isinstance(raw_type, str) else "json"
if req_type not in {"json", "form"}:
LOG.warning("Skipping notification '%s' due to invalid type '%s'.", name, req_type)
return None
@ -94,7 +104,8 @@ class Migration(FeatureMigration):
headers = self._normalize_headers(request.get("headers"))
enabled: bool = item.get("enabled") if isinstance(item.get("enabled"), bool) else True
enabled_value = item.get("enabled")
enabled: bool = enabled_value if isinstance(enabled_value, bool) else True
events = self._normalize_events(item.get("on"))
if events is None:
LOG.warning("Skipping notification '%s' due to invalid events.", name)