📊Python Google Analytics 4 (GA4) Data Extraction and Analysis Toolkit
# Configuration file for the Sphinx documentation builder.
#
# For the full list of built-in configuration values, see the documentation:
# https://www.sphinx-doc.org/en/master/usage/configuration.html
import os
import sys
from pathlib import Path

sys.path.insert(0, os.path.abspath('../..'))
sys.path.insert(0, os.path.abspath('../../pyga4'))

# -- Project information -----------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#project-information

project = 'pyga4'
copyright = '2023, Lin-jun-xiang'
author = 'Lin-jun-xiang'
release = '0.1.0'

# -- General configuration ---------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration

extensions = [
    'sphinx.ext.intersphinx',
    'sphinx.ext.autodoc',
    'sphinx.ext.autosummary',
    'sphinx.ext.mathjax',
    'sphinx.ext.viewcode',
    'sphinx_rtd_theme',
    'sphinx.ext.githubpages',
]

templates_path = ['_templates']
exclude_patterns = []

# -- Options for HTML output -------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output

html_theme = "sphinx_rtd_theme"
html_static_path = ['_static']

# Replace 'view source code' with 'Edit on Github'
# https://stackoverflow.com/questions/62904172/how-do-i-replace-view-page-source-with-edit-on-github-links-in-sphinx-rtd-th
html_context = {
  'display_github': True,
  'github_user': 'Lin-jun-xiang',
  'github_repo': 'pyga4',
  'github_version': 'main/docs/source/',
}

docs/source/conf.py
Lin-jun-xiang/pyga4
from pyga4.analytic.analytic import DeviceAnalytic, EventAnalytic, UserAnalytic
from pyga4.analytic.data_transform import Transformer

pyga4/analytic/__init__.py
Lin-jun-xiang/pyga4
from collections import Counter
from typing import Any

from pyga4.model.bigquery import Ga4Table


class BaseAnalytic:
    def __init__(self, table: Ga4Table) -> None:
        self.table = table

    @staticmethod
    def attribute_distribution(table: Ga4Table, attribute_name: str) -> Counter:
        attribute_list = getattr(table, attribute_name)
        counts = Counter(attribute_list)

        return counts
   

class UserAnalytic(BaseAnalytic):
    """The features of user"""

    def __init__(self, table: Ga4Table) -> None:
        super().__init__(table)

    @property
    def user_id_distribution(self) -> Counter:
        return self.attribute_distribution(
            self.table, 'user_id_list'
        )

    @property
    def countries_distribution(self) -> Counter:
        return self.attribute_distribution(
            self.table, 'geo_country_list'
        )


class DeviceAnalytic(BaseAnalytic):
    """The features of technology"""

    def __init__(self, table: Ga4Table) -> None:
        super().__init__(table)

    @property
    def os_distribution(self) -> Counter:
        return self.attribute_distribution(
            self.table, 'device_operating_system_list'
        )

    @property
    def category_distribution(self) -> Counter:
        return self.attribute_distribution(
            self.table, 'device_category_list'
        )

    @property
    def browser_distribution(self) -> Counter:
        return self.attribute_distribution(
            self.table, 'device_web_info_browser_list'
        )

    @property
    def mobile_brand_distribution(self) -> Counter:
        return self.attribute_distribution(
            self.table, 'device_mobile_brand_name_list'
        )

    @property
    def mobile_model_distribution(self) -> Counter:
        return self.attribute_distribution(
            self.table, 'device_mobile_model_name_list'
        )


class EventAnalytic(BaseAnalytic):
    """The features of event"""

    def __init__(self, table: Ga4Table) -> None:
        super().__init__(table)

    @property
    def pages_distribution(self) -> Counter:
        """Return most common pages for all users"""
        return self.attribute_distribution(
            self.table, 'page_location_list'
        )

    @staticmethod
    def track_user_loc():
        """Return most common pages for an user"""
        pass

    @staticmethod
    def conversion_rate(engagement_events: Any, target_events: Any) -> float:
        """Calculate the conversion rate

        If your "target" is to measure how many people `place an order`(purchased), 
        so your `ad clicks` would be the "engagement" because the orders resulting 
        from the ad are your "conversions."

        If your "target" is to measure how many people `clicking the ad`,
        then your ad clicks would become the "conversions,"
        and `ad impressions` would become the "engagement."

        Parameters
        ----------
        action_events:
            The engagement events that users perform.

        target_events:
            The target events that represent successful conversions.

        Returns
        -------
        float
            The calculated conversion rate as a percentage.

        Examples
        --------
        If `action_events` contains 1000 page views and `target_events` contains 200 purchases,
        the conversion rate would be calculated as:

        >>> conversion_rate([...,1000], [...,200])
        20.0

        If you have multiple engagement events and target events, you can calculate the combined
        conversion rate as well:

        >>> conversion_rate([1000, 500], [200, 100])
        18.18

        Note that the input lists should contain counts or frequencies of events.

    """
        return len(target_events) / len(engagement_events) * 100.0

pyga4/analytic/analytic.py
Lin-jun-xiang/pyga4
import base64
from datetime import datetime, timedelta, timezone
from typing import Union, Iterable


class Transformer:
    """Transform the data of GA4"""
    @staticmethod
    def timestamp_to_datetime(
        ga4_timestamp: Union[str, Iterable]
    ) -> Union[str, list]:
        if isinstance(ga4_timestamp, str):
            ga4_timestamp_seconds = int(ga4_timestamp) / 1e6

            # Create a datetime object in UTC timezone
            utc_datetime = datetime.fromtimestamp(ga4_timestamp_seconds, tz=timezone.utc)

            # Convert to Taiwan timezone
            taiwan_timezone = timezone(timedelta(hours=8))
            taiwan_datetime = utc_datetime.astimezone(taiwan_timezone)

            return taiwan_datetime

        timestamp_list = []
        for timestamp in ga4_timestamp:
            timestamp_list.append(Transformer.timestamp_to_datetime(timestamp))
        return timestamp_list

pyga4/analytic/data_transform.py
Lin-jun-xiang/pyga4
import pandas as pd
import matplotlib.pyplot as plt


def event_distribution(time_data: list, time_interval: str = '5T'):
    timestamps = pd.to_datetime(time_data)

    timestamps_series = pd.Series(timestamps)
    timestamps_series.index = timestamps
    event_counts = timestamps_series.groupby(pd.Grouper(freq=time_interval)).size()

    num_intervals = len(event_counts)
    display_points = 10
    display_interval = int(num_intervals / display_points)

    plt.figure(figsize=(10, 6))
    event_counts.plot(kind='bar')

    plt.xlabel('Time Interval')
    plt.ylabel('Event Count')
    plt.title('Event Counts by Time Interval')

    x_labels = [str(time.time().strftime('%H:%M')) for time in event_counts.index[::display_interval]]
    x_ticks = list(range(0, len(event_counts), display_interval))

    plt.xticks(x_ticks, x_labels, rotation=45, ha='right', fontsize=10)
    plt.tight_layout()
    plt.show()

pyga4/analytic/time_analysis.py
Lin-jun-xiang/pyga4
from pyga4.model.bigquery import BaseTable, Ga4Table

pyga4/model/__init__.py
Lin-jun-xiang/pyga4
from functools import wraps

from google.cloud import bigquery


def calculate_bytes_processed(func):
    @wraps(func)
    def wrapper(self, *args, **kwargs):
        result = func(self, *args, **kwargs)
        query_job = getattr(self, "_query_job", None)
        total_bytes_processed = query_job.total_bytes_processed / (1024 * 1024)
        print(f"Total bytes processed: {total_bytes_processed} MB")

        return result

    return wrapper


class Client:
    """Instance the bigquery client and define its attributes

    Parameters
    ----------
    query_config:
        if `dry_run=True`,
        the query will not response results,
        but return the bytes usage which indicate cost.

    Examples
    --------
    ```python
    Client().query_config.dry_run = True
    ```
    """
    query_config = bigquery.QueryJobConfig(
        dry_run = False,
        use_query_cache = False
    )
    def __init__(self, client, project_id: str) -> None:
        self.client = client
        self.project_id = project_id


class BaseTable(Client):
    """Connect to the data set from bigquery"""
    def __init__(self, client, project_id: str, dataset_name: str) -> None:
        super().__init__(client, project_id)
        self.dataset_name = dataset_name
        self._table_id = None

    @property
    def table_id(self) -> str:
        return self._table_id

    @table_id.setter
    def table_id(self, table_id) -> None:
        self._table_id = table_id

    @property
    def all_tables_list(self) -> list:
        """Return all tables id from data set"""
        dataset_ref = self.client.dataset(self.dataset_name)
        tables = list(self.client.list_tables(dataset_ref))

        return [table.table_id for table in tables]

    @calculate_bytes_processed
    def to_dataframe(self):
        query = f"""
            SELECT *
            FROM `{self.project_id}.{self.dataset_name}.{self.table_id}`
        """
        query_job = self.client.query(query)
        df = query_job.to_dataframe()

        return df


class Ga4Table(BaseTable):
    def __init__(self, client, project_id: str, dataset_name: str) -> None:
        super().__init__(client, project_id, dataset_name)
        self._query_job = None

    def _query_template(self, query_target: str) -> list:
        query = f"""
            SELECT {query_target}
            FROM `{self.project_id}.{self.dataset_name}.{self.table_id}`
        """
        self._query_job = self.client.query(query, job_config=self.query_config)
        results = self._query_job.result()

        field_name = query_target.split('.')[-1] if '.' in query_target else query_target

        return [getattr(row, field_name) for row in results]

    @property
    @calculate_bytes_processed
    def user_id_list(self) -> list:
        """Return all user_id from data table"""
        return self._query_template('user_id')

    @property
    @calculate_bytes_processed
    def event_date_list(self) -> list:
        """Return all event_date from data table"""
        return self._query_template('event_date')

    @property
    @calculate_bytes_processed
    def event_timestamp_list(self) -> list:
        """Return all event_timestamp from data table"""
        return self._query_template('event_timestamp')

    @property
    @calculate_bytes_processed
    def event_name_list(self) -> list:
        """Return all event_name from data table"""
        return self._query_template('event_name')

    @property
    @calculate_bytes_processed
    def device_category_list(self) -> list:
        """Return all device_category from data table
        ex: mobile, desktop
        """
        return self._query_template('device.category')

    @property
    @calculate_bytes_processed
    def device_mobile_brand_name_list(self) -> list:
        """Return all device_mobile_brand_name from data table
        ex: Apple
        """
        return self._query_template('device.mobile_brand_name')

    @property
    @calculate_bytes_processed
    def device_mobile_model_name_list(self) -> list:
        """Return all device_mobile_model_name from data table
        ex: iPhone
        """
        return self._query_template('device.mobile_model_name')

    @property
    @calculate_bytes_processed
    def device_operating_system_list(self) -> list:
        """Return all device_operating_system from data table
        ex: iOS
        """
        return self._query_template('device.operating_system')

    @property
    @calculate_bytes_processed
    def device_language_list(self) -> list:
        """Return all device_language from data table"""
        return self._query_template('device.language')

    @property
    @calculate_bytes_processed
    def device_language_list(self) -> list:
        """Return all device_language from data table"""
        return self._query_template('device.language')

    @property
    @calculate_bytes_processed
    def device_web_info_browser_list(self) -> list:
        """Return all device_web_info_browser from data table"""
        return self._query_template('device.web_info.browser')

    @property
    @calculate_bytes_processed
    def geo_country_list(self) -> list:
        """Return all user countries from data table"""
        return self._query_template('geo.country')

    @property
    @calculate_bytes_processed
    def geo_region_list(self) -> list:
        """Return all user region from data table"""
        return self._query_template('geo.region')

    @property
    @calculate_bytes_processed
    def geo_city_list(self) -> list:
        """Return all user city from data table"""
        return self._query_template('geo.city')

    @property
    @calculate_bytes_processed
    def geo_sub_continent_list(self) -> list:
        """Return all user sub_continent from data table"""
        return self._query_template('geo.sub_continent')

    @property
    @calculate_bytes_processed
    def geo_metro_list(self) -> list:
        """Return all user metro from data table"""
        return self._query_template('geo.metro')

    @property
    @calculate_bytes_processed
    def page_location_list(self) -> list:
        """Return all event of page location from data table"""
        query = f"""
            SELECT
                param.value.string_value AS param_string_value
            FROM
                `{self.project_id}.{self.dataset_name}.{self.table_id}`,
                UNNEST(event_params) AS param
            WHERE
                param.key = "page_location" OR param.key IS NULL
        """
        self._query_job = self.client.query(query, job_config=self.query_config)
        results = self._query_job.result()

        return [row.param_string_value for row in results]

    @calculate_bytes_processed
    def query(self, query: str) -> list:
        """Custom query from data table"""
        self._query_job = self.client.query(query, job_config=self.query_config)
        results = self._query_job.result()

        return results

pyga4/model/bigquery.py
Lin-jun-xiang/pyga4