mem0ai--mem0/scripts/oss-to-platform-migrate.sh

#!/usr/bin/env bash
set -euo pipefail

if ! command -v python3 >/dev/null 2>&1; then
  printf '%s\n' "Error: python3 is required to run the Mem0 migration. Install Python 3, then rerun this command." >&2
  exit 1
fi

exec python3 - "$@" <<'PY'
from __future__ import annotations

import argparse
from collections import Counter
import getpass
import json
import os
import platform
import re
import stat
import sys
import uuid
from datetime import datetime, timezone
from hashlib import sha256
from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Tuple
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen


DEFAULT_BASE_URL = "https://api.mem0.ai"
POSTHOG_API_KEY = "phc_hgJkUVJFYtmaJqrvf6CYN67TIQ8yhXAkWzUn9AMU4yX"
POSTHOG_CAPTURE_URL = "https://us.i.posthog.com/i/v0/e/"
SCRIPT_VERSION = "oss-to-platform-migrate-v1"
EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
DEFAULT_QDRANT_COLLECTION = "mem0"
DEFAULT_EXPORT_PAGE_SIZE = 100
DEFAULT_PLATFORM_PAGE_SIZE = 100


class MigrationError(Exception):
    pass


class AuthError(MigrationError):
    pass


class HTTPStatusError(MigrationError):
    def __init__(self, status_code: int, detail: str):
        self.status_code = status_code
        self.detail = detail
        super().__init__(detail)


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")


def print_info(message: str) -> None:
    print(message)


def print_error(message: str) -> None:
    print(f"Error: {message}", file=sys.stderr)


def mem0_dir() -> Path:
    configured = os.environ.get("MEM0_DIR")
    if configured:
        return Path(configured).expanduser()
    return Path.home() / ".mem0"


def config_path() -> Path:
    return mem0_dir() / "config.json"


def load_config(path: Path) -> Dict[str, Any]:
    if not path.exists():
        return {}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}


def write_config(path: Path, config: Dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    os.chmod(path.parent, stat.S_IRWXU)
    path.write_text(json.dumps(config, indent=4) + "\n", encoding="utf-8")
    os.chmod(path, stat.S_IRUSR | stat.S_IWUSR)


def ensure_oss_anon_id(path: Path, config: Dict[str, Any]) -> str:
    existing = config.get("user_id")
    if isinstance(existing, str) and existing:
        return existing

    generated = str(uuid.uuid4())
    config["user_id"] = generated
    try:
        write_config(path, config)
    except Exception:
        # Keep the run usable even on read-only home directories. Telemetry can
        # still use the transient ID for this run.
        pass
    return generated


def read_anon_ids(config: Dict[str, Any], fallback_oss_id: str) -> Dict[str, Any]:
    telemetry = config.get("telemetry")
    if not isinstance(telemetry, dict):
        telemetry = {}
    aliased_pairs = telemetry.get("aliased_pairs")
    if not isinstance(aliased_pairs, list):
        aliased_pairs = []
    return {
        "oss": config.get("user_id") if isinstance(config.get("user_id"), str) else fallback_oss_id,
        "cli": telemetry.get("anonymous_id") if isinstance(telemetry.get("anonymous_id"), str) else None,
        "aliased_pairs": [item for item in aliased_pairs if isinstance(item, str)],
    }


def primary_anon_id(anon_ids: Dict[str, Any]) -> str:
    return anon_ids.get("oss") or anon_ids.get("cli") or "anonymous_user"


def unique_anon_ids(anon_ids: Dict[str, Any]) -> List[str]:
    seen: Set[str] = set()
    result: List[str] = []
    for anon_id in (anon_ids.get("oss"), anon_ids.get("cli")):
        if not anon_id or anon_id in seen:
            continue
        seen.add(anon_id)
        result.append(anon_id)
    return result


def alias_pair_marker(anon_id: str, email: str) -> str:
    return sha256(f"{anon_id}\0{email}".encode("utf-8")).hexdigest()


def is_aliased(config: Dict[str, Any], anon_id: str, email: str) -> bool:
    telemetry = config.get("telemetry")
    if not isinstance(telemetry, dict):
        return False
    aliased_pairs = telemetry.get("aliased_pairs")
    if not isinstance(aliased_pairs, list):
        return False
    return alias_pair_marker(anon_id, email) in aliased_pairs


def mark_aliased(path: Path, config: Dict[str, Any], anon_id: str, email: str) -> None:
    telemetry = config.get("telemetry")
    if not isinstance(telemetry, dict):
        telemetry = {}
    aliased_pairs = telemetry.get("aliased_pairs")
    if not isinstance(aliased_pairs, list):
        aliased_pairs = []
    marker = alias_pair_marker(anon_id, email)
    if marker not in aliased_pairs:
        aliased_pairs.append(marker)
    telemetry["aliased_pairs"] = aliased_pairs
    config["telemetry"] = telemetry
    write_config(path, config)


def normalize_email(email: str) -> str:
    normalized = email.strip().lower()
    if not EMAIL_RE.match(normalized):
        raise MigrationError(f"Invalid email address: {email!r}")
    return normalized


def parse_error_body(raw: bytes, fallback: str) -> str:
    text = raw.decode("utf-8", errors="replace").strip()
    if not text:
        return fallback
    try:
        parsed = json.loads(text)
        if isinstance(parsed, dict):
            for key in ("error", "detail", "message"):
                value = parsed.get(key)
                if value:
                    return str(value)
    except Exception:
        pass
    return text


def request_json(
    method: str,
    url: str,
    *,
    payload: Optional[Dict[str, Any]] = None,
    headers: Optional[Dict[str, str]] = None,
    timeout: float = 30.0,
) -> Dict[str, Any]:
    request_headers = {"Accept": "application/json"}
    if headers:
        request_headers.update(headers)

    data = None
    if payload is not None:
        data = json.dumps(payload).encode("utf-8")
        request_headers["Content-Type"] = "application/json"

    req = Request(url, data=data, headers=request_headers, method=method)
    try:
        with urlopen(req, timeout=timeout) as resp:
            raw = resp.read()
    except HTTPError as exc:
        detail = parse_error_body(exc.read(), exc.reason or f"HTTP {exc.code}")
        raise HTTPStatusError(exc.code, detail) from exc
    except URLError as exc:
        raise MigrationError(f"Network request failed for {url}: {exc.reason}. Check your connection and rerun the command.") from exc

    if not raw:
        return {}
    try:
        parsed = json.loads(raw.decode("utf-8"))
    except json.JSONDecodeError as exc:
        raise MigrationError(f"Expected JSON response from {url}") from exc
    return parsed if isinstance(parsed, dict) else {}


def post_json_best_effort(url: str, payload: Dict[str, Any], timeout: float = 5.0) -> bool:
    data = json.dumps(payload).encode("utf-8")
    req = Request(url, data=data, headers={"Content-Type": "application/json"}, method="POST")
    try:
        with urlopen(req, timeout=timeout) as resp:
            resp.read()
            return 200 <= resp.status < 300
    except Exception:
        return False


def telemetry_enabled() -> bool:
    raw = os.environ.get("MEM0_TELEMETRY", "true")
    return raw.lower() in ("true", "1", "yes")


def telemetry_url() -> str:
    return os.environ.get("MEM0_MIGRATE_TELEMETRY_URL", POSTHOG_CAPTURE_URL)


def base_event_properties(anon_ids: Dict[str, Any], email: Optional[str] = None) -> Dict[str, Any]:
    properties: Dict[str, Any] = {
        "client_source": "python",
        "client_version": SCRIPT_VERSION,
        "migration_phase": "migration",
        "local_anonymous_id": primary_anon_id(anon_ids),
        "oss_anonymous_id": anon_ids.get("oss"),
        "cli_anonymous_id": anon_ids.get("cli"),
        "python_version": sys.version,
        "os": sys.platform,
        "os_version": platform.version(),
        "os_release": platform.release(),
        "machine": platform.machine(),
        "timestamp": utc_now(),
        "$lib": "posthog-python",
    }
    if email:
        properties["authenticated_email"] = email
    return properties


def capture_migration_event(
    event_name: str,
    anon_ids: Dict[str, Any],
    *,
    email: Optional[str] = None,
    additional: Optional[Dict[str, Any]] = None,
) -> bool:
    if not telemetry_enabled():
        return False
    distinct_id = email or primary_anon_id(anon_ids)
    properties = base_event_properties(anon_ids, email)
    if additional:
        properties.update(additional)
    payload = {
        "api_key": POSTHOG_API_KEY,
        "distinct_id": distinct_id,
        "event": event_name,
        "properties": properties,
    }
    return post_json_best_effort(telemetry_url(), payload)


def capture_identify(anon_id: str, email: str) -> bool:
    if not telemetry_enabled() or not anon_id or not email or anon_id == email:
        return False
    payload = {
        "api_key": POSTHOG_API_KEY,
        "distinct_id": email,
        "event": "$identify",
        "properties": {
            "$anon_distinct_id": anon_id,
            "client_source": "python",
            "client_version": SCRIPT_VERSION,
            "$lib": "posthog-python",
        },
    }
    return post_json_best_effort(telemetry_url(), payload)


def stitch_identities(config_path_value: Path, config: Dict[str, Any], anon_ids: Dict[str, Any], email: str) -> None:
    for anon_id in unique_anon_ids(anon_ids):
        if anon_id == email or is_aliased(config, anon_id, email):
            continue
        if capture_identify(anon_id, email):
            try:
                mark_aliased(config_path_value, config, anon_id, email)
            except Exception:
                pass


def auth_headers(language: str = "python") -> Dict[str, str]:
    return {
        "X-Mem0-Source": "migration",
        "X-Mem0-Client-Language": language,
        "X-Mem0-Client-Version": SCRIPT_VERSION,
    }


def platform_headers(api_key: str, language: str = "python") -> Dict[str, str]:
    return {
        "Authorization": f"Token {api_key}",
        "X-Mem0-Source": "migration",
        "X-Mem0-Client-Language": language,
        "X-Mem0-Client-Version": SCRIPT_VERSION,
    }


def api_base_url(args: argparse.Namespace, config: Dict[str, Any]) -> str:
    platform_config = config.get("platform") if isinstance(config.get("platform"), dict) else {}
    value = args.base_url or os.environ.get("MEM0_BASE_URL") or platform_config.get("base_url") or DEFAULT_BASE_URL
    return str(value).rstrip("/")


def configured_api_key(args: argparse.Namespace, config: Dict[str, Any]) -> Tuple[Optional[str], str]:
    if args.api_key:
        return args.api_key, "flag"
    env_key = os.environ.get("MEM0_API_KEY")
    if env_key:
        return env_key, "env"
    platform_config = config.get("platform") if isinstance(config.get("platform"), dict) else {}
    file_key = platform_config.get("api_key")
    if isinstance(file_key, str) and file_key:
        return file_key, "config"
    return None, ""


def validate_api_key(api_key: str, base_url: str) -> str:
    try:
        result = request_json(
            "GET",
            f"{base_url}/v1/ping/",
            headers={"Authorization": f"Token {api_key}"},
            timeout=5.0,
        )
    except HTTPStatusError as exc:
        if exc.status_code == 401:
            raise AuthError("Invalid or expired API key.") from exc
        raise

    email = result.get("user_email")
    if not isinstance(email, str) or not email:
        raise MigrationError("Platform authenticated the API key but did not return an account email.")
    return normalize_email(email)


def prompt_line(prompt: str) -> str:
    try:
        with open("/dev/tty", "r", encoding="utf-8") as tty_in, open("/dev/tty", "w", encoding="utf-8") as tty_out:
            tty_out.write(prompt)
            tty_out.flush()
            line = tty_in.readline()
    except OSError as exc:
        raise MigrationError(
            "No interactive terminal is available. Re-run with --email and --code, or use --yes with a valid API key."
        ) from exc

    if line == "":
        raise MigrationError("No input received from terminal.")
    return line.strip()


def prompt_line_default(prompt: str, default: str) -> str:
    answer = prompt_line(f"{prompt} [{default}]: ")
    return answer or default


def prompt_secret(prompt: str) -> str:
    ensure_interactive_terminal()
    try:
        with open("/dev/tty", "w", encoding="utf-8") as tty_out:
            return getpass.getpass(prompt, stream=tty_out).strip()
    except OSError as exc:
        raise MigrationError(
            "No interactive terminal is available. Re-run with the required flags or environment variables."
        ) from exc


def has_interactive_terminal() -> bool:
    try:
        with open("/dev/tty", "r", encoding="utf-8"), open("/dev/tty", "w", encoding="utf-8"):
            return True
    except OSError:
        return False


def ensure_interactive_terminal() -> None:
    try:
        with open("/dev/tty", "r", encoding="utf-8"), open("/dev/tty", "w", encoding="utf-8"):
            return
    except OSError as exc:
        raise MigrationError(
            "No interactive terminal is available. Re-run with --email and --code, or use --yes with a valid API key."
        ) from exc


def prompt_yes_no(prompt: str, *, default: bool = True) -> bool:
    suffix = " [Y/n]: " if default else " [y/N]: "
    answer = prompt_line(prompt + suffix).strip().lower()
    if not answer:
        return default
    return answer in ("y", "yes")


def request_email_code(email: str, base_url: str) -> None:
    try:
        request_json(
            "POST",
            f"{base_url}/api/v1/auth/email_code/",
            payload={"email": email},
            headers=auth_headers(),
        )
    except HTTPStatusError as exc:
        if exc.status_code == 429:
            raise MigrationError("Too many attempts. Try again in a few minutes.") from exc
        raise MigrationError(f"Failed to send verification code: {exc.detail}") from exc


def verify_email_code(email: str, code: str, base_url: str) -> str:
    try:
        result = request_json(
            "POST",
            f"{base_url}/api/v1/auth/email_code/verify/",
            payload={"email": email, "code": code.strip()},
            headers=auth_headers(),
        )
    except HTTPStatusError as exc:
        if exc.status_code == 429:
            raise MigrationError("Too many attempts. Try again in a few minutes.") from exc
        raise MigrationError(f"Verification failed: {exc.detail}") from exc

    api_key = result.get("api_key")
    if not isinstance(api_key, str) or not api_key:
        raise MigrationError("Auth succeeded but no API key was returned. Contact support.")
    return api_key


def email_code_auth(args: argparse.Namespace, base_url: str) -> Tuple[str, str, str]:
    email = normalize_email(args.email) if args.email else normalize_email(prompt_line("Email: "))
    code = args.code

    if not code:
        ensure_interactive_terminal()
        request_email_code(email, base_url)
        print_info("Verification code sent. Check your email.")
        code = prompt_line("Verification code: ")
        if not code:
            raise MigrationError("Verification code is required.")

    api_key = verify_email_code(email, code, base_url)
    resolved_email = validate_api_key(api_key, base_url)
    return api_key, resolved_email, "email_code"


def existing_session_auth(args: argparse.Namespace, api_key: str, key_source: str, base_url: str) -> Optional[Tuple[str, str, str]]:
    try:
        email = validate_api_key(api_key, base_url)
    except AuthError:
        if key_source in ("flag", "env"):
            raise
        print_info("Stored Mem0 Platform API key is invalid or expired. Continuing with email login.")
        return None
    except MigrationError:
        if key_source in ("flag", "env"):
            raise
        print_info("Could not validate stored Mem0 Platform session. Continuing with email login.")
        return None

    if args.email:
        return None

    if not args.yes:
        print_info("")
        print_info("Found an existing Mem0 Platform session.")
        print_info("")
        print_info(f"Account: {email}")
        print_info("This migration will copy your local OSS memories into this Platform account.")
        print_info("")
        if not prompt_yes_no(f"Continue with {email}?", default=True):
            return None

    return api_key, email, "existing_api_key"


def resolve_auth(args: argparse.Namespace, config: Dict[str, Any], base_url: str) -> Tuple[str, str, str]:
    if args.api_key and args.email:
        raise MigrationError("Cannot use both --api-key and --email.")
    if args.code and not args.email:
        raise MigrationError("--code requires --email.")

    api_key, key_source = configured_api_key(args, config)
    if api_key:
        result = existing_session_auth(args, api_key, key_source, base_url)
        if result:
            return result

    return email_code_auth(args, base_url)


def default_export_path() -> Path:
    timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    return mem0_dir() / "migrations" / f"mem0-qdrant-export-{timestamp}.json"


def strip_env_assignment(value: str, name: str) -> str:
    prefix = f"{name}="
    return value[len(prefix):] if value.startswith(prefix) else value


def qdrant_url(args: argparse.Namespace) -> str:
    value = args.qdrant_url
    env_value = os.environ.get("QDRANT_URL")
    if not value and env_value:
        if has_interactive_terminal():
            normalized_env_value = strip_env_assignment(env_value.strip(), "QDRANT_URL")
            print_info("")
            print_info("Found QDRANT_URL in your environment:")
            print_info(normalized_env_value)
            if prompt_yes_no("Use this Qdrant URL?", default=True):
                value = env_value
        else:
            value = env_value
    if not value:
        if not has_interactive_terminal():
            raise MigrationError("Hosted Qdrant export requires --qdrant-url or QDRANT_URL.")
        print_info("")
        print_info("Python OSS hosted Qdrant export setup")
        print_info("Enter the Qdrant Cloud details used by your OSS Memory configuration.")
        value = prompt_line("Qdrant URL: ")
    value = strip_env_assignment(value.strip(), "QDRANT_URL")
    if not value:
        raise MigrationError("Qdrant URL is required.")
    return value.rstrip("/")


def qdrant_api_key(args: argparse.Namespace) -> str:
    value = args.qdrant_api_key
    env_value = os.environ.get("QDRANT_API_KEY")
    if not value and env_value:
        if has_interactive_terminal():
            print_info("")
            print_info("Found QDRANT_API_KEY in your environment.")
            if prompt_yes_no("Use this Qdrant API key?", default=True):
                value = env_value
        else:
            value = env_value
    if not value:
        if not has_interactive_terminal():
            raise MigrationError("Hosted Qdrant export requires --qdrant-api-key or QDRANT_API_KEY.")
        value = prompt_secret("Qdrant API key: ")
    value = strip_env_assignment(value.strip(), "QDRANT_API_KEY")
    if not value:
        raise MigrationError("Qdrant API key is required.")
    return value


def qdrant_collection(args: argparse.Namespace) -> str:
    value = args.qdrant_collection or os.environ.get("QDRANT_COLLECTION")
    if not value:
        if has_interactive_terminal():
            value = prompt_line_default("Qdrant collection", DEFAULT_QDRANT_COLLECTION)
        else:
            value = DEFAULT_QDRANT_COLLECTION
    value = value.strip()
    if not value:
        raise MigrationError("Qdrant collection name cannot be empty.")
    return value


def qdrant_headers(api_key: str) -> Dict[str, str]:
    return {"api-key": api_key}


def qdrant_request_json(
    method: str,
    url: str,
    *,
    api_key: str,
    payload: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    try:
        return request_json(method, url, payload=payload, headers=qdrant_headers(api_key), timeout=30.0)
    except HTTPStatusError as exc:
        if exc.status_code in (401, 403):
            raise MigrationError("Qdrant authentication failed. Check your QDRANT_API_KEY.") from exc
        if exc.status_code == 404:
            raise MigrationError(f"Qdrant resource not found: {url}") from exc
        raise MigrationError(f"Qdrant request failed: {exc.detail}") from exc


def preflight_qdrant_collection(base_url: str, api_key: str, collection_name: str) -> None:
    qdrant_request_json(
        "GET",
        f"{base_url}/collections/{collection_name}",
        api_key=api_key,
    )


def qdrant_filter(args: argparse.Namespace) -> Optional[Dict[str, Any]]:
    filters = []
    for key, value in (("user_id", args.user_id), ("agent_id", args.agent_id), ("run_id", args.run_id)):
        if value:
            filters.append({"key": key, "match": {"value": value}})

    if not filters:
        if args.all:
            return None
        if not has_interactive_terminal():
            raise MigrationError("Export requires --user-id, --agent-id, --run-id, or --all.")
        print_info("")
        print_info("Choose which OSS memories to export.")
        print_info("Enter a user_id to export one user, or press Enter to export the full collection.")
        user_id = prompt_line("User ID to export (press Enter for full collection): ")
        if user_id:
            args.user_id = user_id
            filters.append({"key": "user_id", "match": {"value": user_id}})
        else:
            args.all = True
            return None

    return {"must": filters}


def scroll_qdrant_points(
    base_url: str,
    api_key: str,
    collection_name: str,
    scroll_filter: Optional[Dict[str, Any]],
    page_size: int,
) -> List[Dict[str, Any]]:
    points: List[Dict[str, Any]] = []
    offset = None

    while True:
        payload: Dict[str, Any] = {
            "limit": page_size,
            "with_payload": True,
            "with_vector": False,
        }
        if scroll_filter is not None:
            payload["filter"] = scroll_filter
        if offset is not None:
            payload["offset"] = offset

        response = qdrant_request_json(
            "POST",
            f"{base_url}/collections/{collection_name}/points/scroll",
            api_key=api_key,
            payload=payload,
        )
        result = response.get("result")
        if not isinstance(result, dict):
            raise MigrationError("Qdrant scroll response did not include a result object.")

        page_points = result.get("points")
        if not isinstance(page_points, list):
            raise MigrationError("Qdrant scroll response did not include a points list.")
        points.extend(point for point in page_points if isinstance(point, dict))

        offset = result.get("next_page_offset")
        if offset is None:
            break

    return points


def normalize_qdrant_point(point: Dict[str, Any]) -> Dict[str, Any]:
    payload = point.get("payload")
    if not isinstance(payload, dict):
        payload = {}

    promoted_payload_keys = ("user_id", "agent_id", "run_id", "actor_id", "role")
    core_and_promoted_keys = {
        "data",
        "hash",
        "created_at",
        "updated_at",
        "id",
        "text_lemmatized",
        "attributed_to",
        *promoted_payload_keys,
    }

    record: Dict[str, Any] = {
        "id": str(point.get("id")),
        "memory": payload.get("data", ""),
        "hash": payload.get("hash"),
        "created_at": payload.get("created_at"),
        "updated_at": payload.get("updated_at"),
        "source": "python_oss_qdrant",
        "source_payload": payload,
    }

    for key in promoted_payload_keys:
        if key in payload:
            record[key] = payload[key]

    metadata = {key: value for key, value in payload.items() if key not in core_and_promoted_keys}
    if metadata:
        record["metadata"] = metadata

    return record


def write_export_file(output_path: Path, artifact: Dict[str, Any]) -> None:
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(json.dumps(artifact, indent=2) + "\n", encoding="utf-8")
    os.chmod(output_path, stat.S_IRUSR | stat.S_IWUSR)


def export_qdrant_memories(args: argparse.Namespace, anon_ids: Dict[str, Any]) -> Tuple[Path, int]:
    base_url = qdrant_url(args)
    api_key = qdrant_api_key(args)
    collection_name = qdrant_collection(args)
    output_path = Path(args.output).expanduser() if args.output else default_export_path()
    page_size = args.qdrant_page_size
    if page_size < 1:
        raise MigrationError("--qdrant-page-size must be greater than 0.")

    scroll_filter = qdrant_filter(args)
    preflight_qdrant_collection(base_url, api_key, collection_name)
    points = scroll_qdrant_points(base_url, api_key, collection_name, scroll_filter, page_size)
    records = [normalize_qdrant_point(point) for point in points]

    artifact = {
        "version": 1,
        "kind": "mem0_oss_qdrant_export",
        "exported_at": utc_now(),
        "source": {
            "sdk": "python",
            "vector_store": "qdrant",
            "storage": "hosted",
            "collection": collection_name,
            "url": base_url,
            "filters": {
                "user_id": args.user_id,
                "agent_id": args.agent_id,
                "run_id": args.run_id,
                "all": args.all,
            },
        },
        "local_anonymous_id": primary_anon_id(anon_ids),
        "anonymous_ids": {
            "oss": anon_ids.get("oss"),
            "cli": anon_ids.get("cli"),
        },
        "record_count": len(records),
        "records": records,
    }
    write_export_file(output_path, artifact)
    return output_path, len(records)


def platform_request_json(
    method: str,
    url: str,
    *,
    api_key: str,
    payload: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    try:
        return request_json(method, url, payload=payload, headers=platform_headers(api_key), timeout=30.0)
    except HTTPStatusError as exc:
        if exc.status_code in (401, 403):
            raise AuthError("Platform authentication failed. Check your MEM0_API_KEY or rerun email login.") from exc
        raise MigrationError(f"Platform request failed: {exc.detail}") from exc


def import_input_path(args: argparse.Namespace) -> Path:
    if not args.input:
        raise MigrationError("--input is required for --import-only.")
    return Path(args.input).expanduser()


def read_import_records(input_path: Path) -> Tuple[Dict[str, Any], List[Dict[str, Any]]]:
    try:
        parsed = json.loads(input_path.read_text(encoding="utf-8"))
    except FileNotFoundError as exc:
        raise MigrationError(f"Import file not found: {input_path}") from exc
    except json.JSONDecodeError as exc:
        raise MigrationError(f"Import file is not valid JSON: {input_path}") from exc

    if isinstance(parsed, list):
        records = parsed
        artifact: Dict[str, Any] = {}
    elif isinstance(parsed, dict):
        artifact = parsed
        raw_records = parsed.get("records")
        if not isinstance(raw_records, list):
            raise MigrationError("Import artifact must contain a records array.")
        records = raw_records
    else:
        raise MigrationError("Import file must be a JSON object or array.")

    return artifact, [record for record in records if isinstance(record, dict)]


def migration_import_key(source: Dict[str, Any], record: Dict[str, Any]) -> str:
    raw = ":".join(
        [
            str(source.get("sdk") or "unknown"),
            str(source.get("vector_store") or "unknown"),
            str(source.get("collection") or "unknown"),
            str(record.get("id") or ""),
        ]
    )
    return sha256(raw.encode("utf-8")).hexdigest()


def parse_timestamp(value: Any) -> Optional[int]:
    if not isinstance(value, str) or not value:
        return None
    try:
        return int(datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp())
    except ValueError:
        return None


def build_import_metadata(source: Dict[str, Any], record: Dict[str, Any]) -> Dict[str, Any]:
    existing_metadata = record.get("metadata") if isinstance(record.get("metadata"), dict) else {}
    metadata = dict(existing_metadata)
    metadata.update(
        {
            "mem0_migration_source": record.get("source") or "python_oss_qdrant",
            "mem0_migration_collection": source.get("collection"),
            "mem0_migration_local_id": record.get("id"),
            "mem0_migration_local_hash": record.get("hash"),
            "mem0_migration_import_key": migration_import_key(source, record),
        }
    )
    return {key: value for key, value in metadata.items() if value is not None}


def build_import_payload(source: Dict[str, Any], record: Dict[str, Any]) -> Dict[str, Any]:
    memory = record.get("memory")
    if not isinstance(memory, str) or not memory.strip():
        raise MigrationError("record is missing memory text")

    payload: Dict[str, Any] = {
        "messages": [{"role": "user", "content": memory}],
        "metadata": build_import_metadata(source, record),
        "infer": False,
        "source": "migration",
    }

    for key in ("user_id", "agent_id", "run_id"):
        value = record.get(key)
        if value:
            payload[key] = value

    timestamp = parse_timestamp(record.get("created_at"))
    if timestamp is not None:
        payload["timestamp"] = timestamp

    if not any(payload.get(key) for key in ("user_id", "agent_id", "run_id")):
        raise MigrationError("record must have at least one of user_id, agent_id, or run_id")

    return payload


def payload_scope(payload: Dict[str, Any]) -> Tuple[str, str]:
    for key in ("user_id", "agent_id", "run_id"):
        value = payload.get(key)
        if value:
            return key, str(value)
    raise MigrationError("import payload must have at least one of user_id, agent_id, or run_id")


def platform_get_all_scope(base_url: str, api_key: str, scope_field: str, scope_value: str) -> List[Dict[str, Any]]:
    memories: List[Dict[str, Any]] = []
    page = 1
    while True:
        response = platform_request_json(
            "POST",
            f"{base_url}/v3/memories/?page={page}&page_size={DEFAULT_PLATFORM_PAGE_SIZE}",
            api_key=api_key,
            payload={"filters": {scope_field: scope_value}, "source": "migration"},
        )
        page_results = response.get("results")
        if not isinstance(page_results, list):
            raise MigrationError("Platform get_all response did not include a results list.")
        memories.extend(memory for memory in page_results if isinstance(memory, dict))
        if not response.get("next"):
            return memories
        page += 1


def existing_migration_memories(
    base_url: str,
    api_key: str,
    payloads: List[Dict[str, Any]],
) -> Dict[str, Dict[str, Any]]:
    scopes: List[Tuple[str, str]] = []
    seen_scopes: Set[Tuple[str, str]] = set()
    for payload in payloads:
        scope = payload_scope(payload)
        if scope not in seen_scopes:
            seen_scopes.add(scope)
            scopes.append(scope)

    existing: Dict[str, Dict[str, Any]] = {}
    for scope_field, scope_value in scopes:
        print_info(f"Checking existing Platform memories for {scope_field}={scope_value!r}...")
        for memory in platform_get_all_scope(base_url, api_key, scope_field, scope_value):
            metadata = memory.get("metadata")
            if not isinstance(metadata, dict):
                continue
            key = metadata.get("mem0_migration_import_key")
            if isinstance(key, str) and key:
                existing[key] = memory
    return existing


def review_file_path(input_path: Path) -> Path:
    timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    return input_path.with_name(f"{input_path.stem}-import-review-{timestamp}.json")


def write_import_review_file(input_path: Path, review_records: List[Dict[str, Any]], summary: Dict[str, Any]) -> Optional[Path]:
    if not review_records:
        return None
    path = review_file_path(input_path)
    payload = {
        "version": 1,
        "kind": "mem0_platform_import_review",
        "created_at": utc_now(),
        "summary": summary,
        "records": review_records,
    }
    path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
    os.chmod(path, stat.S_IRUSR | stat.S_IWUSR)
    return path


def import_platform_memories(input_path: Path, api_key: str, base_url: str) -> Dict[str, Any]:
    artifact, records = read_import_records(input_path)
    source = artifact.get("source") if isinstance(artifact.get("source"), dict) else {}

    payloads: List[Tuple[Dict[str, Any], Dict[str, Any]]] = []
    review_records: List[Dict[str, Any]] = []
    invalid = 0
    for record in records:
        try:
            payloads.append((record, build_import_payload(source, record)))
        except MigrationError as exc:
            invalid += 1
            review_records.append({"status": "invalid", "error": str(exc), "record": record})

    existing = existing_migration_memories(base_url, api_key, [payload for _record, payload in payloads]) if payloads else {}

    imported = 0
    skipped_existing = 0
    changed_existing = 0
    failed = 0
    statuses: Counter[str] = Counter()

    for record, payload in payloads:
        key = payload["metadata"]["mem0_migration_import_key"]
        existing_memory = existing.get(key)
        if existing_memory:
            existing_metadata = existing_memory.get("metadata") if isinstance(existing_memory.get("metadata"), dict) else {}
            existing_hash = existing_metadata.get("mem0_migration_local_hash")
            local_hash = payload["metadata"].get("mem0_migration_local_hash")
            if existing_hash == local_hash:
                skipped_existing += 1
            else:
                changed_existing += 1
                review_records.append(
                    {
                        "status": "changed_existing",
                        "platform_memory_id": existing_memory.get("id"),
                        "existing_hash": existing_hash,
                        "local_hash": local_hash,
                        "record": record,
                    }
                )
            continue

        try:
            response = platform_request_json("POST", f"{base_url}/v3/memories/add/", api_key=api_key, payload=payload)
            imported += 1
            statuses[str(response.get("status", "unknown"))] += 1
        except MigrationError as exc:
            failed += 1
            review_records.append({"status": "failed", "error": str(exc), "record": record})

    summary = {
        "input_records": len(records),
        "valid_records": len(payloads),
        "invalid": invalid,
        "imported": imported,
        "skipped_existing": skipped_existing,
        "changed_existing": changed_existing,
        "failed": failed,
        "response_statuses": dict(statuses),
    }
    review_path = write_import_review_file(input_path, review_records, summary)
    summary["review_path"] = str(review_path) if review_path else None
    return summary


def print_import_summary(summary: Dict[str, Any]) -> None:
    print_info(f"Imported: {summary['imported']}")
    print_info(f"Skipped existing identical: {summary['skipped_existing']}")
    print_info(f"Changed existing: {summary['changed_existing']}")
    print_info(f"Invalid: {summary['invalid']}")
    print_info(f"Failed: {summary['failed']}")
    if summary.get("review_path"):
        print_info(f"Review file: {summary['review_path']}")


def parse_args(argv: List[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        prog="oss-to-platform-migrate.sh",
        description="Migrate Python OSS hosted-Qdrant memories to a Mem0 Platform account.",
    )
    parser.add_argument("--auth-only", action="store_true", help="Run only auth/account resolution.")
    parser.add_argument("--export-only", action="store_true", help="Run only Python OSS hosted-Qdrant JSON export.")
    parser.add_argument("--import-only", action="store_true", help="Run only Platform import from a migration JSON file.")
    parser.add_argument("--email", help="Email address for email-code authentication.")
    parser.add_argument("--code", help="Verification code for non-interactive email-code authentication.")
    parser.add_argument("--api-key", help="Platform API key to use for this run only.")
    parser.add_argument("--base-url", help=f"Mem0 Platform API base URL. Defaults to {DEFAULT_BASE_URL}.")
    parser.add_argument("--yes", action="store_true", help="Accept an existing valid Platform session without prompting.")
    parser.add_argument("--qdrant-url", help="Hosted Qdrant URL for Python OSS memory export.")
    parser.add_argument("--qdrant-api-key", help="Hosted Qdrant API key for this run only.")
    parser.add_argument("--qdrant-collection", help=f"Qdrant collection name. Defaults to {DEFAULT_QDRANT_COLLECTION}.")
    parser.add_argument("--qdrant-page-size", type=int, default=DEFAULT_EXPORT_PAGE_SIZE, help="Qdrant scroll page size.")
    parser.add_argument("--user-id", help="Export memories matching this user_id.")
    parser.add_argument("--agent-id", help="Export memories matching this agent_id.")
    parser.add_argument("--run-id", help="Export memories matching this run_id.")
    parser.add_argument("--all", action="store_true", help="Export the full Qdrant collection without a scope filter.")
    parser.add_argument("--output", help="Path for the migration JSON export file.")
    parser.add_argument("--input", help="Path to a migration JSON file to import into Platform.")
    return parser.parse_args(argv)


def main(argv: List[str]) -> int:
    args = parse_args(argv)
    exclusive_modes = [args.auth_only, args.export_only, args.import_only]
    if sum(1 for enabled in exclusive_modes if enabled) > 1:
        print_error("Use only one of --auth-only, --export-only, or --import-only.")
        return 1

    path = config_path()
    config = load_config(path)
    oss_anon_id = ensure_oss_anon_id(path, config)
    anon_ids = read_anon_ids(config, oss_anon_id)
    base_url = api_base_url(args, config)

    print_info("Mem0 OSS to Platform migration")
    if args.export_only:
        print_info("Export phase")
    elif args.import_only:
        print_info("Import phase")
    else:
        print_info("Auth phase")
    print_info("Supported environments: macOS, Linux, and Windows via WSL/Git Bash with bash and python3.")
    if args.auth_only:
        print_info("No memories are imported during this auth step.")
    elif args.export_only:
        print_info("No memories are imported to Platform during this step.")
    else:
        print_info("Memories will be imported to your authenticated Mem0 Platform account.")
    print_info("")

    capture_migration_event("oss.migrate.started", anon_ids, additional={"base_url": base_url})

    try:
        email = None
        api_key = None
        output_path = None
        export_count = 0
        import_summary: Optional[Dict[str, Any]] = None

        if args.import_only:
            input_path = import_input_path(args)
            api_key, key_source = configured_api_key(args, config)
            if not api_key:
                raise MigrationError("--import-only requires --api-key, MEM0_API_KEY, or a stored Platform API key.")
            try:
                email = validate_api_key(api_key, base_url)
            except MigrationError:
                if key_source in ("flag", "env"):
                    raise
                email = None
            if email:
                capture_migration_event(
                    "oss.migrate.authenticated",
                    anon_ids,
                    email=email,
                    additional={"auth_method": f"{key_source}_api_key", "base_url": base_url},
                )
                stitch_identities(path, config, anon_ids, email)
            print_info("Phase 1/1: Import memories into Mem0 Platform")
            import_summary = import_platform_memories(input_path, api_key, base_url)
        elif not args.export_only:
            phase_label = "Phase 1/1" if args.auth_only else "Phase 1/3"
            print_info(f"{phase_label}: Authenticate with Mem0 Platform")
            api_key, email, auth_method = resolve_auth(args, config, base_url)
            capture_migration_event(
                "oss.migrate.authenticated",
                anon_ids,
                email=email,
                additional={"auth_method": auth_method, "base_url": base_url},
            )
            stitch_identities(path, config, anon_ids, email)
            print_info(f"Authenticated as {email}")
            print_info("")

        if not args.auth_only:
            if args.export_only:
                print_info("Phase 1/1: Export Python OSS memories from hosted Qdrant")
                output_path, export_count = export_qdrant_memories(args, anon_ids)
            elif not args.import_only:
                print_info("Phase 2/3: Export Python OSS memories from hosted Qdrant")
                output_path, export_count = export_qdrant_memories(args, anon_ids)
                print_info(f"Exported {export_count} memories to {output_path}")
                print_info("")
                print_info("Phase 3/3: Import memories into Mem0 Platform")
                import_summary = import_platform_memories(output_path, api_key or "", base_url)

        if import_summary is not None:
            capture_migration_event(
                "oss.migrate.completed",
                anon_ids,
                email=email,
                additional={
                    "base_url": base_url,
                    "phase": "import",
                    "exported": export_count,
                    "imported": import_summary["imported"],
                    "skipped_existing": import_summary["skipped_existing"],
                    "changed_existing": import_summary["changed_existing"],
                    "invalid": import_summary["invalid"],
                    "failed": import_summary["failed"],
                },
            )
    except MigrationError as exc:
        capture_migration_event(
            "oss.migrate.failed",
            anon_ids,
            email=email,
            additional={"error": str(exc), "base_url": base_url},
        )
        print_error(str(exc))
        return 1

    if args.auth_only:
        print_info("Auth-only mode complete. Export and import were not run.")
    elif args.export_only:
        print_info(f"Exported {export_count} memories to {output_path}")
        print_info("Export-only mode complete. Platform import was not run.")
    elif args.import_only:
        print_import_summary(import_summary or {})
        print_info("Import-only mode complete. Export was not run.")
    else:
        print_info("")
        print_info(f"Export file: {output_path}")
        print_info(f"Exported: {export_count}")
        print_import_summary(import_summary or {})
        print_info("Migration complete.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))
PY