mirror of
https://github.com/Second-Hand-Friends/kleinanzeigen-bot.git
synced 2026-07-09 12:41:05 +02:00
refactor: extract local path renaming primitives and orchestrator into domain module (#1069)
## ℹ️ Description Extract local path renaming logic from `kleinanzeigen_bot/__init__.py` into a pure domain module `kleinanzeigen_bot/local_path_renaming.py`. The new module has explicit `__all__`, public operation names, and zero dependency on `KleinanzeigenBot`. Bot methods are reduced to thin wrapper adapters that derive config values from `self.config`. ## 📋 Changes Summary - **New:** `src/kleinanzeigen_bot/local_path_renaming.py` — types, template-slot replacement, path-rename helpers, file/folder/image orchestrator functions - **Modified:** `src/kleinanzeigen_bot/__init__.py` — bot methods reduced to thin wrappers; coordinator (~30 lines → 6 line adapter) - **Modified:** `src/kleinanzeigen_bot/resources/translations.de.yaml` — moved translations for renamed functions - **New:** `tests/unit/test_local_path_renaming.py` — 22 focused tests importing directly from the module (no bot construction) - **Modified:** `tests/unit/test_init.py` — removed moved tests; kept compat alias verification ### ⚙️ Type of Change - [x] ✨ Refactor (non-breaking internal restructuring) ## ✅ Checklist - [x] I have reviewed my changes to ensure they meet the project's standards. - [x] I have tested my changes and ensured that all tests pass (`pdm run test`). - [x] I have formatted the code (`pdm run format`). - [x] I have verified that linting passes (`pdm run lint`). - [x] I have updated documentation where necessary. <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * **Refactor** * Moved local path renaming into a dedicated component; bot now delegates renames via thin wrappers and re-exports renaming outcomes. * Added template-based ID matching and safer rename behavior for ad files, ad folders, and referenced images (avoids overwrites, handles symlink/collision cases). * **Tests** * Expanded unit tests for template matching, collision/symlink handling, image-reference updates, per-image skip rules, and end-to-end rename scenarios; init tests adjusted. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
@@ -1,15 +1,13 @@
|
||||
# SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
|
||||
# SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
|
||||
import atexit, asyncio, enum, functools, json, os, re, signal, sys, textwrap # isort: skip
|
||||
import atexit, asyncio, enum, json, os, re, signal, sys, textwrap # isort: skip
|
||||
import getopt # pylint: disable=deprecated-module
|
||||
import urllib.parse as urllib_parse
|
||||
from dataclasses import dataclass, replace
|
||||
from dataclasses import field as dc_field # noqa: I001
|
||||
from datetime import datetime
|
||||
from gettext import gettext as _
|
||||
from pathlib import Path
|
||||
from string import Formatter
|
||||
from typing import Any, Final, Literal, NamedTuple, Sequence, cast
|
||||
|
||||
import certifi, colorama, nodriver # isort: skip
|
||||
@@ -19,6 +17,39 @@ from wcmatch import glob
|
||||
|
||||
from . import extract, resources
|
||||
from ._version import __version__
|
||||
from .local_path_renaming import (
|
||||
DOWNLOAD_IMAGE_FILENAME_RE as _DOWNLOAD_IMAGE_FILENAME_RE, # noqa: F401
|
||||
)
|
||||
from .local_path_renaming import (
|
||||
ImageRenameResult as ImageRenameResult,
|
||||
)
|
||||
from .local_path_renaming import (
|
||||
LocalPathRenameResult as LocalPathRenameResult,
|
||||
)
|
||||
from .local_path_renaming import (
|
||||
RenamePathResult as RenamePathResult,
|
||||
)
|
||||
from .local_path_renaming import (
|
||||
RenameStatus as RenameStatus,
|
||||
)
|
||||
from .local_path_renaming import (
|
||||
rename_local_ad_file_after_id_change as _rename_local_ad_file_after_id_change_impl,
|
||||
)
|
||||
from .local_path_renaming import (
|
||||
rename_local_ad_file_and_folder_after_id_change as _rename_local_ad_file_and_folder_after_id_change_impl,
|
||||
)
|
||||
from .local_path_renaming import (
|
||||
rename_local_ad_folder_after_id_change as _rename_local_ad_folder_after_id_change_impl,
|
||||
)
|
||||
from .local_path_renaming import (
|
||||
rename_path_if_target_is_free as _rename_path_if_target_is_free, # noqa: F401
|
||||
)
|
||||
from .local_path_renaming import (
|
||||
rename_referenced_local_image_files_after_id_change as _rename_referenced_local_image_files_after_id_change_impl,
|
||||
)
|
||||
from .local_path_renaming import (
|
||||
replace_template_id_slot as _replace_template_id_slot, # noqa: F401
|
||||
)
|
||||
from .model.ad_model import (
|
||||
CARRIER_CODE_BY_OPTION,
|
||||
CARRIER_CODES_BY_SIZE,
|
||||
@@ -73,7 +104,6 @@ LOG.setLevel(loggers.INFO)
|
||||
|
||||
SUBMISSION_MAX_RETRIES:Final[int] = 3
|
||||
_NUMERIC_IDS_RE:Final[re.Pattern[str]] = re.compile(r"^\d+(,\d+)*$")
|
||||
_DOWNLOAD_IMAGE_FILENAME_RE:Final[re.Pattern[str]] = re.compile(r"^(?P<prefix>.+?)(?P<image_suffix>__img\d+(?:\..*)?)$")
|
||||
_SPECIAL_ATTRIBUTE_TOKEN_RE:Final[re.Pattern[str]] = re.compile(r"^[A-Za-z0-9_]+$")
|
||||
_LOGIN_ENV_PATTERN:Final[re.Pattern[str]] = re.compile(r"^\$\{(?P<var>\w+)(?::-(?P<default>.*))?\}$")
|
||||
# See issue #930 for migrating __select_button_combobox to web_select_button_combobox
|
||||
@@ -146,100 +176,6 @@ def _xpath_literal(value:str) -> str:
|
||||
return "concat(" + ', "\'", '.join(f"'{part}'" for part in value.split("'")) + ")"
|
||||
|
||||
|
||||
@functools.lru_cache(maxsize = 32)
|
||||
def _build_id_slot_regex(template:str) -> re.Pattern[str]:
|
||||
"""Build a compiled regex from a download template with named groups.
|
||||
|
||||
Returns a regex where {id} captures any digit run as group 'id'
|
||||
and {title} matches any text non-greedily.
|
||||
"""
|
||||
regex_parts:list[str] = []
|
||||
parsed_template = list(Formatter().parse(template))
|
||||
for literal_text, field_name, _format_spec, _conversion in parsed_template:
|
||||
regex_parts.append(re.escape(literal_text))
|
||||
if field_name == "id":
|
||||
regex_parts.append(r"(?P<id>\d+)")
|
||||
elif field_name == "title":
|
||||
regex_parts.append(".*?")
|
||||
return re.compile("".join(regex_parts))
|
||||
|
||||
|
||||
def _replace_template_id_slot(template:str, name:str, new_id:int) -> tuple[str | None, int | None]:
|
||||
"""Replace the numeric ID in the template-defined {id} slot with new_id.
|
||||
|
||||
The {id} slot matches any sequence of digits; renaming is skipped if the
|
||||
matched ID already equals new_id. The {title} slot uses non-greedy
|
||||
matching so the {id} slot greedily captures the maximal digit run — even
|
||||
when {title} and {id} are adjacent in the template. This preserves
|
||||
user-edited or previously truncated title fragments instead of re-rendering.
|
||||
|
||||
Returns ``(new_name, old_id)`` where ``new_name`` is None when no rename
|
||||
is needed and ``old_id`` is the integer ID extracted from the path.
|
||||
"""
|
||||
match = _build_id_slot_regex(template).fullmatch(name)
|
||||
if match is None:
|
||||
return None, None
|
||||
|
||||
old_id_in_path = int(match.group("id"))
|
||||
if old_id_in_path == new_id:
|
||||
return None, old_id_in_path
|
||||
|
||||
id_start, id_end = match.span("id")
|
||||
return f"{name[:id_start]}{new_id}{name[id_end:]}", old_id_in_path
|
||||
|
||||
|
||||
class RenameStatus(enum.Enum):
|
||||
"""Outcome of an attempted local path rename."""
|
||||
RENAMED = enum.auto()
|
||||
SAME = enum.auto()
|
||||
NO_MATCH = enum.auto()
|
||||
TARGET_EXISTS = enum.auto()
|
||||
ERROR = enum.auto()
|
||||
|
||||
|
||||
@dataclass(frozen = True)
|
||||
class RenamePathResult:
|
||||
"""Result of a local path rename operation."""
|
||||
path:Path
|
||||
status:RenameStatus
|
||||
|
||||
|
||||
@dataclass(frozen = True)
|
||||
class LocalPathRenameResult:
|
||||
"""Aggregate result of local path renaming after a successful republish."""
|
||||
ad_file:Path
|
||||
file_status:RenameStatus
|
||||
folder_status:RenameStatus
|
||||
renamed_image_count:int = 0
|
||||
blocked_image_count:int = 0
|
||||
path_old_id:int | None = None
|
||||
yaml_old_id:int | None = None
|
||||
|
||||
|
||||
@dataclass(frozen = True)
|
||||
class ImageRenameResult:
|
||||
"""Outcome of renaming referenced local image files."""
|
||||
renamed_count:int = 0
|
||||
blocked_count:int = 0
|
||||
updated_images:list[object] | None = None
|
||||
renamed_paths:list[tuple[Path, Path]] = dc_field(default_factory = list)
|
||||
|
||||
|
||||
def _rename_path_if_target_is_free(source:Path, target:Path, *, label:str) -> RenamePathResult:
|
||||
if source == target:
|
||||
return RenamePathResult(source, RenameStatus.SAME)
|
||||
if target.exists() or target.is_symlink():
|
||||
LOG.debug("Skipping local %s rename because target already exists: %s", label, target)
|
||||
return RenamePathResult(source, RenameStatus.TARGET_EXISTS)
|
||||
try:
|
||||
source.rename(target)
|
||||
except OSError as ex:
|
||||
LOG.warning("Could not rename local %s from %s to %s: %s", label, source, target, ex)
|
||||
return RenamePathResult(source, RenameStatus.ERROR)
|
||||
LOG.debug("Renamed local %s from %s to %s", label, source, target)
|
||||
return RenamePathResult(target, RenameStatus.RENAMED)
|
||||
|
||||
|
||||
class LoginDetectionReason(enum.Enum):
|
||||
USER_INFO_MATCH = enum.auto()
|
||||
CTA_MATCH = enum.auto()
|
||||
@@ -1990,51 +1926,17 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904
|
||||
return await self.web_check(By.ID, "checking-done", Is.DISPLAYED) or await self.web_check(By.ID, "not-completed", Is.DISPLAYED)
|
||||
|
||||
def _rename_local_ad_file_and_folder_after_id_change(self, ad_file:Path, old_id:int | None, new_id:int) -> LocalPathRenameResult:
|
||||
if old_id is None or old_id == new_id or self.config.publishing.local_path_renaming.mode != "TEMPLATE_MATCH":
|
||||
return LocalPathRenameResult(
|
||||
ad_file = ad_file,
|
||||
file_status = RenameStatus.SAME,
|
||||
folder_status = RenameStatus.SAME,
|
||||
)
|
||||
|
||||
# resolve() is used intentionally: if the ad file path contains symlinks,
|
||||
# we rename the actual target path, not the symlink. This is the expected
|
||||
# behavior for local ad file management — we want to manage real files.
|
||||
ad_file = ad_file.resolve()
|
||||
download_config = self.config.download
|
||||
|
||||
# Extract the old ID from the path before renaming (used for logging provenance).
|
||||
# Try the file stem first; fall back to the parent folder name.
|
||||
__, path_old_id = _replace_template_id_slot(download_config.ad_file_name_template, ad_file.stem, new_id)
|
||||
if path_old_id is None:
|
||||
__, path_old_id = _replace_template_id_slot(download_config.folder_name_template, ad_file.parent.name, new_id)
|
||||
|
||||
file_result = self._rename_local_ad_file_after_id_change(
|
||||
return _rename_local_ad_file_and_folder_after_id_change_impl(
|
||||
ad_file,
|
||||
old_id = old_id,
|
||||
new_id = new_id,
|
||||
ad_file_name_template = download_config.ad_file_name_template,
|
||||
)
|
||||
folder_result = self._rename_local_ad_folder_after_id_change(
|
||||
file_result.path,
|
||||
new_id = new_id,
|
||||
folder_name_template = download_config.folder_name_template,
|
||||
)
|
||||
return LocalPathRenameResult(
|
||||
ad_file = folder_result.path,
|
||||
file_status = file_result.status,
|
||||
folder_status = folder_result.status,
|
||||
path_old_id = path_old_id,
|
||||
ad_file_name_template = self.config.download.ad_file_name_template,
|
||||
folder_name_template = self.config.download.folder_name_template,
|
||||
enabled = self.config.publishing.local_path_renaming.mode == "TEMPLATE_MATCH",
|
||||
)
|
||||
|
||||
def _rename_local_ad_file_after_id_change(self, ad_file:Path, *, new_id:int, ad_file_name_template:str) -> RenamePathResult:
|
||||
renamed_stem, path_id = _replace_template_id_slot(ad_file_name_template, ad_file.stem, new_id)
|
||||
if renamed_stem is None:
|
||||
if path_id == new_id:
|
||||
LOG.debug("Skipping local ad file rename because name already contains new ID: %s", ad_file)
|
||||
return RenamePathResult(ad_file, RenameStatus.SAME)
|
||||
LOG.debug("Skipping local ad file rename because name does not match configured template: %s", ad_file)
|
||||
return RenamePathResult(ad_file, RenameStatus.NO_MATCH)
|
||||
return _rename_path_if_target_is_free(ad_file, ad_file.with_name(f"{renamed_stem}{ad_file.suffix}"), label = _("ad file"))
|
||||
return _rename_local_ad_file_after_id_change_impl(ad_file, new_id = new_id, ad_file_name_template = ad_file_name_template)
|
||||
|
||||
def _rename_referenced_local_image_files_after_id_change(
|
||||
self,
|
||||
@@ -2043,93 +1945,17 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904
|
||||
old_id:int | None,
|
||||
new_id:int,
|
||||
) -> ImageRenameResult:
|
||||
if old_id is None or old_id == new_id or self.config.publishing.local_path_renaming.mode != "TEMPLATE_MATCH":
|
||||
return ImageRenameResult()
|
||||
|
||||
images = ad_cfg_orig.get("images")
|
||||
if not isinstance(images, list):
|
||||
return ImageRenameResult()
|
||||
|
||||
updated_images:list[object] = []
|
||||
renamed_count = 0
|
||||
blocked_count = 0
|
||||
renamed_paths:list[tuple[Path, Path]] = []
|
||||
for image_ref in images:
|
||||
updated_image_ref, status = self._rename_referenced_local_image_file_after_id_change(
|
||||
return _rename_referenced_local_image_files_after_id_change_impl(
|
||||
ad_file,
|
||||
image_ref,
|
||||
ad_cfg_orig.get("images"),
|
||||
old_id = old_id,
|
||||
new_id = new_id,
|
||||
ad_file_name_template = self.config.download.ad_file_name_template,
|
||||
enabled = self.config.publishing.local_path_renaming.mode == "TEMPLATE_MATCH",
|
||||
)
|
||||
updated_images.append(updated_image_ref)
|
||||
if status == RenameStatus.RENAMED:
|
||||
renamed_count += 1
|
||||
renamed_paths.append(
|
||||
(ad_file.parent / Path(str(image_ref)), ad_file.parent / Path(str(updated_image_ref)))
|
||||
)
|
||||
elif status in {RenameStatus.TARGET_EXISTS, RenameStatus.ERROR}:
|
||||
blocked_count += 1
|
||||
|
||||
if renamed_count > 0:
|
||||
return ImageRenameResult(
|
||||
renamed_count = renamed_count,
|
||||
blocked_count = blocked_count,
|
||||
updated_images = updated_images,
|
||||
renamed_paths = renamed_paths,
|
||||
)
|
||||
|
||||
return ImageRenameResult(renamed_count = renamed_count, blocked_count = blocked_count)
|
||||
|
||||
def _rename_referenced_local_image_file_after_id_change(
|
||||
self,
|
||||
ad_file:Path,
|
||||
image_ref:object,
|
||||
*,
|
||||
new_id:int,
|
||||
ad_file_name_template:str,
|
||||
) -> tuple[object, RenameStatus | None]:
|
||||
if not isinstance(image_ref, str):
|
||||
return image_ref, None
|
||||
|
||||
original_path = Path(image_ref)
|
||||
if original_path.is_absolute():
|
||||
return image_ref, None
|
||||
|
||||
image_path = ad_file.parent / original_path
|
||||
if not image_path.is_file():
|
||||
return image_ref, None
|
||||
if not image_path.resolve().is_relative_to(ad_file.parent.resolve()):
|
||||
return image_ref, None
|
||||
|
||||
match = _DOWNLOAD_IMAGE_FILENAME_RE.fullmatch(image_path.name)
|
||||
if match is None:
|
||||
return image_ref, None
|
||||
|
||||
renamed_prefix, path_id = _replace_template_id_slot(ad_file_name_template, match.group("prefix"), new_id)
|
||||
if renamed_prefix is None:
|
||||
return image_ref, RenameStatus.SAME if path_id == new_id else None
|
||||
|
||||
renamed_name = f"{renamed_prefix}{match.group('image_suffix')}"
|
||||
rename_result = _rename_path_if_target_is_free(image_path, image_path.with_name(renamed_name), label = _("image file"))
|
||||
if rename_result.status != RenameStatus.RENAMED:
|
||||
return image_ref, rename_result.status
|
||||
|
||||
return original_path.with_name(renamed_name).as_posix(), RenameStatus.RENAMED
|
||||
|
||||
def _rename_local_ad_folder_after_id_change(self, ad_file:Path, *, new_id:int, folder_name_template:str) -> RenamePathResult:
|
||||
parent = ad_file.parent
|
||||
renamed_folder_name, path_id = _replace_template_id_slot(folder_name_template, parent.name, new_id)
|
||||
if renamed_folder_name is None:
|
||||
if path_id == new_id:
|
||||
LOG.debug("Skipping local ad folder rename because name already contains new ID: %s", parent)
|
||||
return RenamePathResult(ad_file, RenameStatus.SAME)
|
||||
LOG.debug("Skipping local ad folder rename because name does not match configured template: %s", parent)
|
||||
return RenamePathResult(ad_file, RenameStatus.NO_MATCH)
|
||||
|
||||
result = _rename_path_if_target_is_free(parent, parent.with_name(renamed_folder_name), label = _("ad folder"))
|
||||
if result.status != RenameStatus.RENAMED:
|
||||
return RenamePathResult(ad_file, result.status)
|
||||
return RenamePathResult(result.path / ad_file.name, RenameStatus.RENAMED)
|
||||
return _rename_local_ad_folder_after_id_change_impl(ad_file, new_id = new_id, folder_name_template = folder_name_template)
|
||||
|
||||
def _log_local_path_rename_result(self, result:LocalPathRenameResult, ad_id:int) -> None:
|
||||
"""Log a human-readable summary of local path renaming after a republish."""
|
||||
|
||||
304
src/kleinanzeigen_bot/local_path_renaming.py
Normal file
304
src/kleinanzeigen_bot/local_path_renaming.py
Normal file
@@ -0,0 +1,304 @@
|
||||
# SPDX-FileCopyrightText: © Jens Bergmann and contributors
|
||||
# SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
|
||||
|
||||
"""Local path renaming helpers for ad files, folders, and referenced images."""
|
||||
|
||||
import enum, functools, re # isort: skip
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import field as dc_field
|
||||
from gettext import gettext as _
|
||||
from pathlib import Path
|
||||
from string import Formatter
|
||||
from typing import Final
|
||||
|
||||
from kleinanzeigen_bot.utils import loggers
|
||||
|
||||
LOG:Final[loggers.Logger] = loggers.get_logger(__name__)
|
||||
|
||||
__all__ = [
|
||||
"ImageRenameResult",
|
||||
"LocalPathRenameResult",
|
||||
"RenamePathResult",
|
||||
"RenameStatus",
|
||||
"rename_local_ad_file_after_id_change",
|
||||
"rename_local_ad_file_and_folder_after_id_change",
|
||||
"rename_local_ad_folder_after_id_change",
|
||||
"rename_path_if_target_is_free",
|
||||
"rename_referenced_local_image_file_after_id_change",
|
||||
"rename_referenced_local_image_files_after_id_change",
|
||||
"replace_template_id_slot",
|
||||
]
|
||||
|
||||
DOWNLOAD_IMAGE_FILENAME_RE:Final[re.Pattern[str]] = re.compile(r"^(?P<prefix>.+?)(?P<image_suffix>__img\d+(?:\..*)?)$")
|
||||
|
||||
|
||||
@functools.lru_cache(maxsize = 32)
|
||||
def _build_id_slot_regex(template:str) -> re.Pattern[str]:
|
||||
"""Build a compiled regex from a download template with named groups.
|
||||
|
||||
Returns a regex where {id} captures any digit run as group 'id'
|
||||
and {title} matches any text non-greedily.
|
||||
"""
|
||||
regex_parts:list[str] = []
|
||||
parsed_template = list(Formatter().parse(template))
|
||||
for literal_text, field_name, _format_spec, _conversion in parsed_template:
|
||||
regex_parts.append(re.escape(literal_text))
|
||||
if field_name == "id":
|
||||
regex_parts.append(r"(?P<id>\d+)")
|
||||
elif field_name == "title":
|
||||
regex_parts.append(".*?")
|
||||
return re.compile("".join(regex_parts))
|
||||
|
||||
|
||||
def replace_template_id_slot(template:str, name:str, new_id:int) -> tuple[str | None, int | None]:
|
||||
"""Replace the numeric ID in the template-defined {id} slot with new_id.
|
||||
|
||||
The {id} slot matches any sequence of digits; renaming is skipped if the
|
||||
matched ID already equals new_id. The {title} slot uses non-greedy
|
||||
matching so the {id} slot greedily captures the maximal digit run — even
|
||||
when {title} and {id} are adjacent in the template. This preserves
|
||||
user-edited or previously truncated title fragments instead of re-rendering.
|
||||
|
||||
Returns ``(new_name, old_id)`` where ``new_name`` is None when no rename
|
||||
is needed and ``old_id`` is the integer ID extracted from the path.
|
||||
|
||||
.. note::
|
||||
|
||||
The *template* must have already passed download-config validation
|
||||
(i.e. it contains an ``{id}`` placeholder and each placeholder
|
||||
appears at most once). Malformed templates may raise ``IndexError``.
|
||||
"""
|
||||
match = _build_id_slot_regex(template).fullmatch(name)
|
||||
if match is None:
|
||||
return None, None
|
||||
|
||||
old_id_in_path = int(match.group("id"))
|
||||
if old_id_in_path == new_id:
|
||||
return None, old_id_in_path
|
||||
|
||||
id_start, id_end = match.span("id")
|
||||
return f"{name[:id_start]}{new_id}{name[id_end:]}", old_id_in_path
|
||||
|
||||
|
||||
class RenameStatus(enum.Enum):
|
||||
"""Outcome of an attempted local path rename."""
|
||||
RENAMED = enum.auto()
|
||||
SAME = enum.auto()
|
||||
NO_MATCH = enum.auto()
|
||||
TARGET_EXISTS = enum.auto()
|
||||
ERROR = enum.auto()
|
||||
|
||||
|
||||
@dataclass(frozen = True)
|
||||
class RenamePathResult:
|
||||
"""Result of a local path rename operation."""
|
||||
path:Path
|
||||
status:RenameStatus
|
||||
|
||||
|
||||
@dataclass(frozen = True)
|
||||
class LocalPathRenameResult:
|
||||
"""Aggregate result of local path renaming after a successful republish."""
|
||||
ad_file:Path
|
||||
file_status:RenameStatus
|
||||
folder_status:RenameStatus
|
||||
renamed_image_count:int = 0
|
||||
blocked_image_count:int = 0
|
||||
path_old_id:int | None = None
|
||||
yaml_old_id:int | None = None
|
||||
|
||||
|
||||
@dataclass(frozen = True)
|
||||
class ImageRenameResult:
|
||||
"""Outcome of renaming referenced local image files."""
|
||||
renamed_count:int = 0
|
||||
blocked_count:int = 0
|
||||
updated_images:list[object] | None = None
|
||||
renamed_paths:list[tuple[Path, Path]] = dc_field(default_factory = list)
|
||||
|
||||
|
||||
def rename_path_if_target_is_free(source:Path, target:Path, *, label:str) -> RenamePathResult:
|
||||
if source == target:
|
||||
return RenamePathResult(source, RenameStatus.SAME)
|
||||
if target.exists() or target.is_symlink():
|
||||
LOG.debug("Skipping local %s rename because target already exists: %s", label, target)
|
||||
return RenamePathResult(source, RenameStatus.TARGET_EXISTS)
|
||||
try:
|
||||
source.rename(target)
|
||||
except OSError as ex:
|
||||
LOG.warning("Could not rename local %s from %s to %s: %s", label, source, target, ex)
|
||||
return RenamePathResult(source, RenameStatus.ERROR)
|
||||
LOG.debug("Renamed local %s from %s to %s", label, source, target)
|
||||
return RenamePathResult(target, RenameStatus.RENAMED)
|
||||
|
||||
|
||||
def rename_local_ad_file_after_id_change(ad_file:Path, *, new_id:int, ad_file_name_template:str) -> RenamePathResult:
|
||||
"""Rename a local ad file when the ad ID changes.
|
||||
|
||||
Matches the file stem against the configured download template, replaces
|
||||
the {id} slot, and renames the file when the target name is free.
|
||||
"""
|
||||
renamed_stem, path_id = replace_template_id_slot(ad_file_name_template, ad_file.stem, new_id)
|
||||
if renamed_stem is None:
|
||||
if path_id == new_id:
|
||||
LOG.debug("Skipping local ad file rename because name already contains new ID: %s", ad_file)
|
||||
return RenamePathResult(ad_file, RenameStatus.SAME)
|
||||
LOG.debug("Skipping local ad file rename because name does not match configured template: %s", ad_file)
|
||||
return RenamePathResult(ad_file, RenameStatus.NO_MATCH)
|
||||
return rename_path_if_target_is_free(ad_file, ad_file.with_name(f"{renamed_stem}{ad_file.suffix}"), label = _("ad file"))
|
||||
|
||||
|
||||
def rename_referenced_local_image_file_after_id_change(
|
||||
ad_file:Path,
|
||||
image_ref:object,
|
||||
*,
|
||||
new_id:int,
|
||||
ad_file_name_template:str,
|
||||
) -> tuple[object, RenameStatus | None]:
|
||||
"""Rename a single referenced local image when the ad ID changes.
|
||||
|
||||
Only renames images whose filename matches the DOWNLOAD_IMAGE_FILENAME_RE
|
||||
pattern and that live within the ad directory. Returns
|
||||
``(post_rename_image_ref, status)``.
|
||||
"""
|
||||
if not isinstance(image_ref, str):
|
||||
return image_ref, None
|
||||
|
||||
original_path = Path(image_ref)
|
||||
if original_path.is_absolute():
|
||||
return image_ref, None
|
||||
|
||||
image_path = ad_file.parent / original_path
|
||||
if not image_path.is_file():
|
||||
return image_ref, None
|
||||
if not image_path.resolve().is_relative_to(ad_file.parent.resolve()):
|
||||
return image_ref, None
|
||||
|
||||
match = DOWNLOAD_IMAGE_FILENAME_RE.fullmatch(image_path.name)
|
||||
if match is None:
|
||||
return image_ref, None
|
||||
|
||||
renamed_prefix, path_id = replace_template_id_slot(ad_file_name_template, match.group("prefix"), new_id)
|
||||
if renamed_prefix is None:
|
||||
return image_ref, RenameStatus.SAME if path_id == new_id else None
|
||||
|
||||
renamed_name = f"{renamed_prefix}{match.group('image_suffix')}"
|
||||
rename_result = rename_path_if_target_is_free(image_path, image_path.with_name(renamed_name), label = _("image file"))
|
||||
if rename_result.status != RenameStatus.RENAMED:
|
||||
return image_ref, rename_result.status
|
||||
|
||||
return original_path.with_name(renamed_name).as_posix(), RenameStatus.RENAMED
|
||||
|
||||
|
||||
def rename_referenced_local_image_files_after_id_change(
|
||||
ad_file:Path,
|
||||
images:object,
|
||||
*,
|
||||
old_id:int | None,
|
||||
new_id:int,
|
||||
ad_file_name_template:str,
|
||||
enabled:bool,
|
||||
) -> ImageRenameResult:
|
||||
"""Rename all referenced local image files when the ad ID changes.
|
||||
|
||||
``images`` is the image-ref list from the ad config. ``enabled`` should
|
||||
be ``True`` when ``publishing.local_path_renaming.mode == "TEMPLATE_MATCH"``.
|
||||
"""
|
||||
if old_id is None or old_id == new_id or not enabled:
|
||||
return ImageRenameResult()
|
||||
if images is None or not isinstance(images, list):
|
||||
return ImageRenameResult()
|
||||
|
||||
updated_images:list[object] = []
|
||||
renamed_count = 0
|
||||
blocked_count = 0
|
||||
renamed_paths:list[tuple[Path, Path]] = []
|
||||
for image_ref in images:
|
||||
updated_image_ref, status = rename_referenced_local_image_file_after_id_change(
|
||||
ad_file,
|
||||
image_ref,
|
||||
new_id = new_id,
|
||||
ad_file_name_template = ad_file_name_template,
|
||||
)
|
||||
updated_images.append(updated_image_ref)
|
||||
if status == RenameStatus.RENAMED:
|
||||
renamed_count += 1
|
||||
renamed_paths.append(
|
||||
(ad_file.parent / Path(str(image_ref)), ad_file.parent / Path(str(updated_image_ref)))
|
||||
)
|
||||
elif status in {RenameStatus.TARGET_EXISTS, RenameStatus.ERROR}:
|
||||
blocked_count += 1
|
||||
|
||||
if renamed_count > 0:
|
||||
return ImageRenameResult(
|
||||
renamed_count = renamed_count,
|
||||
blocked_count = blocked_count,
|
||||
updated_images = updated_images,
|
||||
renamed_paths = renamed_paths,
|
||||
)
|
||||
|
||||
return ImageRenameResult(renamed_count = renamed_count, blocked_count = blocked_count)
|
||||
|
||||
|
||||
def rename_local_ad_folder_after_id_change(ad_file:Path, *, new_id:int, folder_name_template:str) -> RenamePathResult:
|
||||
"""Rename the parent folder of an ad file when the ad ID changes.
|
||||
|
||||
The returned ``RenamePathResult.path`` always points to the (possibly
|
||||
renamed) ad file inside its (possibly renamed) parent folder.
|
||||
"""
|
||||
parent = ad_file.parent
|
||||
renamed_folder_name, path_id = replace_template_id_slot(folder_name_template, parent.name, new_id)
|
||||
if renamed_folder_name is None:
|
||||
if path_id == new_id:
|
||||
LOG.debug("Skipping local ad folder rename because name already contains new ID: %s", parent)
|
||||
return RenamePathResult(ad_file, RenameStatus.SAME)
|
||||
LOG.debug("Skipping local ad folder rename because name does not match configured template: %s", parent)
|
||||
return RenamePathResult(ad_file, RenameStatus.NO_MATCH)
|
||||
|
||||
result = rename_path_if_target_is_free(parent, parent.with_name(renamed_folder_name), label = _("ad folder"))
|
||||
if result.status != RenameStatus.RENAMED:
|
||||
return RenamePathResult(ad_file, result.status)
|
||||
return RenamePathResult(result.path / ad_file.name, RenameStatus.RENAMED)
|
||||
|
||||
|
||||
def rename_local_ad_file_and_folder_after_id_change(
|
||||
ad_file:Path,
|
||||
*,
|
||||
old_id:int | None,
|
||||
new_id:int,
|
||||
ad_file_name_template:str,
|
||||
folder_name_template:str,
|
||||
enabled:bool,
|
||||
) -> LocalPathRenameResult:
|
||||
"""Rename an ad file and its parent folder when the ad ID changes.
|
||||
|
||||
This is the main coordinator: it gates on ``enabled`` / ``old_id``,
|
||||
resolves symlinks, extracts the old path ID, then renames the file
|
||||
and folder in sequence.
|
||||
"""
|
||||
if old_id is None or old_id == new_id or not enabled:
|
||||
return LocalPathRenameResult(
|
||||
ad_file = ad_file,
|
||||
file_status = RenameStatus.SAME,
|
||||
folder_status = RenameStatus.SAME,
|
||||
)
|
||||
|
||||
# resolve() is used intentionally: if the ad file path contains symlinks,
|
||||
# we rename the actual target path, not the symlink.
|
||||
ad_file = ad_file.resolve()
|
||||
|
||||
# Extract the old ID from the path before renaming (used for logging provenance).
|
||||
# Try the file stem first; fall back to the parent folder name.
|
||||
__, path_old_id = replace_template_id_slot(ad_file_name_template, ad_file.stem, new_id)
|
||||
if path_old_id is None:
|
||||
__, path_old_id = replace_template_id_slot(folder_name_template, ad_file.parent.name, new_id)
|
||||
|
||||
file_result = rename_local_ad_file_after_id_change(ad_file, new_id = new_id, ad_file_name_template = ad_file_name_template)
|
||||
folder_result = rename_local_ad_folder_after_id_change(file_result.path, new_id = new_id, folder_name_template = folder_name_template)
|
||||
return LocalPathRenameResult(
|
||||
ad_file = folder_result.path,
|
||||
file_status = file_result.status,
|
||||
folder_status = folder_result.status,
|
||||
path_old_id = path_old_id,
|
||||
)
|
||||
@@ -191,20 +191,6 @@ kleinanzeigen_bot/__init__.py:
|
||||
"DONE: (Re-)published %s": "FERTIG: %s (erneut) veröffentlicht"
|
||||
"ad": "Anzeige"
|
||||
|
||||
_rename_path_if_target_is_free:
|
||||
"Skipping local %s rename because target already exists: %s": "Lokale %s-Umbenennung übersprungen, da das Ziel bereits existiert: %s"
|
||||
"Could not rename local %s from %s to %s: %s": "Lokale %s konnte nicht von %s nach %s umbenannt werden: %s"
|
||||
"Renamed local %s from %s to %s": "Lokale %s von %s nach %s umbenannt"
|
||||
|
||||
_rename_local_ad_file_after_id_change:
|
||||
"ad file": "Anzeigendatei"
|
||||
|
||||
_rename_referenced_local_image_file_after_id_change:
|
||||
"image file": "Bilddatei"
|
||||
|
||||
_rename_local_ad_folder_after_id_change:
|
||||
"ad folder": "Anzeigenordner"
|
||||
|
||||
publish_ad:
|
||||
"Publishing ad '%s'...": "Veröffentliche Anzeige '%s'..."
|
||||
"Updating ad '%s'...": "Aktualisiere Anzeige '%s'..."
|
||||
@@ -371,6 +357,21 @@ kleinanzeigen_bot/__init__.py:
|
||||
"Processing %s/%s: '%s' from [%s]...": "Verarbeite %s/%s: '%s' von [%s]..."
|
||||
"ad": "Anzeige"
|
||||
|
||||
#################################################
|
||||
kleinanzeigen_bot/local_path_renaming.py:
|
||||
#################################################
|
||||
rename_path_if_target_is_free:
|
||||
"Could not rename local %s from %s to %s: %s": "Lokale %s konnte nicht von %s nach %s umbenannt werden: %s"
|
||||
|
||||
rename_local_ad_file_after_id_change:
|
||||
"ad file": "Anzeigendatei"
|
||||
|
||||
rename_referenced_local_image_file_after_id_change:
|
||||
"image file": "Bilddatei"
|
||||
|
||||
rename_local_ad_folder_after_id_change:
|
||||
"ad folder": "Anzeigenordner"
|
||||
|
||||
#################################################
|
||||
kleinanzeigen_bot/price_reduction.py:
|
||||
#################################################
|
||||
|
||||
@@ -13,7 +13,7 @@ import pytest
|
||||
from nodriver.core.connection import ProtocolException
|
||||
from pydantic import ValidationError
|
||||
|
||||
from kleinanzeigen_bot import (
|
||||
from kleinanzeigen_bot import ( # type: ignore[attr-defined]
|
||||
LOG,
|
||||
SUBMISSION_MAX_RETRIES,
|
||||
KleinanzeigenBot,
|
||||
@@ -165,26 +165,6 @@ def test_root_re_exports_resolve_correctly() -> None:
|
||||
assert PriceReductionDecision is _PriceReductionDecision_src
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("template", "name", "expected"),
|
||||
[
|
||||
("ad_{id}", "ad_123", "ad_456"),
|
||||
("ad_{id}_{title}", "ad_123_User edited title", "ad_456_User edited title"),
|
||||
("{title} ({id})", "User edited title (123)", "User edited title (456)"),
|
||||
("{id}", "123", "456"),
|
||||
("{title}{id}", "Bike123", "Bike456"),
|
||||
("{title}{id}", "123", "456"),
|
||||
("{id}{title}", "123Bike", "456Bike"),
|
||||
],
|
||||
)
|
||||
def test_replace_template_id_slot_preserves_non_id_text(template:str, name:str, expected:str) -> None:
|
||||
assert _replace_template_id_slot(template, name, 456)[0] == expected
|
||||
|
||||
|
||||
def test_replace_template_id_slot_skips_non_matching_name() -> None:
|
||||
assert _replace_template_id_slot("ad_{id}_{title}", "manual_123_Title", 456)[0] is None
|
||||
|
||||
|
||||
def test_rename_local_ad_file_and_folder_after_id_change_is_disabled_by_default(test_bot:KleinanzeigenBot, tmp_path:Path) -> None:
|
||||
folder = tmp_path / "ad_123_Title"
|
||||
folder.mkdir()
|
||||
@@ -228,129 +208,6 @@ def test_rename_local_ad_file_and_folder_after_id_change_renames_template_matche
|
||||
assert not folder.exists()
|
||||
|
||||
|
||||
def test_rename_referenced_local_image_files_after_id_change_updates_config_paths(test_bot:KleinanzeigenBot, tmp_path:Path) -> None:
|
||||
test_bot.config = Config.model_validate(
|
||||
{
|
||||
"login": {"username": "dummy", "password": "dummy"}, # noqa: S106
|
||||
"publishing": {"local_path_renaming": {"mode": "TEMPLATE_MATCH"}},
|
||||
}
|
||||
)
|
||||
folder = tmp_path / "ad_123_Title"
|
||||
nested = folder / "nested"
|
||||
nested.mkdir(parents = True)
|
||||
ad_file = folder / "ad_123.yaml"
|
||||
(folder / "ad_123__img1.jpeg").write_bytes(b"img1")
|
||||
(nested / "ad_123__img2.png").write_bytes(b"img2")
|
||||
(folder / "manual_123__img3.jpeg").write_bytes(b"manual")
|
||||
ad_cfg_orig:dict[str, Any] = {"images": ["ad_123__img1.jpeg", "nested/ad_123__img2.png", "manual_123__img3.jpeg"]}
|
||||
|
||||
image_result = test_bot._rename_referenced_local_image_files_after_id_change(ad_file, ad_cfg_orig, 123, 456)
|
||||
|
||||
assert image_result.updated_images == ["ad_456__img1.jpeg", "nested/ad_456__img2.png", "manual_123__img3.jpeg"]
|
||||
assert (folder / "ad_456__img1.jpeg").exists()
|
||||
assert (nested / "ad_456__img2.png").exists()
|
||||
assert (folder / "manual_123__img3.jpeg").exists()
|
||||
assert image_result.renamed_count == 2
|
||||
assert image_result.blocked_count == 0
|
||||
|
||||
|
||||
def test_rename_referenced_local_image_files_after_id_change_ignores_unreferenced_images(test_bot:KleinanzeigenBot, tmp_path:Path) -> None:
|
||||
test_bot.config = Config.model_validate(
|
||||
{
|
||||
"login": {"username": "dummy", "password": "dummy"}, # noqa: S106
|
||||
"publishing": {"local_path_renaming": {"mode": "TEMPLATE_MATCH"}},
|
||||
}
|
||||
)
|
||||
folder = tmp_path / "ad_123_Title"
|
||||
folder.mkdir()
|
||||
ad_file = folder / "ad_123.yaml"
|
||||
(folder / "ad_123__img1.jpeg").write_bytes(b"referenced")
|
||||
(folder / "ad_123__img2.jpeg").write_bytes(b"unreferenced")
|
||||
ad_cfg_orig:dict[str, Any] = {"images": ["ad_123__img1.jpeg"]}
|
||||
|
||||
image_result = test_bot._rename_referenced_local_image_files_after_id_change(ad_file, ad_cfg_orig, 123, 456)
|
||||
|
||||
assert image_result.updated_images == ["ad_456__img1.jpeg"]
|
||||
assert (folder / "ad_456__img1.jpeg").exists()
|
||||
assert (folder / "ad_123__img2.jpeg").exists()
|
||||
assert image_result.renamed_count == 1
|
||||
assert image_result.blocked_count == 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"scenario",
|
||||
["target_exists", "absolute_path", "outside_ad_folder"],
|
||||
)
|
||||
def test_rename_referenced_local_image_files_skips_when_unsafe(
|
||||
test_bot:KleinanzeigenBot,
|
||||
tmp_path:Path,
|
||||
scenario:str,
|
||||
) -> None:
|
||||
"""Image renames that could affect non-owned files are safely skipped."""
|
||||
test_bot.config = Config.model_validate(
|
||||
{
|
||||
"login": {"username": "dummy", "password": "dummy"}, # noqa: S106
|
||||
"publishing": {"local_path_renaming": {"mode": "TEMPLATE_MATCH"}},
|
||||
}
|
||||
)
|
||||
folder = tmp_path / "ad_123_Title"
|
||||
folder.mkdir()
|
||||
ad_file = folder / "ad_123.yaml"
|
||||
|
||||
source_file:Path | None = None
|
||||
target_file:Path | None = None
|
||||
if scenario == "target_exists":
|
||||
(folder / "ad_123__img1.jpeg").write_bytes(b"old")
|
||||
(folder / "ad_456__img1.jpeg").write_bytes(b"existing")
|
||||
images_config = ["ad_123__img1.jpeg"]
|
||||
source_file = folder / "ad_123__img1.jpeg"
|
||||
target_file = folder / "ad_456__img1.jpeg"
|
||||
elif scenario == "absolute_path":
|
||||
img = tmp_path / "ad_123__img1.jpeg"
|
||||
img.write_bytes(b"img")
|
||||
images_config = [str(img)]
|
||||
source_file = img
|
||||
elif scenario == "outside_ad_folder":
|
||||
img = tmp_path / "other" / "ad_123__img1.jpeg"
|
||||
img.parent.mkdir(parents = True)
|
||||
img.write_bytes(b"img")
|
||||
images_config = ["../other/ad_123__img1.jpeg"]
|
||||
source_file = img
|
||||
else:
|
||||
raise AssertionError(f"Unknown scenario: {scenario}")
|
||||
|
||||
ad_cfg_orig:dict[str, object] = {"images": images_config}
|
||||
image_result = test_bot._rename_referenced_local_image_files_after_id_change(ad_file, ad_cfg_orig, 123, 456)
|
||||
|
||||
assert image_result.updated_images is None
|
||||
assert source_file is not None
|
||||
assert source_file.exists()
|
||||
if target_file is not None:
|
||||
assert target_file.exists()
|
||||
assert image_result.renamed_count == 0
|
||||
if scenario == "target_exists":
|
||||
assert image_result.blocked_count == 1
|
||||
else:
|
||||
assert image_result.blocked_count == 0
|
||||
|
||||
|
||||
def test_rename_path_if_target_is_free_treats_broken_symlink_as_collision(tmp_path:Path) -> None:
|
||||
source = tmp_path / "source.txt"
|
||||
target = tmp_path / "target.txt"
|
||||
source.write_text("source", encoding = "utf-8")
|
||||
try:
|
||||
target.symlink_to(tmp_path / "missing.txt")
|
||||
except (OSError, NotImplementedError):
|
||||
pytest.skip("symlink not supported on this platform")
|
||||
|
||||
result = _rename_path_if_target_is_free(source, target, label = "test file")
|
||||
|
||||
assert result.path == source
|
||||
assert result.status == RenameStatus.TARGET_EXISTS
|
||||
assert source.exists()
|
||||
assert target.is_symlink()
|
||||
|
||||
|
||||
def test_rename_local_ad_file_and_folder_after_id_change_skips_manual_names(test_bot:KleinanzeigenBot, tmp_path:Path) -> None:
|
||||
test_bot.config = Config.model_validate(
|
||||
{
|
||||
@@ -400,6 +257,27 @@ def test_rename_local_ad_file_and_folder_after_id_change_skips_collisions(test_b
|
||||
assert not folder.exists()
|
||||
|
||||
|
||||
def test_replace_template_id_slot_underscore_alias_is_same_object() -> None:
|
||||
"""Root-package alias _replace_template_id_slot resolves to the module function."""
|
||||
from kleinanzeigen_bot.local_path_renaming import replace_template_id_slot # noqa: PLC0415
|
||||
|
||||
assert _replace_template_id_slot is replace_template_id_slot
|
||||
|
||||
|
||||
def test_rename_path_if_target_is_free_underscore_alias_works(tmp_path:Path) -> None:
|
||||
"""Underscored root-package alias _rename_path_if_target_is_free is callable."""
|
||||
source = tmp_path / "a.txt"
|
||||
target = tmp_path / "b.txt"
|
||||
source.write_text("hello", encoding = "utf-8")
|
||||
|
||||
result = _rename_path_if_target_is_free(source, target, label = "test item")
|
||||
|
||||
assert result.path == target
|
||||
assert result.status == RenameStatus.RENAMED
|
||||
assert target.read_text(encoding = "utf-8") == "hello"
|
||||
assert not source.exists()
|
||||
|
||||
|
||||
class TestKleinanzeigenBotInitialization:
|
||||
"""Tests for KleinanzeigenBot initialization and basic functionality."""
|
||||
|
||||
|
||||
359
tests/unit/test_local_path_renaming.py
Normal file
359
tests/unit/test_local_path_renaming.py
Normal file
@@ -0,0 +1,359 @@
|
||||
# SPDX-FileCopyrightText: © Jens Bergmann and contributors
|
||||
# SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from kleinanzeigen_bot.local_path_renaming import (
|
||||
DOWNLOAD_IMAGE_FILENAME_RE,
|
||||
ImageRenameResult,
|
||||
LocalPathRenameResult,
|
||||
RenamePathResult,
|
||||
RenameStatus,
|
||||
rename_local_ad_file_after_id_change,
|
||||
rename_local_ad_folder_after_id_change,
|
||||
rename_path_if_target_is_free,
|
||||
rename_referenced_local_image_file_after_id_change,
|
||||
rename_referenced_local_image_files_after_id_change,
|
||||
replace_template_id_slot,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("template", "name", "new_id", "expected_name", "expected_old_id"),
|
||||
[
|
||||
("ad_{id}", "ad_123", 456, "ad_456", 123),
|
||||
("ad_{id}_{title}", "ad_123_User edited title", 789, "ad_789_User edited title", 123),
|
||||
("{title} ({id})", "User edited title (123)", 456, "User edited title (456)", 123),
|
||||
("{id}", "123", 456, "456", 123),
|
||||
("{title}{id}", "Bike123", 456, "Bike456", 123),
|
||||
("{title}{id}", "123", 456, "456", 123),
|
||||
("{id}{title}", "123Bike", 456, "456Bike", 123),
|
||||
],
|
||||
)
|
||||
def test_replace_template_id_slot(template:str, name:str, new_id:int, expected_name:str, expected_old_id:int) -> None:
|
||||
result = replace_template_id_slot(template, name, new_id)
|
||||
assert result == (expected_name, expected_old_id)
|
||||
|
||||
|
||||
def test_replace_template_id_slot_skips_non_matching_name() -> None:
|
||||
assert replace_template_id_slot("ad_{id}_{title}", "manual_123_Title", 456)[0] is None
|
||||
|
||||
|
||||
def test_rename_path_if_target_is_free_treats_broken_symlink_as_collision(tmp_path:Path) -> None:
|
||||
source = tmp_path / "source.txt"
|
||||
target = tmp_path / "target.txt"
|
||||
source.write_text("source", encoding = "utf-8")
|
||||
try:
|
||||
target.symlink_to(tmp_path / "missing.txt")
|
||||
except (OSError, NotImplementedError):
|
||||
pytest.skip("symlink not supported on this platform")
|
||||
|
||||
result = rename_path_if_target_is_free(source, target, label = "test file")
|
||||
|
||||
assert result.path == source
|
||||
assert result.status == RenameStatus.TARGET_EXISTS
|
||||
assert source.exists()
|
||||
assert target.is_symlink()
|
||||
|
||||
|
||||
def test_download_image_filename_re_matches_image_suffix() -> None:
|
||||
"""DOWNLOAD_IMAGE_FILENAME_RE matches filenames with __imgN suffix."""
|
||||
assert DOWNLOAD_IMAGE_FILENAME_RE.fullmatch("ad_123__img1.jpeg") is not None
|
||||
assert DOWNLOAD_IMAGE_FILENAME_RE.fullmatch("ad_123__img42.png") is not None
|
||||
assert DOWNLOAD_IMAGE_FILENAME_RE.fullmatch("manual_123.txt") is None
|
||||
assert DOWNLOAD_IMAGE_FILENAME_RE.fullmatch("ad_123.jpeg") is None
|
||||
|
||||
|
||||
def test_public_api_types_are_exported() -> None:
|
||||
"""All result types are importable from kleinanzeigen_bot.local_path_renaming."""
|
||||
assert RenameStatus.RENAMED is not None
|
||||
assert RenamePathResult is not None
|
||||
assert LocalPathRenameResult is not None
|
||||
assert ImageRenameResult is not None
|
||||
|
||||
|
||||
def test_rename_local_ad_file_change_renames_matching_file(tmp_path:Path) -> None:
|
||||
folder = tmp_path / "ad_123_Title"
|
||||
folder.mkdir()
|
||||
ad_file = folder / "ad_123.yaml"
|
||||
ad_file.write_text("id: 456\n", encoding = "utf-8")
|
||||
|
||||
result = rename_local_ad_file_after_id_change(ad_file, new_id = 456, ad_file_name_template = "ad_{id}")
|
||||
|
||||
assert result.status == RenameStatus.RENAMED
|
||||
assert result.path == folder / "ad_456.yaml"
|
||||
assert result.path.exists()
|
||||
assert not ad_file.exists()
|
||||
|
||||
|
||||
def test_rename_local_ad_file_change_skips_when_id_already_matches(tmp_path:Path) -> None:
|
||||
folder = tmp_path / "ad_456_Title"
|
||||
folder.mkdir()
|
||||
ad_file = folder / "ad_456.yaml"
|
||||
ad_file.write_text("id: 456\n", encoding = "utf-8")
|
||||
|
||||
result = rename_local_ad_file_after_id_change(ad_file, new_id = 456, ad_file_name_template = "ad_{id}")
|
||||
|
||||
assert result.status == RenameStatus.SAME
|
||||
assert result.path == ad_file
|
||||
assert ad_file.exists()
|
||||
|
||||
|
||||
def test_rename_local_ad_file_change_skips_non_template_name(tmp_path:Path) -> None:
|
||||
folder = tmp_path / "manual_123"
|
||||
folder.mkdir()
|
||||
ad_file = folder / "manual_123.yaml"
|
||||
ad_file.write_text("id: 456\n", encoding = "utf-8")
|
||||
|
||||
result = rename_local_ad_file_after_id_change(ad_file, new_id = 456, ad_file_name_template = "ad_{id}")
|
||||
|
||||
assert result.status == RenameStatus.NO_MATCH
|
||||
assert result.path == ad_file
|
||||
assert ad_file.exists()
|
||||
|
||||
|
||||
def test_rename_local_ad_folder_change_renames_matching_folder(tmp_path:Path) -> None:
|
||||
folder = tmp_path / "ad_123_Title"
|
||||
folder.mkdir()
|
||||
ad_file = folder / "ad_123.yaml"
|
||||
ad_file.write_text("id: 456\n", encoding = "utf-8")
|
||||
|
||||
result = rename_local_ad_folder_after_id_change(ad_file, new_id = 456, folder_name_template = "ad_{id}_{title}")
|
||||
|
||||
assert result.status == RenameStatus.RENAMED
|
||||
assert result.path == tmp_path / "ad_456_Title" / "ad_123.yaml"
|
||||
assert result.path.parent.exists()
|
||||
assert not folder.exists()
|
||||
|
||||
|
||||
def test_rename_local_ad_folder_change_skips_when_id_already_matches(tmp_path:Path) -> None:
|
||||
folder = tmp_path / "ad_456_Title"
|
||||
folder.mkdir()
|
||||
ad_file = folder / "ad_123.yaml"
|
||||
ad_file.write_text("id: 456\n", encoding = "utf-8")
|
||||
|
||||
result = rename_local_ad_folder_after_id_change(ad_file, new_id = 456, folder_name_template = "ad_{id}_{title}")
|
||||
|
||||
assert result.status == RenameStatus.SAME
|
||||
assert result.path == ad_file
|
||||
assert folder.exists()
|
||||
|
||||
|
||||
def test_rename_referenced_local_image_files_updates_config_paths(tmp_path:Path) -> None:
|
||||
folder = tmp_path / "ad_123_Title"
|
||||
nested = folder / "nested"
|
||||
nested.mkdir(parents = True)
|
||||
ad_file = folder / "ad_123.yaml"
|
||||
(folder / "ad_123__img1.jpeg").write_bytes(b"img1")
|
||||
(nested / "ad_123__img2.png").write_bytes(b"img2")
|
||||
(folder / "manual_123__img3.jpeg").write_bytes(b"manual")
|
||||
images:list[object] = ["ad_123__img1.jpeg", "nested/ad_123__img2.png", "manual_123__img3.jpeg"]
|
||||
|
||||
image_result = rename_referenced_local_image_files_after_id_change(
|
||||
ad_file,
|
||||
images,
|
||||
old_id = 123,
|
||||
new_id = 456,
|
||||
ad_file_name_template = "ad_{id}",
|
||||
enabled = True,
|
||||
)
|
||||
|
||||
assert image_result.updated_images == ["ad_456__img1.jpeg", "nested/ad_456__img2.png", "manual_123__img3.jpeg"]
|
||||
assert (folder / "ad_456__img1.jpeg").exists()
|
||||
assert (nested / "ad_456__img2.png").exists()
|
||||
assert (folder / "manual_123__img3.jpeg").exists()
|
||||
assert image_result.renamed_count == 2
|
||||
assert image_result.blocked_count == 0
|
||||
|
||||
|
||||
def test_rename_referenced_local_image_files_ignores_unreferenced_images(tmp_path:Path) -> None:
|
||||
folder = tmp_path / "ad_123_Title"
|
||||
folder.mkdir()
|
||||
ad_file = folder / "ad_123.yaml"
|
||||
(folder / "ad_123__img1.jpeg").write_bytes(b"referenced")
|
||||
(folder / "ad_123__img2.jpeg").write_bytes(b"unreferenced")
|
||||
images:list[object] = ["ad_123__img1.jpeg"]
|
||||
|
||||
image_result = rename_referenced_local_image_files_after_id_change(
|
||||
ad_file,
|
||||
images,
|
||||
old_id = 123,
|
||||
new_id = 456,
|
||||
ad_file_name_template = "ad_{id}",
|
||||
enabled = True,
|
||||
)
|
||||
|
||||
assert image_result.updated_images == ["ad_456__img1.jpeg"]
|
||||
assert (folder / "ad_456__img1.jpeg").exists()
|
||||
assert (folder / "ad_123__img2.jpeg").exists()
|
||||
assert image_result.renamed_count == 1
|
||||
assert image_result.blocked_count == 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"scenario",
|
||||
["target_exists", "absolute_path", "outside_ad_folder"],
|
||||
)
|
||||
def test_rename_referenced_local_image_files_skips_when_unsafe(
|
||||
tmp_path:Path,
|
||||
scenario:str,
|
||||
) -> None:
|
||||
"""Image renames that could affect non-owned files are safely skipped."""
|
||||
folder = tmp_path / "ad_123_Title"
|
||||
folder.mkdir()
|
||||
ad_file = folder / "ad_123.yaml"
|
||||
|
||||
source_file:Path | None = None
|
||||
target_file:Path | None = None
|
||||
if scenario == "target_exists":
|
||||
(folder / "ad_123__img1.jpeg").write_bytes(b"old")
|
||||
(folder / "ad_456__img1.jpeg").write_bytes(b"existing")
|
||||
images_config = ["ad_123__img1.jpeg"]
|
||||
source_file = folder / "ad_123__img1.jpeg"
|
||||
target_file = folder / "ad_456__img1.jpeg"
|
||||
elif scenario == "absolute_path":
|
||||
img = tmp_path / "ad_123__img1.jpeg"
|
||||
img.write_bytes(b"img")
|
||||
images_config = [str(img)]
|
||||
source_file = img
|
||||
elif scenario == "outside_ad_folder":
|
||||
img = tmp_path / "other" / "ad_123__img1.jpeg"
|
||||
img.parent.mkdir(parents = True)
|
||||
img.write_bytes(b"img")
|
||||
images_config = ["../other/ad_123__img1.jpeg"]
|
||||
source_file = img
|
||||
else:
|
||||
raise AssertionError(f"Unknown scenario: {scenario}")
|
||||
|
||||
image_result = rename_referenced_local_image_files_after_id_change(
|
||||
ad_file,
|
||||
images_config,
|
||||
old_id = 123,
|
||||
new_id = 456,
|
||||
ad_file_name_template = "ad_{id}",
|
||||
enabled = True,
|
||||
)
|
||||
|
||||
assert image_result.updated_images is None
|
||||
assert source_file is not None
|
||||
assert source_file.exists()
|
||||
if target_file is not None:
|
||||
assert target_file.exists()
|
||||
assert image_result.renamed_count == 0
|
||||
if scenario == "target_exists":
|
||||
assert image_result.blocked_count == 1
|
||||
else:
|
||||
assert image_result.blocked_count == 0
|
||||
|
||||
|
||||
def test_rename_referenced_local_image_files_disabled_returns_empty(tmp_path:Path) -> None:
|
||||
"""enabled=False is an immediate no-op."""
|
||||
folder = tmp_path / "ad_123_Title"
|
||||
folder.mkdir()
|
||||
(folder / "ad_123__img1.jpeg").write_bytes(b"img")
|
||||
images = ["ad_123__img1.jpeg"]
|
||||
|
||||
result = rename_referenced_local_image_files_after_id_change(
|
||||
folder / "ad_123.yaml",
|
||||
images,
|
||||
old_id = 123,
|
||||
new_id = 456,
|
||||
ad_file_name_template = "ad_{id}",
|
||||
enabled = False,
|
||||
)
|
||||
|
||||
assert result.renamed_count == 0
|
||||
assert result.blocked_count == 0
|
||||
assert result.updated_images is None
|
||||
|
||||
|
||||
def test_rename_referenced_local_image_files_old_id_none_returns_empty(tmp_path:Path) -> None:
|
||||
"""old_id=None is an immediate no-op."""
|
||||
folder = tmp_path / "ad_123_Title"
|
||||
folder.mkdir()
|
||||
(folder / "ad_123__img1.jpeg").write_bytes(b"img")
|
||||
images = ["ad_123__img1.jpeg"]
|
||||
|
||||
result = rename_referenced_local_image_files_after_id_change(
|
||||
folder / "ad_123.yaml",
|
||||
images,
|
||||
old_id = None,
|
||||
new_id = 456,
|
||||
ad_file_name_template = "ad_{id}",
|
||||
enabled = True,
|
||||
)
|
||||
|
||||
assert result.renamed_count == 0
|
||||
assert result.blocked_count == 0
|
||||
assert result.updated_images is None
|
||||
|
||||
|
||||
def test_rename_path_if_target_is_free_skips_when_source_equals_target(tmp_path:Path) -> None:
|
||||
source = tmp_path / "file.txt"
|
||||
source.write_text("data", encoding = "utf-8")
|
||||
|
||||
result = rename_path_if_target_is_free(source, source, label = "test")
|
||||
|
||||
assert result.status == RenameStatus.SAME
|
||||
assert result.path == source
|
||||
|
||||
|
||||
def test_rename_referenced_local_image_file_skips_non_string_ref(tmp_path:Path) -> None:
|
||||
folder = tmp_path / "ad_123_Title"
|
||||
folder.mkdir()
|
||||
ad_file = folder / "ad_123.yaml"
|
||||
|
||||
result, status = rename_referenced_local_image_file_after_id_change(
|
||||
ad_file, 42, new_id = 456, ad_file_name_template = "ad_{id}",
|
||||
)
|
||||
|
||||
assert result == 42
|
||||
assert status is None
|
||||
|
||||
|
||||
def test_rename_referenced_local_image_file_skips_non_matching_filename(tmp_path:Path) -> None:
|
||||
folder = tmp_path / "ad_123_Title"
|
||||
folder.mkdir()
|
||||
ad_file = folder / "ad_123.yaml"
|
||||
(folder / "manual.txt").write_bytes(b"data")
|
||||
|
||||
result, status = rename_referenced_local_image_file_after_id_change(
|
||||
ad_file, "manual.txt", new_id = 456, ad_file_name_template = "ad_{id}",
|
||||
)
|
||||
|
||||
assert result == "manual.txt"
|
||||
assert status is None
|
||||
|
||||
|
||||
def test_rename_referenced_local_image_files_skips_non_list_images(tmp_path:Path) -> None:
|
||||
folder = tmp_path / "ad_123_Title"
|
||||
folder.mkdir()
|
||||
|
||||
result = rename_referenced_local_image_files_after_id_change(
|
||||
folder / "ad_123.yaml",
|
||||
"not_a_list",
|
||||
old_id = 123,
|
||||
new_id = 456,
|
||||
ad_file_name_template = "ad_{id}",
|
||||
enabled = True,
|
||||
)
|
||||
|
||||
assert result.renamed_count == 0
|
||||
assert result.blocked_count == 0
|
||||
assert result.updated_images is None
|
||||
|
||||
|
||||
def test_rename_local_ad_folder_change_skips_when_target_exists(tmp_path:Path) -> None:
|
||||
folder = tmp_path / "ad_123_Title"
|
||||
folder.mkdir()
|
||||
ad_file = folder / "ad_123.yaml"
|
||||
ad_file.write_text("id: 456\n", encoding = "utf-8")
|
||||
(tmp_path / "ad_456_Title").mkdir()
|
||||
|
||||
result = rename_local_ad_folder_after_id_change(ad_file, new_id = 456, folder_name_template = "ad_{id}_{title}")
|
||||
|
||||
assert result.status == RenameStatus.TARGET_EXISTS
|
||||
assert result.path == ad_file
|
||||
assert folder.exists()
|
||||
Reference in New Issue
Block a user