romfast/gomag-vending
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: dual pricing policies + discount VAT splitting

Browse Source 
Add production pricing policy (id_pol_productie) for articles with cont 341/345,
smart discount VAT splitting across multiple rates, per-article id_pol support,
and mapped SKU price validation. Settings UI updated with new controls.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Agent
2026-03-18 15:10:05 +00:00
parent 69a3088579
commit 695dafacd5
11 changed files with 428 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

172
api/app/services/validation_service.py
View File

@@ -28,10 +28,10 @@ def check_orders_in_roa(min_date, conn) -> dict:
return existing
def resolve_codmat_ids(codmats: set[str], id_gestiune: int = None, conn=None) -> dict[str, int]:
"""Resolve CODMATs to best id_articol: prefers article with stock, then MAX(id_articol).
def resolve_codmat_ids(codmats: set[str], id_gestiune: int = None, conn=None) -> dict[str, dict]:
"""Resolve CODMATs to best id_articol + cont: prefers article with stock, then MAX(id_articol).
Filters: sters=0 AND inactiv=0.
Returns: {codmat: id_articol}
Returns: {codmat: {"id_articol": int, "cont": str|None}}
"""
if not codmats:
return {}
@@ -58,8 +58,8 @@ def resolve_codmat_ids(codmats: set[str], id_gestiune: int = None, conn=None) ->
params["id_gestiune"] = id_gestiune
cur.execute(f"""
SELECT codmat, id_articol FROM (
SELECT na.codmat, na.id_articol,
SELECT codmat, id_articol, cont FROM (
SELECT na.codmat, na.id_articol, na.cont,
ROW_NUMBER() OVER (
PARTITION BY na.codmat
ORDER BY
@@ -79,7 +79,7 @@ def resolve_codmat_ids(codmats: set[str], id_gestiune: int = None, conn=None) ->
) WHERE rn = 1
""", params)
for row in cur:
result[row[0]] = row[1]
result[row[0]] = {"id_articol": row[1], "cont": row[2]}
finally:
if own_conn:
database.pool.release(conn)
@@ -90,11 +90,11 @@ def resolve_codmat_ids(codmats: set[str], id_gestiune: int = None, conn=None) ->
def validate_skus(skus: set[str], conn=None, id_gestiune: int = None) -> dict:
"""Validate a set of SKUs against Oracle.
Returns: {mapped: set, direct: set, missing: set, direct_id_map: {codmat: id_articol}}
Returns: {mapped: set, direct: set, missing: set, direct_id_map: {codmat: {"id_articol": int, "cont": str|None}}}
- mapped: found in ARTICOLE_TERTI (active)
- direct: found in NOM_ARTICOLE by codmat (not in ARTICOLE_TERTI)
- missing: not found anywhere
- direct_id_map: {codmat: id_articol} for direct SKUs (saves a round-trip in validate_prices)
- direct_id_map: {codmat: {"id_articol": int, "cont": str|None}} for direct SKUs
"""
if not skus:
return {"mapped": set(), "direct": set(), "missing": set(), "direct_id_map": {}}
@@ -129,6 +129,7 @@ def validate_skus(skus: set[str], conn=None, id_gestiune: int = None) -> dict:
else:
direct_id_map = {}
direct = set()
finally:
if own_conn:
database.pool.release(conn)
@@ -136,7 +137,8 @@ def validate_skus(skus: set[str], conn=None, id_gestiune: int = None) -> dict:
missing = skus - mapped - direct
logger.info(f"SKU validation: {len(mapped)} mapped, {len(direct)} direct, {len(missing)} missing")
return {"mapped": mapped, "direct": direct, "missing": missing, "direct_id_map": direct_id_map}
return {"mapped": mapped, "direct": direct, "missing": missing,
"direct_id_map": direct_id_map}
def classify_orders(orders, validation_result):
"""Classify orders as importable or skipped based on SKU validation.
@@ -158,6 +160,19 @@ def classify_orders(orders, validation_result):
return importable, skipped
def _extract_id_map(direct_id_map: dict) -> dict:
"""Extract {codmat: id_articol} from either enriched or simple format."""
if not direct_id_map:
return {}
result = {}
for cm, val in direct_id_map.items():
if isinstance(val, dict):
result[cm] = val["id_articol"]
else:
result[cm] = val
return result
def validate_prices(codmats: set[str], id_pol: int, conn=None, direct_id_map: dict=None) -> dict:
"""Check which CODMATs have a price entry in CRM_POLITICI_PRET_ART for the given policy.
If direct_id_map is provided, skips the NOM_ARTICOLE lookup for those CODMATs.
@@ -166,7 +181,7 @@ def validate_prices(codmats: set[str], id_pol: int, conn=None, direct_id_map: di
if not codmats:
return {"has_price": set(), "missing_price": set()}
codmat_to_id = dict(direct_id_map) if direct_id_map else {}
codmat_to_id = _extract_id_map(direct_id_map)
ids_with_price = set()
own_conn = conn is None
@@ -199,14 +214,18 @@ def validate_prices(codmats: set[str], id_pol: int, conn=None, direct_id_map: di
logger.info(f"Price validation (policy {id_pol}): {len(has_price)} have price, {len(missing_price)} missing price")
return {"has_price": has_price, "missing_price": missing_price}
def ensure_prices(codmats: set[str], id_pol: int, conn=None, direct_id_map: dict=None):
def ensure_prices(codmats: set[str], id_pol: int, conn=None, direct_id_map: dict=None,
cota_tva: float = None):
"""Insert price 0 entries for CODMATs missing from the given price policy.
Uses batch executemany instead of individual INSERTs.
Relies on TRG_CRM_POLITICI_PRET_ART trigger for ID_POL_ART sequence.
cota_tva: VAT rate from settings (e.g. 21) — used for PROC_TVAV metadata.
"""
if not codmats:
return
proc_tvav = 1 + (cota_tva / 100) if cota_tva else 1.21
own_conn = conn is None
if own_conn:
conn = database.get_oracle_connection()
@@ -224,7 +243,7 @@ def ensure_prices(codmats: set[str], id_pol: int, conn=None, direct_id_map: dict
# Build batch params using direct_id_map (already resolved via resolve_codmat_ids)
batch_params = []
codmat_id_map = dict(direct_id_map) if direct_id_map else {}
codmat_id_map = _extract_id_map(direct_id_map)
for codmat in codmats:
id_articol = codmat_id_map.get(codmat)
@@ -234,7 +253,8 @@ def ensure_prices(codmats: set[str], id_pol: int, conn=None, direct_id_map: dict
batch_params.append({
"id_pol": id_pol,
"id_articol": id_articol,
"id_valuta": id_valuta
"id_valuta": id_valuta,
"proc_tvav": proc_tvav
})
if batch_params:
@@ -244,9 +264,9 @@ def ensure_prices(codmats: set[str], id_pol: int, conn=None, direct_id_map: dict
ID_UTIL, DATAORA, PROC_TVAV, PRETFTVA, PRETCTVA)
VALUES
(:id_pol, :id_articol, 0, :id_valuta,
-3, SYSDATE, 1.19, 0, 0)
-3, SYSDATE, :proc_tvav, 0, 0)
""", batch_params)
logger.info(f"Batch inserted {len(batch_params)} price entries for policy {id_pol}")
logger.info(f"Batch inserted {len(batch_params)} price entries for policy {id_pol} (PROC_TVAV={proc_tvav})")
conn.commit()
finally:
@@ -254,3 +274,125 @@ def ensure_prices(codmats: set[str], id_pol: int, conn=None, direct_id_map: dict
database.pool.release(conn)
logger.info(f"Ensure prices done: {len(codmats)} CODMATs processed for policy {id_pol}")
def validate_and_ensure_prices_dual(codmats: set[str], id_pol_vanzare: int,
id_pol_productie: int, conn, direct_id_map: dict,
cota_tva: float = 21) -> dict[str, int]:
"""Dual-policy price validation: assign each CODMAT to sales or production policy.
Logic:
1. Check both policies in one SQL
2. If article in one policy → use that
3. If article in BOTH → prefer id_pol_vanzare
4. If article in NEITHER → check cont: 341/345 → production, else → sales; insert price 0
Returns: codmat_policy_map = {codmat: assigned_id_pol}
"""
if not codmats:
return {}
codmat_policy_map = {}
id_map = _extract_id_map(direct_id_map)
# Collect all id_articol values we need to check
id_to_codmats = {} # {id_articol: [codmat, ...]}
for cm in codmats:
aid = id_map.get(cm)
if aid:
id_to_codmats.setdefault(aid, []).append(cm)
if not id_to_codmats:
return {}
# Query both policies in one SQL
existing = {} # {id_articol: set of id_pol}
id_list = list(id_to_codmats.keys())
with conn.cursor() as cur:
for i in range(0, len(id_list), 500):
batch = id_list[i:i+500]
placeholders = ",".join([f":a{j}" for j in range(len(batch))])
params = {f"a{j}": aid for j, aid in enumerate(batch)}
params["id_pol_v"] = id_pol_vanzare
params["id_pol_p"] = id_pol_productie
cur.execute(f"""
SELECT pa.ID_ARTICOL, pa.ID_POL FROM CRM_POLITICI_PRET_ART pa
WHERE pa.ID_POL IN (:id_pol_v, :id_pol_p) AND pa.ID_ARTICOL IN ({placeholders})
""", params)
for row in cur:
existing.setdefault(row[0], set()).add(row[1])
# Classify each codmat
missing_vanzare = set() # CODMATs needing price 0 in sales policy
missing_productie = set() # CODMATs needing price 0 in production policy
for aid, cms in id_to_codmats.items():
pols = existing.get(aid, set())
for cm in cms:
if pols:
if id_pol_vanzare in pols:
codmat_policy_map[cm] = id_pol_vanzare
elif id_pol_productie in pols:
codmat_policy_map[cm] = id_pol_productie
else:
# Not in any policy — classify by cont
info = direct_id_map.get(cm, {})
cont = info.get("cont", "") if isinstance(info, dict) else ""
cont_str = str(cont or "").strip()
if cont_str in ("341", "345"):
codmat_policy_map[cm] = id_pol_productie
missing_productie.add(cm)
else:
codmat_policy_map[cm] = id_pol_vanzare
missing_vanzare.add(cm)
# Ensure prices for missing articles in each policy
if missing_vanzare:
ensure_prices(missing_vanzare, id_pol_vanzare, conn, direct_id_map, cota_tva=cota_tva)
if missing_productie:
ensure_prices(missing_productie, id_pol_productie, conn, direct_id_map, cota_tva=cota_tva)
logger.info(
f"Dual-policy: {len(codmat_policy_map)} CODMATs assigned "
f"(vanzare={sum(1 for v in codmat_policy_map.values() if v == id_pol_vanzare)}, "
f"productie={sum(1 for v in codmat_policy_map.values() if v == id_pol_productie)})"
)
return codmat_policy_map
def resolve_mapped_codmats(mapped_skus: set[str], conn) -> dict[str, list[dict]]:
"""For mapped SKUs, get their underlying CODMATs from ARTICOLE_TERTI + nom_articole.
Returns: {sku: [{"codmat": str, "id_articol": int, "cont": str|None}]}
"""
if not mapped_skus:
return {}
result = {}
sku_list = list(mapped_skus)
with conn.cursor() as cur:
for i in range(0, len(sku_list), 500):
batch = sku_list[i:i+500]
placeholders = ",".join([f":s{j}" for j in range(len(batch))])
params = {f"s{j}": sku for j, sku in enumerate(batch)}
cur.execute(f"""
SELECT at.sku, at.codmat, na.id_articol, na.cont
FROM ARTICOLE_TERTI at
JOIN NOM_ARTICOLE na ON na.codmat = at.codmat AND na.sters = 0 AND na.inactiv = 0
WHERE at.sku IN ({placeholders}) AND at.activ = 1 AND at.sters = 0
""", params)
for row in cur:
sku = row[0]
if sku not in result:
result[sku] = []
result[sku].append({
"codmat": row[1],
"id_articol": row[2],
"cont": row[3]
})
logger.info(f"resolve_mapped_codmats: {len(result)} SKUs → {sum(len(v) for v in result.values())} CODMATs")
return result