from __future__ import annotations from fastapi import FastAPI, UploadFile, File, HTTPException from fastapi.responses import StreamingResponse import math import pandas as pd from typing import Dict, List, Optional, Tuple import io import os import zipfile import re from dotenv import load_dotenv from fastapi import FastAPI, UploadFile, File, HTTPException, Request, status, Response from starlette.responses import JSONResponse from starlette.middleware.base import BaseHTTPMiddleware from datetime import datetime load_dotenv() #router = APIRouter(prefix="/api/plytix", tags=["Plytix"]) # FastAPI application setup app = FastAPI( openapi_url=None # Completely disable OpenAPI schema ) # ---------------------------- # Normalization helpers # ---------------------------- def _norm_col(col: str) -> str: """ Normalize column names from Plytix: - trim, lowercase - replace spaces and slashes with underscore - collapse repeated underscores """ col = (col or "").strip().replace("\ufeff", "").lower() col = re.sub(r"[\/\s]+", "_", col) # "Variant of" -> "variant_of" col = re.sub(r"[^a-z0-9_]+", "_", col) # remove punctuation col = re.sub(r"_+", "_", col).strip("_") return col def canonicalize_plytix_columns(df: pd.DataFrame) -> pd.DataFrame: """ Renames incoming columns to canonical snake_case and applies alias mapping. Also handles duplicate columns (e.g. Brand + brand). """ # Step 1: normalize all incoming headers normed = [_norm_col(c) for c in df.columns] # Step 2: if duplicate normalized names exist, keep the first and drop the rest # (Pandas can keep duplicate column names; that becomes painful later.) seen = set() keep_idx = [] for i, c in enumerate(normed): if c not in seen: keep_idx.append(i) seen.add(c) df = df.iloc[:, keep_idx] normed = [normed[i] for i in keep_idx] df.columns = normed # Step 3: alias mapping (map Plytix-style names to what your code expects) aliases = { "sku": ["sku"], # "SKU" -> sku after normalization "name": ["name"], "price": ["price"], "categories": ["categories"], "product_online": ["product_online"], "status": ["status"], # Variants "variant_of": ["variant_of", "variantof"], "variant_list": ["variant_list", "variants"], # Dimensions "width_in": ["width_in", "width"], "height_in": ["height_in", "height"], "depth_in": ["depth_in", "depth"], # Text "short_description": ["short_description"], "tax_class_name": ["tax_class_name"], # Media "base_image": ["base_image", "base_image_url", "base_imageurl", "base_image_"], } # Inject canonical columns from any alias that exists for canon, keys in aliases.items(): if canon in df.columns: continue for k in keys: if k in df.columns: df[canon] = df[k] break return df def normalize_num(v): if v is None or (isinstance(v, float) and pd.isna(v)): return "" s = str(v).strip() if not s: return "" try: x = float(s) if x.is_integer(): return str(int(x)) return str(x).rstrip("0").rstrip(".") except Exception: import re m = re.search(r"(\d+(\.\d+)?)", s) return m.group(1) if m else "" def normalize_categories(v): if v is None or (isinstance(v, float) and pd.isna(v)): return "" s = str(v).strip() if not s: return "" parts = [p.strip() for p in s.split(",") if p.strip()] return ",".join(p.replace(">", "/") for p in parts) def status_to_magento(v): s = str(v).strip().lower() return 1 if s in ("1", "enabled", "true", "yes") else 2 def parse_variant_list(v): if v is None or (isinstance(v, float) and pd.isna(v)): return [] import re return [x.strip() for x in re.split(r"[,\|;]+", str(v)) if x.strip()] # ---------------------------- # Core normalization logic # ---------------------------- M2_KEEP_COLS: List[str] = [ "sku", "attribute_set_code", "product_type", "categories", "product_websites", "name", "description", "short_description", "weight", "product_online", "tax_class_name", "visibility", "price", "base_image", "base_image_label", "small_image", "thumbnail_image", "created_at", "updated_at", "display_product_options_in", "gift_message_available", "msrp_display_actual_price_type", "use_config_min_qty", "is_qty_decimal", "allow_backorders", "use_config_backorders", "min_cart_qty", "use_config_min_sale_qty", "max_cart_qty", "use_config_max_sale_qty", "notify_on_stock_below", "use_config_notify_stock_qty", "manage_stock", "use_config_manage_stock", "use_config_qty_increments", "qty_increments", "use_config_enable_qty_inc", "enable_qty_increments", "is_decimal_divided", "website_id", "configurable_variations", # seed in -> magento-ready out "configurable_variation_labels", # used to infer axis codes "additional_attributes", ] DEFAULT_HARDCODED_ADDITIONAL: Dict[str, str] = { # put your defaults here (examples) "am_recurring_enable": "No", "am_subscription_only": "No", "dc_sample_allow": "0", } def _is_empty(v: object) -> bool: if v is None: return True try: if isinstance(v, float) and math.isnan(v): return True except Exception: pass s = str(v).strip() return s == "" or s.lower() in {"nan", "none", "null"} def _escape_value(v: str) -> str: v = v.replace("\\", "\\\\") v = v.replace(",", "\\,") v = v.replace("=", "\\=") return v def _case_insensitive_col(df: pd.DataFrame, wanted: str) -> Optional[str]: w = wanted.strip().lower() for c in df.columns: if str(c).strip().lower() == w: return c return None def _parse_kv_blob(blob: object) -> Dict[str, str]: if _is_empty(blob): return {} s = str(blob) # split on unescaped commas parts: List[str] = [] cur: List[str] = [] esc = False for ch in s: if esc: cur.append(ch) esc = False elif ch == "\\": esc = True elif ch == ",": parts.append("".join(cur)) cur = [] else: cur.append(ch) parts.append("".join(cur)) out: Dict[str, str] = {} for p in parts: if not p: continue # split on first unescaped '=' key: List[str] = [] val: List[str] = [] esc = False saw_eq = False for ch in p: if esc: (val if saw_eq else key).append(ch) esc = False elif ch == "\\": esc = True elif ch == "=" and not saw_eq: saw_eq = True else: (val if saw_eq else key).append(ch) k = "".join(key).strip() v = "".join(val).strip() if k: out[k] = v return out def _format_kv_blob(d: Dict[str, str]) -> str: pairs: List[str] = [] for k, v in d.items(): if _is_empty(v): continue pairs.append(f"{k}={_escape_value(str(v).strip())}") return ",".join(pairs) def _extract_axis_codes_from_labels(labels: object) -> List[str]: """ Accept: 'width=Width (inches)' 'width=Width (inches),height=Height (inches)' Returns: ['width'] or ['width','height'] """ if _is_empty(labels): return [] s = str(labels).strip() axes: List[str] = [] for seg in [x.strip() for x in s.split(",") if x.strip()]: if "=" not in seg: continue axis = seg.split("=", 1)[0].strip() if axis: axes.append(axis) # de-dupe but preserve order seen = set() out = [] for a in axes: if a not in seen: out.append(a) seen.add(a) return out def _extract_child_skus_from_seed(seed: object) -> List[str]: """ Seed format: comma-separated SKUs (confirmed by you) 'SKU1,SKU2,SKU3' """ if _is_empty(seed): return [] return [x.strip() for x in str(seed).split(",") if x.strip()] def _get_child_value(df: pd.DataFrame, child_sku: str, axis: str) -> Optional[str]: """ Try: - axis column - axis_in column (common in your data: width_in, height_in) """ sku_col = _case_insensitive_col(df, "sku") or "sku" child = df[df[sku_col].astype(str) == str(child_sku)] if child.empty: return None axis_col = _case_insensitive_col(df, axis) if axis_col: v = child.iloc[0].get(axis_col, "") if not _is_empty(v): return str(v).strip() axis_in_col = _case_insensitive_col(df, f"{axis}_in") if axis_in_col: v = child.iloc[0].get(axis_in_col, "") if not _is_empty(v): return str(v).strip() return None def normalize_plytix_df( df: pd.DataFrame, *, keep_cols: List[str] = M2_KEEP_COLS, seed_children_col: str = "configurable_variations", labels_col: str = "configurable_variation_axis", additional_col: str = "additional_attributes", also_pack_cols: Optional[List[str]] = None, hardcoded_additional: Optional[Dict[str, str]] = None, drop_before_pack: Optional[List[str]] = None, drop_after_compute: Optional[List[str]] = None, ) -> Tuple[pd.DataFrame, pd.DataFrame]: """ Returns (normalized_df, quarantine_df). - seed_children_col contains a comma-separated list of child SKUs (seed input) - We overwrite configurable_variations with Magento-ready: 'sku=CHILD,axis=val|...' - additional_attributes merges: existing additional + extra columns + also_pack + hardcoded - quarantine_df contains rows that cannot be normalized correctly (with a reason) """ also_pack_cols = also_pack_cols or [] hardcoded_additional = hardcoded_additional or DEFAULT_HARDCODED_ADDITIONAL out = df.copy() # --- NEW: drop columns early (they will not be packed, not used, not output) --- drop_before_pack = drop_before_pack or [] before_actual = [] for c in drop_before_pack: actual = _case_insensitive_col(out, c) or c if actual in out.columns: before_actual.append(actual) if before_actual: out = out.drop(columns=before_actual) # Ensure keep columns exist for c in keep_cols: if c not in out.columns: out[c] = "" sku_col = _case_insensitive_col(out, "sku") or "sku" seed_col = _case_insensitive_col(out, seed_children_col) or seed_children_col lbl_col = _case_insensitive_col(out, labels_col) or labels_col add_col = _case_insensitive_col(out, additional_col) or additional_col # identify extra columns keep_set = {c.lower() for c in keep_cols} extra_cols = [c for c in out.columns if str(c).strip().lower() not in keep_set] quarantine_rows: List[Dict[str, object]] = [] # PRESERVE seed values before overwriting seed_series = out[seed_col].copy() if seed_col in out.columns else pd.Series("", index=out.index) # rebuild configurable_variations out["configurable_variations"] = "" for idx, row in out.iterrows(): # parents are the only ones that should carry seed list; but we won't assume product_type is perfect seed_children = _extract_child_skus_from_seed(seed_series.loc[idx]) if not seed_children: continue # nothing to build axes = _extract_axis_codes_from_labels(row.get(lbl_col, "")) if not axes: quarantine_rows.append({ "row_index": idx, "sku": row.get(sku_col, ""), "reason": "Missing configurable_variation_labels (cannot infer axis codes)", "seed_children": row.get(seed_col, ""), }) continue parts: List[str] = [] missing: List[str] = [] for child_sku in seed_children: axis_pairs: List[str] = [] for axis in axes: val = _get_child_value(out, child_sku, axis) if _is_empty(val): missing.append(f"{child_sku}:{axis}") continue axis_pairs.append(f"{axis}={val}") # If any axis missing for this child, skip it to avoid malformed variation record if len(axis_pairs) != len(axes): continue parts.append(f"sku={child_sku}," + ",".join(axis_pairs)) if not parts: quarantine_rows.append({ "row_index": idx, "sku": row.get(sku_col, ""), "reason": "Unable to build configurable_variations (no child had complete axis values)", "seed_children": row.get(seed_col, ""), "missing_axis_values": ";".join(missing)[:5000], }) continue out.at[idx, "configurable_variations"] = "|".join(parts) # build merged additional_attributes additional_out: List[str] = [] for _, row in out.iterrows(): merged: Dict[str, str] = {} # existing additional_attributes merged.update(_parse_kv_blob(row.get(add_col, ""))) # extra cols for c in extra_cols: v = row.get(c, "") if _is_empty(v): continue merged[str(c).strip()] = str(v).strip() # also pack selected keep-cols into additional_attributes for c in also_pack_cols: actual = _case_insensitive_col(out, c) or c if actual not in out.columns: continue v = row.get(actual, "") if _is_empty(v): continue merged[str(c).strip()] = str(v).strip() # hardcoded k/v for k, v in hardcoded_additional.items(): if _is_empty(v): continue merged[str(k).strip()] = str(v).strip() additional_out.append(_format_kv_blob(merged)) out["additional_attributes"] = additional_out out["attribute_set_code"] = out["attribute_set_code"].fillna("").astype(str).str.strip() out["attribute_set_code"] = out["attribute_set_code"].replace("", "Default") #normalized = out[keep_cols].copy() quarantine_df = pd.DataFrame(quarantine_rows) if quarantine_rows else pd.DataFrame(columns=["row_index", "sku", "reason", "seed_children", "missing_axis_values"]) # --- NEW: drop columns at the end (can be used for compute but not exported) --- drop_after_compute = drop_after_compute or [] after_actual = [] for c in drop_after_compute: actual = _case_insensitive_col(out, c) or c if actual in out.columns and actual not in keep_cols: after_actual.append(actual) if after_actual: out = out.drop(columns=after_actual) normalized = out[keep_cols].copy() return normalized, quarantine_df # ---------------------------- # FastAPI endpoint # ---------------------------- allowed_origins = [ "https://plytixdash.dev.piesol.com", "http://dev.piesol.com", "http://localhost:3000" ] # Function to check if a domain matches the wildcard subdomain pattern def is_amplify_preview_url(origin: str) -> bool: # Regex pattern for matching any subdomain of amplifyapp.com #amplify_preview_pattern = r"^https://([a-zA-Z0-9\-]+)\.([a-zA-Z0-9\-]+)\.amplifyapp\.com$" #return bool(re.match(amplify_preview_pattern, origin)) return origin.endswith(".amplifyapp.com") # Combine static and dynamic origins def get_allowed_origins(): # You can also dynamically fetch preview URLs if needed (e.g., from environment variables or API) # For now, we're assuming you've manually defined them in `allowed_origins` above. dynamic_origins = [url for url in os.getenv(" ", "").split(",") if is_amplify_preview_url(url)] return allowed_origins + dynamic_origins class DynamicCORSMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): origin = request.headers.get("origin") request_headers = request.headers.get("access-control-request-headers", "") if origin and (origin in get_allowed_origins() or is_amplify_preview_url(origin)): # Preflight request → return immediately if request.method == "OPTIONS": return Response( status_code=204, headers={ "Access-Control-Allow-Origin": origin, "Access-Control-Allow-Credentials": "true", "Access-Control-Allow-Methods": "GET, POST, PUT, PATCH, DELETE, OPTIONS", "Access-Control-Allow-Headers": request_headers or "Content-Type, Authorization", }, ) # Normal request → add CORS headers after processing response: Response = await call_next(request) response.headers["Access-Control-Allow-Origin"] = origin response.headers["Access-Control-Allow-Credentials"] = "true" response.headers["Access-Control-Allow-Methods"] = "GET, POST, PUT, PATCH, DELETE, OPTIONS" response.headers["Access-Control-Allow-Headers"] = request_headers or "Content-Type, Authorization" return response # Origin not allowed if request.method == "OPTIONS": return JSONResponse({"detail": "CORS origin not allowed"}, status_code=403) return await call_next(request) app.add_middleware(DynamicCORSMiddleware) @app.get("/") async def root(): return {"message": "Pie PLANS Server"} @app.get("/api/") async def root(): return {"message": "Pie PLANS API Server"} @app.post("/api/plytix/normalize-magento") async def normalize_magento_csv(file: UploadFile = File(...)): if not file.filename.lower().endswith(".csv"): raise HTTPException(status_code=400, detail="Only CSV files are supported") content = await file.read() try: df = pd.read_csv( io.BytesIO(content), dtype=str, # critical: prevents NaN conversions keep_default_na=False, # keep blanks as '' ) except Exception as e: raise HTTPException(status_code=400, detail=f"Failed to read CSV: {e}") df = canonicalize_plytix_columns(df) try: normalized_df, quarantine_df = normalize_plytix_df(df,drop_after_compute=["configurable_variation_axis", "price"]) except Exception as e: raise HTTPException(status_code=400, detail=str(e)) magento_buf = io.StringIO() quarantine_buf = io.StringIO() normalized_df.to_csv(magento_buf, index=False) quarantine_df.to_csv(quarantine_buf, index=False) # optional: stats/debug stats = { "input_rows": len(df), "normalized_rows": len(normalized_df), "quarantine_rows": len(quarantine_df), } stats_buf = io.StringIO() stats_buf.write("\n".join([f"{k}={v}" for k, v in stats.items()])) zip_buf = io.BytesIO() with zipfile.ZipFile(zip_buf, "w", zipfile.ZIP_DEFLATED) as z: z.writestr("magento_products.csv", magento_buf.getvalue()) z.writestr("quarantine.csv", quarantine_buf.getvalue()) z.writestr("stats.txt", stats_buf.getvalue()) zip_buf.seek(0) timestamp = datetime.now().strftime("%Y%m%d%H%M%S") return StreamingResponse( zip_buf, media_type="application/zip", headers={"Content-Disposition": f"attachment; filename=magento_normalized_{timestamp}.zip"}, ) if __name__ == "__main__": import uvicorn #Cast as int Port = int(os.getenv('PORT', 3983)) uvicorn.run("main:app", host = "127.0.0.1", port = Port, reload = True)