| import pandas as pd | |
| import numpy as np | |
| from config import WEIGHTS | |
| REQUIRED_COLS = [ | |
| "Name of Product", | |
| "Components Used", | |
| "Quantity of Each Component", | |
| "Oldest Product Required First", | |
| "Priority Assigned", | |
| ] | |
| def _normalize_columns(df: pd.DataFrame) -> pd.DataFrame: | |
| df = df.copy() | |
| df.columns = [str(c).strip() for c in df.columns] | |
| return df | |
| def _component_count(series: pd.Series) -> pd.Series: | |
| def _count(x): | |
| if pd.isna(x): | |
| return 0 | |
| parts = [p.strip() for p in str(x).split(",") if str(p).strip()] | |
| return len(set(parts)) if parts else 0 | |
| return series.apply(_count) | |
| def compute_priority(df: pd.DataFrame, min_qty: int = 50) -> pd.DataFrame: | |
| df = _normalize_columns(df) | |
| missing = [c for c in REQUIRED_COLS if c not in df.columns] | |
| if missing: | |
| raise ValueError(f"Missing required columns: {missing}") | |
| out = df.copy() | |
| out["Oldest Product Required First"] = pd.to_datetime(out["Oldest Product Required First"], errors="coerce") | |
| today = pd.Timestamp.now().normalize() | |
| out["OrderAgeDays"] = (today - pd.to_datetime(out["Oldest Product Required First"], errors="coerce")).dt.days | |
| out["OrderAgeDays"] = out["OrderAgeDays"].fillna(0).clip(lower=0) | |
| if out["OrderAgeDays"].max() > 0: | |
| age_score_01 = out["OrderAgeDays"] / out["OrderAgeDays"].max() | |
| else: | |
| age_score_01 = 0 | |
| qty = pd.to_numeric(out["Quantity of Each Component"], errors="coerce").fillna(0) | |
| age_score_01 = np.where(qty >= min_qty, age_score_01, 0) | |
| comp_count = _component_count(out["Components Used"]) | |
| if comp_count.max() > 0: | |
| comp_simplicity_01 = 1 - (comp_count / comp_count.max()) | |
| else: | |
| comp_simplicity_01 = 0 | |
| def manual_to01(x): | |
| if pd.isna(x): | |
| return 0.0 | |
| s = str(x).strip().lower() | |
| if s in {"high", "urgent", "yes", "y", "true"}: | |
| return 1.0 | |
| try: | |
| return 1.0 if float(s) > 0 else 0.0 | |
| except: | |
| return 0.0 | |
| manual_01 = out["Priority Assigned"].apply(manual_to01) | |
| w_age = WEIGHTS["AGE_WEIGHT"] / 100.0 | |
| w_comp = WEIGHTS["COMPONENT_WEIGHT"] / 100.0 | |
| w_man = WEIGHTS["MANUAL_WEIGHT"] / 100.0 | |
| out["AgeScore"] = age_score_01 * w_age | |
| if isinstance(comp_simplicity_01, pd.Series): | |
| out["SimplicityScore"] = comp_simplicity_01 * w_comp | |
| else: | |
| out["SimplicityScore"] = comp_simplicity_01 | |
| out["ManualScore"] = manual_01 * w_man | |
| out["PriorityScore"] = out["AgeScore"] + out["SimplicityScore"] + out["ManualScore"] | |
| out["ComponentCount"] = comp_count | |
| out["QtyThresholdOK"] = qty >= min_qty | |
| out = out.sort_values(["PriorityScore", "OrderAgeDays"], ascending=[False, False]) | |
| return out |