Karakeep migrate highlights

_readme.md

Migrate Karakeep highlights

1. Obtain API tokens for source and target systems
2. Export env variables:

   export KARAKEEP_SRC_SERVER="http://source.xyz:9000"
   export KARAKEEP_DST_SERVER="http://karakeep.dest.com"
   export KARAKEEP_SRC_API_TOKEN="ak2_..."
   export KARAKEEP_DST_API_TOKEN="ak2_..."

3. Run script:

   python migrate_highlights.py

Example output

PHASE 1: populate highlights
PHASE 2: get all bookmarks
PHASE 3: match bookmarks
PHASE 4: add 79 highlights

Adding highlight 51/79
Skipping highlight ajko4uzq05vgfie35btcjyiq: no matching target bookmark found
Adding highlight 79/79
Done

migrate_highlights.py

import os

import requests

SRC_SERVER = os.getenv("KARAKEEP_SRC_SERVER")
SRC_TOKEN = os.getenv("KARAKEEP_SRC_API_TOKEN")
DST_SERVER = os.getenv("KARAKEEP_DST_SERVER")
DST_TOKEN = os.getenv("KARAKEEP_DST_API_TOKEN")


def get_highlights(limit=20, cursor=None):
    url = f"{SRC_SERVER}/api/v1/highlights"
    params = {"limit": limit}
    if cursor:
        params["cursor"] = cursor
    headers = {
        "Authorization": f"Bearer {SRC_TOKEN}",
        "Content-Type": "application/json",
    }
    r = requests.get(url, params=params, headers=headers)
    r.raise_for_status()
    return r.json()


def add_highlight(data):
    url = f"{DST_SERVER}/api/v1/highlights"
    headers = {
        "Authorization": f"Bearer {DST_TOKEN}",
        "Accept": "application/json",
        "Content-Type": "application/json",
    }
    r = requests.post(url, headers=headers, json=data)
    r.raise_for_status()
    return r.json()


def get_bookmarks(server, token, limit=10, cursor=None):
    url = f"{server}/api/v1/bookmarks"
    params = {"limit": limit}
    if cursor:
        params["cursor"] = cursor
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
    }
    r = requests.get(url, params=params, headers=headers)
    r.raise_for_status()
    return r.json()


def populate_all(field, fn):
    result = []
    cursor = None
    while True:
        res = fn(cursor)
        page = res.get(field, [])
        result.extend(page)
        cursor = res.get("nextCursor")
        if not cursor:
            return result


def get_type(bm):
    return bm.get("content", {}).get("type")


def main():
    print("PHASE 1: populate highlights")
    highlights = populate_all(
        "highlights",
        lambda cur: get_highlights(cursor=cur),
    )

    print("PHASE 2: get all bookmarks")
    src_bookmarks = populate_all(
        "bookmarks",
        lambda cur: get_bookmarks(SRC_SERVER, SRC_TOKEN, cursor=cur),
    )
    dst_bookmarks = populate_all(
        "bookmarks",
        lambda cur: get_bookmarks(DST_SERVER, DST_TOKEN, cursor=cur),
    )

    print("PHASE 3: match bookmarks")
    # filter only bookmarks with .content.type = "link"
    src_bookmarks = [bm for bm in src_bookmarks if get_type(bm) == "link"]
    dst_bookmarks = [bm for bm in dst_bookmarks if get_type(bm) == "link"]
    src_by_url = {bm["content"]["url"]: bm["id"] for bm in src_bookmarks}
    src_by_id = {v: k for k, v in src_by_url.items()}
    dst_by_url = {bm["content"]["url"]: bm["id"] for bm in dst_bookmarks}

    total = len(highlights)
    print(f"PHASE 4: add {total} highlights\n")
    for i, highlight in enumerate(highlights, 1):
        src_id = highlight["bookmarkId"]
        src_url = src_by_id.get(src_id)
        dst_id = dst_by_url.get(src_url)
        if not dst_id:
            print(f"\nSkipping highlight {src_id}: no matching target bookmark found")
            continue

        print(f"Adding highlight {i}/{total}", end="\r", flush=True)
        data = {
            k: v for k, v in highlight.items() if k not in ["id", "userId", "createdAt"]
        }
        data["bookmarkId"] = dst_id
        try:
            add_highlight(data)
        except Exception as e:
            print(f"\nError adding highlight {src_id}: {e}")

    print("\nDone")


if __name__ == "__main__":
    main()