pinboard_linkding_migration.py

import datetime
import json
import os

import grequests

TIME_SPEC = "%Y-%m-%dT%H:%M:%SZ"
YOUR_LINKDING_URL = "https://linkding.example.com/"
LINKDING_ENDPOINT_URL = YOUR_LINKDING_URL + "api/bookmarks/"


class PinboardEntry:
    @staticmethod
    def parse_tags(tag_str):
        return tag_str.split(" ")

    @staticmethod
    def parse_time(time_str):
        return datetime.datetime.strptime(time_str,TIME_SPEC)

    @staticmethod
    def parse_bool(bool_str):
        return bool_str == "yes"

    def __init__(self, href, description, meta, hash, shared, to_read, time, tags, extended):
        # TODO: Talk about why we rename here
        self.url = href
        self.title = description
        self.description = extended

        self.time = PinboardEntry.parse_time(time)

        self.shared = PinboardEntry.parse_bool(shared)
        self.to_read = PinboardEntry.parse_bool(to_read)

        self.tags = PinboardEntry.parse_tags(tags)

        self.meta = meta
        self.hash = hash

    @classmethod
    def from_json(cls, json_obj):
        args = [
            json_obj["href"],
            json_obj["description"],
            json_obj["meta"],
            json_obj["hash"],
            json_obj["shared"],
            json_obj["toread"],
            json_obj["time"],
            json_obj["tags"],
            json_obj["extended"],
        ]
        return cls(*args)

    def __repr__(self):
        return f"<PinboardEntry: {self.url}>"

    @property
    def notes(self):
        d = {
            "meta": self.meta,
            "hash": self.hash,
            "time": self.time.strftime(TIME_SPEC),
        }

        note_lines = [
            "IMPORTED FROM PINBOARD",
            json.dumps(d, sort_keys=True, indent=2),
        ]

        return "\n".join(note_lines)

    def to_linkding_format(self):
        return {
            "url": self.url,
            "title": self.title,
            "description": self.description,
            "tag_names": self.tags,
            #
            "notes": self.notes,
            "shared": self.shared,
            "unread": True,
            "is_archived": False,
        }


def read_file(file_name):
    with open(file_name, "r") as f:
        text = f.read()

    return text


def main():
    # Get API key & file name
    LINKDING_TOKEN = os.environ["LINKDING_TOKEN"]
    PINBOARD_EXPORT_FILE_NAME = os.environ["PINBOARD_EXPORT_FILE_NAME"]

    auth_header = {"Authorization": f"Token {LINKDING_TOKEN}"}

    # Turn JSON entries into `PinboardEntry` objects
    json_text = read_file(PINBOARD_EXPORT_FILE_NAME)
    json_objs = json.loads(json_text)

    pinboard_entries = [PinboardEntry.from_json(json_obj) for json_obj in json_objs]

    # Send each `PinboardEntry` to the Linkding API
    rs = (
        grequests.post(
            LINKDING_ENDPOINT_URL,
            data=pinboard_entry.to_linkding_format(),
            headers=auth_header,
        )
        for pinboard_entry in pinboard_entries
    )

    grequests.map(rs, size=100)


if __name__ == "__main__":
    main()