"""Pure helpers for classifying and presenting collection skip reasons.""" from __future__ import annotations ALIAS_UNMATCHED = "alias_unmatched" LOGIN_REQUIRED = "login_required" OTHER = "other" SKIP_REASON_ORDER = (ALIAS_UNMATCHED, LOGIN_REQUIRED, OTHER) SKIP_REASON_LABELS = { ALIAS_UNMATCHED: "别名未匹配", LOGIN_REQUIRED: "账号未登录", OTHER: "其他原因", } def empty_skip_reason_counts(): return {reason: 0 for reason in SKIP_REASON_ORDER} def classify_skip_reason(reason="", *, alias_matched=True): """Classify historical skip text without treating uncertain login checks as logout.""" if not alias_matched: return ALIAS_UNMATCHED text = str(reason or "").strip().lower() if "别名未匹配" in text or "alias_unmatched" in text: return ALIAS_UNMATCHED if ( "账号未登录" in text or "采集中途掉登录" in text or "login_page" in text or ("accounts.shopee." in text and "/seller/login" in text) ): return LOGIN_REQUIRED return OTHER def skipped_stage_text(reason="", *, alias_matched=True): reason_code = classify_skip_reason(reason, alias_matched=alias_matched) return f"略过 · {SKIP_REASON_LABELS[reason_code]}" def normalize_skip_reason_counts(counts=None, *, skipped_total=0): normalized = empty_skip_reason_counts() source = counts if isinstance(counts, dict) else {} for reason in SKIP_REASON_ORDER: try: normalized[reason] = max(0, int(source.get(reason, 0) or 0)) except (TypeError, ValueError): normalized[reason] = 0 total = max(0, int(skipped_total or 0)) accounted = sum(normalized.values()) if accounted < total: normalized[OTHER] += total - accounted return normalized def format_skip_reason_summary(counts=None, *, skipped_total=0): total = max(0, int(skipped_total or 0)) if total == 0: return "" normalized = normalize_skip_reason_counts(counts, skipped_total=total) parts = [ f"{SKIP_REASON_LABELS[reason]}{normalized[reason]}" for reason in SKIP_REASON_ORDER if normalized[reason] > 0 ] return f"略过{total}:" + ",".join(parts)