Migrate from Google Analytics to PostHog

from google.cloud import bigquery
from google.cloud.bigquery import Row

# Set the project ID explicitly
project_id = 'cool'

client = bigquery.Client(project=project_id)

QUERY = (
    'SELECT * FROM `analytics_123456789.events_20240731` '
    'WHERE event_name NOT IN ("session_start", "first_visit") '
    'LIMIT 100')
query_job = client.query(QUERY)
rows = query_job.result()

from posthog import Posthog
from datetime import datetime
from google.cloud import bigquery


posthog = Posthog(
  '<ph_project_api_key>', 
  host='https://us.i.posthog.com',
  debug=True,
  historical_migration=True
)

key_mapping = {
  'page_title': 'title',
  'page_location': '$current_url',
  'page_referrer': '$referrer',
  'category': '$device_type',
  'operating_system': '$os',
  'operating_system_version': '$os_version',
  'language': '$browser_language',
  'web_info.browser': '$browser',
  'web_info.browser_version': '$browser_version',
  'web_info.hostname': '$host',
  'city': '$geoip_city_name',
  'country': '$geoip_country_name',
  'continent': '$geoip_continent_name',
  'region': '$geoip_subdivision_1_name'
}

omitted_keys = [
  'event_date',
  'event_timestamp',
  'event_name',
  'event_bundle_sequence_id',
  'event_server_timestamp_offset',
  'user_id',
  'user_pseudo_id',
  'stream_id',
  'platform',
  'batch_event_index',
  'batch_page_id',
  'batch_ordering_id',
  'session_traffic_source_last_click'
]

def get_record_key_value(param):
  record_key = param['key']
  value_dict = param['value']
  record_value = next((v for v in value_dict.values() if v is not None), None)
  return record_key, record_value

def flatten_row(row):
  flat_row = {}
  for key, value in row.items():
    print(f"{key}: {value}")
    if value == None:
      continue
    elif isinstance(value, list) and len(value) == 0:
      continue
    elif key in omitted_keys:
      continue
    elif key in ['user_properties', 'user_first_touch_timestamp', 'user_ltv']:
      # We'll handle user properties separately
      continue
    elif key in ['event_params', 'items']:
      # Flatten records
      for param in value:
        record_key, record_value = get_record_key_value(param)
        flat_row[record_key] = record_value
    elif key in ['privacy_info', 'device', 'geo', 'app_info', 'collected_traffic_source', 'traffic_source', 'event_dimensions', 'ecommerce']:
      # Flatten dicts and nested dicts
      for key, val in value.items():
        if isinstance(val, dict):
          for sub_key, sub_val in val.items():
            # We might not want to do this
            flat_row[f"{key}.{sub_key}"] = sub_val
        else:
          flat_row[key] = val
    else:
      flat_row[key] = value
  return flat_row

def get_person_properties(row):
  person_properties = {}
  user_properties = row.get('user_properties', [])
  for prop in user_properties:
    key, value = get_record_key_value(prop)
    person_properties[key] = value
  user_first_touch_timestamp = row.get('user_first_touch_timestamp')
  user_ltv = row.get('user_ltv')
  if user_first_touch_timestamp is not None:
    person_properties['user_first_touch_timestamp'] = user_first_touch_timestamp
  if user_ltv is not None:
    person_properties['user_ltv'] = user_ltv
  return person_properties

def get_properties(row):
  for key, value in flat_row.items():
    if value == None:
      continue
    elif key in omitted_keys:
      continue
    elif value in ['(not provided)', '(not set)']:
      continue
    elif key in key_mapping:
      properties[key_mapping[key]] = value
    else:
      properties[key] = value
  return properties

# Get events from BigQuery
project_id = 'cool'
client = bigquery.Client(project=project_id)

QUERY = (
    'SELECT * FROM `analytics_123456789.events_20240731` '
    'WHERE event_name NOT IN ("session_start", "first_visit") '
    'LIMIT 100')
query_job = client.query(QUERY)
rows = query_job.result()

# Convert and capture events in PostHog
for row in rows:
  distinct_id = row.get('user_id') or row.get('user_pseudo_id')

  ph_event_name = row.get('event_name')
  if ph_event_name == 'page_view':
    ph_event_name = '$pageview'
  if ph_event_name == 'user_engagement':
    ph_event_name = '$autocapture'

  event_timestamp = row.get('event_timestamp')
  ph_timestamp = datetime.fromtimestamp(event_timestamp / 1000000)

  flat_row = flatten_row(row)
  properties = get_properties(flat_row)
  properties['$set'] = get_person_properties(row)

  posthog.capture(
    distinct_id=distinct_id,
    event=ph_event_name,
    properties=properties,
    timestamp=ph_timestamp
  )