import json
import logging
from datetime import date, datetime
from typing import Optional, List, Iterator, Tuple, Dict, Any

from google.cloud import bigquery
from google.oauth2 import service_account
from django.conf import settings
from django.db import connection

logger = logging.getLogger(__name__)


GCP_PROJECT = "ee-internet"
GCP_GA4_LOCATION = "US"
GCP_GA4_DATASET = "analytics_495069065"


def _get_bigquery_client() -> bigquery.Client:
    credentials = service_account.Credentials.from_service_account_info(
        json.loads(settings.GA_GOOGLE_CREDENTIALS_JSON)
    )
    return bigquery.Client(
        project=GCP_PROJECT,
        location=GCP_GA4_LOCATION,
        credentials=credentials,
    )


def _build_ga_table_ref(date_str: str, intraday: bool) -> str:
    suffix = f"events_intraday_{date_str}" if intraday else f"events_{date_str}"
    return f"`{GCP_PROJECT}.{GCP_GA4_DATASET}.{suffix}`"


BASE_SELECT = """
SELECT
  event_date,
  event_timestamp,
  event_name,
  user_pseudo_id,
  user_id,
  (SELECT value.int_value FROM UNNEST(event_params) WHERE key = 'ga_session_id') AS ga_session_id,
  (SELECT value.int_value FROM UNNEST(event_params) WHERE key = 'ga_session_number') AS ga_session_number,
  platform,
  traffic_source,
  device,
  geo,
  event_params
FROM {table_ref}
WHERE event_date = @date
  AND user_id IS NOT NULL
  AND user_id != ''
"""


def _iterate_rows_paged(
    bq_client: bigquery.Client,
    table_ref: str,
    date_str: str,
    page_size: int = 1000,
    where: str = "",
    query_params: Optional[List[bigquery.ScalarQueryParameter]] = None,
) -> Iterator[bigquery.Row]:
    query = BASE_SELECT.format(table_ref=table_ref)
    if where:
        query += f" {where}"

    job_config = bigquery.QueryJobConfig(
        query_parameters=[bigquery.ScalarQueryParameter("date", "STRING", date_str)]
        + (query_params or [])
    )
    it = bq_client.query(query, job_config=job_config).result(page_size=page_size)
    for page in it.pages:
        for row in page:
            yield row


def _fetch_rows_from_ga_table(
    bq_client: bigquery.Client,
    table_ref: str,
    date_str: str,
    hash_shards: int = 16,
    page_size: int = 1000,
) -> Iterator[bigquery.Row]:
    where = "AND MOD(ABS(FARM_FINGERPRINT(COALESCE(user_pseudo_id, ''))), @shard_count) = @shard"
    for shard in range(hash_shards):
        params = [
            bigquery.ScalarQueryParameter("shard_count", "INT64", hash_shards),
            bigquery.ScalarQueryParameter("shard", "INT64", shard),
        ]
        for row in _iterate_rows_paged(
            bq_client, table_ref, date_str, page_size, where, params
        ):
            yield row


def _pick_param_value(v: Dict[str, Any]) -> Any:
    # GA4 BigQuery export has value.{string_value,int_value,float_value,double_value}
    for k in ("string_value", "int_value", "double_value", "float_value"):
        if k in v and v[k] is not None:
            return v[k]
    return None


def _params_array_to_obj(arr: Optional[List[Dict[str, Any]]]) -> Dict[str, Any]:
    out = {}
    if not arr:
        return out
    for item in arr:
        key = item.get("key")
        val = item.get("value", {})
        out[key] = _pick_param_value(val)
    return out


def _extract_common_fields(event_params: Dict[str, Any]):
    page_location = event_params.get("page_location")
    page_referrer = event_params.get("page_referrer")
    return page_location, page_referrer


def _microseconds_to_datetime(microseconds: int) -> datetime:
    return datetime.fromtimestamp(microseconds / 1_000_000)


INSERT_SQL_TEMPLATE = """
    INSERT INTO user_events(
      event_source, created_at, user_id, event_name, event_body,
      page_location, page_referrer, device, geo, user_pseudo_id,
      lab_code
    ) VALUES (
      0, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
    )
    ON DUPLICATE KEY UPDATE
      event_body=VALUES(event_body),
      page_location=VALUES(page_location),
      page_referrer=VALUES(page_referrer),
      device=VALUES(device),
      geo=VALUES(geo),
      user_pseudo_id=VALUES(user_pseudo_id),
      lab_code=VALUES(lab_code);
    """


def _build_insert_rows(bq_row) -> Tuple:
    ep_obj = _params_array_to_obj(bq_row.get("event_params"))

    page_location, page_referrer = _extract_common_fields(ep_obj)

    event_name = bq_row["event_name"]
    event_timestamp = int(bq_row["event_timestamp"])
    event_datetime = _microseconds_to_datetime(event_timestamp)
    user_id = int(bq_row["user_id"])
    lab_code = ep_obj.get("lab_code", "")[:32]

    record = (
        event_datetime,
        user_id,
        event_name,
        json.dumps(ep_obj),
        (page_location or "")[:255],
        (page_referrer or "")[:255],
        json.dumps(bq_row.get("device")) if bq_row.get("device") is not None else None,
        json.dumps(bq_row.get("geo")) if bq_row.get("geo") is not None else None,
        bq_row.get("user_pseudo_id", "")[:128],
        lab_code,
    )
    return record


def sync_ga_events_from_bigquery(
    event_date: date,
    intraday: bool = False,
    sql_batch_size: int = 200,
    bq_batch_size: int = 200,
):
    try:
        bq_client = _get_bigquery_client()
    except Exception as e:
        logger.error(f"Failed to get bigquery client, error: {e}")
        return

    batch = []
    count = 0
    event_date_str = event_date.strftime("%Y%m%d")
    table_ref = _build_ga_table_ref(event_date_str, intraday)
    for row in _fetch_rows_from_ga_table(
        bq_client=bq_client,
        table_ref=table_ref,
        date_str=event_date_str,
        page_size=bq_batch_size,
    ):
        record = _build_insert_rows(row)
        batch.append(record)
        if len(batch) >= sql_batch_size:
            count += len(batch)
            with connection.cursor() as cur:
                cur.executemany(INSERT_SQL_TEMPLATE, batch)
            batch = []

    if batch:
        count += len(batch)
        with connection.cursor() as cur:
            cur.executemany(INSERT_SQL_TEMPLATE, batch)
    logger.info(f"Synced {count} events from GA for {event_date_str}")