Browse Source

First ipfshttpclient based version

main
Victor Roest 10 months ago
parent
commit
b2b5a578b0
Signed by: 0x76 GPG Key ID: A3923C699D1A3BDA
  1. 3
      .gitmodules
  2. 25
      README.md
  3. 0
      ipfs-share2.py
  4. 1
      ipfs_share/__init__.py
  5. 44
      ipfs_share/cli.py
  6. 3
      ipfs_share/clipboard.py
  7. 11
      ipfs_share/ipfs_share.py
  8. 17
      ipfs_share/pinner.py

3
.gitmodules

@ -1,6 +1,3 @@
[submodule "ipfshttpclient"]
path = ipfshttpclient
url = git@github.com:NULLx76/py-ipfs-http-client.git
[submodule "ipfs_share/ipfshttpclient"]
path = ipfs_share/ipfshttpclient
url = git@github.com:NULLx76/py-ipfs-http-client.git

25
README.md

@ -4,24 +4,22 @@ and copying the url to your clipboard.
### Usage
```shell
usage: ipfs-share [-h] [--no-clipboard] [--nocopy] [-p] [-g URL] [-r URL] [-t {node,cluster}] path
usage: ipfs-share [-h] [-p] [--no-clipboard] path
Share a file using IPFS
positional arguments:
path the file or folder to share
path The file or folder to share
optional arguments:
-h, --help show this help message and exit
--no-clipboard disable clipboard support
--nocopy Use the experimental ipfs 'no copy' feature
-p, --pin Pin file/folder to a remote pinner
-g URL, --gateway URL
gateway(s) to use for url generation (repetition allowed). You can also use the 'IPFS_GATEWAYS' environment variable
-r URL, --remote-pinner-url URL
Url of a remote pinner. You can also use the 'IPFS_REMOTE_PINNER' environment variable. Required when using '--pin'
-t {node,cluster}, --pinner-type {node,cluster}
Remote pinner type to use. You can also use the 'IPFS_REMOTE_PINNER_TYPE' environment variable. Defaults to node
-h, --help show this help message and exit
-p, --pin Pin target to a remote node or cluster
--no-clipboard Disable clipboard support
Environment:
IPFS_GATEWAYS A list of IPFS Gateway URLs to be used for generating urls
IPFS_REMOTE_PINNER_TYPE Either 'node' or 'cluster' depending on what remote pinner you want to use
IPFS_REMOTE_PINNER_URL The URL for a remote pinner
```
### Example
@ -36,6 +34,5 @@ https://ipfs.xirion.net/ipfs/QmTeLU7tgi82xU9Hmmp4GwTV11XDPF6Ts5qvCciPNKhs3r/inde
### Requirements
* `python` version 3.9 (untested on older versions)
* `ipfs` binary in your path
* `python-requests` for sending http requests
* (optional) `tk` for clipboard support
* (optional) `tk` for clipboard support

0
ipfs-share2.py

1
ipfs_share/__init__.py

@ -1 +1,2 @@
from ipfs_share.ipfshttpclient import ipfshttpclient
from ipfs_share.ipfs_share import ipfs_share, IpfsShare, BASE_FOLDER

44
ipfs_share/cli.py

@ -3,13 +3,13 @@ import argparse
from urllib.parse import urlparse
from ipfs_share.ipfs_share import ipfs_share
from ipfs_share.pinner import RemotePinner, RemotePinnerType
GATEWAYS = ["https://cloudflare-ipfs.com", "https://ipfs.xirion.net"]
def path_t(path: str) -> str:
"""Checker for argparse to verify that the path exists"""
if os.path.isdir(path) or os.path.isfile(path):
return path
else:
@ -18,7 +18,6 @@ def path_t(path: str) -> str:
def url_t(url: str) -> str:
"""Checker for argparse to validate urls"""
parsed = urlparse(url)
if all([parsed.scheme, parsed.netloc]):
return parsed.geturl()
@ -27,29 +26,30 @@ def url_t(url: str) -> str:
def main():
gateways = [url_t(x) for x in env_gateways.split()] if (env_gateways := os.environ.get("IPFS_GATEWAYS")) is not None else None
# remote_pinner_url = url_t(env_remote_pinner) if (env_remote_pinner := os.environ.get("IPFS_REMOTE_PINNER")) is not None else None
# pinner_type = RemotePinnerType(env_pinner_type) if (env_pinner_type := os.environ.get("IPFS_REMOTE_PINNER_TYPE")) is not None else RemotePinnerType.Node
parser = argparse.ArgumentParser(description="Share a file using IPFS")
parser.add_argument("path", type=path_t, help="the file or folder to share")
parser.add_argument("--no-clipboard", action="store_true", help="disable clipboard support")
parser.add_argument("-g", "--gateway", metavar="URL", action="append", type=url_t, default=gateways,
help="gateway(s) to use for url generation (repetition allowed). You can also use the 'IPFS_GATEWAYS' environment variable")
# parser.add_argument("-p", "--pin", action="store_true", help="Pin file/folder to a remote pinner")
# parser.add_argument("-r", "--remote-pinner-url", metavar="URL", type=url_t, default=remote_pinner_url,
# required="-p" in sys.argv and remote_pinner_url is None,
# help="Url of a remote pinner. You can also use the 'IPFS_REMOTE_PINNER' environment variable. Required when using '--pin'")
#
# parser.add_argument("-t", "--pinner-type", type=RemotePinnerType, default=pinner_type, choices=list(RemotePinnerType),
# help="Remote pinner type to use. You can also use the 'IPFS_REMOTE_PINNER_TYPE' environment variable. Defaults to node")
env_help = """
Environment:
IPFS_GATEWAYS A list of IPFS Gateway URLs to be used for generating urls
IPFS_REMOTE_PINNER_TYPE Either 'node' or 'cluster' depending on what remote pinner you want to use
IPFS_REMOTE_PINNER_URL The URL for a remote pinner
"""
gateways = [url_t(x) for x in eg.split()] if (eg := os.environ.get("IPFS_GATEWAYS")) else GATEWAYS
pinner_type = RemotePinnerType(ept) if (ept := os.environ.get("IPFS_REMOTE_PINNER_TYPE")) else RemotePinnerType.Node
remote_pinner_url = url_t(erp) if (erp := os.environ.get("IPFS_REMOTE_PINNER_URL")) else None
parser = argparse.ArgumentParser(description="Share a file using IPFS", epilog=env_help,
formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("path", type=path_t, help="The file or folder to share")
parser.add_argument("-p", "--pin", action="store_true", help="Pin target to a remote node or cluster")
parser.add_argument("--no-clipboard", action="store_true", help="Disable clipboard support")
args = parser.parse_args()
# pinner = RemotePinner.from_type(args.pinner_type, args.remote_pinner_url)
ipfs_share(args.path, not args.no_clipboard, args.gateway or GATEWAYS)
if args.pin and remote_pinner_url is None:
raise ValueError("Can't pin without IPFS_REMOTE_PINNER_URL")
pinner = RemotePinner.from_type(pinner_type, remote_pinner_url) if remote_pinner_url is not None else None
ipfs_share(args.path, not args.no_clipboard, gateways, pinner)
if __name__ == "__main__":

3
ipfs_share/clipboard.py

@ -6,10 +6,11 @@ except ImportError:
def copy_to_clipboard(string: str):
"""Copies a string to clipboard using Tk"""
if Tk is not None:
r = Tk()
r.withdraw()
r.clipboard_clear()
r.clipboard_append(string)
r.update()
r.destroy()
r.destroy()

11
ipfs_share/ipfs_share.py

@ -2,17 +2,19 @@ from __future__ import annotations
import os
from urllib.parse import urljoin
from typing import Optional
from ipfs_share import ipfshttpclient as ipfs
from ipfs_share.clipboard import copy_to_clipboard
from ipfs_share.pinner import RemotePinner
BASE_FOLDER = "/ipfs-share"
class IpfsShare:
def __init__(self, addr: str = ipfs.DEFAULT_ADDR, base_folder: str = BASE_FOLDER):
def __init__(self, addr: str = ipfs.DEFAULT_ADDR, base_folder: Optional[str] = None):
self._client = ipfs.connect(addr=addr)
self._base_folder = base_folder
self._base_folder = base_folder if base_folder else BASE_FOLDER
def __enter__(self) -> IpfsShare:
return self
@ -66,10 +68,13 @@ class IpfsShare:
return f"{self.root_cid()}/{abs_path.split(os.sep)[-1]}"
def ipfs_share(path: str, clipboard: bool, gateways: [str]):
def ipfs_share(path: str, clipboard: bool, gateways: [str], remote_pinner: Optional[RemotePinner] = None):
with IpfsShare() as share:
cid = share.add(path)
if remote_pinner is not None:
remote_pinner.pin(cid)
urls = [urljoin(g, f"ipfs/{cid}") for g in gateways]
if clipboard:
copy_to_clipboard(urls[-1])

17
ipfs_share/pinner.py

@ -5,6 +5,7 @@ from enum import Enum
from urllib.parse import urljoin
from abc import ABC
from typing import Union
from ipfs_share import ipfshttpclient as ipfs
class RemotePinnerType(Enum):
@ -30,22 +31,16 @@ class RemotePinner(ABC):
class NodeAPI(RemotePinner):
url: str
def __init__(self, addr: str):
self._client = ipfs.connect(addr=addr)
def pin(self, cid: str) -> None:
res = requests.post(urljoin(self.url, f"/api/v0/pin/add?arg={cid}"))
if res.status_code != 200:
raise Exception(f"Failed remote pin: {res.text}")
def __init__(self, url: str):
self.url = url
self._client.pin.add(cid)
class ClusterAPI(RemotePinner):
url: str
def __init__(self, addr: str):
self._addr = addr
def pin(self, cid: str) -> None:
raise NotImplementedError
def __init__(self, url: str):
self.url = url

Loading…
Cancel
Save