Tezos Account Addres -> Get account OBJKT list -> Get IPFS addresses -> IPFS Pinning

pin_myobjkt.py

# Tezos Account Addres ("tz...") -> Get account OBJKT list
#     -> Get IPFS addresses -> IPFS Pinning
#
# Requirements
# ------------
#
# - IPFS Desktop (https://docs.ipfs.io/install/ipfs-desktop/) or CLI tool
#
# - python package
#
#     ```
#     pip install requests
#     ```
#
# Usage
# -----
#
# see `--help`
#
# ```
# $ python pin_myobjkt.py <account_address> --ipfs-task pin
# $ python pin_myobjkt.py <account_address> --ipfs-task pin --limit 5
# $ python pin_myobjkt.py <account_address> --ipfs-task pin --creations-only
# $ python pin_myobjkt.py <account_address> --ipfs-task pin --ignore-empty
# $ python pin_myobjkt.py <account_address> --ipfs-task unpin
# $ python pin_myobjkt.py <account_address> --ipfs-task none
# ```
#
# License
# -------
#
# CC0
# https://creativecommons.org/publicdomain/zero/1.0/

import argparse
import subprocess
import time
from enum import Enum
from typing import List

import requests


class IpfsTask(Enum):
    NONE = "none"
    PIN = "pin"
    UNPIN = "unpin"

    def __str__(self):
        return self.value

    def __repr__(self):
        return str(self)


IPFS_BIN = "ipfs"
NETWORK = "mainnet"
SYMBOL = "OBJKT"

parser = argparse.ArgumentParser(
    description='Tezos Account Addres ("tz...") -> Get account OBJKT list -> Get IPFS addresses -> IPFS Pinning')
parser.add_argument('tz', metavar='TZ_ADDRESS', help='tezos account address')
parser.add_argument('--ipfs-task', type=IpfsTask, metavar=list(IpfsTask),
                    choices=list(IpfsTask), required=True, help='ipfs command type')
parser.add_argument('--limit', type=int, metavar='LIMIT',
                    default=None, help='limit the number of items for pinning')
parser.add_argument('--creations-only', action='store_true',
                    help='own creations only')
parser.add_argument('--ignore-empty', action='store_true',
                    help='ignore zero quantity tokens (such as burned all)')
args = parser.parse_args()

has_limit = args.limit is not None


class TokenMetadata:
    def __init__(self, token_balance_record={}, tz=None):
        self.token_id = token_balance_record.get("token_id")
        self.name = token_balance_record.get("name")
        self.creators = token_balance_record.get("creators")
        self.artifact_uri = token_balance_record.get("artifact_uri")
        self.display_uri = token_balance_record.get("display_uri")
        self.thumbnail_uri = token_balance_record.get("thumbnail_uri")
        self.is_creation = None if tz is None else tz in self.creators

    def get_page_uri(self) -> str:
        return f"https://www.hicetnunc.xyz/objkt/{self.token_id}/"


def fetch_token_metadata_list(network, symbol, tz) -> List[TokenMetadata]:
    SIZE = 50
    offset = 0
    total = None

    token_metadata_list: List[TokenMetadata] = []

    limit_reached = False

    while (total is None or offset < total) and not limit_reached:
        # Reference: https://better-call.dev/docs#operation/get-account-token-balances
        res = requests.get(
            f"https://api.better-call.dev/v1/account/{network}/{tz}/token_balances?size={SIZE}&offset={offset}&hide_empty={args.ignore_empty}")
        json = res.json()

        time.sleep(0.5)

        total = json.get("total")
        assert(total is not None)

        for rec in json["balances"]:
            offset += 1

            if rec.get("symbol") != symbol:
                continue

            metadata = TokenMetadata(rec, tz)
            print(
                f"METADATA {offset}/{total} #{metadata.token_id} '{metadata.name}' {metadata.artifact_uri} {metadata.get_page_uri()} {1 if metadata.is_creation else 0}")
            token_metadata_list.append(metadata)

            if has_limit and len(token_metadata_list) >= args.limit:
                limit_reached = True
                break

    print(f"objkt token count: {len(token_metadata_list)}")
    print(f"all token count: {total}")
    return token_metadata_list


def ipfs_uri_to_hash(ipfs_uri):
    return ipfs_uri.replace("ipfs://", "")


def pin(ipfs_uri):
    if ipfs_uri:
        hash = ipfs_uri_to_hash(ipfs_uri)
        return subprocess.run([IPFS_BIN, "pin", "add", hash], shell=True)


def unpin(ipfs_uri):
    if ipfs_uri:
        hash = ipfs_uri_to_hash(ipfs_uri)
        return subprocess.run([IPFS_BIN, "pin", "rm", hash], shell=True)


def main():
    assert(args.tz.startswith("tz"))

    token_metadata_list = fetch_token_metadata_list(NETWORK, SYMBOL, args.tz)

    t = args.ipfs_task
    if t == IpfsTask.NONE:
        exit(0)
    elif t == IpfsTask.PIN:
        task = pin
        task_name = "PIN"
    elif t == IpfsTask.UNPIN:
        task = unpin
        task_name = "UNPIN"
    else:
        raise ValueError("Invalid IpfsTask type")

    count = 0
    total = len(token_metadata_list)
    for metadata in token_metadata_list:
        if args.creations_only and not metadata.is_creation:
            continue

        count += 1

        print(f"{task_name} {count}/{total} #{metadata.token_id} '{metadata.name}'")

        task(metadata.artifact_uri)
        task(metadata.display_uri)
        task(metadata.thumbnail_uri)

    print(f"processed objkt count: {len(token_metadata_list)}")


if __name__ == "__main__":
    main()