Create linkedin_follower_stats.py

linkedin_follower_stats.py

@@ -0,0 +1,320 @@
+# -- coding: utf-8 --
+import json
+import requests
+import logging
+from datetime import datetime, timezone
+from dateutil.relativedelta import relativedelta # For robust month arithmetic
+from urllib.parse import quote
+
+# Assuming you have a sessions.py with create_session
+# If sessions.py or create_session is not found, it will raise an ImportError,
+# which is appropriate for a module that depends on it.
+from sessions import create_session
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+
+API_V2_BASE = 'https://api.linkedin.com/v2'
+API_REST_BASE = "https://api.linkedin.com/rest"
+LINKEDIN_API_VERSION = "202502" # As per user's example for follower stats
+
+# --- ID to Name Mapping Helper Functions ---
+
+def _fetch_linkedin_names(session, url, params, result_key_path, name_key_path, id_key="id"):
+    """
+    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:
+        logging.debug(f"Fetching names from URL: {url} with params: {params}")
+        response = session.get(url, params=params)
+        response.raise_for_status()
+        data = response.json()
+
+        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
+                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
+
+        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
+                    if isinstance(name, dict):
+                        name = name.get(key_nav)
+                    else:
+                        name = None # Path broken
+                        break
+                if name:
+                    mapping[item_id_str] = name 
+                else:
+                    logging.warning(f"No name found for ID {item_id_str} at path {name_key_path} in item: {item_data} from URL {url}")
+        elif isinstance(items, list): # For list responses like functions/seniorities
+            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
+                    if isinstance(name, dict):
+                        name = name.get(key_nav)
+                    else:
+                        name = None # Path broken
+                        break
+                if item_id_val is not None and name:
+                    mapping[str(item_id_val)] = name # Ensure ID is string for consistency
+                else:
+                    logging.warning(f"No ID ('{id_key}') or name found at path {name_key_path} in item: {item} from URL {url}")
+        else:
+            logging.warning(f"Expected list or dict of items at {result_key_path} from URL {url}, got {type(items)}")
+            
+    except requests.exceptions.RequestException as e:
+        status_code = getattr(e.response, 'status_code', 'N/A')
+        error_text = getattr(e.response, 'text', str(e))
+        logging.error(f"Error fetching names from {url} (Status: {status_code}): {error_text}")
+    except json.JSONDecodeError as e:
+        logging.error(f"Error decoding JSON for names from {url}: {e}. Response: {response.text if 'response' in locals() else 'N/A'}")
+    except Exception as e:
+        logging.error(f"Unexpected error fetching names from {url}: {e}", exc_info=True)
+    return mapping
+
+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'} 
+    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'}
+    logging.info("Fetching all LinkedIn seniorities.")
+    return _fetch_linkedin_names(session, url, params, ["elements"], ["name", "localized", "en_US"], "id")
+
+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
+    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'}
+    logging.info(f"Fetching names for {len(unique_ids)} unique industry IDs.")
+    return _fetch_linkedin_names(session, url, params, ["results"], ["name", "localized", "en_US"])
+
+
+def get_geo_map(session, geo_urns):
+    """Fetches names for a list of geo URNs. Returns a map {id: name}."""
+    if not geo_urns: return {}
+    geo_ids = [_parse_urn_to_id(urn) for urn in geo_urns if urn]
+    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"
+    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}")
+        return None
+    try:
+        return urn_string.split(':')[-1]
+    except IndexError: # Handle cases where split doesn't yield enough parts
+        logging.warning(f"Could not parse ID from URN: {urn_string}")
+        return None
+    except Exception as e:
+        logging.error(f"Unexpected error parsing URN {urn_string}: {e}")
+        return None
+
+# --- Follower Data Fetching Functions ---
+
+def fetch_monthly_follower_gains(session, org_urn):
+    """
+    Fetches monthly follower gains for the last 12-13 months to ensure full coverage.
+    """
+    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)
+    
+    url = (
+        f"{API_REST_BASE}/organizationalEntityFollowerStatistics"
+        f"?q=organizationalEntity"
+        f"&organizationalEntity={quote(org_urn)}"
+        f"&timeIntervals.timeGranularityType=MONTH"
+        f"&timeIntervals.timeRange.start={start_ms}"
+    )
+    logging.info(f"Fetching monthly follower gains from: {url}")
+
+    try:
+        response = session.get(url)
+        response.raise_for_status()
+        data = response.json()
+
+        for item in data.get("elements", []):
+            time_range = item.get("timeRange", {})
+            start_timestamp_ms = time_range.get("start")
+            if start_timestamp_ms is None:
+                logging.warning("Skipping item due to missing start timestamp in monthly gains.")
+                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
+            
+            follower_gains = item.get("followerGains", {})
+            organic_gain = follower_gains.get("organicFollowerGain", 0)
+            paid_gain = follower_gains.get("paidFollowerGain", 0)
+
+            results.append({
+                "category_name": date_str,
+                "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
+            })
+        logging.info(f"Fetched {len(results)} monthly follower gain entries for org URN {org_urn}.")
+    except requests.exceptions.RequestException as e:
+        status_code = getattr(e.response, 'status_code', 'N/A')
+        error_text = getattr(e.response, 'text', str(e))
+        logging.error(f"Error fetching monthly follower gains for {org_urn} (Status: {status_code}): {error_text}")
+    except json.JSONDecodeError as e:
+        logging.error(f"Error decoding JSON for monthly follower gains for {org_urn}: {e}. Response: {response.text if 'response' in locals() else 'N/A'}")
+    except Exception as e:
+        logging.error(f"Unexpected error fetching monthly follower gains for {org_urn}: {e}", exc_info=True)
+    return results
+
+
+def fetch_follower_demographics(session, org_urn, functions_map, seniorities_map):
+    """
+    Fetches current follower demographics (seniority, industry, function, geo, association).
+    """
+    results = []
+    url = (
+        f"{API_REST_BASE}/organizationalEntityFollowerStatistics"
+        f"?q=organizationalEntity&organizationalEntity={quote(org_urn)}"
+    )
+    logging.info(f"Fetching follower demographics from: {url} for org URN {org_urn}")
+
+    try:
+        response = session.get(url)
+        response.raise_for_status()
+        data = response.json()
+        
+        elements = data.get("elements", [])
+        if not elements:
+            logging.warning(f"No elements found in follower demographics response for {org_urn}.")
+            return []
+        
+        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
+
+            for item in 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")
+                else: # For URN-based categories
+                    urn_val = item.get(id_field_name)
+                    entity_id = _parse_urn_to_id(urn_val)
+                    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({
+                    "category_name": category_name_val,
+                    "follower_count_organic": counts.get("organicFollowerCount", 0),
+                    "follower_count_paid": counts.get("paidFollowerCount", 0),
+                    "follower_count_type": type_name,
+                    "organization_urn": org_urn_val
+                })
+
+        _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)
+        
+        logging.info(f"Processed follower demographics for {org_urn}. Total entries from this type: {len(results)}")
+
+    except requests.exceptions.RequestException as e:
+        status_code = getattr(e.response, 'status_code', 'N/A')
+        error_text = getattr(e.response, 'text', str(e))
+        logging.error(f"Error fetching follower demographics for {org_urn} (Status: {status_code}): {error_text}")
+    except json.JSONDecodeError as e:
+        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
+
+# --- Main Orchestration Function ---
+
+def get_linkedin_follower_stats(comm_client_id, community_token, org_urn):
+    """
+    Main function to fetch all follower statistics (monthly gains and demographics)
+    and format them for Bubble.
+    """
+    if not all([comm_client_id, community_token, org_urn]):
+        logging.error("Client ID, token, or Organization URN is missing for get_linkedin_follower_stats.")
+        return []
+
+    token_dict = community_token if isinstance(community_token, dict) else {'access_token': community_token, 'token_type': 'Bearer'}
+    
+    session = None # Initialize session to 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 
+        })
+    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
+
+    logging.info(f"Starting follower stats retrieval for org: {org_urn}")
+
+    # These maps are fetched once per call to get_linkedin_follower_stats
+    functions_map = get_functions_map(session)
+    seniorities_map = get_seniorities_map(session)
+    
+    if not functions_map: logging.warning(f"Functions map is empty for org {org_urn}. Function names might not be resolved.")
+    if not seniorities_map: logging.warning(f"Seniorities map is empty for org {org_urn}. Seniority names might not be resolved.")
+
+    all_follower_data = []
+
+    monthly_gains = fetch_monthly_follower_gains(session, org_urn)
+    all_follower_data.extend(monthly_gains)
+
+    demographics = fetch_follower_demographics(session, org_urn, functions_map, seniorities_map)
+    all_follower_data.extend(demographics)
+
+    logging.info(f"Successfully compiled {len(all_follower_data)} total follower stat entries for {org_urn}.")
+    return all_follower_data
+