in data store the create page the upload doc fixed

.devcontainer/docker-compose.yml

@@ -46,6 +46,7 @@ services:
     environment:
       MINIO_ROOT_USER: ${MINIO_USERNAME}
       MINIO_ROOT_PASSWORD: ${MINIO_PASSWORD}
+      MINIO_API_CORS_ALLOW_ORIGIN: "*"
     command: server /data/s3 --console-address ":9001"
     restart: unless-stopped
     network_mode: host

backend/app/api/endpoints/datasets/datasets.py

@@ -1,3 +1,4 @@
+import os
 import uuid
 import mimetypes
 import pandas as pd
@@ -27,6 +28,15 @@
 def get_presigned_url(filename: str, current_user: dict = Depends(get_current_user)) -> PresignedURLResponse:
     storage_service = get_storage_service()
     user_id = str(current_user.get("_id"))
+
+    # Check if filename is an existing file_id ObjectId in files_collection
+    if ObjectId.is_valid(filename):
+        file_doc = files_collection.find_one({"_id": ObjectId(filename)})
+        if file_doc and file_doc.get("file_location"):
+            file_location = file_doc["file_location"]
+            download_url = storage_service.generate_download_url(file_location)
+            return PresignedURLResponse(upload_url=download_url, object_name=file_location)
+
     url, object_name = storage_service.generate_presigned_url(
         filename=filename, user_id=user_id)
     return PresignedURLResponse(upload_url=url, object_name=object_name)

backend/app/api/endpoints/users/role_check.py

@@ -95,13 +95,27 @@ def check_user_role_access(request: RoleCheckRequest, fastapi_request: Request,
             raise HTTPException(status_code=404, detail="User not found")
 
         # Get user's role
-        role_id = user.get("role_id")
+        role_ids = user.get("role_id")
         role_name = "user"
-        if role_id:
-            role = get_role_by_id(role_id)
-            if role and role.get("role_name"):
-                role_name = role["role_name"]
-        logger.debug("User role resolved: role_id=%s role_name=%s", role_id, role_name)
+        if role_ids:
+            # Handle both list of IDs and single ID cases
+            if not isinstance(role_ids, list):
+                role_ids = [role_ids]
+
+            roles_found = []
+            for r_id in role_ids:
+                role = get_role_by_id(r_id)
+                if role and role.get("role_name"):
+                    roles_found.append(role["role_name"])
+
+            # Select the most privileged role if multiple exist
+            if "superadmin" in roles_found:
+                role_name = "superadmin"
+            elif "admin" in roles_found:
+                role_name = "admin"
+            elif roles_found:
+                role_name = roles_found[0]
+        logger.debug("User role resolved: role_ids=%s role_name=%s", role_ids, role_name)
 
         # Find matching endpoint access rule
         endpoint_access = find_matching_endpoint_access(role_name, request.path)

backend/app/config/pipeline_mapping.py

@@ -6,15 +6,142 @@
 uses specific dataset names, while our system uses pipeline IDs.
 """
 
-# Mapping of pipeline IDs to ERP dataset names
+from typing import Optional, Dict, Any
+
+# Mapping of pipeline IDs to ERP configuration (source name, source type, sync strategy, identity key, default filters)
+PIPELINE_CONFIG = {
+    # -------------------------------------------------------------------------
+    # Dashboard Pipelines (Phase 3 Verified Mappings)
+    # -------------------------------------------------------------------------
+    "nf_coordinator_activities": {
+        "erp_base_url": "http://erp.csa-india.org",
+        "source_name": "CC Daily Reports",
+        "source_type": "doctype",
+        "sync_strategy": "timestamp",
+        "identity_key": "date",
+        "target_collection": "nf_coordinator_activities",
+        "mapper": "NFCoordinatorMapper",
+        "fetch_full_docs": True,
+    },
+    "territory_transactions": {
+        "erp_base_url": "http://erp.fpohub.com",
+        "source_name": "Purchase Invoice",
+        "source_type": "doctype",
+        "sync_strategy": "timestamp",
+        "identity_key": "date",
+        "target_collection": "territory_transactions",
+        "mapper": "TerritoryTransactionsMapper",
+    },
+    "farmer_income_visits": {
+        "erp_base_url": "http://erp.csa-india.org",
+        "source_name": "CC Daily Reports",
+        "source_type": "doctype",
+        "sync_strategy": "timestamp",
+        "identity_key": "coordinator_name",
+        "target_collection": "farmer_income_visits",
+        "mapper": "FarmerIncomeVisitsMapper",
+        "fetch_full_docs": True,
+    },
+    "stock_movement": {
+        "erp_base_url": "http://erp.fpohub.com",
+        "source_name": "Stock Balance",
+        "source_type": "query_report",
+        "sync_strategy": "snapshot",
+        "identity_key": None,
+        "target_collection": "stock_movement",
+        "mapper": "StockMovementMapper",
+        "default_filters": {
+            "company": "Nelathalli Farmer Producer Company Limited",
+            "from_date": "2024-01-01",
+            "to_date": "2026-12-31",
+        },
+    },
+    "stock_inventory": {
+        "erp_base_url": "http://erp.fpohub.com",
+        "source_name": "Stock Balance",
+        "source_type": "query_report",
+        "sync_strategy": "snapshot",
+        "identity_key": None,
+        "target_collection": "stock_inventory",
+        "mapper": "StockInventoryMapper",
+        "default_filters": {
+            "company": "Nelathalli Farmer Producer Company Limited",
+            "from_date": "2024-01-01",
+            "to_date": "2026-12-31",
+        },
+    },
+    "revenue_analysis": {
+        "erp_base_url": "http://erp.fpohub.com",
+        "source_name": "Purchase Invoice",
+        "source_type": "doctype",
+        "sync_strategy": "timestamp",
+        "identity_key": "territory",
+        "target_collection": "revenue_analysis",
+        "mapper": "RevenueAnalysisMapper",
+    },
+
+    # -------------------------------------------------------------------------
+    # Generic Pipelines (Phase 1 & 2)
+    # -------------------------------------------------------------------------
+    "soil_collection": {
+        "erp_base_url": "http://erp.csa-india.org",
+        "source_name": "Soil Collection Data",
+        "source_type": "doctype",
+        "sync_strategy": "timestamp",
+        "identity_key": "name",
+    },
+    "weather_data": {
+        "erp_base_url": "http://erp.csa-india.org",
+        "source_name": "Weather Data",
+        "source_type": "doctype",
+        "sync_strategy": "timestamp",
+        "identity_key": "name",
+    },
+    "crop_yield": {
+        "erp_base_url": "http://erp.csa-india.org",
+        "source_name": "Crop Yield Data",
+        "source_type": "doctype",
+        "sync_strategy": "timestamp",
+        "identity_key": "name",
+    },
+    "sales_invoice": {
+        "erp_base_url": "http://erp.csa-india.org",
+        "source_name": "Sales Invoice",
+        "source_type": "doctype",
+        "sync_strategy": "timestamp",
+        "identity_key": "name",
+    },
+    "purchase_invoice": {
+        "erp_base_url": "http://erp.fpohub.com",
+        "source_name": "Purchase Invoice",
+        "source_type": "doctype",
+        "sync_strategy": "timestamp",
+        "identity_key": "name",
+    },
+    "cc_daily_reports": {
+        "erp_base_url": "http://erp.csa-india.org",
+        "source_name": "CC Daily Reports",
+        "source_type": "doctype",
+        "sync_strategy": "timestamp",
+        "identity_key": "name",
+    },
+    "stock_balance": {
+        "erp_base_url": "http://erp.fpohub.com",
+        "source_name": "Stock Balance",
+        "source_type": "query_report",
+        "sync_strategy": "snapshot",
+        "identity_key": None,
+        "default_filters": {
+            "company": "Nelathalli Farmer Producer Company Limited",
+            "from_date": "2024-01-01",
+            "to_date": "2026-12-31",
+        },
+    },
+}
+
+# Legacy dictionary for backward compatibility
 PIPELINE_DATASET_MAPPING = {
-    # Add your pipeline mappings here
-    # "pipeline_id": "ERP Dataset Name"
-    # Example mappings (replace with actual mappings)
-    "soil_collection": "Soil Collection Data",
-    "weather_data": "Weather Data",
-    "crop_yield": "Crop Yield Data",
-    # Add more mappings as needed
+    pid: cfg["source_name"] for pid, cfg in PIPELINE_CONFIG.items()
 }
 
 
@@ -28,4 +155,92 @@ def get_dataset_name_for_pipeline(pipeline_id: str) -> str:
     Returns:
         The corresponding dataset name, or the pipeline_id itself if no mapping exists
     """
+    if pipeline_id in PIPELINE_CONFIG:
+        return PIPELINE_CONFIG[pipeline_id]["source_name"]
     return PIPELINE_DATASET_MAPPING.get(pipeline_id, pipeline_id)
+
+
+def get_pipeline_source_type(pipeline_id: str) -> str:
+    """
+    Get the ERP source type ('doctype' or 'query_report') for a given pipeline ID.
+
+    Args:
+        pipeline_id: The pipeline ID to look up
+
+    Returns:
+        'doctype' or 'query_report' (defaults to 'doctype')
+    """
+    if pipeline_id in PIPELINE_CONFIG:
+        return PIPELINE_CONFIG[pipeline_id].get("source_type", "doctype")
+    name = get_dataset_name_for_pipeline(pipeline_id).lower()
+    if "report" in name or name in ("stock balance", "general ledger"):
+        return "query_report"
+    return "doctype"
+
+
+def get_pipeline_sync_strategy(pipeline_id: str) -> str:
+    """
+    Get the sync strategy ('timestamp' or 'snapshot') for a given pipeline ID.
+    """
+    if pipeline_id in PIPELINE_CONFIG:
+        return PIPELINE_CONFIG[pipeline_id].get("sync_strategy", "timestamp")
+    stype = get_pipeline_source_type(pipeline_id)
+    return "snapshot" if stype == "query_report" else "timestamp"
+
+
+def get_pipeline_identity_key(pipeline_id: str) -> Optional[str]:
+    """
+    Get the primary record identity key for upserting records (e.g. 'name' for DocTypes).
+    """
+    if pipeline_id in PIPELINE_CONFIG:
+        return PIPELINE_CONFIG[pipeline_id].get("identity_key", "name")
+    stype = get_pipeline_source_type(pipeline_id)
+    return None if stype == "query_report" else "name"
+
+
+def get_pipeline_erp_url(pipeline_id: str) -> Optional[str]:
+    """
+    Get the configured ERP base URL for a pipeline (http://erp.csa-india.org vs http://erp.fpohub.com).
+    """
+    if pipeline_id in PIPELINE_CONFIG:
+        return PIPELINE_CONFIG[pipeline_id].get("erp_base_url")
+    return None
+
+
+def get_pipeline_mapper(pipeline_id: str) -> Optional[str]:
+    """
+    Get the configured mapper name for a pipeline (e.g. 'NFCoordinatorMapper').
+    """
+    if pipeline_id in PIPELINE_CONFIG:
+        return PIPELINE_CONFIG[pipeline_id].get("mapper")
+    return None
+
+
+def get_pipeline_target_collection(pipeline_id: str) -> str:
+    """
+    Get the target MongoDB collection name for a pipeline.
+    """
+    if pipeline_id in PIPELINE_CONFIG:
+        return PIPELINE_CONFIG[pipeline_id].get("target_collection", pipeline_id)
+    return pipeline_id
+
+
+def get_pipeline_config(pipeline_id: str) -> dict:
+    """
+    Get the full pipeline configuration for a given pipeline ID.
+    """
+    return PIPELINE_CONFIG.get(
+        pipeline_id,
+        {
+            "source_name": get_dataset_name_for_pipeline(pipeline_id),
+            "source_type": get_pipeline_source_type(pipeline_id),
+            "sync_strategy": get_pipeline_sync_strategy(pipeline_id),
+            "identity_key": get_pipeline_identity_key(pipeline_id),
+            "erp_base_url": get_pipeline_erp_url(pipeline_id),
+            "mapper": get_pipeline_mapper(pipeline_id),
+            "target_collection": get_pipeline_target_collection(pipeline_id),
+        },
+    )
+
+
+

backend/app/dashboards/farmer_income_dashboard.py

@@ -17,7 +17,7 @@
 """
 
 import base64
-from utilities import initialize_page
+from utilities import initialize_page, load_dashboard_data_from_mongodb
 
 import pandas as pd
 import plotly.express as px
@@ -36,16 +36,24 @@
 initialize_page()
 
 
-# ── Data loading — swap this function for MongoDB when ready ───────────────────
+# ── Data loading — MongoDB primary with reference CSV fallback ────────────────
 @st.cache_data
 def load_data() -> pd.DataFrame:
     """
-    Load farmer income & visits data.
-
-    Current source : CSV  (data/farmer_income_data.csv)
-    Future source  : MongoDB collection  "farmer_income_visits"
+    Load farmer income & visits data from MongoDB collection 'farmer_income_visits'.
+    Falls back to local CSV reference data only if MongoDB collection is empty.
     """
-    df = pd.read_csv(DATA_CSV, encoding="ISO-8859-1")
+    df = load_dashboard_data_from_mongodb("farmer_income_visits")
+
+    if df is None or df.empty:
+        if DATA_CSV.exists():
+            df = pd.read_csv(DATA_CSV, encoding="ISO-8859-1")
+        else:
+            return pd.DataFrame(columns=[
+                "coordinator_name", "month", "village", "farmers_met",
+                "visits", "score", "income", "net_income", "yield"
+            ])
+
     return df
 
 

backend/app/dashboards/nf_coordinator_dashboard.py

@@ -18,7 +18,7 @@
 """
 
 import base64
-from utilities import initialize_page
+from utilities import initialize_page, load_dashboard_data_from_mongodb
 
 import pandas as pd
 import plotly.express as px
@@ -41,16 +41,24 @@
 initialize_page()
 
 
-# ── Data loading — swap this function for MongoDB when ready ───────────────────
+# ── Data loading — MongoDB primary with reference CSV fallback ────────────────
 @st.cache_data
 def load_nf_data() -> pd.DataFrame:
     """
-    Load NF Coordinator activity data.
-
-    Current source : CSV  (data/nf_coordinator_data.csv)
-    Future source  : MongoDB collection  "nf_coordinator_activities"
+    Load NF Coordinator activity data from MongoDB collection 'nf_coordinator_activities'.
+    Falls back to local CSV reference data only if MongoDB collection is empty.
     """
-    df = pd.read_csv(NF_CSV, encoding="ISO-8859-1")
+    df = load_dashboard_data_from_mongodb("nf_coordinator_activities")
+
+    if df is None or df.empty:
+        if NF_CSV.exists():
+            df = pd.read_csv(NF_CSV, encoding="ISO-8859-1")
+        else:
+            return pd.DataFrame(columns=[
+                "coordinator_name", "district", "date", "type_of_activity",
+                "planned_activities", "actual_activities", "total_score"
+            ])
+
     df["date"]        = pd.to_datetime(df["date"], format="%d-%m-%Y", errors="coerce")
     df["month_label"] = df["date"].dt.strftime("%b %Y")
     df["month_short"] = df["date"].dt.strftime("%B")