refactor(myopencre): move CSV validation to parser and detect noop imports

Author: PRAteek-singHWY

application/utils/spreadsheet_parsers.py

@@ -141,8 +141,8 @@ def validate_export_csv_rows(rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]
 
     headers = list(rows[0].keys())
 
-    if not headers or len(headers) < 2:
-        raise ValueError("Invalid CSV format or missing header row")
+    if not headers:
+        raise ValueError("CSV header row is missing")
 
     if not any(h.startswith("CRE") for h in headers):
         raise ValueError("At least one CRE column is required")

application/web/web_main.py

@@ -844,136 +844,7 @@ def import_from_cre_csv() -> Any:
     rows = list(csv.DictReader(decoded_contents.splitlines()))
 
     # ------------------------
-    # Schema / header validation
-    # ------------------------
-
-    headers = [h.strip() for h in csv_read.fieldnames]
-
-    if not headers:
-        return (
-            jsonify(
-                {
-                    "success": False,
-                    "type": "SCHEMA_ERROR",
-                    "message": "CSV header row is missing",
-                }
-            ),
-            400,
-        )
-
-    has_cre_column = any(h.startswith("CRE") for h in headers)
-    if not has_cre_column:
-        return (
-            jsonify(
-                {
-                    "success": False,
-                    "type": "SCHEMA_ERROR",
-                    "message": "At least one CRE column is required",
-                }
-            ),
-            400,
-        )
-
-    required_columns = ["standard|name", "standard|id"]
-    for col in required_columns:
-        if col not in headers:
-            return (
-                jsonify(
-                    {
-                        "success": False,
-                        "type": "SCHEMA_ERROR",
-                        "message": f"Missing required column: {col}",
-                    }
-                ),
-                400,
-            )
-
-    # ------------------------
-    # Row-level validation (export-compatible)
-    # ------------------------
-
-    rows = list(csv_read)
-    errors = []
-
-    # 🚨 NEW: guard against misaligned rows (extra columns)
-    for row_index, row in enumerate(rows, start=2):
-        if None in row:
-            return (
-                jsonify(
-                    {
-                        "success": False,
-                        "type": "SCHEMA_ERROR",
-                        "message": (
-                            f"Row {row_index} has more columns than header. "
-                            "Please ensure the CSV matches the exported template."
-                        ),
-                    }
-                ),
-                400,
-            )
-
-    for row_index, row in enumerate(rows, start=2):  # header is row 1
-        normalized_row = {
-            k: (v.strip() if isinstance(v, str) else v) for k, v in row.items()
-        }
-
-        # Skip completely empty rows (exported templates contain them)
-        if all(not v for v in normalized_row.values()):
-            continue
-
-        cre_values = [normalized_row.get(h) for h in headers if h.startswith("CRE")]
-        cre_values = [v for v in cre_values if v]
-
-        # Rows without CRE are allowed by export format → skip
-        if not cre_values:
-            continue
-
-        # Validate CRE format
-        for cre in cre_values:
-            if "|" not in cre:
-                errors.append(
-                    {
-                        "row": row_index,
-                        "code": "INVALID_CRE_FORMAT",
-                        "message": (
-                            f"Invalid CRE entry '{cre}', expected '<CRE-ID>|<Name>'"
-                        ),
-                    }
-                )
-
-    if errors:
-        return (
-            jsonify(
-                {
-                    "success": False,
-                    "type": "ROW_VALIDATION_ERROR",
-                    "errors": errors,
-                }
-            ),
-            400,
-        )
-
-    # ------------------------
-    # No-op import guard (IMPORTANT)
-    # ------------------------
-
-    importable_rows = []
-    for row in rows:
-        if any(v for v in row.values()):
-            importable_rows.append(row)
-
-    if not importable_rows:
-        return jsonify(
-            {
-                "status": "success",
-                "new_cres": [],
-                "new_standards": 0,
-                "import_type": "empty",
-            }
-        )
-
-    # ------------------------
-    # Import execution
+    # Delegate validation + parsing
     # ------------------------
 
     try: