mirror of
https://github.com/Second-Hand-Friends/kleinanzeigen-bot.git
synced 2026-03-12 18:41:50 +01:00
feat: Add descriptive comments and examples to create-config output (#805)
This commit is contained in:
Jens
GitHub
@@ -579,14 +579,15 @@ class KleinanzeigenBot(WebScrapingMixin): # noqa: PLR0904
|
||||
default_config = Config.model_construct()
|
||||
default_config.login.username = "changeme" # noqa: S105 placeholder for default config, not a real username
|
||||
default_config.login.password = "changeme" # noqa: S105 placeholder for default config, not a real password
|
||||
dicts.save_dict(
|
||||
dicts.save_commented_model(
|
||||
self.config_file_path,
|
||||
default_config.model_dump(exclude_none = True, exclude = {"ad_defaults": {"description"}}),
|
||||
default_config,
|
||||
header=(
|
||||
"# yaml-language-server: $schema="
|
||||
"https://raw.githubusercontent.com/Second-Hand-Friends/kleinanzeigen-bot"
|
||||
"/refs/heads/main/schemas/config.schema.json"
|
||||
),
|
||||
exclude = {"ad_defaults": {"description"}},
|
||||
)
|
||||
|
||||
def load_config(self) -> None:
|
||||
|
||||
@@ -23,12 +23,19 @@ _MAX_PERCENTAGE:Final[int] = 100
|
||||
class AutoPriceReductionConfig(ContextualModel):
|
||||
enabled:bool = Field(default = False, description = "automatically lower the price of reposted ads")
|
||||
strategy:Literal["FIXED", "PERCENTAGE"] | None = Field(
|
||||
default = None, description = "PERCENTAGE reduces by a percentage of the previous price, FIXED reduces by a fixed amount"
|
||||
default = None,
|
||||
description = "reduction strategy (required when enabled: true). PERCENTAGE = % of price, FIXED = absolute amount",
|
||||
examples = ["PERCENTAGE", "FIXED"],
|
||||
)
|
||||
amount:float | None = Field(
|
||||
default = None, gt = 0, description = "magnitude of the reduction; interpreted as percent for PERCENTAGE or currency units for FIXED"
|
||||
default = None,
|
||||
gt = 0,
|
||||
description = "reduction amount (required when enabled: true). For PERCENTAGE: use percent value (e.g., 10 = 10%%). For FIXED: use currency amount",
|
||||
examples = [10.0, 5.0, 20.0],
|
||||
)
|
||||
min_price:float | None = Field(
|
||||
default = None, ge = 0, description = "minimum price floor (required when enabled: true). Use 0 for no minimum", examples = [1.0, 5.0, 10.0]
|
||||
)
|
||||
min_price:float | None = Field(default = None, ge = 0, description = "required when enabled is true; minimum price floor (use 0 for no lower bound)")
|
||||
delay_reposts:int = Field(default = 0, ge = 0, description = "number of reposts to wait before applying the first automatic price reduction")
|
||||
delay_days:int = Field(default = 0, ge = 0, description = "number of days to wait after publication before applying automatic price reductions")
|
||||
|
||||
@@ -47,34 +54,50 @@ class AutoPriceReductionConfig(ContextualModel):
|
||||
|
||||
|
||||
class ContactDefaults(ContextualModel):
|
||||
name:str | None = None
|
||||
street:str | None = None
|
||||
zipcode:int | str | None = None
|
||||
location:str | None = Field(
|
||||
default = None, description = "city or locality of the listing (can include multiple districts)", examples = ["Sample Town - District One"]
|
||||
name:str = Field(default = "", description = "contact name displayed on the ad")
|
||||
street:str = Field(default = "", description = "street address for the listing")
|
||||
zipcode:int | str = Field(default = "", description = "postal/ZIP code for the listing location")
|
||||
location:str = Field(
|
||||
default = "",
|
||||
description = "city or locality of the listing (can include multiple districts)",
|
||||
examples = ["Sample Town - District One"],
|
||||
)
|
||||
phone:str = Field(
|
||||
default = "",
|
||||
description = "phone number for contact - only available for commercial accounts, personal accounts no longer support this",
|
||||
examples = ['"01234 567890"'],
|
||||
)
|
||||
phone:str | None = None
|
||||
|
||||
|
||||
@deprecated("Use description_prefix/description_suffix instead")
|
||||
class DescriptionAffixes(ContextualModel):
|
||||
prefix:str | None = None
|
||||
suffix:str | None = None
|
||||
prefix:str | None = Field(default = None, description = "text to prepend to the ad description (deprecated, use description_prefix)")
|
||||
suffix:str | None = Field(default = None, description = "text to append to the ad description (deprecated, use description_suffix)")
|
||||
|
||||
|
||||
class AdDefaults(ContextualModel):
|
||||
active:bool = True
|
||||
type:Literal["OFFER", "WANTED"] = "OFFER"
|
||||
description:DescriptionAffixes | None = None
|
||||
description_prefix:str | None = Field(default = None, description = "prefix for the ad description")
|
||||
description_suffix:str | None = Field(default = None, description = "suffix for the ad description")
|
||||
price_type:Literal["FIXED", "NEGOTIABLE", "GIVE_AWAY", "NOT_APPLICABLE"] = "NEGOTIABLE"
|
||||
auto_price_reduction:AutoPriceReductionConfig = Field(default_factory = AutoPriceReductionConfig, description = "automatic price reduction configuration")
|
||||
shipping_type:Literal["PICKUP", "SHIPPING", "NOT_APPLICABLE"] = "SHIPPING"
|
||||
sell_directly:bool = Field(default = False, description = "requires shipping_type SHIPPING to take effect")
|
||||
images:list[str] | None = Field(default = None)
|
||||
contact:ContactDefaults = Field(default_factory = ContactDefaults)
|
||||
republication_interval:int = 7
|
||||
active:bool = Field(default = True, description = "whether the ad should be published (false = skip this ad)")
|
||||
type:Literal["OFFER", "WANTED"] = Field(default = "OFFER", description = "type of the ad listing", examples = ["OFFER", "WANTED"])
|
||||
description:DescriptionAffixes | None = Field(default = None, description = "DEPRECATED: Use description_prefix/description_suffix instead")
|
||||
description_prefix:str | None = Field(default = "", description = "text to prepend to each ad (optional)")
|
||||
description_suffix:str | None = Field(default = "", description = "text to append to each ad (optional)")
|
||||
price_type:Literal["FIXED", "NEGOTIABLE", "GIVE_AWAY", "NOT_APPLICABLE"] = Field(
|
||||
default = "NEGOTIABLE", description = "pricing strategy for the listing", examples = ["FIXED", "NEGOTIABLE", "GIVE_AWAY", "NOT_APPLICABLE"]
|
||||
)
|
||||
auto_price_reduction:AutoPriceReductionConfig = Field(
|
||||
default_factory = AutoPriceReductionConfig, description = "automatic price reduction configuration for reposted ads"
|
||||
)
|
||||
shipping_type:Literal["PICKUP", "SHIPPING", "NOT_APPLICABLE"] = Field(
|
||||
default = "SHIPPING", description = "shipping method for the item", examples = ["PICKUP", "SHIPPING", "NOT_APPLICABLE"]
|
||||
)
|
||||
sell_directly:bool = Field(default = False, description = "enable direct purchase option (only works when shipping_type is SHIPPING)")
|
||||
images:list[str] | None = Field(
|
||||
default_factory = list,
|
||||
description = "default image glob patterns (optional). Leave empty for no default images",
|
||||
examples = ['"images/*.jpg"', '"photos/*.{png,jpg}"'],
|
||||
)
|
||||
contact:ContactDefaults = Field(default_factory = ContactDefaults, description = "default contact information for ads")
|
||||
republication_interval:int = Field(default = 7, description = "number of days between automatic republication of ads")
|
||||
|
||||
@model_validator(mode = "before")
|
||||
@classmethod
|
||||
@@ -99,7 +122,8 @@ class DownloadConfig(ContextualModel):
|
||||
)
|
||||
excluded_shipping_options:list[str] = Field(
|
||||
default_factory = list,
|
||||
description = "list of shipping options to exclude, e.g. ['DHL_2', 'DHL_5']",
|
||||
description = ("shipping options to exclude (optional). Leave as [] to include all. Add items like 'DHL_2' to exclude specific carriers"),
|
||||
examples = ['"DHL_2"', '"DHL_5"', '"Hermes"'],
|
||||
)
|
||||
folder_name_max_length:int = Field(
|
||||
default = 100,
|
||||
@@ -117,36 +141,49 @@ class BrowserConfig(ContextualModel):
|
||||
arguments:list[str] = Field(
|
||||
default_factory = list,
|
||||
description=(
|
||||
"See https://peter.sh/experiments/chromium-command-line-switches/. "
|
||||
"Browser profile path is auto-configured based on installation mode (portable/XDG)."
|
||||
"additional Chromium command line switches (optional). Leave as [] for default behavior. "
|
||||
"See https://peter.sh/experiments/chromium-command-line-switches/ "
|
||||
"Common: --headless (no GUI), --disable-dev-shm-usage (Docker fix), --user-data-dir=/path"
|
||||
),
|
||||
examples = ['"--headless"', '"--disable-dev-shm-usage"', '"--user-data-dir=/path/to/profile"'],
|
||||
)
|
||||
binary_location:str | None = Field(default = None, description = "path to custom browser executable, if not specified will be looked up on PATH")
|
||||
extensions:list[str] = Field(default_factory = list, description = "a list of .crx extension files to be loaded")
|
||||
use_private_window:bool = True
|
||||
binary_location:str | None = Field(default = "", description = "path to custom browser executable (optional). Leave empty to use system default")
|
||||
extensions:list[str] = Field(
|
||||
default_factory = list,
|
||||
description = "Chrome extensions to load (optional). Leave as [] for no extensions. Add .crx file paths relative to config file",
|
||||
examples = ['"extensions/adblock.crx"', '"/absolute/path/to/extension.crx"'],
|
||||
)
|
||||
use_private_window:bool = Field(default = True, description = "open browser in private/incognito mode (recommended to avoid cookie conflicts)")
|
||||
user_data_dir:str | None = Field(
|
||||
default = None,
|
||||
description=(
|
||||
"See https://github.com/chromium/chromium/blob/main/docs/user_data_dir.md. "
|
||||
"If not specified, defaults to XDG cache directory in XDG mode or .temp/browser-profile in portable mode."
|
||||
),
|
||||
default = "",
|
||||
description = "custom browser profile directory (optional). Leave empty for auto-configured default",
|
||||
)
|
||||
profile_name:str | None = Field(
|
||||
default = "",
|
||||
description = "browser profile name (optional). Leave empty for default profile",
|
||||
examples = ['"Profile 1"'],
|
||||
)
|
||||
profile_name:str | None = None
|
||||
|
||||
|
||||
class LoginConfig(ContextualModel):
|
||||
username:str = Field(..., min_length = 1)
|
||||
password:str = Field(..., min_length = 1)
|
||||
username:str = Field(..., min_length = 1, description = "kleinanzeigen.de login email or username")
|
||||
password:str = Field(..., min_length = 1, description = "kleinanzeigen.de login password")
|
||||
|
||||
|
||||
class PublishingConfig(ContextualModel):
|
||||
delete_old_ads:Literal["BEFORE_PUBLISH", "AFTER_PUBLISH", "NEVER"] | None = "AFTER_PUBLISH"
|
||||
delete_old_ads_by_title:bool = Field(default = True, description = "only works if delete_old_ads is set to BEFORE_PUBLISH")
|
||||
delete_old_ads:Literal["BEFORE_PUBLISH", "AFTER_PUBLISH", "NEVER"] | None = Field(
|
||||
default = "AFTER_PUBLISH", description = "when to delete old versions of republished ads", examples = ["BEFORE_PUBLISH", "AFTER_PUBLISH", "NEVER"]
|
||||
)
|
||||
delete_old_ads_by_title:bool = Field(default = True, description = "match old ads by title when deleting (only works with BEFORE_PUBLISH)")
|
||||
|
||||
|
||||
class CaptchaConfig(ContextualModel):
|
||||
auto_restart:bool = False
|
||||
restart_delay:str = "6h"
|
||||
auto_restart:bool = Field(
|
||||
default = False, description = "if true, abort when captcha is detected and auto-retry after restart_delay (if false, wait for manual solving)"
|
||||
)
|
||||
restart_delay:str = Field(
|
||||
default = "6h", description = "duration to wait before retrying after captcha detection (e.g., 1h30m, 6h, 30m)", examples = ["6h", "1h30m", "30m"]
|
||||
)
|
||||
|
||||
|
||||
class TimeoutConfig(ContextualModel):
|
||||
@@ -291,15 +328,12 @@ if relative paths are specified, then they are relative to this configuration fi
|
||||
|
||||
categories:dict[str, str] = Field(
|
||||
default_factory = dict,
|
||||
description = """
|
||||
additional name to category ID mappings, see default list at
|
||||
https://github.com/Second-Hand-Friends/kleinanzeigen-bot/blob/main/src/kleinanzeigen_bot/resources/categories.yaml
|
||||
|
||||
Example:
|
||||
categories:
|
||||
Elektronik > Notebooks: 161/278
|
||||
Jobs > Praktika: 102/125
|
||||
""",
|
||||
description=(
|
||||
"additional name to category ID mappings (optional). Leave as {} if not needed. "
|
||||
"See full list at: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/blob/main/src/kleinanzeigen_bot/resources/categories.yaml "
|
||||
"To add: use format 'Category > Subcategory': 'ID'"
|
||||
),
|
||||
examples = ['"Elektronik > Notebooks": "161/278"', '"Jobs > Praktika": "102/125"'],
|
||||
)
|
||||
|
||||
download:DownloadConfig = Field(default_factory = DownloadConfig)
|
||||
@@ -309,7 +343,7 @@ Example:
|
||||
captcha:CaptchaConfig = Field(default_factory = CaptchaConfig)
|
||||
update_check:UpdateCheckConfig = Field(default_factory = UpdateCheckConfig, description = "Update check configuration")
|
||||
timeouts:TimeoutConfig = Field(default_factory = TimeoutConfig, description = "Centralized timeout configuration.")
|
||||
diagnostics:DiagnosticsConfig | None = Field(default = None, description = "Optional failure-only diagnostics capture.")
|
||||
diagnostics:DiagnosticsConfig = Field(default_factory = DiagnosticsConfig, description = "diagnostics capture configuration for troubleshooting")
|
||||
|
||||
def with_values(self, values:dict[str, Any]) -> Config:
|
||||
return Config.model_validate(dicts.apply_defaults(copy.deepcopy(values), defaults = self.model_dump()))
|
||||
|
||||
@@ -6,22 +6,22 @@ from __future__ import annotations
|
||||
|
||||
from typing import Literal
|
||||
|
||||
from pydantic import Field
|
||||
|
||||
from kleinanzeigen_bot.utils.pydantics import ContextualModel
|
||||
|
||||
|
||||
class UpdateCheckConfig(ContextualModel):
|
||||
"""Configuration for update checking functionality.
|
||||
|
||||
Attributes:
|
||||
enabled: Whether update checking is enabled.
|
||||
channel: Which release channel to check ('latest' for stable, 'preview' for prereleases).
|
||||
interval: How often to check for updates (e.g. '7d', '1d').
|
||||
If the interval is invalid, too short (<1d), or too long (>30d),
|
||||
the bot will log a warning and use a default interval for this run:
|
||||
- 1d for 'preview' channel
|
||||
- 7d for 'latest' channel
|
||||
The config file is not changed automatically; please fix your config to avoid repeated warnings.
|
||||
"""
|
||||
enabled:bool = True
|
||||
channel:Literal["latest", "preview"] = "latest"
|
||||
interval:str = "7d" # Default interval of 7 days
|
||||
enabled:bool = Field(default = True, description = "whether to check for updates on startup")
|
||||
channel:Literal["latest", "preview"] = Field(
|
||||
default = "latest", description = "which release channel to check (latest = stable, preview = prereleases)", examples = ["latest", "preview"]
|
||||
)
|
||||
interval:str = Field(
|
||||
default = "7d",
|
||||
description=(
|
||||
"how often to check for updates (e.g., 7d, 1d). "
|
||||
"If invalid, too short (<1d), or too long (>30d), "
|
||||
"uses defaults: 1d for 'preview' channel, 7d for 'latest' channel"
|
||||
),
|
||||
examples = ["7d", "1d", "14d"],
|
||||
)
|
||||
|
||||
@@ -350,6 +350,8 @@ kleinanzeigen_bot/utils/dicts.py:
|
||||
"Unsupported file type. The filename \"%s\" must end with *.json, *.yaml, or *.yml": "Nicht unterstützter Dateityp. Der Dateiname \"%s\" muss mit *.json, *.yaml oder *.yml enden"
|
||||
save_dict:
|
||||
"Saving [%s]...": "Speichere [%s]..."
|
||||
save_commented_model:
|
||||
"Saving [%s]...": "Speichere [%s]..."
|
||||
load_dict_from_module:
|
||||
"Loading %s[%s.%s]...": "Lade %s[%s.%s]..."
|
||||
|
||||
|
||||
@@ -8,7 +8,7 @@ from gettext import gettext as _
|
||||
from importlib.resources import read_text as get_resource_as_string
|
||||
from pathlib import Path
|
||||
from types import ModuleType
|
||||
from typing import Any, Final, TypeVar
|
||||
from typing import Any, Final, TypeVar, cast, get_origin
|
||||
|
||||
from ruamel.yaml import YAML
|
||||
|
||||
@@ -25,7 +25,7 @@ def apply_defaults(
|
||||
target:dict[Any, Any],
|
||||
defaults:dict[Any, Any],
|
||||
ignore:Callable[[Any, Any], bool] = lambda _k, _v: False,
|
||||
override:Callable[[Any, Any], bool] = lambda _k, _v: False
|
||||
override:Callable[[Any, Any], bool] = lambda _k, _v: False,
|
||||
) -> dict[Any, Any]:
|
||||
"""
|
||||
>>> apply_defaults({}, {'a': 'b'})
|
||||
@@ -48,12 +48,7 @@ def apply_defaults(
|
||||
for key, default_value in defaults.items():
|
||||
if key in target:
|
||||
if isinstance(target[key], dict) and isinstance(default_value, dict):
|
||||
apply_defaults(
|
||||
target = target[key],
|
||||
defaults = default_value,
|
||||
ignore = ignore,
|
||||
override = override
|
||||
)
|
||||
apply_defaults(target = target[key], defaults = default_value, ignore = ignore, override = override)
|
||||
elif override(key, target[key]): # force overwrite if override says so
|
||||
target[key] = copy.deepcopy(default_value)
|
||||
elif not ignore(key, default_value): # only set if not explicitly ignored
|
||||
@@ -111,6 +106,24 @@ def load_dict_from_module(module:ModuleType, filename:str, content_label:str = "
|
||||
return json.loads(content) if filename.endswith(".json") else YAML().load(content) # type: ignore[no-any-return] # mypy
|
||||
|
||||
|
||||
def _configure_yaml() -> YAML:
|
||||
"""
|
||||
Configure and return a YAML instance with standard settings.
|
||||
|
||||
Returns:
|
||||
Configured YAML instance ready for dumping
|
||||
"""
|
||||
yaml = YAML()
|
||||
yaml.indent(mapping = 2, sequence = 4, offset = 2)
|
||||
yaml.representer.add_representer(
|
||||
str, # use YAML | block style for multi-line strings
|
||||
lambda dumper, data: dumper.represent_scalar("tag:yaml.org,2002:str", data, style = "|" if "\n" in data else None),
|
||||
)
|
||||
yaml.allow_duplicate_keys = False
|
||||
yaml.explicit_start = False
|
||||
return yaml
|
||||
|
||||
|
||||
def save_dict(filepath:str | Path, content:dict[str, Any], *, header:str | None = None) -> None:
|
||||
# Normalize filepath to NFC for cross-platform consistency (issue #728)
|
||||
# Ensures file paths match NFC-normalized directory names from sanitize_folder_name()
|
||||
@@ -128,14 +141,7 @@ def save_dict(filepath:str | Path, content:dict[str, Any], *, header:str | None
|
||||
if filepath.suffix == ".json":
|
||||
file.write(json.dumps(content, indent = 2, ensure_ascii = False))
|
||||
else:
|
||||
yaml = YAML()
|
||||
yaml.indent(mapping = 2, sequence = 4, offset = 2)
|
||||
yaml.representer.add_representer(str, # use YAML | block style for multi-line strings
|
||||
lambda dumper, data:
|
||||
dumper.represent_scalar("tag:yaml.org,2002:str", data, style = "|" if "\n" in data else None)
|
||||
)
|
||||
yaml.allow_duplicate_keys = False
|
||||
yaml.explicit_start = False
|
||||
yaml = _configure_yaml()
|
||||
yaml.dump(content, file)
|
||||
|
||||
|
||||
@@ -153,3 +159,206 @@ def safe_get(a_map:dict[Any, Any], *keys:str) -> Any:
|
||||
except (KeyError, TypeError):
|
||||
return None
|
||||
return a_map
|
||||
|
||||
|
||||
def _should_exclude(field_name:str, exclude:set[str] | dict[str, Any] | None) -> bool:
|
||||
"""Check if a field should be excluded based on exclude rules."""
|
||||
if exclude is None:
|
||||
return False
|
||||
if isinstance(exclude, set):
|
||||
return field_name in exclude
|
||||
if isinstance(exclude, dict):
|
||||
# If the value is None, it means exclude this field entirely
|
||||
# If the value is a dict/set, it means nested exclusion rules
|
||||
if field_name in exclude:
|
||||
return exclude[field_name] is None
|
||||
return False
|
||||
|
||||
|
||||
def _get_nested_exclude(field_name:str, exclude:set[str] | dict[str, Any] | None) -> set[str] | dict[str, Any] | None:
|
||||
"""Get nested exclude rules for a field."""
|
||||
if exclude is None:
|
||||
return None
|
||||
if isinstance(exclude, dict) and field_name in exclude:
|
||||
nested = exclude[field_name]
|
||||
# If nested is None, it means exclude entirely - no nested rules to pass down
|
||||
# If nested is a set or dict, pass it down as nested exclusion rules
|
||||
if nested is None:
|
||||
return None
|
||||
return cast(set[str] | dict[str, Any], nested)
|
||||
return None
|
||||
|
||||
|
||||
def model_to_commented_yaml(
|
||||
model_instance:Any,
|
||||
*,
|
||||
indent_level:int = 0,
|
||||
exclude:set[str] | dict[str, Any] | None = None,
|
||||
) -> Any:
|
||||
"""
|
||||
Convert a Pydantic model instance to a structure with YAML comments.
|
||||
|
||||
This function recursively processes a Pydantic model and creates a
|
||||
CommentedMap/CommentedSeq structure with comments based on field descriptions.
|
||||
The comments are added as block comments above each field.
|
||||
|
||||
Args:
|
||||
model_instance: A Pydantic model instance to convert
|
||||
indent_level: Current indentation level (for recursive calls)
|
||||
exclude: Optional set of field names to exclude, or dict for nested exclusion
|
||||
|
||||
Returns:
|
||||
A CommentedMap, CommentedSeq, or primitive value suitable for YAML output
|
||||
|
||||
Example:
|
||||
>>> from pydantic import BaseModel, Field
|
||||
>>> class Config(BaseModel):
|
||||
... name: str = Field(default="test", description="The name")
|
||||
>>> config = Config()
|
||||
>>> result = model_to_commented_yaml(config)
|
||||
"""
|
||||
# Delayed import to avoid circular dependency
|
||||
from pydantic import BaseModel # noqa: PLC0415
|
||||
from ruamel.yaml.comments import CommentedMap, CommentedSeq # noqa: PLC0415
|
||||
|
||||
# Handle primitive types
|
||||
if model_instance is None or isinstance(model_instance, (str, int, float, bool)):
|
||||
return model_instance
|
||||
|
||||
# Handle lists/sequences
|
||||
if isinstance(model_instance, (list, tuple)):
|
||||
seq = CommentedSeq()
|
||||
for item in model_instance:
|
||||
seq.append(model_to_commented_yaml(item, indent_level = indent_level + 1, exclude = exclude))
|
||||
return seq
|
||||
|
||||
# Handle dictionaries (not from Pydantic models)
|
||||
if isinstance(model_instance, dict) and not isinstance(model_instance, BaseModel):
|
||||
cmap = CommentedMap()
|
||||
for key, value in model_instance.items():
|
||||
if _should_exclude(key, exclude):
|
||||
continue
|
||||
cmap[key] = model_to_commented_yaml(value, indent_level = indent_level + 1, exclude = exclude)
|
||||
return cmap
|
||||
|
||||
# Handle Pydantic models
|
||||
if isinstance(model_instance, BaseModel):
|
||||
cmap = CommentedMap()
|
||||
model_class = model_instance.__class__
|
||||
field_count = 0
|
||||
|
||||
# Get field information from the model class
|
||||
for field_name, field_info in model_class.model_fields.items():
|
||||
# Skip excluded fields
|
||||
if _should_exclude(field_name, exclude):
|
||||
continue
|
||||
|
||||
# Get the value from the instance, handling unset required fields
|
||||
try:
|
||||
value = getattr(model_instance, field_name)
|
||||
except AttributeError:
|
||||
# Field is not set (e.g., required field with no default)
|
||||
continue
|
||||
|
||||
# Add visual separators
|
||||
if indent_level == 0 and field_count > 0:
|
||||
# Major section: blank line + prominent separator with 80 # characters
|
||||
cmap.yaml_set_comment_before_after_key(field_name, before = "\n" + "#" * 80, indent = 0)
|
||||
elif indent_level > 0:
|
||||
# Nested fields: always add blank line separator (both between siblings and before first child)
|
||||
cmap.yaml_set_comment_before_after_key(field_name, before = "", indent = 0)
|
||||
|
||||
# Get nested exclude rules for this field
|
||||
nested_exclude = _get_nested_exclude(field_name, exclude)
|
||||
|
||||
# Process the value recursively
|
||||
processed_value = model_to_commented_yaml(value, indent_level = indent_level + 1, exclude = nested_exclude)
|
||||
cmap[field_name] = processed_value
|
||||
field_count += 1
|
||||
|
||||
# Build comment from description and examples
|
||||
comment_parts = []
|
||||
|
||||
# Add description if available
|
||||
description = field_info.description
|
||||
if description:
|
||||
comment_parts.append(description)
|
||||
|
||||
# Add examples if available
|
||||
examples = field_info.examples
|
||||
if examples:
|
||||
# Check if this is a list field by inspecting type annotation first (handles empty lists),
|
||||
# then fall back to runtime value type check
|
||||
is_list_field = get_origin(field_info.annotation) is list or isinstance(value, list)
|
||||
|
||||
if is_list_field:
|
||||
# For list fields, show YAML syntax with field name for clarity
|
||||
examples_lines = [
|
||||
"Example usage:",
|
||||
f" {field_name}:",
|
||||
*[f" - {ex}" for ex in examples]
|
||||
]
|
||||
comment_parts.append("\n".join(examples_lines))
|
||||
elif len(examples) == 1:
|
||||
# Single example for scalar field: use singular form without list marker
|
||||
comment_parts.append(f"Example: {examples[0]}")
|
||||
else:
|
||||
# Multiple examples for scalar field: show as alternatives (not list items)
|
||||
# Use bullets (•) instead of hyphens to distinguish from YAML list syntax
|
||||
examples_lines = ["Examples (choose one):", *[f" • {ex}" for ex in examples]]
|
||||
comment_parts.append("\n".join(examples_lines))
|
||||
|
||||
# Set the comment above the key
|
||||
if comment_parts:
|
||||
full_comment = "\n".join(comment_parts)
|
||||
cmap.yaml_set_comment_before_after_key(field_name, before = full_comment, indent = indent_level * 2)
|
||||
|
||||
return cmap
|
||||
|
||||
# Fallback: return as-is
|
||||
return model_instance
|
||||
|
||||
|
||||
def save_commented_model(
|
||||
filepath:str | Path,
|
||||
model_instance:Any,
|
||||
*,
|
||||
header:str | None = None,
|
||||
exclude:set[str] | dict[str, Any] | None = None,
|
||||
) -> None:
|
||||
"""
|
||||
Save a Pydantic model to a YAML file with field descriptions as comments.
|
||||
|
||||
This function converts a Pydantic model to a commented YAML structure
|
||||
where each field has its description (and optionally examples) as a
|
||||
block comment above the key.
|
||||
|
||||
Args:
|
||||
filepath: Path to the output YAML file
|
||||
model_instance: Pydantic model instance to save
|
||||
header: Optional header string to write at the top of the file
|
||||
exclude: Optional set of field names to exclude, or dict for nested exclusion
|
||||
|
||||
Example:
|
||||
>>> from kleinanzeigen_bot.model.config_model import Config
|
||||
>>> from pathlib import Path
|
||||
>>> import tempfile
|
||||
>>> config = Config()
|
||||
>>> with tempfile.TemporaryDirectory() as tmpdir:
|
||||
... save_commented_model(Path(tmpdir) / "config.yaml", config, header="# Config file")
|
||||
"""
|
||||
filepath = Path(unicodedata.normalize("NFC", str(filepath)))
|
||||
filepath.parent.mkdir(parents = True, exist_ok = True)
|
||||
|
||||
LOG.info("Saving [%s]...", filepath)
|
||||
|
||||
# Convert to commented structure directly from model (preserves metadata)
|
||||
commented_data = model_to_commented_yaml(model_instance, exclude = exclude)
|
||||
|
||||
with open(filepath, "w", encoding = "utf-8") as file:
|
||||
if header:
|
||||
file.write(header)
|
||||
file.write("\n")
|
||||
|
||||
yaml = _configure_yaml()
|
||||
yaml.dump(commented_data, file)
|
||||
|
||||
Reference in New Issue
Block a user