import asyncio
import functools
import json
import os

import aiohttp
import boto3
import pandas as pd

S3_BUCKET = os.getenv("S3_BUCKET")
MIXPANEL_PROJECT = os.getenv("MIXPANEL_PROJECT")
MIXPANEL_USER = os.getenv("MIXPANEL_USER")
MIXPANEL_PASS = os.getenv("MIXPANEL_PASS")


async def sync_mixpanel(date, checkpoint=None):
    s3 = boto3.client("s3")

    async with aiohttp.ClientSession(trust_env=True) as session:
        response = await session.get(
            "https://data-eu.mixpanel.com/api/2.0/export",
            auth=aiohttp.BasicAuth(MIXPANEL_USER, MIXPANEL_PASS),
            params={
                "project_id": MIXPANEL_PROJECT,
                "from_date": date.strftime("%Y-%m-%d"),
                "to_date": date.strftime("%Y-%m-%d"),
            },
        )
        text = await response.text()

    xs = []
    for line in text.splitlines():
        data = json.loads(line)
        data["properties"]["event"] = data["event"]
        xs.append(data["properties"])

    buf = pd.DataFrame.from_records(xs).to_csv(index=False)
    await asyncio.get_running_loop().run_in_executor(
        None,
        functools.partial(
            s3.put_object,
            Body=buf,
            Bucket=S3_BUCKET,
            Key=f"data/mixpanel/{date}.csv",
            ContentType="text/csv",
        ),
    )
    print(f"sync data/mixpanel/{date}.csv success")

    # make sure this is an async generator
    yield None