0f9c2a1cbd
- add backend README and refresh API map and smoke regression docs - add full backend smoke regression script - add admin pricing cleanup preview and dry-run endpoints - add helper script for test pricing cleanup - verify typed error contracts, draft flow, publish readiness and preview flows - verify publish preview retention and clean backend startup behavior
128 lines
3.9 KiB
Python
128 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
from app.repositories.pricing_cleanup import (
|
|
delete_pricing_categories_by_ids,
|
|
list_pricing_categories_with_rule_counts,
|
|
)
|
|
|
|
|
|
def _matches_any_prefix(value: str | None, prefixes: list[str]) -> list[str]:
|
|
if not value:
|
|
return []
|
|
matches: list[str] = []
|
|
lower_value = value.lower()
|
|
for prefix in prefixes:
|
|
if lower_value.startswith(prefix.lower()):
|
|
matches.append(prefix)
|
|
return matches
|
|
|
|
|
|
async def build_pricing_cleanup_preview(
|
|
*,
|
|
scheme_id: str,
|
|
code_prefixes: list[str],
|
|
name_prefixes: list[str],
|
|
pricing_category_ids: list[str],
|
|
delete_only_without_rules: bool,
|
|
) -> dict:
|
|
rows = await list_pricing_categories_with_rule_counts(scheme_id=scheme_id)
|
|
requested_ids = set(pricing_category_ids)
|
|
|
|
items: list[dict] = []
|
|
safe_to_delete_count = 0
|
|
|
|
for row in rows:
|
|
matched_by: list[str] = []
|
|
|
|
for prefix in _matches_any_prefix(row["code"], code_prefixes):
|
|
matched_by.append(f"code_prefix:{prefix}")
|
|
|
|
for prefix in _matches_any_prefix(row["name"], name_prefixes):
|
|
matched_by.append(f"name_prefix:{prefix}")
|
|
|
|
if row["pricing_category_id"] in requested_ids:
|
|
matched_by.append("pricing_category_id")
|
|
|
|
if not matched_by:
|
|
continue
|
|
|
|
deletable = True
|
|
if delete_only_without_rules and row["rules_count"] > 0:
|
|
deletable = False
|
|
|
|
if deletable:
|
|
safe_to_delete_count += 1
|
|
|
|
items.append(
|
|
{
|
|
"pricing_category_id": row["pricing_category_id"],
|
|
"name": row["name"],
|
|
"code": row["code"],
|
|
"rules_count": row["rules_count"],
|
|
"matched_by": matched_by,
|
|
"deletable": deletable,
|
|
}
|
|
)
|
|
|
|
return {
|
|
"scheme_id": scheme_id,
|
|
"code_prefixes": code_prefixes,
|
|
"name_prefixes": name_prefixes,
|
|
"pricing_category_ids": pricing_category_ids,
|
|
"delete_only_without_rules": delete_only_without_rules,
|
|
"total_candidates": len(items),
|
|
"safe_to_delete_count": safe_to_delete_count,
|
|
"items": items,
|
|
}
|
|
|
|
|
|
async def execute_pricing_cleanup(
|
|
*,
|
|
scheme_id: str,
|
|
code_prefixes: list[str],
|
|
name_prefixes: list[str],
|
|
pricing_category_ids: list[str],
|
|
delete_only_without_rules: bool,
|
|
dry_run: bool,
|
|
) -> dict:
|
|
preview = await build_pricing_cleanup_preview(
|
|
scheme_id=scheme_id,
|
|
code_prefixes=code_prefixes,
|
|
name_prefixes=name_prefixes,
|
|
pricing_category_ids=pricing_category_ids,
|
|
delete_only_without_rules=delete_only_without_rules,
|
|
)
|
|
|
|
deletable_items = [item for item in preview["items"] if item["deletable"]]
|
|
skipped_items = [item for item in preview["items"] if not item["deletable"]]
|
|
|
|
would_delete_ids = [item["pricing_category_id"] for item in deletable_items]
|
|
deleted_ids: list[str] = []
|
|
|
|
if not dry_run and would_delete_ids:
|
|
await delete_pricing_categories_by_ids(
|
|
scheme_id=scheme_id,
|
|
pricing_category_ids=would_delete_ids,
|
|
)
|
|
deleted_ids = list(would_delete_ids)
|
|
|
|
return {
|
|
"scheme_id": scheme_id,
|
|
"dry_run": dry_run,
|
|
"delete_only_without_rules": delete_only_without_rules,
|
|
"requested_total": len(pricing_category_ids) + len(code_prefixes) + len(name_prefixes),
|
|
"matched_total": preview["total_candidates"],
|
|
"would_delete_count": len(would_delete_ids),
|
|
"deleted_count": 0 if dry_run else len(deleted_ids),
|
|
"skipped_count": len(skipped_items),
|
|
"would_delete_category_ids": would_delete_ids,
|
|
"deleted_category_ids": deleted_ids,
|
|
"skipped": [
|
|
{
|
|
"pricing_category_id": item["pricing_category_id"],
|
|
"reason": "category_has_rules",
|
|
}
|
|
for item in skipped_items
|
|
],
|
|
}
|