Merge pull request #35 from gnanarepo/ujjawal_core

druva splits handling

Author: ujjawal-aviso

Dockerfile

@@ -60,4 +60,4 @@ USER appuser
 
 EXPOSE 8000
 
-CMD ["gunicorn", "aviso_core.wsgi:application", "--bind", "0.0.0.0:8000", "--workers", "1", "--timeout", "900"]
+CMD ["gunicorn", "aviso_core.wsgi:application", "--bind", "0.0.0.0:8000", "--workers", "3", "--timeout", "900"]

gbm_apis/api/data_load.py

@@ -1,7 +1,9 @@
+import json
 import os
 
 from aviso.settings import sec_context
 from aviso.utils.dateUtils import TimeHorizon
+from aviso.utils import is_true
 
 import logging
 import collections
@@ -73,7 +75,9 @@ def __init__(
             period,
             run_type='chipotle',
             from_timestamp=0,
-            changed_fields_only=False):
+            changed_fields_only=False,
+            return_oppids_only=False):
+        
         self.id_list = id_list
         self.from_timestamp = from_timestamp
         self.tenant_name = tenant_name
@@ -84,6 +88,7 @@ def __init__(
         self.changed_fields_only = changed_fields_only
         self.run_type = run_type
         self.period = period
+        self.return_oppids_only = return_oppids_only
 
     def get_basic_results(self):
         ds = Dataset.getByNameAndStage('OppDS', None)
@@ -113,7 +118,27 @@ def get_basic_results(self):
         deals = list(coll.find(criteria, {'_id': 0}))
 
         logger.info(f"got result length of deals: {len(deals)}")
-        
+
+        if self.return_oppids_only:
+            try:
+                #opp_ids_records = sec_context.etl.uip('UIPIterator', dataset='OppDS', record_filter=criteria, fields_requested=['ID'])
+                if deals:
+                    oppids_list = [deal['object']['extid'] for deal in deals]
+                else:
+                    oppids_list = []
+                logger.info("Number of oppids fetched: %s", len(oppids_list))
+                
+                return {
+                    'oppids': oppids_list
+                }
+            except Exception as e:
+                logger.error("Exception occurred while fetching opp_ids_records: %s", str(e))
+                error_response = {
+                    "success": False,
+                    "error": str(e)
+                }
+                return error_response
+            
         final_deals = []
         from_timestamp_xl = (self.from_timestamp / 1000.0) / 86400 + 25569
         for deal in deals:
@@ -320,12 +345,23 @@ def revenue_schedule(self):
                 expiration_date_fld = rev_schedule_config.get('expiration_date_fld', 'ExpirationDate')
                 type_fld = rev_schedule_config.get('type_fld', 'Type')
                 renewal_vals = rev_schedule_config.get('renewal_vals', ['Renewal'])
+
+                viewgen_config = oppds.models['common'].config.get('viewgen_config', {})
+                splitted_fields = []
+                if viewgen_config['split_config']:
+                    split_config = viewgen_config['split_config'][list(viewgen_config['split_config'])[0]]
+                    for _, dtls in split_config.items():
+                        if type(dtls) != list:
+                            dtls = [dtls]
+                        for element in dtls:
+                            splitted_fields += element['num_fields']
+                
                 basic_results_dict_copy = deepcopy(basic_results_dict)
                 basic_results_dict = {}
                 for opp_id, res in basic_results_dict_copy.items():
                     output_dict = generate_expiration_date_renewal_rec(rev_period, renewal_drilldown, close_date_fld,
                                                                        expiration_date_fld, type_fld, renewal_vals,
-                                                                       opp_id, res)
+                                                                       opp_id, res, splitted_fields)
                     basic_results_dict.update(output_dict)
             else:
                 basic_results_dict = generate_revenue_recs(rev_period, drilldown, close_date_fld, basic_results_dict)
@@ -414,6 +450,8 @@ def get(self, request, *args, **kwargs):
             changed_fields_param = request.GET.get('changed_fields_only', 'false')
             changed_fields_only = changed_fields_param.lower() == 'true'
 
+            return_oppids_only = is_true(request.GET.get('return_oppids_only', False))
+
             if not all([tenant_name, stack]):
                 return JsonResponse(
                     {"error": "Missing required fields (tenant_name, stack)"},
@@ -433,12 +471,16 @@ def get(self, request, *args, **kwargs):
                 period=period,
                 run_type=run_type,
                 from_timestamp=from_timestamp,
-                changed_fields_only=changed_fields_only
+                changed_fields_only=changed_fields_only,
+                return_oppids_only=return_oppids_only
             )
 
             # 5. Get Basic Results
             basic_results = loader.get_basic_results()
 
+            if return_oppids_only:
+                return JsonResponse(basic_results, safe=False, status=200)
+
             # 6. Initialize RevenueSchedule
             scheduler = RevenueSchedule(period=period, basic_results=basic_results)
             final_results = scheduler.revenue_schedule()

gbm_apis/api/drilldown_fields_v2.py

@@ -1,5 +1,5 @@
 import itertools
-import json
+import json, os
 import logging
 
 from collections import defaultdict
@@ -17,6 +17,9 @@
 from django.utils.decorators import method_decorator
 from django.views.decorators.csrf import csrf_exempt
 
+from aviso.framework.tenant_mongo_resolver import TenantMongoResolver
+from gbm_apis.deal_result.result_Utils import deals_results_by_period
+
 logger = logging.getLogger("aviso-core.%s" % __name__)
 
 
@@ -177,9 +180,28 @@ def build_global_territory_owners(groups: Iterable[Iterable[str]], period: Optio
         logger.warning("sec_context.etl not available; returning empty drilldown set")
         return {}
 
-    records = sec_context.etl.uip(
-        "UIPIterator", dataset="OppDS", record_filter=record_filter, fields_requested=list(required_fields)
-    )
+    # records = sec_context.etl.uip(
+    #     "UIPIterator", dataset="OppDS", record_filter=record_filter, fields_requested=list(required_fields),
+    #     skip_prepare=True
+    # )
+
+    cname = os.environ.get("CNAME", "preprod")
+    etl_db = TenantMongoResolver().ms_connection_mongo_client_db(tenant=sec_context.name, db_type='etl', cname=cname)
+
+    coll = etl_db[sec_context.name + '.OppDS._uip._data']
+
+    projection = {
+        "object.terminal_date": 1 
+    }
+
+    for field in required_fields:
+        projection[f"object.history.{field}"] = 1
+
+    records = coll.find(
+        record_filter,
+        projection=projection,
+        no_cursor_timeout=True
+    ).batch_size(50000)
 
     hierarchy = _fetch_raw_hierarchy()
     parents = {}
@@ -229,36 +251,79 @@ def build_global_territory_owners(groups: Iterable[Iterable[str]], period: Optio
     processed_count = 0
     skipped_count = 0
 
+    # for record in records:
+    #     #print("Processing record:", record)
+    #     # raise
+    #     feat_map = record['object']['history']
+    #     #print("Feature map keys:", feat_map.keys())
+    #     # raise
+    #     if territory_key not in feat_map:
+    #         skipped_count += 1
+    #         continue
+
+    #     territory_data = feat_map[territory_key]
+    #     if not territory_data:
+    #         skipped_count += 1
+    #         continue
+        
+    #     #print("Territory data:", territory_data)
+    #     leaf_node = territory_data[-1][1]
+    #     if leaf_node not in global_children_set:
+    #         skipped_count += 1
+    #         continue
+
+    #     processed_count += 1
+    #     for group_name, field_mappings in group_configs:
+    #         current_row = []
+    #         for fld_label, map_key in field_mappings:
+    #             if map_key in feat_map:
+    #                 field_data = feat_map[map_key]
+    #                 if field_data:
+    #                     current_row.append([fld_label, field_data[-1][1]])
+    #                 else:
+    #                     current_row.append([fld_label, "N/A"])
+    #             else:
+    #                 current_row.append([fld_label, "N/A"])
+    #         staging[group_name][leaf_node].add(tuple(tuple(pair) for pair in current_row))
+
     for record in records:
-        feat_map = record.featMap
-        if territory_key not in feat_map:
+        obj = record.get("object")
+        if not obj:
+            skipped_count += 1
+            continue
+
+        history = obj.get("history")
+        if not history:
             skipped_count += 1
             continue
 
-        territory_data = feat_map[territory_key]
+        # Get territory data (single lookup only)
+        territory_data = history.get(territory_key)
         if not territory_data:
             skipped_count += 1
             continue
 
+        # Get leaf node
         leaf_node = territory_data[-1][1]
+
         if leaf_node not in global_children_set:
             skipped_count += 1
             continue
 
         processed_count += 1
+
+        # Process groups
         for group_name, field_mappings in group_configs:
-            current_row = []
+            row = []
+
             for fld_label, map_key in field_mappings:
-                if map_key in feat_map:
-                    field_data = feat_map[map_key]
-                    if field_data:
-                        current_row.append([fld_label, field_data[-1][1]])
-                    else:
-                        current_row.append([fld_label, "N/A"])
-                else:
-                    current_row.append([fld_label, "N/A"])
-            staging[group_name][leaf_node].add(tuple(tuple(pair) for pair in current_row))
+                field_data = history.get(map_key)
+                value = field_data[-1][1] if field_data else "N/A"
+                row.append((fld_label, value))  # use tuple directly
 
+            staging[group_name][leaf_node].add(tuple(row)) 
+
+            
     logger.info(
         "build_global_territory_owners: processed %s records, skipped %s records",
         processed_count,
@@ -302,32 +367,32 @@ def drilldown_values_by_owner_v2(drilldown: str, periods: Iterable[str]):
     return output
 
 
-def deals_results_by_period(periods: Iterable[str]) -> Dict[str, Dict[str, Any]]:
-    """
-    Lightweight helper to fetch deal results for the requested periods.
-    """
-    results: Dict[str, Dict[str, Any]] = {}
-    if sec_context and getattr(sec_context, "gbm", None):
-        gbm_client = sec_context.gbm
-        for period in periods:
-            try:
-                url = f"/gbm/deals_results?period={period}"
-                results[period] = gbm_client.api(url, None) or {"results": {}}
-            except Exception as exc:
-                logger.exception("Failed to fetch deals results for %s", period, exc_info=exc)
-                results[period] = {"results": {}}
-    else:
-        logger.warning("GBM client not configured; returning empty results for periods %s", list(periods))
-        for period in periods:
-            results[period] = {"results": {}}
-    return results
-
-
-def drilldown_values_by_period(periods: Iterable[str], groups: Iterable[Iterable[str]]):
+# def deals_results_by_period(periods: Iterable[str]) -> Dict[str, Dict[str, Any]]:
+#     """
+#     Lightweight helper to fetch deal results for the requested periods.
+#     """
+#     results: Dict[str, Dict[str, Any]] = {}
+#     if sec_context and getattr(sec_context, "gbm", None):
+#         gbm_client = sec_context.gbm
+#         for period in periods:
+#             try:
+#                 url = f"/gbm/deals_results?period={period}"
+#                 results[period] = gbm_client.api(url, None) or {"results": {}}
+#             except Exception as exc:
+#                 logger.exception("Failed to fetch deals results for %s", period, exc_info=exc)
+#                 results[period] = {"results": {}}
+#     else:
+#         logger.warning("GBM client not configured; returning empty results for periods %s", list(periods))
+#         for period in periods:
+#             results[period] = {"results": {}}
+#     return results
+
+
+def drilldown_values_by_period(periods, groups,):
     """
     Given a set of drilldown groups, return all possible combos of those field values.
     """
-    output: Dict[str, Dict[str, List[List[tuple]]]] = {}
+    output = {}
     curr_period = current_period().mnemonic
 
     for period in periods:

gbm_apis/data_load/splitter_service.py

@@ -83,6 +83,9 @@ def create_splits(self, record):
                 output[field][split_id] = self.try_num_val(record[field], ratio)
             for field in prim_str_fields:
                 output[field][split_id] = self.try_str_val(record[field], split_id)
+                if field + '_ratios' not in output:
+                    output[field + '_ratios'] = {}
+                output[field + '_ratios'][split_id] = ratio
             # for field in set(output) & self.ignore_fields:
             #    output[field][split_id] = record[field]
 
@@ -98,6 +101,9 @@ def create_splits(self, record):
                     output[field][split_id] = self.try_num_val(record[field], ratio)
                 for field in o_str_fields:
                     output[field][split_id] = self.try_num_ratio(record[field], split_id)
+                    if field + '_ratios' not in output:
+                        output[field + '_ratios'] = {}
+                    output[field + '_ratios'][split_id] = ratio
         return output, split_ids
 
     def try_num_val(self, val, ratio):