导出 Mixpanel 中的所有用户

Question

我尝试导出所有用户；但我无法这样做。我尝试根据页数导出；但没有成功。 此代码仅导出 1000 个用户并返回 Error500。

import requests
import json

url = "https://mixpanel.com/api/2.0/engage?project_id="

headers = {
    "accept": "application/json",
    "content-type": "application/x-www-form-urlencoded",
    "authorization": "Basic "
}

all_users = []

page = 0
while True:
    payload = {
        "include_all_users": True,
        "page": page
    }

    response = requests.post(url, data=payload, headers=headers)

    if response.status_code == 200:
        data_json = json.loads(response.text)
        users = data_json['results']

        if not users:
            break

        all_users.extend(users)
        page += 1
    else:
        print("Error:", response.status_code)
        break

with open('exported_User_data.json', 'w', encoding='utf-8') as jsonfile:
    json.dump(all_users, jsonfile, ensure_ascii=False)

print(f"Exported {len(all_users)} users to exported_User_data.json")

非常感谢您帮助解决这个问题。谢谢！

Answer 1

希望您已经解决了这个问题。如果您不这样做，我将与您分享我用来导出所有用户数据的解决方案。它非常相似，但根据 Query Profile 文档，您需要在后续请求中同时提供

session_id

和

page

参数以继续迭代，直到完成。

import argparse
import json
import logging
import math
from datetime import datetime
from typing import Any, Dict

import requests

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s:%(lineno)d - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)


def export_all_users(
    project_id: int,
    output_filename: str,
    auth_key: str,
    limit_pages: int,
) -> Dict[str, Any]:
    url = f"https://mixpanel.com/api/query/engage?project_id={project_id}"
    headers = {
        "authorization": f"Basic {auth_key}",
        "accept": "application/json",
        "content-type": "application/x-www-form-urlencoded",
    }

    all_users = []
    total_users = 0
    page = 0
    iterations = 0
    payload = {"include_all_users": True}
    while True:
        response = requests.post(url=url, headers=headers, json=payload)

        logger.info(f"Page: {page + 1} - status code: {response.status_code}")
        if response.status_code == 200:
            resp_json = response.json()
        else:
            logger.error(f"{response.status_code}: {response.text}")
            break

        if page == 0:
            total_users = resp_json["total"]
            iterations = math.ceil(total_users / 1000)
            payload["session_id"] = resp_json["session_id"]

        data = resp_json["results"]
        if not data:
            break

        all_users += data
        page += 1
        if page > iterations:
            break
        payload["page"] = page

        if limit_pages > 0 and limit_pages == page:
            break

    export_filepath = f"data/{output_filename}"
    with open(export_filepath, "w", encoding="utf-8") as file:
        json.dump(all_users, file, ensure_ascii=False)

    logger.info(f"Exported {len(all_users)} users to {export_filepath}")
    return {"filepath": export_filepath, "total_users_export": len(all_users)}


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Export all users from Mixpanel")
    parser.add_argument("--key", type=str, help="Mixpanel API key", required=True)
    parser.add_argument(
        "--project_id",
        type=int,
        help="Mixpanel project ID",
        required=True,
    )
    parser.add_argument(
        "--limit_pages",
        type=int,
        help="Limit the number of pages to export",
        default=0,
        required=False,
    )
    args = parser.parse_args()

    _now = datetime.now().strftime("%Y%m%d%H%M%S%f")
    _output_filename = f"mixpanel_users_export_{_now}.json"

    export_metadata = export_all_users(
        project_id=args.project_id,
        output_filename=_output_filename,
        auth_key=args.key,
        limit_pages=args.limit_pages,
    )

在终端中我只需运行：

python -m mixpanel__user_profiles --key <mixpanel_api_key> --project_id <project_id>