cyber-valley/cve/tools/pin.py

python
π 0.0% 230 words · 2 min 
#!/usr/bin/env python3
"""Pin files to Pinata and record their CIDs.

    python3 pin.py cids.json  <name>=<path> …

Reads PINATA_JWT from ~/.config/cyber/env. Each file is pinned with the
metadata name given; the CID goes into cids.json under that name. Idempotent:
a name already in cids.json is skipped.
"""
import json
import mimetypes
import sys
import urllib.request
import uuid
from pathlib import Path


def env():
    out = {}
    for line in (Path.home() / ".config/cyber/env").read_text().splitlines():
        if "=" in line and not line.startswith("#"):
            k, v = line.split("=", 1)
            out[k.strip()] = v.strip()
    return out


def pin(jwt: str, path: Path, name: str) -> str:
    """Pinata Files API v3, public network — returns the CID (v1)."""
    boundary = uuid.uuid4().hex
    mime = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
    def field(k, v):
        return f'--{boundary}\r\nContent-Disposition: form-data; name="{k}"\r\n\r\n{v}\r\n'.encode()
    body = b"".join(
        [
            f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="{path.name}"\r\nContent-Type: {mime}\r\n\r\n'.encode(),
            path.read_bytes(),
            b"\r\n",
            field("network", "public"),
            field("name", name),
            field("keyvalues", json.dumps({"site": "cyberia.my", "pack": "leasehold"})),
            f"--{boundary}--\r\n".encode(),
        ]
    )
    req = urllib.request.Request(
        "https://uploads.pinata.cloud/v3/files",
        data=body,
        headers={"Authorization": f"Bearer {jwt}", "Content-Type": f"multipart/form-data; boundary={boundary}", "User-Agent": "cyberia-pin/1.0"},
    )
    with urllib.request.urlopen(req, timeout=900) as r:
        return json.load(r)["data"]["cid"]


def main() -> int:
    out = Path(sys.argv[1])
    cids = json.load(out.open()) if out.exists() else {}
    jwt = env()["PINATA_JWT"]
    for arg in sys.argv[2:]:
        name, _, path = arg.partition("=")
        if name in cids:
            print(f"= {name} {cids[name]}")
            continue
        p = Path(path)
        cid = pin(jwt, p, name)
        cids[name] = cid
        out.write_text(json.dumps(cids, indent=1, ensure_ascii=False))
        print(f"+ {name} {cid} ({p.stat().st_size // 1024} KB)")
    return 0


if __name__ == "__main__":
    sys.exit(main())

Graph