""" import_utils.py Utility module for importing and validating LeafDoctor AI datasets. Responsibilities ---------------- 1. Read CSV files 2. Clean data 3. Validate data 4. Import into PostgreSQL 5. Generate crop-label mapping 6. Generate import report Author: LeafDoctor AI """ from __future__ import annotations import json from dataclasses import dataclass, field from pathlib import Path from typing import Dict, List, Optional, Set import pandas as pd from sqlalchemy.orm import Session from config import settings from models import ( ProtectionLabel, TreatmentMapping, ) # ---------------------------------------------------------- # File Locations # ---------------------------------------------------------- LABEL_FILE = settings.data_path / "crop_protection_labels.csv" TREATMENT_FILE = ( settings.data_path / "normalized_class_treatment_mapping.csv" ) CROP_MAPPING_FILE = ( settings.data_path / "crop_label_mapping.json" ) IMPORT_REPORT_FILE = ( settings.data_path / "import_report.json" ) # ---------------------------------------------------------- # Report # ---------------------------------------------------------- @dataclass class ImportReport: master_labels: int = 0 treatment_records: int = 0 imported_labels: int = 0 imported_treatments: int = 0 duplicate_labels: int = 0 duplicate_class_ids: int = 0 orphan_class_ids: int = 0 label_mismatches: int = 0 skipped_rows: int = 0 warnings: List[dict] = field(default_factory=list) errors: List[str] = field(default_factory=list) def to_dict(self): return { "master_labels": self.master_labels, "treatment_records": self.treatment_records, "imported_labels": self.imported_labels, "imported_treatments": self.imported_treatments, "duplicate_labels": self.duplicate_labels, "duplicate_class_ids": self.duplicate_class_ids, "orphan_class_ids": self.orphan_class_ids, "label_mismatches": self.label_mismatches, "skipped_rows": self.skipped_rows, "warnings": self.warnings, "errors": self.errors, } # ---------------------------------------------------------- # Import Utility # ---------------------------------------------------------- class ImportUtils: def __init__(self, db: Session): self.db = db self.report = ImportReport() self.labels_df: Optional[pd.DataFrame] = None self.treatment_df: Optional[pd.DataFrame] = None self.master_label_map: Dict[int, str] = {} self.crop_label_map: Dict[str, Set[str]] = {} # ------------------------------------------------------ # CSV Loading # ------------------------------------------------------ def load_csv_files(self): if not LABEL_FILE.exists(): raise FileNotFoundError(LABEL_FILE) if not TREATMENT_FILE.exists(): raise FileNotFoundError(TREATMENT_FILE) self.labels_df = pd.read_csv(LABEL_FILE) self.treatment_df = pd.read_csv(TREATMENT_FILE) print("\n===== LABELS DTYPES =====") print(self.labels_df.dtypes) print("\n===== TREATMENT DTYPES =====") print(self.treatment_df.dtypes) print("\n===== TREATMENT HEAD =====") print(self.treatment_df.head()) self.report.master_labels = len(self.labels_df) self.report.treatment_records = len( self.treatment_df ) # ------------------------------------------------------ # Data Cleaning # ------------------------------------------------------ @staticmethod def clean_string(value): if pd.isna(value): return None value = str(value) value = value.strip() if value == "": return None value = " ".join(value.split()) return value @staticmethod def clean_integer(value): if pd.isna(value): return None try: return int(value) except Exception: return None def clean_dataframes(self): # ------------------------- # Labels # ------------------------- for column in self.labels_df.columns: self.labels_df[column] = ( self.labels_df[column] .apply(self.clean_string) ) self.labels_df["Class_ID"] = ( self.labels_df["Class_ID"] .apply(self.clean_integer) ) # ------------------------- # Treatment # ------------------------- for column in self.treatment_df.columns: if column == "Class_ID": continue self.treatment_df[column] = ( self.treatment_df[column] .apply(self.clean_string) ) self.treatment_df["Class_ID"] = ( self.treatment_df["Class_ID"] .apply(self.clean_integer) ) # ------------------------------------------------------ # Master Lookup # ------------------------------------------------------ def build_master_lookup(self): self.master_label_map.clear() for _, row in self.labels_df.iterrows(): class_id = row["Class_ID"] label = row["Label_Name"] if class_id is None: continue self.master_label_map[class_id] = label # ------------------------------------------------------ # Helper # ------------------------------------------------------ def master_label(self, class_id: int): return self.master_label_map.get(class_id) def print_header(self): print() print("=" * 70) print("LeafDoctor AI Dataset Import") print("=" * 70) print() def print_footer(self): print() print("=" * 70) print("Dataset Loading Finished") print("=" * 70) print() # ------------------------------------------------------ # Validation # ------------------------------------------------------ def validate_master_labels(self): """ Validate master label dataset. """ print("Validating master labels...") # ------------------------------------ # Duplicate Class_ID # ------------------------------------ duplicate_ids = ( self.labels_df[ self.labels_df["Class_ID"].duplicated(keep=False) ] .sort_values("Class_ID") ) if not duplicate_ids.empty: self.report.duplicate_class_ids = ( duplicate_ids["Class_ID"].nunique() ) for _, row in duplicate_ids.iterrows(): self.report.errors.append( f"Duplicate Class_ID : {row['Class_ID']}" ) # ------------------------------------ # Duplicate Label Names # ------------------------------------ duplicate_labels = ( self.labels_df[ self.labels_df["Label_Name"].duplicated(keep=False) ] .sort_values("Label_Name") ) if not duplicate_labels.empty: self.report.duplicate_labels = ( duplicate_labels["Label_Name"].nunique() ) for _, row in duplicate_labels.iterrows(): self.report.errors.append( f"Duplicate Label : {row['Label_Name']}" ) # ------------------------------------ # Missing Class_ID # ------------------------------------ missing_ids = self.labels_df[ self.labels_df["Class_ID"].isna() ] if not missing_ids.empty: self.report.skipped_rows += len(missing_ids) self.report.errors.append( f"Missing Class_ID rows : {len(missing_ids)}" ) # ------------------------------------ # Missing Label # ------------------------------------ missing_labels = self.labels_df[ self.labels_df["Label_Name"].isna() ] if not missing_labels.empty: self.report.skipped_rows += len(missing_labels) self.report.errors.append( f"Missing Label rows : {len(missing_labels)}" ) # ------------------------------------------------------ # Treatment Validation # ------------------------------------------------------ def validate_treatment_mapping(self): """ Validate treatment mapping dataset. """ print("Validating treatment mapping...") orphan_ids = set() for index, row in self.treatment_df.iterrows(): class_id = row["Class_ID"] crop = row.get("Crop") if class_id is None: self.report.skipped_rows += 1 self.report.errors.append( f"Row {index}: Missing Class_ID" ) continue if class_id not in self.master_label_map: orphan_ids.add(class_id) self.report.warnings.append( { "row": int(index), "class_id": class_id, "issue": "Orphan Class_ID", } ) if crop is None: self.report.warnings.append( { "row": int(index), "class_id": class_id, "issue": "Missing Crop", } ) if row.get("Chemical & Formulation") is None: self.report.warnings.append( { "row": int(index), "class_id": class_id, "issue": "Missing Chemical", } ) if row.get("SMS : Growth Stage 1") is None: self.report.warnings.append( { "row": int(index), "class_id": class_id, "issue": "Missing SMS", } ) self.report.orphan_class_ids = len(orphan_ids) # ------------------------------------------------------ # Label Mapping Validation # ------------------------------------------------------ def detect_mapping_mismatches(self): """ Compare the label in treatment CSV with the master label table. """ print("Checking label mapping...") mismatches = 0 for index, row in self.treatment_df.iterrows(): class_id = row["Class_ID"] if class_id is None: continue if class_id not in self.master_label_map: continue master_label = self.master_label_map[class_id] csv_label = self.clean_string( row.get("Master_Label_Name") ) if csv_label is None: continue if master_label != csv_label: mismatches += 1 self.report.warnings.append( { "row": int(index), "class_id": class_id, "master_label": master_label, "csv_label": csv_label, } ) self.report.label_mismatches = mismatches # ------------------------------------------------------ # Validation Runner # ------------------------------------------------------ def validate(self): """ Run every validation step. """ self.validate_master_labels() self.validate_treatment_mapping() self.detect_mapping_mismatches() print() print("Validation Complete") print( f"Duplicate Labels : " f"{self.report.duplicate_labels}" ) print( f"Duplicate Class IDs : " f"{self.report.duplicate_class_ids}" ) print( f"Orphan Class IDs : " f"{self.report.orphan_class_ids}" ) print( f"Label Mismatches : " f"{self.report.label_mismatches}" ) print( f"Skipped Rows : " f"{self.report.skipped_rows}" ) print() # ------------------------------------------------------ # Database Cleanup # ------------------------------------------------------ def clear_database(self): """ Clear master tables before importing. Import is idempotent, so every execution starts from a clean state. """ print("Clearing existing data...") try: self.db.query(TreatmentMapping).delete() self.db.query(ProtectionLabel).delete() self.db.commit() except Exception: self.db.rollback() raise # ------------------------------------------------------ # Import Protection Labels # ------------------------------------------------------ def import_protection_labels(self): """ Import master label table. """ print("Importing protection labels...") imported = 0 for _, row in self.labels_df.iterrows(): class_id = row["Class_ID"] label = row["Label_Name"] if class_id is None: continue if label is None: continue obj = ProtectionLabel( class_id=class_id, label_name_en=label, label_name_bn=None, primary_category_en=row.get( "Primary_Category" ), primary_category_bn=None, ) self.db.add(obj) imported += 1 self.db.flush() self.report.imported_labels = imported print(f"Imported {imported} labels.") # ------------------------------------------------------ # Import Treatment Mapping # ------------------------------------------------------ def import_treatment_mapping(self): """ Import treatment mapping. IMPORTANT We intentionally ignore Master_Label_Name from the CSV because the master table already contains the authoritative label. """ print("Importing treatment mapping...") imported = 0 skipped = 0 for _, row in self.treatment_df.iterrows(): class_id = row["Class_ID"] if class_id is None: skipped += 1 continue if class_id not in self.master_label_map: skipped += 1 continue treatment = TreatmentMapping( class_id=class_id, crop_en=row.get("Crop"), crop_bn=None, problem_type_en=row.get( "Type of Problem" ), problem_type_bn=None, chemical=row.get( "Chemical & Formulation" ), chemical_type=row.get( "Chemical Type" ), dosage=row.get( "Revised Dosages" ), sms_en=row.get( "SMS : Growth Stage 1" ), sms_bn=None, ) self.db.add(treatment) # Temporary diagnostic test self.db.flush() imported += 1 print(vars(treatment)) self.db.flush() self.report.imported_treatments = imported self.report.skipped_rows += skipped print(f"Imported {imported} treatments.") print(f"Skipped {skipped} rows.") # ------------------------------------------------------ # Database Import # ------------------------------------------------------ def import_database(self): """ Complete import inside one transaction. """ print() print("=" * 60) print("Database Import Started") print("=" * 60) try: self.clear_database() self.import_protection_labels() self.import_treatment_mapping() self.db.commit() print() print("Database Import Completed.") except Exception as ex: self.db.rollback() print() print("Database Import Failed") raise RuntimeError(str(ex)) # ------------------------------------------------------ # Crop -> Label Mapping # ------------------------------------------------------ def generate_crop_label_mapping(self): """ Generate crop -> valid labels mapping. Example { "Rice": { "labels": [...], "class_ids": [...] } } """ print("Generating crop-label mapping...") mapping = {} for _, row in self.treatment_df.iterrows(): class_id = row["Class_ID"] if class_id not in self.master_label_map: continue crop = self.clean_string(row.get("Crop")) if crop is None: continue label = self.master_label_map[class_id] if crop not in mapping: mapping[crop] = { "labels": set(), "class_ids": set() } mapping[crop]["labels"].add(label) mapping[crop]["class_ids"].add(class_id) output = {} for crop, values in sorted(mapping.items()): output[crop] = { "labels": sorted(values["labels"]), "class_ids": sorted(values["class_ids"]) } with open( CROP_MAPPING_FILE, "w", encoding="utf-8" ) as fp: json.dump( output, fp, indent=4, ensure_ascii=False ) print( f"Crop mapping saved to " f"{CROP_MAPPING_FILE}" ) # ------------------------------------------------------ # Save Import Report # ------------------------------------------------------ def save_import_report(self): with open( IMPORT_REPORT_FILE, "w", encoding="utf-8" ) as fp: json.dump( self.report.to_dict(), fp, indent=4, ensure_ascii=False ) print( f"Import report saved to " f"{IMPORT_REPORT_FILE}" ) # ------------------------------------------------------ # Print Summary # ------------------------------------------------------ def print_summary(self): print() print("=" * 70) print("LeafDoctor AI Import Summary") print("=" * 70) print() print( f"Master Labels : " f"{self.report.master_labels}" ) print( f"Treatment Records : " f"{self.report.treatment_records}" ) print( f"Imported Labels : " f"{self.report.imported_labels}" ) print( f"Imported Treatments : " f"{self.report.imported_treatments}" ) print( f"Duplicate Labels : " f"{self.report.duplicate_labels}" ) print( f"Duplicate Class IDs : " f"{self.report.duplicate_class_ids}" ) print( f"Orphan Class IDs : " f"{self.report.orphan_class_ids}" ) print( f"Label Mismatches : " f"{self.report.label_mismatches}" ) print( f"Skipped Rows : " f"{self.report.skipped_rows}" ) print( f"Warnings : " f"{len(self.report.warnings)}" ) print( f"Errors : " f"{len(self.report.errors)}" ) print() print("=" * 70) print() # ------------------------------------------------------ # Run Complete Import # ------------------------------------------------------ def run_import(self): """ Complete import workflow. """ self.print_header() self.load_csv_files() self.clean_dataframes() self.build_master_lookup() self.validate() self.import_database() self.generate_crop_label_mapping() self.save_import_report() self.print_summary() self.print_footer()