117 lines
3.7 KiB
Python
117 lines
3.7 KiB
Python
# wc_period_payload.py – Woo Bridge: Periodenfelder aus Provision/Sync-Payload ableiten
|
||
# (ohne /wc/v3/subscriptions REST).
|
||
from __future__ import annotations
|
||
|
||
from typing import Any, Optional, Tuple
|
||
|
||
from aza_wc_period_sync import _parse_wc_datetime, _subtract_billing_period
|
||
|
||
# Plausible Unix range: 2000-01-01 … 2100-01-01 UTC
|
||
_TS_MIN = 946684800
|
||
_TS_MAX = 4102444800
|
||
|
||
|
||
def _normalize_int_ts(v: Optional[int]) -> Optional[int]:
|
||
if v is None:
|
||
return None
|
||
try:
|
||
x = int(v)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
if x > 10**12: # ms
|
||
x //= 1000
|
||
if _TS_MIN <= x <= _TS_MAX:
|
||
return x
|
||
return None
|
||
|
||
|
||
def coerce_next_payment_unix(value: Any) -> Optional[int]:
|
||
"""Unix-Sekunden aus next_payment (int, Ziffernstring oder ISO-Datum)."""
|
||
if value is None:
|
||
return None
|
||
if isinstance(value, bool):
|
||
return None
|
||
if isinstance(value, int):
|
||
return _normalize_int_ts(value)
|
||
if isinstance(value, float):
|
||
return _normalize_int_ts(int(value))
|
||
s = str(value).strip()
|
||
if not s:
|
||
return None
|
||
if s.isdigit():
|
||
return _normalize_int_ts(int(s))
|
||
parsed = _parse_wc_datetime(s)
|
||
return _normalize_int_ts(parsed) if parsed is not None else None
|
||
|
||
|
||
def _normalize_billing_period_word(raw: Optional[str]) -> str:
|
||
"""
|
||
WooCommerce REST liefert oft 'month'/'day'/…
|
||
WP Desk Flexible Subscriptions API nutzt Einzelbuchstaben: D, W, M, Y.
|
||
"""
|
||
if raw is None or (isinstance(raw, str) and not str(raw).strip()):
|
||
return "month"
|
||
s = str(raw).strip().lower()
|
||
if len(s) == 1:
|
||
return {"d": "day", "w": "week", "m": "month", "y": "year"}.get(s, s)
|
||
return s
|
||
|
||
|
||
def resolve_wc_period_fields(
|
||
*,
|
||
current_period_start: Optional[int] = None,
|
||
current_period_end: Optional[int] = None,
|
||
next_payment_date: Any = None,
|
||
billing_period: Optional[str] = None,
|
||
billing_interval: Optional[int] = None,
|
||
) -> Tuple[Optional[int], Optional[int]]:
|
||
"""
|
||
Liefert (period_start, period_end) oder (None, None) wenn nicht ableitbar/ungültig.
|
||
Priorität:
|
||
1) Beide current_period_* gesetzt und start < end
|
||
2) next_payment_date + billing_period/billing_interval → Ende = Zahlungszeitpunkt,
|
||
Start = Ende − Abrechnungsintervall (wie Woo-REST-Logik)
|
||
"""
|
||
cs = _normalize_int_ts(current_period_start)
|
||
ce = _normalize_int_ts(current_period_end)
|
||
|
||
if cs is not None and ce is not None:
|
||
if cs < ce:
|
||
return cs, ce
|
||
return None, None
|
||
|
||
# Nur period_end gesetzt: Start aus Abrechnungsintervall rückwärts (Default month/1)
|
||
if ce is not None and cs is None:
|
||
bp = _normalize_billing_period_word(billing_period)
|
||
try:
|
||
bi = int(billing_interval) if billing_interval is not None else 1
|
||
except (TypeError, ValueError):
|
||
bi = 1
|
||
if bi < 1:
|
||
bi = 1
|
||
ps = _subtract_billing_period(ce, bp, bi)
|
||
if ps >= ce:
|
||
ps = ce - 86400
|
||
if ps < ce and _TS_MIN <= ps <= _TS_MAX:
|
||
return ps, ce
|
||
return None, None
|
||
|
||
pe = coerce_next_payment_unix(next_payment_date)
|
||
if pe is not None and cs is None and ce is None:
|
||
bp = _normalize_billing_period_word(billing_period)
|
||
try:
|
||
bi = int(billing_interval) if billing_interval is not None else 1
|
||
except (TypeError, ValueError):
|
||
bi = 1
|
||
if bi < 1:
|
||
bi = 1
|
||
ps = _subtract_billing_period(pe, bp, bi)
|
||
if ps >= pe:
|
||
ps = pe - 86400
|
||
if ps < pe and _TS_MIN <= ps <= _TS_MAX:
|
||
return ps, pe
|
||
return None, None
|
||
|
||
# Nur eines der Felder – nicht raten
|
||
return None, None
|