Update linkedin_follower_stats.py

linkedin_follower_stats.py

@@ -24,8 +24,6 @@ def _fetch_linkedin_names(session, url, params, result_key_path, name_key_path,
     Generic helper to fetch and map IDs to names from a LinkedIn API endpoint.
     result_key_path: list of keys to navigate to the list of items (e.g., ["elements"])
     name_key_path: list of keys to navigate to the name within an item (e.g., ["name", "localized", "en_US"])
-    
-    Revised: Removed locale_needed parameter; calling functions should provide locale in params if required.
     """
     mapping = {}
     try:
@@ -37,19 +35,19 @@ def _fetch_linkedin_names(session, url, params, result_key_path, name_key_path,
         items = data
         for key in result_key_path: # Navigate to the list/dict of items
             if isinstance(items, dict):
-                items = items.get(key, []) # Default to empty list if key not found
-            else: # If items is already not a dict
+                items = items.get(key, []) 
+            else: 
                 logging.warning(f"Expected dict to get key '{key}' but got {type(items)} at path {result_key_path} for URL {url}. Check result_key_path.")
-                return mapping # Cannot proceed with this path
+                return mapping 
 
         if isinstance(items, dict): # For batch responses like geo/industry (where keys are IDs)
             for item_id_str, item_data in items.items():
                 name = item_data
-                for key_nav in name_key_path: # Navigate to the name string
+                for key_nav in name_key_path: 
                     if isinstance(name, dict):
                         name = name.get(key_nav)
                     else:
-                        name = None # Path broken
+                        name = None 
                         break
                 if name:
                     mapping[item_id_str] = name 
@@ -59,14 +57,14 @@ def _fetch_linkedin_names(session, url, params, result_key_path, name_key_path,
             for item in items:
                 item_id_val = item.get(id_key)
                 name = item
-                for key_nav in name_key_path: # Navigate to the name string
+                for key_nav in name_key_path: 
                     if isinstance(name, dict):
                         name = name.get(key_nav)
                     else:
-                        name = None # Path broken
+                        name = None 
                         break
                 if item_id_val is not None and name:
-                    mapping[str(item_id_val)] = name # Ensure ID is string for consistency
+                    mapping[str(item_id_val)] = name 
                 else:
                     logging.warning(f"No ID ('{id_key}') or name found at path {name_key_path} in item: {item} from URL {url}")
         else:
@@ -85,14 +83,16 @@ def _fetch_linkedin_names(session, url, params, result_key_path, name_key_path,
 def get_functions_map(session):
     """Fetches all LinkedIn functions and returns a map of {id: name}."""
     url = f"{API_V2_BASE}/functions"
-    params = {'locale': 'en_US'} 
+    # Rely on Accept-Language header from session for localization
+    params = {} 
     logging.info("Fetching all LinkedIn functions.")
     return _fetch_linkedin_names(session, url, params, ["elements"], ["name", "localized", "en_US"], "id")
 
 def get_seniorities_map(session):
     """Fetches all LinkedIn seniorities and returns a map of {id: name}."""
     url = f"{API_V2_BASE}/seniorities"
-    params = {'locale': 'en_US'}
+    # Rely on Accept-Language header from session for localization
+    params = {} 
     logging.info("Fetching all LinkedIn seniorities.")
     return _fetch_linkedin_names(session, url, params, ["elements"], ["name", "localized", "en_US"], "id")
 
@@ -100,13 +100,11 @@ def get_industries_map(session, industry_urns, version="DEFAULT"):
     """Fetches names for a list of industry URNs. Returns a map {id: name}."""
     if not industry_urns: return {}
     industry_ids = [_parse_urn_to_id(urn) for urn in industry_urns if urn]
-    unique_ids = list(set(filter(None, industry_ids))) # Filter out None IDs from parsing
+    unique_ids = list(set(filter(None, industry_ids))) 
     if not unique_ids: return {}
         
     url = f"{API_V2_BASE}/industryTaxonomyVersions/{version}/industries"
-    # LinkedIn API for batch industries expects ids as repeated query parameters: ids=1&ids=23
-    # The requests library handles lists in params by creating repeated query parameters.
-    params = {'ids': unique_ids, 'locale.language': 'en', 'locale.country': 'US'}
+    params = {'ids': unique_ids, 'locale': 'en_US'} # Corrected locale parameter
     logging.info(f"Fetching names for {len(unique_ids)} unique industry IDs.")
     return _fetch_linkedin_names(session, url, params, ["results"], ["name", "localized", "en_US"])
 
@@ -118,24 +116,21 @@ def get_geo_map(session, geo_urns):
     unique_ids = list(set(filter(None, geo_ids)))
     if not unique_ids: return {}
 
-    # API expects ids=List(123,456) format in query string.
-    ids_param_value = "List(" + ",".join(map(str,unique_ids)) + ")" # Ensure IDs are strings
-    # Parameters are embedded in the URL for this specific format
-    # Note: locale params are added here directly as part of the URL construction for this specific endpoint style.
-    url = f"{API_V2_BASE}/geo?ids={quote(ids_param_value)}&locale.language=en&locale.country=US"
+    ids_param_value = "List(" + ",".join(map(str,unique_ids)) + ")" 
+    locale_param = "en_US" # Corrected locale parameter
+    url = f"{API_V2_BASE}/geo?ids={quote(ids_param_value)}&locale={locale_param}"
     logging.info(f"Fetching names for {len(unique_ids)} unique geo IDs using URL: {url}")
-    # Params dict is empty as all params are in the URL string for this call.
     return _fetch_linkedin_names(session, url, {}, ["results"], ["defaultLocalizedName", "value"])
 
 
 def _parse_urn_to_id(urn_string):
     """Helper to get the last part (ID) from a URN string."""
     if not isinstance(urn_string, str):
-        logging.warning(f"Invalid URN type: {type(urn_string)}, value: {urn_string}")
+        logging.debug(f"Invalid URN type: {type(urn_string)}, value: {urn_string}. Cannot parse ID.")
         return None
     try:
         return urn_string.split(':')[-1]
-    except IndexError: # Handle cases where split doesn't yield enough parts
+    except IndexError: 
         logging.warning(f"Could not parse ID from URN: {urn_string}")
         return None
     except Exception as e:
@@ -150,7 +145,6 @@ def fetch_monthly_follower_gains(session, org_urn):
     """
     results = []
     now = datetime.now(timezone.utc)
-    # Go back 13 months to ensure we capture at least 12 full previous months
     thirteen_months_ago = now - relativedelta(months=13) 
     start_of_period = thirteen_months_ago.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
     start_ms = int(start_of_period.timestamp() * 1000)
@@ -159,8 +153,8 @@ def fetch_monthly_follower_gains(session, org_urn):
         f"{API_REST_BASE}/organizationalEntityFollowerStatistics"
         f"?q=organizationalEntity"
         f"&organizationalEntity={quote(org_urn)}"
-        f"&timeIntervals.timeGranularityType=MONTH"
-        f"&timeIntervals.timeRange.start={start_ms}"
+        f"&timeGranularity=MONTH"  # Corrected parameter name
+        f"&startTime={start_ms}"    # Corrected parameter name
     )
     logging.info(f"Fetching monthly follower gains from: {url}")
 
@@ -177,7 +171,7 @@ def fetch_monthly_follower_gains(session, org_urn):
                 continue
 
             date_obj = datetime.fromtimestamp(start_timestamp_ms / 1000, tz=timezone.utc)
-            date_str = date_obj.strftime('%Y-%m-%d') # First day of the month
+            date_str = date_obj.strftime('%Y-%m-%d') 
             
             follower_gains = item.get("followerGains", {})
             organic_gain = follower_gains.get("organicFollowerGain", 0)
@@ -188,7 +182,7 @@ def fetch_monthly_follower_gains(session, org_urn):
                 "follower_count_organic": organic_gain,
                 "follower_count_paid": paid_gain,
                 "follower_count_type": "follower_gains_monthly",
-                "organization_urn": org_urn # Add org_urn for consistency
+                "organization_urn": org_urn 
             })
         logging.info(f"Fetched {len(results)} monthly follower gain entries for org URN {org_urn}.")
     except requests.exceptions.RequestException as e:
@@ -204,9 +198,9 @@ def fetch_monthly_follower_gains(session, org_urn):
 
 def fetch_follower_demographics(session, org_urn, functions_map, seniorities_map):
     """
-    Fetches current follower demographics (seniority, industry, function, geo, association).
+    Fetches current follower demographics, applying Top-N for specified categories.
     """
-    results = []
+    final_demographics_results = []
     url = (
         f"{API_REST_BASE}/organizationalEntityFollowerStatistics"
         f"?q=organizationalEntity&organizationalEntity={quote(org_urn)}"
@@ -225,44 +219,69 @@ def fetch_follower_demographics(session, org_urn, functions_map, seniorities_map
         
         stat_element = elements[0] # Data is usually in the first element
 
-        # Collect URNs for batch mapping
-        industry_urns_to_map = [item.get("industry") for item in stat_element.get("followerCountsByIndustry", []) if item.get("industry")]
-        geo_urns_to_map = [item.get("geo") for item in stat_element.get("followerCountsByGeoCountry", []) if item.get("geo")]
-        
-        industries_map = get_industries_map(session, industry_urns_to_map)
-        geo_map = get_geo_map(session, geo_urns_to_map)
-
-        # Helper to create demographic entries
-        def _add_demographic_entry(items_list, type_name, id_map, id_field_name, org_urn_val):
-            if not items_list:
-                logging.info(f"No items found for demographic type '{type_name}' for org {org_urn_val}.")
-                return
+        # Helper to convert raw API items for a single demographic type into our structured list
+        def _get_entries_for_type(raw_items_list, type_name, id_map, id_field_name_in_item, org_urn_val):
+            current_type_entries = []
+            if not raw_items_list:
+                logging.debug(f"No raw items for demographic type '{type_name}' for org {org_urn_val}.")
+                return current_type_entries
 
-            for item in items_list:
+            for item in raw_items_list:
                 category_name_val = "Unknown"
-                if type_name == "follower_association": # associationType is directly the name
-                    category_name_val = item.get("associationType", f"Unknown AssociationType")
+                # For associationType, the id_field_name_in_item is the direct name
+                if type_name == "follower_association":
+                    category_name_val = item.get(id_field_name_in_item, f"Unknown {id_field_name_in_item}")
                 else: # For URN-based categories
-                    urn_val = item.get(id_field_name)
+                    urn_val = item.get(id_field_name_in_item)
                     entity_id = _parse_urn_to_id(urn_val)
+                    # Use str(entity_id) for map lookup as map keys were stored as strings
                     category_name_val = id_map.get(str(entity_id), f"Unknown {type_name.split('_')[-1].capitalize()} (ID: {entity_id if entity_id else urn_val})")
                 
                 counts = item.get("followerCounts", {})
-                results.append({
+                organic_count = counts.get("organicFollowerCount", 0)
+                paid_count = counts.get("paidFollowerCount", 0)
+
+                current_type_entries.append({
                     "category_name": category_name_val,
-                    "follower_count_organic": counts.get("organicFollowerCount", 0),
-                    "follower_count_paid": counts.get("paidFollowerCount", 0),
+                    "follower_count_organic": organic_count,
+                    "follower_count_paid": paid_count,
                     "follower_count_type": type_name,
                     "organization_urn": org_urn_val
                 })
+            return current_type_entries
 
-        _add_demographic_entry(stat_element.get("followerCountsByAssociationType", []), "follower_association", {}, "associationType", org_urn)
-        _add_demographic_entry(stat_element.get("followerCountsBySeniority", []), "follower_seniority", seniorities_map, "seniority", org_urn)
-        _add_demographic_entry(stat_element.get("followerCountsByFunction", []), "follower_function", functions_map, "function", org_urn)
-        _add_demographic_entry(stat_element.get("followerCountsByIndustry", []), "follower_industry", industries_map, "industry", org_urn)
-        _add_demographic_entry(stat_element.get("followerCountsByGeoCountry", []), "follower_geo", geo_map, "geo", org_urn)
+        # Fetch live maps for industries and geo as they depend on URNs from the current API response
+        industry_urns_to_map = [item.get("industry") for item in stat_element.get("followerCountsByIndustry", []) if item.get("industry")]
+        geo_urns_to_map = [item.get("geo") for item in stat_element.get("followerCountsByGeoCountry", []) if item.get("geo")]
         
-        logging.info(f"Processed follower demographics for {org_urn}. Total entries from this type: {len(results)}")
+        live_industries_map = get_industries_map(session, industry_urns_to_map)
+        live_geo_map = get_geo_map(session, geo_urns_to_map)
+
+        demographic_configs = [
+            {"items_key": "followerCountsBySeniority", "type_name": "follower_seniority", "id_map": seniorities_map, "id_field": "seniority", "top_n": 10},
+            {"items_key": "followerCountsByFunction", "type_name": "follower_function", "id_map": functions_map, "id_field": "function", "top_n": 10},
+            {"items_key": "followerCountsByIndustry", "type_name": "follower_industry", "id_map": live_industries_map, "id_field": "industry", "top_n": 10},
+            {"items_key": "followerCountsByGeoCountry", "type_name": "follower_geo", "id_map": live_geo_map, "id_field": "geo", "top_n": 10},
+            {"items_key": "followerCountsByAssociationType", "type_name": "follower_association", "id_map": {}, "id_field": "associationType", "top_n": None} # Keep all associations
+        ]
+
+        for config in demographic_configs:
+            raw_items = stat_element.get(config["items_key"], [])
+            processed_entries = _get_entries_for_type(raw_items, config["type_name"], config["id_map"], config["id_field"], org_urn)
+            
+            if config["top_n"] is not None and processed_entries:
+                # Sort by organic follower count (ensure it's numeric)
+                for entry in processed_entries: # Ensure numeric for sorting
+                    if not isinstance(entry.get("follower_count_organic"), (int, float)):
+                        entry["follower_count_organic"] = 0
+                sorted_entries = sorted(processed_entries, key=lambda x: x.get("follower_count_organic", 0), reverse=True)
+                final_demographics_results.extend(sorted_entries[:config["top_n"]])
+                logging.debug(f"Added top {config['top_n']} for {config['type_name']}. Count: {len(sorted_entries[:config['top_n']])}")
+            else:
+                final_demographics_results.extend(processed_entries) # Add all if top_n is None or no entries
+                logging.debug(f"Added all for {config['type_name']}. Count: {len(processed_entries)}")
+        
+        logging.info(f"Processed follower demographics for {org_urn}. Total entries from all types: {len(final_demographics_results)}")
 
     except requests.exceptions.RequestException as e:
         status_code = getattr(e.response, 'status_code', 'N/A')
@@ -272,7 +291,7 @@ def fetch_follower_demographics(session, org_urn, functions_map, seniorities_map
         logging.error(f"Error decoding JSON for follower demographics for {org_urn}: {e}. Response: {response.text if 'response' in locals() else 'N/A'}")
     except Exception as e:
         logging.error(f"Unexpected error fetching follower demographics for {org_urn}: {e}", exc_info=True)
-    return results
+    return final_demographics_results
 
 # --- Main Orchestration Function ---
 
@@ -287,20 +306,23 @@ def get_linkedin_follower_stats(comm_client_id, community_token, org_urn):
 
     token_dict = community_token if isinstance(community_token, dict) else {'access_token': community_token, 'token_type': 'Bearer'}
     
-    session = None # Initialize session to None
+    session = None 
     try:
         session = create_session(comm_client_id, token=token_dict)
         session.headers.update({
             "X-Restli-Protocol-Version": "2.0.0", 
-            "LinkedIn-Version": LINKEDIN_API_VERSION 
+            "LinkedIn-Version": LINKEDIN_API_VERSION,
+            # "Accept-Language": "en_US" # Consider adding if not set by create_session and locale issues persist for v2 name lookups
         })
     except Exception as e:
         logging.error(f"Failed to create session or update headers for org {org_urn}: {e}", exc_info=True)
-        return [] # Cannot proceed without a session
+        return [] 
 
     logging.info(f"Starting follower stats retrieval for org: {org_urn}")
 
     # These maps are fetched once per call to get_linkedin_follower_stats
+    # For industries and geo, the maps will be fetched live within fetch_follower_demographics
+    # as they depend on URNs from the API response itself.
     functions_map = get_functions_map(session)
     seniorities_map = get_seniorities_map(session)
     
@@ -312,6 +334,7 @@ def get_linkedin_follower_stats(comm_client_id, community_token, org_urn):
     monthly_gains = fetch_monthly_follower_gains(session, org_urn)
     all_follower_data.extend(monthly_gains)
 
+    # Pass pre-fetched function and seniority maps. Industry and Geo maps are fetched inside.
     demographics = fetch_follower_demographics(session, org_urn, functions_map, seniorities_map)
     all_follower_data.extend(demographics)