mirror of
https://github.com/Second-Hand-Friends/kleinanzeigen-bot.git
synced 2026-07-09 12:41:05 +02:00
feat: add ad status command (#1177)
## ℹ️ Description - Link to the related issue(s): Issue # - Adds a CLI status overview inspired by useful lifecycle features from kleinanzeigen-bot-ui, while keeping this project CLI-only. - The new command helps users see local ad state at a glance without replacing the existing verify/APR preflight flow. ## 📋 Changes Summary - Add `kleinanzeigen-bot status` with an ASCII table and summary for local ad states: disabled, draft, changed, due, and published-local. - Add shared pure helpers for content-change and republication-due checks. - Add a small neutral `load_ad_configs()` loader reused by `load_ads()` and `status` without broad loader redesign. - Add German translations and focused unit/guardrail tests. - No schema/config model changes or generated artifacts. ### ⚙️ Type of Change Select the type(s) of change(s) included in this pull request: - [ ] 🐞 Bug fix (non-breaking change which fixes an issue) - [x] ✨ New feature (adds new functionality without breaking existing usage) - [ ] 💥 Breaking change (changes that might break existing user setups, scripts, or configurations) ## ✅ Checklist Before requesting a review, confirm the following: - [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. By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice. Validation: - `pdm run format` passed - `pdm run lint` passed - `pdm run test`: 1315 passed, 4 skipped; coverage 93.18% <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * **New Features** * Added a new `status` command to view the current state of configured ads. * Introduced a status summary that shows whether ads are disabled, drafts, changed, due, or published locally. * Added cleaner handling for ad loading and republication checks, improving consistency in displayed results. * **Bug Fixes** * Improved change detection for ads, including better handling when stored data is missing or empty. * Enhanced due-date checks so timing-based status is more reliable. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
@@ -96,7 +96,45 @@ def is_valid_ads_selector(ads_selector:str, valid_keywords:set[str]) -> bool:
|
|||||||
|
|
||||||
|
|
||||||
# --------------------------------------------------------------------------- #
|
# --------------------------------------------------------------------------- #
|
||||||
# Republication / change detection
|
# Pure content-hash / due helpers (no I/O, no logging, no mutation)
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
|
||||||
|
|
||||||
|
def has_ad_content_changed(ad_cfg:Ad, ad_cfg_orig:dict[str, Any]) -> bool:
|
||||||
|
"""Return ``True`` when the ad's stored *content_hash* differs from the
|
||||||
|
current computed hash.
|
||||||
|
|
||||||
|
Unlike :func:`check_ad_changed`, this function is **pure** — no logging,
|
||||||
|
no mutation of *ad_cfg_orig*. A missing or empty stored hash is treated
|
||||||
|
as "not changed".
|
||||||
|
"""
|
||||||
|
if ad_cfg.id is None:
|
||||||
|
return False
|
||||||
|
stored_hash = ad_cfg_orig.get("content_hash")
|
||||||
|
if not stored_hash: # None or empty string — hash never computed
|
||||||
|
return False
|
||||||
|
current_hash = AdPartial.model_validate(ad_cfg_orig).update_content_hash().content_hash
|
||||||
|
return current_hash is not None and current_hash != stored_hash
|
||||||
|
|
||||||
|
|
||||||
|
def is_ad_due_for_republication(ad_cfg:Ad, *, now:datetime | None = None) -> bool:
|
||||||
|
"""Return ``True`` when *ad_cfg* is due for republication (pure).
|
||||||
|
|
||||||
|
Unlike :func:`check_ad_republication`, this function is **pure** — no
|
||||||
|
logging, no side effects. Pass *now* for deterministic testing.
|
||||||
|
"""
|
||||||
|
if ad_cfg.updated_on is None and ad_cfg.created_on is None:
|
||||||
|
return True
|
||||||
|
latest = ad_cfg.updated_on or ad_cfg.created_on
|
||||||
|
if latest is None:
|
||||||
|
return True
|
||||||
|
if now is None:
|
||||||
|
now = _misc.now()
|
||||||
|
return (now - latest).days >= ad_cfg.republication_interval
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
# Republication / change detection (with logging / mutation)
|
||||||
# --------------------------------------------------------------------------- #
|
# --------------------------------------------------------------------------- #
|
||||||
|
|
||||||
|
|
||||||
@@ -108,33 +146,28 @@ def check_ad_republication(
|
|||||||
) -> bool:
|
) -> bool:
|
||||||
"""Return ``True`` when *ad_cfg* is due for republication.
|
"""Return ``True`` when *ad_cfg* is due for republication.
|
||||||
|
|
||||||
|
Delegates to :func:`is_ad_due_for_republication` for the pure check,
|
||||||
|
then adds logging when the ad is **not** due (SKIPPED message).
|
||||||
|
|
||||||
The interval is measured against :func:`~kleinanzeigen_bot.utils.misc.now`
|
The interval is measured against :func:`~kleinanzeigen_bot.utils.misc.now`
|
||||||
by default; pass *now* to make the function deterministic in tests.
|
by default; pass *now* to make the function deterministic in tests.
|
||||||
"""
|
"""
|
||||||
if ad_cfg.updated_on:
|
if is_ad_due_for_republication(ad_cfg, now = now):
|
||||||
last_updated_on = ad_cfg.updated_on
|
|
||||||
elif ad_cfg.created_on:
|
|
||||||
last_updated_on = ad_cfg.created_on
|
|
||||||
else:
|
|
||||||
return True
|
|
||||||
|
|
||||||
if not last_updated_on:
|
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
# Log why it was skipped — only reaches here when not due and a date exists
|
||||||
if now is None:
|
if now is None:
|
||||||
now = _misc.now()
|
now = _misc.now()
|
||||||
|
latest = ad_cfg.updated_on or ad_cfg.created_on
|
||||||
ad_age = now - last_updated_on
|
if latest is not None:
|
||||||
if ad_age.days < ad_cfg.republication_interval:
|
ad_age = now - latest
|
||||||
LOG.info(
|
LOG.info(
|
||||||
" -> SKIPPED: ad [%s] was last published %d days ago. republication is only required every %s days",
|
" -> SKIPPED: ad [%s] was last published %d days ago. republication is only required every %s days",
|
||||||
ad_file_relative,
|
ad_file_relative,
|
||||||
ad_age.days,
|
ad_age.days,
|
||||||
ad_cfg.republication_interval,
|
ad_cfg.republication_interval,
|
||||||
)
|
)
|
||||||
return False
|
return False
|
||||||
|
|
||||||
return True
|
|
||||||
|
|
||||||
|
|
||||||
def check_ad_changed(
|
def check_ad_changed(
|
||||||
@@ -150,11 +183,10 @@ def check_ad_changed(
|
|||||||
``ad_cfg_orig["content_hash"]`` with the freshly computed hash when a
|
``ad_cfg_orig["content_hash"]`` with the freshly computed hash when a
|
||||||
change is detected. Callers must be aware of this mutation.
|
change is detected. Callers must be aware of this mutation.
|
||||||
"""
|
"""
|
||||||
if not ad_cfg.id:
|
if not has_ad_content_changed(ad_cfg, ad_cfg_orig):
|
||||||
# New ads are not considered "changed"
|
|
||||||
return False
|
return False
|
||||||
|
|
||||||
# Calculate hash on original config to match what was stored
|
# Recompute hash for logging and mutation
|
||||||
current_hash = AdPartial.model_validate(ad_cfg_orig).update_content_hash().content_hash
|
current_hash = AdPartial.model_validate(ad_cfg_orig).update_content_hash().content_hash
|
||||||
stored_hash = ad_cfg_orig.get("content_hash")
|
stored_hash = ad_cfg_orig.get("content_hash")
|
||||||
|
|
||||||
@@ -162,13 +194,10 @@ def check_ad_changed(
|
|||||||
LOG.debug(" Stored hash: %s", stored_hash)
|
LOG.debug(" Stored hash: %s", stored_hash)
|
||||||
LOG.debug(" Current hash: %s", current_hash)
|
LOG.debug(" Current hash: %s", current_hash)
|
||||||
|
|
||||||
if stored_hash and current_hash != stored_hash:
|
LOG.info("Changes detected in ad [%s], will republish", ad_file_relative)
|
||||||
LOG.info("Changes detected in ad [%s], will republish", ad_file_relative)
|
# Update hash in original configuration
|
||||||
# Update hash in original configuration
|
ad_cfg_orig["content_hash"] = current_hash
|
||||||
ad_cfg_orig["content_hash"] = current_hash
|
return True
|
||||||
return True
|
|
||||||
|
|
||||||
return False
|
|
||||||
|
|
||||||
|
|
||||||
# --------------------------------------------------------------------------- #
|
# --------------------------------------------------------------------------- #
|
||||||
@@ -332,6 +361,27 @@ def _prepare_selected_ad_entry(
|
|||||||
ad_cfg.images = resolve_ad_images(ad_file, ad_cfg.images)
|
ad_cfg.images = resolve_ad_images(ad_file, ad_cfg.images)
|
||||||
|
|
||||||
|
|
||||||
|
def load_ad_configs(
|
||||||
|
*,
|
||||||
|
config_file_path:str,
|
||||||
|
ad_file_patterns:list[str],
|
||||||
|
ad_defaults:Any,
|
||||||
|
) -> list[tuple[str, str, Ad, dict[str, Any]]]:
|
||||||
|
"""Discover, load raw YAML, validate, and apply defaults for every ad file.
|
||||||
|
|
||||||
|
This is a **neutral** loading step — no selector filtering, no category
|
||||||
|
resolution, no image globbing, no ``_prepare_selected_ad_entry()``.
|
||||||
|
Returns a sorted list of ``(abspath, relpath, Ad, raw_dict)`` tuples.
|
||||||
|
"""
|
||||||
|
ad_files = discover_ad_files(config_file_path, ad_file_patterns)
|
||||||
|
result:list[tuple[str, str, Ad, dict[str, Any]]] = []
|
||||||
|
for ad_file, ad_file_relative in sorted(ad_files.items()):
|
||||||
|
ad_cfg_orig = _dicts.load_dict(ad_file, "ad")
|
||||||
|
ad_cfg = load_ad(ad_cfg_orig, ad_defaults)
|
||||||
|
result.append((ad_file, ad_file_relative, ad_cfg, ad_cfg_orig))
|
||||||
|
return result
|
||||||
|
|
||||||
|
|
||||||
def load_ads(
|
def load_ads(
|
||||||
*,
|
*,
|
||||||
config_file_path:str,
|
config_file_path:str,
|
||||||
@@ -354,9 +404,13 @@ def load_ads(
|
|||||||
Tuples of ``(file_path, validated Ad model, original raw data)``.
|
Tuples of ``(file_path, validated Ad model, original raw data)``.
|
||||||
"""
|
"""
|
||||||
LOG.info("Searching for ad config files...")
|
LOG.info("Searching for ad config files...")
|
||||||
ad_files = discover_ad_files(config_file_path, ad_file_patterns)
|
loaded = load_ad_configs(
|
||||||
LOG.info(" -> found %s", pluralize("ad config file", ad_files))
|
config_file_path = config_file_path,
|
||||||
if not ad_files:
|
ad_file_patterns = ad_file_patterns,
|
||||||
|
ad_defaults = ad_defaults,
|
||||||
|
)
|
||||||
|
LOG.info(" -> found %s", pluralize("ad config file", loaded))
|
||||||
|
if not loaded:
|
||||||
return []
|
return []
|
||||||
|
|
||||||
ids, tokens = _parse_ad_selector(ads_selector)
|
ids, tokens = _parse_ad_selector(ads_selector)
|
||||||
@@ -365,9 +419,7 @@ def load_ads(
|
|||||||
LOG.info(" | ".join(str(id_) for id_ in ids))
|
LOG.info(" | ".join(str(id_) for id_ in ids))
|
||||||
|
|
||||||
ads:list[tuple[str, Ad, dict[str, Any]]] = []
|
ads:list[tuple[str, Ad, dict[str, Any]]] = []
|
||||||
for ad_file, ad_file_relative in sorted(ad_files.items()):
|
for ad_file, ad_file_relative, ad_cfg, ad_cfg_orig in loaded:
|
||||||
ad_cfg_orig:dict[str, Any] = _dicts.load_dict(ad_file, "ad")
|
|
||||||
ad_cfg:Ad = load_ad(ad_cfg_orig, ad_defaults)
|
|
||||||
|
|
||||||
# Inactive check runs before numeric ID filtering — an inactive ad
|
# Inactive check runs before numeric ID filtering — an inactive ad
|
||||||
# with a matching numeric ID will still be skipped.
|
# with a matching numeric ID will still be skipped.
|
||||||
|
|||||||
166
src/kleinanzeigen_bot/ad_status.py
Normal file
166
src/kleinanzeigen_bot/ad_status.py
Normal file
@@ -0,0 +1,166 @@
|
|||||||
|
# SPDX-FileCopyrightText: © Jens Bergmann and contributors
|
||||||
|
# SPDX-License-Identifier: AGPL-3.0-or-later
|
||||||
|
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
|
||||||
|
"""Ad status computation and display for the ``status`` CLI command.
|
||||||
|
This module owns status label mapping, row building, and ASCII table rendering.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from datetime import datetime # noqa: TC003 — used in runtime type annotations
|
||||||
|
from gettext import gettext as _
|
||||||
|
from typing import TYPE_CHECKING, Any
|
||||||
|
|
||||||
|
from . import ad_loading
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
from .model.ad_model import Ad
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen = True, slots = True)
|
||||||
|
class StatusRow:
|
||||||
|
"""One row in the status table. Rendered by :func:`render_status_rows`."""
|
||||||
|
|
||||||
|
title:str # ad.title
|
||||||
|
ad_id:str # "-" if None, else str(ad.id)
|
||||||
|
status:str # One of: "disabled", "draft", "changed", "due", "published-local"
|
||||||
|
|
||||||
|
|
||||||
|
def _translate_status(status:str) -> str:
|
||||||
|
"""Return the translated display label for a status string."""
|
||||||
|
if status == "disabled":
|
||||||
|
return _("disabled")
|
||||||
|
if status == "draft":
|
||||||
|
return _("draft")
|
||||||
|
if status == "changed":
|
||||||
|
return _("changed")
|
||||||
|
if status == "due":
|
||||||
|
return _("due")
|
||||||
|
if status == "published-local":
|
||||||
|
return _("published-local")
|
||||||
|
return status
|
||||||
|
|
||||||
|
|
||||||
|
def compute_ad_status(
|
||||||
|
ad:Ad,
|
||||||
|
ad_cfg_orig:dict[str, Any],
|
||||||
|
*,
|
||||||
|
now:datetime | None = None,
|
||||||
|
) -> str:
|
||||||
|
"""Map a single :class:`Ad` to a status string.
|
||||||
|
|
||||||
|
Precedence (first match wins):
|
||||||
|
1. ``disabled`` — *ad.active* is ``False``
|
||||||
|
2. ``draft`` — *ad.id* is ``None``
|
||||||
|
3. ``changed`` — stored *content_hash* exists, non-empty, and differs
|
||||||
|
from recomputed hash
|
||||||
|
4. ``due`` — both timestamp fields are ``None`` or
|
||||||
|
*republication_interval* has elapsed
|
||||||
|
5. ``published-local`` — fallthrough (has id, not disabled/draft/changed/due)
|
||||||
|
"""
|
||||||
|
if not ad.active:
|
||||||
|
return "disabled"
|
||||||
|
if ad.id is None:
|
||||||
|
return "draft"
|
||||||
|
if ad_loading.has_ad_content_changed(ad, ad_cfg_orig):
|
||||||
|
return "changed"
|
||||||
|
if ad_loading.is_ad_due_for_republication(ad, now = now):
|
||||||
|
return "due"
|
||||||
|
return "published-local"
|
||||||
|
|
||||||
|
|
||||||
|
def build_status_rows(
|
||||||
|
ads:list[tuple[str, Ad, dict[str, Any]]],
|
||||||
|
*,
|
||||||
|
now:datetime | None = None,
|
||||||
|
) -> list[StatusRow]:
|
||||||
|
"""Build status rows from ad-file / Ad / raw-dict triples."""
|
||||||
|
rows:list[StatusRow] = []
|
||||||
|
for _ad_file, ad_cfg, ad_cfg_orig in ads:
|
||||||
|
status = compute_ad_status(ad_cfg, ad_cfg_orig, now = now)
|
||||||
|
rows.append(
|
||||||
|
StatusRow(
|
||||||
|
title = ad_cfg.title,
|
||||||
|
ad_id = "-" if ad_cfg.id is None else str(ad_cfg.id),
|
||||||
|
status = status,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
return rows
|
||||||
|
|
||||||
|
|
||||||
|
def render_status_rows(rows:list[StatusRow]) -> str:
|
||||||
|
"""Format status rows into an ASCII table string."""
|
||||||
|
if not rows:
|
||||||
|
return ""
|
||||||
|
|
||||||
|
h_id = _("Ad ID")
|
||||||
|
h_title = _("Title")
|
||||||
|
h_status = _("Status")
|
||||||
|
|
||||||
|
col_id = max(len(h_id), max((len(r.ad_id) for r in rows), default = 0))
|
||||||
|
|
||||||
|
# Translated labels for column width calculation.
|
||||||
|
# Uses ``_translate_status()`` which contains explicit ``_("...")`` calls
|
||||||
|
# that the translation-coverage scanner can find.
|
||||||
|
translated_statuses = [
|
||||||
|
_translate_status("disabled"),
|
||||||
|
_translate_status("draft"),
|
||||||
|
_translate_status("changed"),
|
||||||
|
_translate_status("due"),
|
||||||
|
_translate_status("published-local"),
|
||||||
|
]
|
||||||
|
col_status = max(len(h_status), *[len(s) for s in translated_statuses], 0)
|
||||||
|
|
||||||
|
# Title width is data-driven, but clamp to at least header width
|
||||||
|
col_title = max(len(h_title), max((len(r.title) for r in rows), default = 0))
|
||||||
|
|
||||||
|
sep = (
|
||||||
|
"+"
|
||||||
|
+ "-" * (col_id + 2)
|
||||||
|
+ "+"
|
||||||
|
+ "-" * (col_title + 2)
|
||||||
|
+ "+"
|
||||||
|
+ "-" * (col_status + 2)
|
||||||
|
+ "+"
|
||||||
|
)
|
||||||
|
header = (
|
||||||
|
"| "
|
||||||
|
+ h_id.ljust(col_id)
|
||||||
|
+ " | "
|
||||||
|
+ h_title.ljust(col_title)
|
||||||
|
+ " | "
|
||||||
|
+ h_status.ljust(col_status)
|
||||||
|
+ " |"
|
||||||
|
)
|
||||||
|
|
||||||
|
lines:list[str] = [sep, header, sep]
|
||||||
|
|
||||||
|
for r in rows:
|
||||||
|
label = _translate_status(r.status)
|
||||||
|
lines.append(
|
||||||
|
"| "
|
||||||
|
+ r.ad_id.ljust(col_id)
|
||||||
|
+ " | "
|
||||||
|
+ r.title.ljust(col_title)
|
||||||
|
+ " | "
|
||||||
|
+ label.ljust(col_status)
|
||||||
|
+ " |"
|
||||||
|
)
|
||||||
|
|
||||||
|
lines.append(sep)
|
||||||
|
|
||||||
|
# Summary line
|
||||||
|
counts:dict[str, int] = {}
|
||||||
|
for r in rows:
|
||||||
|
counts[r.status] = counts.get(r.status, 0) + 1
|
||||||
|
|
||||||
|
summary_parts:list[str] = [
|
||||||
|
f"{_translate_status(label)}: {counts[label]}"
|
||||||
|
for label in ("disabled", "draft", "changed", "due", "published-local")
|
||||||
|
if label in counts
|
||||||
|
]
|
||||||
|
|
||||||
|
summary = _("Summary: %s (%d total)") % (", ".join(summary_parts), len(rows))
|
||||||
|
lines.append(summary)
|
||||||
|
|
||||||
|
return "\n".join(lines) + "\n"
|
||||||
@@ -8,7 +8,7 @@ from typing import TYPE_CHECKING, Any, Final, cast
|
|||||||
|
|
||||||
import certifi
|
import certifi
|
||||||
|
|
||||||
from . import ad_loading, delete_flow, download_flow, extend_flow
|
from . import ad_loading, ad_status, delete_flow, download_flow, extend_flow
|
||||||
from . import ad_state as _ad_state
|
from . import ad_state as _ad_state
|
||||||
from . import login_flow as _login_flow
|
from . import login_flow as _login_flow
|
||||||
from . import price_reduction as _price_reduction
|
from . import price_reduction as _price_reduction
|
||||||
@@ -137,6 +137,8 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904
|
|||||||
self._handle_update_check()
|
self._handle_update_check()
|
||||||
case "update-content-hash":
|
case "update-content-hash":
|
||||||
self._handle_update_content_hash()
|
self._handle_update_content_hash()
|
||||||
|
case "status":
|
||||||
|
self._handle_status()
|
||||||
case "publish":
|
case "publish":
|
||||||
await self._handle_publish()
|
await self._handle_publish()
|
||||||
case "update":
|
case "update":
|
||||||
@@ -246,6 +248,26 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904
|
|||||||
LOG.info("DONE: No active ads found.")
|
LOG.info("DONE: No active ads found.")
|
||||||
LOG.info("############################################")
|
LOG.info("############################################")
|
||||||
|
|
||||||
|
def _handle_status(self) -> None:
|
||||||
|
"""Show status overview of all local ads."""
|
||||||
|
self._bootstrap_runtime()
|
||||||
|
self._check_for_updates()
|
||||||
|
|
||||||
|
loaded = ad_loading.load_ad_configs(
|
||||||
|
config_file_path = self.config_file_path,
|
||||||
|
ad_file_patterns = self.config.ad_files,
|
||||||
|
ad_defaults = self.config.ad_defaults,
|
||||||
|
)
|
||||||
|
if not loaded:
|
||||||
|
LOG.info("No ad files found.")
|
||||||
|
return
|
||||||
|
|
||||||
|
now = _misc.now()
|
||||||
|
ads_for_status = [(abspath, ad_cfg, raw) for abspath, _relpath, ad_cfg, raw in loaded]
|
||||||
|
rows = ad_status.build_status_rows(ads_for_status, now = now)
|
||||||
|
output = ad_status.render_status_rows(rows)
|
||||||
|
print(output)
|
||||||
|
|
||||||
async def _handle_publish(self) -> None:
|
async def _handle_publish(self) -> None:
|
||||||
self._bootstrap_runtime()
|
self._bootstrap_runtime()
|
||||||
self._check_for_updates()
|
self._check_for_updates()
|
||||||
|
|||||||
@@ -80,6 +80,7 @@ def _help_text() -> str:
|
|||||||
"geändert" gelten und neu veröffentlicht werden.
|
"geändert" gelten und neu veröffentlicht werden.
|
||||||
create-config - Erstellt eine neue Standard-Konfigurationsdatei, falls noch nicht vorhanden
|
create-config - Erstellt eine neue Standard-Konfigurationsdatei, falls noch nicht vorhanden
|
||||||
diagnose - Diagnostiziert Browser-Verbindungsprobleme und zeigt Troubleshooting-Informationen
|
diagnose - Diagnostiziert Browser-Verbindungsprobleme und zeigt Troubleshooting-Informationen
|
||||||
|
status - Zeigt den Status der Anzeigen an
|
||||||
--
|
--
|
||||||
help - Zeigt diese Hilfe an (Standardbefehl)
|
help - Zeigt diese Hilfe an (Standardbefehl)
|
||||||
version - Zeigt die Version der Anwendung an
|
version - Zeigt die Version der Anwendung an
|
||||||
@@ -135,6 +136,7 @@ def _help_text() -> str:
|
|||||||
use this after changing config.yaml/ad_defaults to avoid every ad being marked "changed" and republished
|
use this after changing config.yaml/ad_defaults to avoid every ad being marked "changed" and republished
|
||||||
create-config - creates a new default configuration file if one does not exist
|
create-config - creates a new default configuration file if one does not exist
|
||||||
diagnose - diagnoses browser connection issues and shows troubleshooting information
|
diagnose - diagnoses browser connection issues and shows troubleshooting information
|
||||||
|
status - shows status overview of ads
|
||||||
--
|
--
|
||||||
help - displays this help (default command)
|
help - displays this help (default command)
|
||||||
version - displays the application version
|
version - displays the application version
|
||||||
|
|||||||
@@ -109,6 +109,9 @@ kleinanzeigen_bot/app.py:
|
|||||||
_handle_download:
|
_handle_download:
|
||||||
"Invalid --ads selector: \"%s\". Valid values: comma-separated keywords (all, new) or numeric IDs.": "Ungültiger --ads-Selektor: \"%s\". Gültige Werte: kommagetrennte Schlüsselwörter (all, new) oder numerische IDs."
|
"Invalid --ads selector: \"%s\". Valid values: comma-separated keywords (all, new) or numeric IDs.": "Ungültiger --ads-Selektor: \"%s\". Gültige Werte: kommagetrennte Schlüsselwörter (all, new) oder numerische IDs."
|
||||||
|
|
||||||
|
_handle_status:
|
||||||
|
"No ad files found.": "Keine Anzeigendateien gefunden."
|
||||||
|
|
||||||
#################################################
|
#################################################
|
||||||
kleinanzeigen_bot/login_flow.py:
|
kleinanzeigen_bot/login_flow.py:
|
||||||
#################################################
|
#################################################
|
||||||
@@ -442,6 +445,22 @@ kleinanzeigen_bot/download_flow.py:
|
|||||||
_download_ad_with_resolved_state:
|
_download_ad_with_resolved_state:
|
||||||
"Ad %d found in overview but not in published profile. Saving as inactive.": "Anzeige %d wurde in der Übersicht gefunden, aber nicht im veröffentlichten Profil. Speichere als inaktiv."
|
"Ad %d found in overview but not in published profile. Saving as inactive.": "Anzeige %d wurde in der Übersicht gefunden, aber nicht im veröffentlichten Profil. Speichere als inaktiv."
|
||||||
|
|
||||||
|
#################################################
|
||||||
|
kleinanzeigen_bot/ad_status.py:
|
||||||
|
#################################################
|
||||||
|
_translate_status:
|
||||||
|
"disabled": "Deaktiviert"
|
||||||
|
"draft": "Entwurf"
|
||||||
|
"changed": "Geändert"
|
||||||
|
"due": "Fällig"
|
||||||
|
"published-local": "Veröffentlicht (lokal)"
|
||||||
|
|
||||||
|
render_status_rows:
|
||||||
|
"Ad ID": "Anzeigen-ID"
|
||||||
|
"Title": "Titel"
|
||||||
|
"Status": "Status"
|
||||||
|
"Summary: %s (%d total)": "Zusammenfassung: %s (%d insgesamt)"
|
||||||
|
|
||||||
#################################################
|
#################################################
|
||||||
kleinanzeigen_bot/ad_loading.py:
|
kleinanzeigen_bot/ad_loading.py:
|
||||||
#################################################
|
#################################################
|
||||||
|
|||||||
@@ -36,7 +36,7 @@ WORKSPACE_FREE_COMMANDS:Final[frozenset[str]] = frozenset({"help", "version", "c
|
|||||||
VALID_COMMANDS:Final[frozenset[str]] = frozenset({
|
VALID_COMMANDS:Final[frozenset[str]] = frozenset({
|
||||||
"help", "version", "create-config", "diagnose", "verify",
|
"help", "version", "create-config", "diagnose", "verify",
|
||||||
"update-check", "update-content-hash",
|
"update-check", "update-content-hash",
|
||||||
"publish", "update", "delete", "extend", "download",
|
"publish", "status", "update", "delete", "extend", "download",
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
317
tests/unit/test_ad_status.py
Normal file
317
tests/unit/test_ad_status.py
Normal file
@@ -0,0 +1,317 @@
|
|||||||
|
# SPDX-FileCopyrightText: © Jens Bergmann and contributors
|
||||||
|
# SPDX-License-Identifier: AGPL-3.0-or-later
|
||||||
|
# SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
|
||||||
|
"""Tests for ad_status module — pure computation, no disk I/O in unit tests."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from datetime import datetime, timedelta, timezone
|
||||||
|
from typing import Any
|
||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from kleinanzeigen_bot.ad_status import (
|
||||||
|
StatusRow,
|
||||||
|
build_status_rows,
|
||||||
|
compute_ad_status,
|
||||||
|
render_status_rows,
|
||||||
|
)
|
||||||
|
from kleinanzeigen_bot.app import KleinanzeigenBot
|
||||||
|
from kleinanzeigen_bot.cli import parse_args
|
||||||
|
from kleinanzeigen_bot.model.ad_model import Ad, AdPartial
|
||||||
|
from kleinanzeigen_bot.model.config_model import Config
|
||||||
|
from kleinanzeigen_bot.runtime_config import VALID_COMMANDS, RuntimeState
|
||||||
|
from kleinanzeigen_bot.utils import xdg_paths
|
||||||
|
|
||||||
|
pytestmark = pytest.mark.unit
|
||||||
|
|
||||||
|
_TZ = timezone.utc
|
||||||
|
|
||||||
|
|
||||||
|
def _now() -> datetime:
|
||||||
|
"""Return a timezone-aware 'now' for reproducible tests."""
|
||||||
|
return datetime.now(tz = _TZ)
|
||||||
|
|
||||||
|
|
||||||
|
def days_ago(n:int) -> datetime:
|
||||||
|
"""Return *n* days before :func:`_now`."""
|
||||||
|
return _now() - timedelta(days = n)
|
||||||
|
|
||||||
|
|
||||||
|
def _ad(**overrides:Any) -> Ad:
|
||||||
|
"""Helper: build an Ad with all required fields."""
|
||||||
|
defaults:dict[str, Any] = {
|
||||||
|
"active": True,
|
||||||
|
"title": "test ad title",
|
||||||
|
"type": "OFFER",
|
||||||
|
"price": 10,
|
||||||
|
"description": "desc",
|
||||||
|
"category": "Möbel",
|
||||||
|
"shipping_type": "PICKUP",
|
||||||
|
"price_type": "FIXED",
|
||||||
|
"sell_directly": False,
|
||||||
|
"contact": {"name": "Test", "zipcode": "12345"},
|
||||||
|
"republication_interval": 7,
|
||||||
|
}
|
||||||
|
defaults.update(overrides)
|
||||||
|
return Ad(**defaults)
|
||||||
|
|
||||||
|
|
||||||
|
def _raw(**overrides:Any) -> dict[str, Any]:
|
||||||
|
"""Helper: build a raw YAML dict with required fields."""
|
||||||
|
d:dict[str, Any] = {
|
||||||
|
"title": "test ad title",
|
||||||
|
"description": "desc",
|
||||||
|
"category": "Möbel",
|
||||||
|
}
|
||||||
|
d.update(overrides)
|
||||||
|
return d
|
||||||
|
|
||||||
|
|
||||||
|
def _content_hash(raw:dict[str, Any]) -> str:
|
||||||
|
"""Compute the actual content_hash for a raw dict."""
|
||||||
|
result = AdPartial.model_validate(raw).update_content_hash().content_hash
|
||||||
|
assert result is not None
|
||||||
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
# compute_ad_status — precedence
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
|
||||||
|
|
||||||
|
def test_disabled() -> None:
|
||||||
|
"""Disabled beats everything, even with id and matching hash."""
|
||||||
|
ad = _ad(active = False, id = 123, content_hash = "abc")
|
||||||
|
raw = _raw(content_hash = "abc")
|
||||||
|
assert compute_ad_status(ad, raw) == "disabled"
|
||||||
|
|
||||||
|
|
||||||
|
def test_draft() -> None:
|
||||||
|
"""Draft when id is None."""
|
||||||
|
ad = _ad(active = True, id = None)
|
||||||
|
raw = _raw()
|
||||||
|
assert compute_ad_status(ad, raw) == "draft"
|
||||||
|
|
||||||
|
|
||||||
|
def test_changed() -> None:
|
||||||
|
"""Changed when stored hash exists, non-empty, and differs."""
|
||||||
|
ad = _ad(active = True, id = 1, content_hash = "old_hash")
|
||||||
|
raw = _raw(content_hash = "old_hash", title = "different title to trigger hash change")
|
||||||
|
result = compute_ad_status(ad, raw)
|
||||||
|
assert result == "changed"
|
||||||
|
|
||||||
|
|
||||||
|
def test_changed_no_hash_not_changed() -> None:
|
||||||
|
"""Missing stored hash is NOT changed — falls through."""
|
||||||
|
ad = _ad(active = True, id = 1, content_hash = None, created_on = _now())
|
||||||
|
raw = _raw()
|
||||||
|
result = compute_ad_status(ad, raw, now = _now())
|
||||||
|
assert result != "changed"
|
||||||
|
assert result == "published-local"
|
||||||
|
|
||||||
|
|
||||||
|
def test_changed_no_id_not_changed() -> None:
|
||||||
|
"""No id means draft before hash check."""
|
||||||
|
ad = _ad(active = True, id = None, content_hash = "some")
|
||||||
|
raw = _raw(content_hash = "some")
|
||||||
|
assert compute_ad_status(ad, raw) == "draft"
|
||||||
|
|
||||||
|
|
||||||
|
def test_changed_empty_hash_not_changed() -> None:
|
||||||
|
"""Empty string stored hash is NOT changed (same as missing)."""
|
||||||
|
ad = _ad(active = True, id = 1, content_hash = "", created_on = _now())
|
||||||
|
raw = _raw(content_hash = "")
|
||||||
|
result = compute_ad_status(ad, raw, now = _now())
|
||||||
|
assert result == "published-local"
|
||||||
|
|
||||||
|
|
||||||
|
def test_due_interval_elapsed() -> None:
|
||||||
|
"""Due when republication_interval elapsed since created_on."""
|
||||||
|
raw = _raw()
|
||||||
|
ch = _content_hash(raw)
|
||||||
|
ad = _ad(
|
||||||
|
active = True,
|
||||||
|
id = 1,
|
||||||
|
content_hash = ch,
|
||||||
|
created_on = days_ago(30),
|
||||||
|
republication_interval = 7,
|
||||||
|
)
|
||||||
|
result = compute_ad_status(ad, raw, now = _now())
|
||||||
|
assert result == "due"
|
||||||
|
|
||||||
|
|
||||||
|
def test_due_no_dates() -> None:
|
||||||
|
"""Due when both updated_on and created_on are None."""
|
||||||
|
raw = _raw()
|
||||||
|
ch = _content_hash(raw)
|
||||||
|
ad = _ad(
|
||||||
|
active = True,
|
||||||
|
id = 1,
|
||||||
|
content_hash = ch,
|
||||||
|
updated_on = None,
|
||||||
|
created_on = None,
|
||||||
|
republication_interval = 7,
|
||||||
|
)
|
||||||
|
result = compute_ad_status(ad, raw, now = _now())
|
||||||
|
assert result == "due"
|
||||||
|
|
||||||
|
|
||||||
|
def test_due_not_due() -> None:
|
||||||
|
"""Not due when republication_interval has not elapsed."""
|
||||||
|
raw = _raw()
|
||||||
|
ch = _content_hash(raw)
|
||||||
|
ad = _ad(
|
||||||
|
active = True,
|
||||||
|
id = 1,
|
||||||
|
content_hash = ch,
|
||||||
|
created_on = days_ago(1),
|
||||||
|
republication_interval = 7,
|
||||||
|
)
|
||||||
|
result = compute_ad_status(ad, raw, now = _now())
|
||||||
|
assert result == "published-local"
|
||||||
|
|
||||||
|
|
||||||
|
def test_published_local() -> None:
|
||||||
|
"""published-local when id exists and nothing else applies."""
|
||||||
|
raw = _raw()
|
||||||
|
ch = _content_hash(raw)
|
||||||
|
ad = _ad(
|
||||||
|
active = True,
|
||||||
|
id = 1,
|
||||||
|
content_hash = ch,
|
||||||
|
created_on = _now(),
|
||||||
|
republication_interval = 7,
|
||||||
|
)
|
||||||
|
result = compute_ad_status(ad, raw, now = _now())
|
||||||
|
assert result == "published-local"
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
# build_status_rows
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
|
||||||
|
|
||||||
|
def test_build_status_rows() -> None:
|
||||||
|
"""build_status_rows produces a StatusRow per ad."""
|
||||||
|
ads:list[tuple[str, Ad, dict[str, Any]]] = [
|
||||||
|
("ads/one.yaml", _ad(active = False, id = 123), _raw()),
|
||||||
|
("ads/two.yaml", _ad(active = True, id = None), _raw()),
|
||||||
|
]
|
||||||
|
rows = build_status_rows(ads, now = _now())
|
||||||
|
assert len(rows) == 2
|
||||||
|
assert rows[0].status == "disabled"
|
||||||
|
assert rows[1].status == "draft"
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
# render_status_rows
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
|
||||||
|
|
||||||
|
class TestRenderStatusRows:
|
||||||
|
def test_empty(self) -> None:
|
||||||
|
assert not render_status_rows([])
|
||||||
|
|
||||||
|
def test_headers_and_rows(self) -> None:
|
||||||
|
rows = [
|
||||||
|
StatusRow(title = "Ad A", ad_id = "-", status = "draft"),
|
||||||
|
StatusRow(title = "Ad B", ad_id = "123", status = "published-local"),
|
||||||
|
]
|
||||||
|
output = render_status_rows(rows)
|
||||||
|
|
||||||
|
# Contains ASCII table borders
|
||||||
|
assert "+" in output
|
||||||
|
assert "-" in output
|
||||||
|
assert "|" in output
|
||||||
|
|
||||||
|
# Contains headers
|
||||||
|
assert "Ad ID" in output
|
||||||
|
assert "Title" in output
|
||||||
|
assert "Status" in output
|
||||||
|
|
||||||
|
# Contains row data
|
||||||
|
assert "-" in output
|
||||||
|
assert "123" in output
|
||||||
|
assert "Ad A" in output
|
||||||
|
assert "Ad B" in output
|
||||||
|
|
||||||
|
# Contains summary line
|
||||||
|
assert "draft:" in output.casefold()
|
||||||
|
assert "published-local:" in output.casefold()
|
||||||
|
|
||||||
|
# Summary ends with total count
|
||||||
|
assert "total" in output.casefold() or "gesamt" in output.casefold()
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
# Guardrail: status does not call load_ads or open browser
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_status_guardrails(
|
||||||
|
monkeypatch:pytest.MonkeyPatch,
|
||||||
|
tmp_path:Any,
|
||||||
|
) -> None:
|
||||||
|
"""Status calls _check_for_updates but NOT load_ads / browser login."""
|
||||||
|
bot = KleinanzeigenBot()
|
||||||
|
bot.config_file_path = str(tmp_path / "config.yaml")
|
||||||
|
workspace = xdg_paths.Workspace.for_config(tmp_path / "config.yaml", "kleinanzeigen-bot")
|
||||||
|
dummy_config = Config()
|
||||||
|
|
||||||
|
update_called:list[bool] = []
|
||||||
|
load_ads_called:list[bool] = []
|
||||||
|
browser_called:list[bool] = []
|
||||||
|
|
||||||
|
def _track_update(*_args:Any, **_kwargs:Any) -> None:
|
||||||
|
update_called.append(True)
|
||||||
|
|
||||||
|
def _track_load_ads(*_args:Any, **_kwargs:Any) -> list[Any]:
|
||||||
|
load_ads_called.append(True)
|
||||||
|
return []
|
||||||
|
|
||||||
|
def _track_browser(*_args:Any, **_kwargs:Any) -> None:
|
||||||
|
browser_called.append(True)
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("kleinanzeigen_bot.runtime_config.resolve_workspace", return_value = workspace),
|
||||||
|
patch(
|
||||||
|
"kleinanzeigen_bot.runtime_config.load_config",
|
||||||
|
return_value = RuntimeState(
|
||||||
|
config = dummy_config, categories = {}, timing_collector = None,
|
||||||
|
),
|
||||||
|
),
|
||||||
|
patch("kleinanzeigen_bot.runtime_config.configure_file_logging", return_value = None),
|
||||||
|
patch("kleinanzeigen_bot.runtime_config.apply_browser_config"),
|
||||||
|
patch("kleinanzeigen_bot.update_checker.UpdateChecker"),
|
||||||
|
patch.object(bot, "_check_for_updates", side_effect = _track_update),
|
||||||
|
patch.object(bot, "load_ads", side_effect = _track_load_ads),
|
||||||
|
patch.object(bot, "create_browser_session", side_effect = _track_browser),
|
||||||
|
patch.object(bot, "login", side_effect = _track_browser),
|
||||||
|
patch.object(bot, "close_browser_session"),
|
||||||
|
patch("kleinanzeigen_bot.ad_loading.load_ad_configs", return_value = []),
|
||||||
|
patch("kleinanzeigen_bot.ad_status.render_status_rows", return_value = ""),
|
||||||
|
):
|
||||||
|
await bot.run(["app", "status"])
|
||||||
|
|
||||||
|
assert update_called, "status must call _check_for_updates()"
|
||||||
|
assert not load_ads_called, "status must not call load_ads()"
|
||||||
|
assert not browser_called, "status must not open browser or login"
|
||||||
|
|
||||||
|
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
# Parser / dispatch
|
||||||
|
# --------------------------------------------------------------------------- #
|
||||||
|
|
||||||
|
|
||||||
|
def test_status_is_valid_command() -> None:
|
||||||
|
"""Status is recognised by the parser."""
|
||||||
|
parsed = parse_args(["app", "status"])
|
||||||
|
assert parsed.command == "status"
|
||||||
|
|
||||||
|
|
||||||
|
def test_status_in_runtime_config() -> None:
|
||||||
|
"""Status is listed in VALID_COMMANDS."""
|
||||||
|
assert "status" in VALID_COMMANDS
|
||||||
Reference in New Issue
Block a user