Browse Source

Split into multiple files

main
Victor Roest 9 months ago
parent
commit
4cd4deda4f
Signed by: 0x76 GPG Key ID: A3923C699D1A3BDA
  1. 4
      README.md
  2. 153
      ipfs-share.py
  3. 2
      ipfs_share/__init__.py
  4. 60
      ipfs_share/cli.py
  5. 102
      ipfs_share/ipfs_share.py
  6. 51
      ipfs_share/pinner.py
  7. 3
      requirements.txt
  8. 9
      setup.py

4
README.md

@ -4,7 +4,7 @@ and copying the url to your clipboard.
### Usage
```shell
usage: ipfs-share.py [-h] [-nc] [-p] [-g URL] [-r URL] [-t {node,cluster}] path
usage: ipfs_share.py [-h] [-nc] [-p] [-g URL] [-r URL] [-t {node,cluster}] path
Share a file using IPFS
@ -25,7 +25,7 @@ optional arguments:
### Example
```shell
> ipfs-share index.html
> ipfs_share index.html
CID: QmTeLU7tgi82xU9Hmmp4GwTV11XDPF6Ts5qvCciPNKhs3r
https://cloudflare-ipfs.com/ipfs/QmTeLU7tgi82xU9Hmmp4GwTV11XDPF6Ts5qvCciPNKhs3r/index.html

153
ipfs-share.py

@ -1,153 +0,0 @@
#!/usr/bin/env python3
import subprocess
import os
import argparse
import requests
import sys
from shutil import which
from typing import List, Tuple
from urllib.parse import urlparse, urljoin
from enum import Enum
# Try importing Tk for clipboard support
try:
from tkinter import Tk
except ImportError:
Tk = None
# Default gateways to generate links for
GATEWAYS = ["https://cloudflare-ipfs.com", "https://ipfs.xirion.net"]
# Types of supported remote pinning APIs
class RemotePinner(Enum):
Node = "node"
Cluster = "cluster"
def __str__(self):
return self.value
# Checker for argparse to verify that the path exists
def path_t(path: str) -> str:
if os.path.isdir(path) or os.path.isfile(path):
return path
else:
raise argparse.ArgumentTypeError(
f"{path} is not a valid path to a file or dir")
# Checker for argparse to validate urls
def url_t(url: str) -> str:
parsed = urlparse(url)
if all([parsed.scheme, parsed.netloc]):
return parsed.geturl()
else:
raise argparse.ArgumentTypeError(f"{url} is not a valid url")
# Upload file to ipfs
def upload_file(file: str, ipfs: str) -> Tuple[str, str]:
# ipfs subcommand
# -w: Wrap, wraps the file in a folder, allows linking to the filename
# -q: Quiet, gives better parsable output
result = subprocess.run([ipfs, "add", "-wq", file],
capture_output=True, text=True)
if result.returncode != 0:
raise Exception(f"ipfs command failed:\n\n{result.stderr}")
folder_hash = result.stdout.splitlines()[-1]
file_name = os.path.basename(file)
return folder_hash, f"{folder_hash}/{file_name}"
# Upload folder to ipfs, returns (folder_cid, file_cid)
def upload_folder(path: str, ipfs: str) -> str:
# ipfs subcommand
# -r: Recursive, required for folders
# -q: Quiet, gives better parsable output
result = subprocess.run([ipfs, "add", "-rq", path],
capture_output=True, text=True)
if result.returncode != 0:
raise Exception(f"ipfs command failed:\n\n{result.stderr}")
folder_hash = result.stdout.splitlines()[-1]
return folder_hash
# Upload folder or file to ipfs, returns (folder_cid, file_cid)
def upload(path: str, ipfs: str = "ipfs") -> Tuple[str, str]:
if os.path.isdir(path):
cid = upload_folder(path, ipfs)
return cid, cid
elif os.path.isfile(path):
return upload_file(path, ipfs)
else:
raise ValueError(f"{path} is an invalid path to file or dir")
def remote_pin(pinner: RemotePinner, host: str, cid: str):
if pinner is RemotePinner.Node:
# curl -X POST "http://ipfs-node:5001/api/v0/pin/add?arg={cid}"
res = requests.post(urljoin(host, f"/api/v0/pin/add?arg={cid}"))
if res.status_code != 200:
raise Exception(f"Failed remote pin: {res.text}")
elif pinner is RemotePinner.Cluster:
raise Exception("Pinning to a IPFS Cluster is not yet supported")
else:
raise ValueError(f"Unknown remote pinner {pinner}")
def copy_to_clipboard(string: str):
if Tk is not None:
r = Tk()
r.withdraw()
r.clipboard_clear()
r.clipboard_append(string)
r.update()
r.destroy()
def ipfs_share(target: str, clipboard: bool, gw: List[str], pin: bool, pinner: RemotePinner, pinner_url: str):
if (ipfs := which("ipfs")) is None:
raise Exception("ipfs binary could not be found")
folder_cid, file_cid = upload(target, ipfs)
if pin:
remote_pin(pinner, pinner_url, folder_cid)
urls = [urljoin(g, f"ipfs/{file_cid}") for g in gw]
if clipboard:
copy_to_clipboard(urls[-1])
print(f"CID: {folder_cid}")
for el in urls:
print(el)
if __name__ == "__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 = RemotePinner(env_pinner_type) if (env_pinner_type := os.environ.get("IPFS_REMOTE_PINNER_TYPE")) is not None else RemotePinner.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("-nc", "--no-clipboard", action="store_true", help="disable clipboard support")
parser.add_argument("-p", "--pin", action="store_true", help="Pin file/folder to a remote pinner")
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("-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=RemotePinner, default=pinner_type, choices=list(RemotePinner),
help="Remote pinner type to use. You can also use the 'IPFS_REMOTE_PINNER_TYPE' environment variable. Defaults to node")
args = parser.parse_args()
ipfs_share(args.path, not args.no_clipboard, args.gateway or GATEWAYS, args.pin, args.pinner_type, args.remote_pinner_url)

2
ipfs_share/__init__.py

@ -0,0 +1,2 @@
from .ipfs_share import ipfs_share, ShareOptions
from .pinner import RemotePinner, RemotePinnerType

60
ipfs_share/cli.py

@ -0,0 +1,60 @@
import os
import sys
import argparse
from urllib.parse import urlparse
from .pinner import RemotePinner, RemotePinnerType
from .ipfs_share import ShareOptions, ipfs_share
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:
raise argparse.ArgumentTypeError(f"{path} is not a valid path to a file or dir")
def url_t(url: str) -> str:
"""Checker for argparse to validate urls"""
parsed = urlparse(url)
if all([parsed.scheme, parsed.netloc]):
return parsed.geturl()
else:
raise argparse.ArgumentTypeError(f"{url} is not a valid url")
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("--nocopy", action="store_true", help="Use the experimental ipfs 'no copy' feature")
parser.add_argument("-p", "--pin", action="store_true", help="Pin file/folder to a remote pinner")
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("-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")
args = parser.parse_args()
pinner = RemotePinner.from_type(args.pinner_type, args.remote_pinner_url)
ipfs_share(ShareOptions(args.path, not args.no_clipboard, args.gateway or GATEWAYS, args.pin, pinner, args.nocopy))
if __name__ == "__main__":
main()

102
ipfs_share/ipfs_share.py

@ -0,0 +1,102 @@
#!/usr/bin/env python3
import subprocess
import os
from .pinner import RemotePinner
from dataclasses import dataclass
from shutil import which
from typing import Tuple, List
from urllib.parse import urljoin
# Try importing Tk for clipboard support
try:
from tkinter import Tk
except ImportError:
Tk = None
@dataclass
class ShareOptions:
target: str
enable_clipboard: bool
gateways: List[str]
pin: bool
pinner: RemotePinner
no_copy: bool
# Upload file to ipfs
def upload_file(file: str, ipfs: str, no_copy: bool) -> Tuple[str, str]:
# ipfs subcommand
# -w: Wrap, wraps the file in a folder, allows linking to the filename
# -q: Quiet, gives better parsable output
cmd = [ipfs, "add", "-wq", file]
if no_copy:
cmd.insert(3, "--nocopy")
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise Exception(f"ipfs command failed:\n\n{result.stderr}")
folder_hash = result.stdout.splitlines()[-1]
file_name = os.path.basename(file)
return str(folder_hash), f"{folder_hash}/{file_name}"
# Upload folder to ipfs, returns (folder_cid, file_cid)
def upload_folder(path: str, ipfs: str, no_copy: bool) -> str:
# ipfs subcommand
# -r: Recursive, required for folders
# -q: Quiet, gives better parsable output
cmd = [ipfs, "add", "-rq", path]
if no_copy:
cmd.insert(3, "--nocopy")
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise Exception(f"ipfs command failed:\n\n{result.stderr}")
folder_hash = result.stdout.splitlines()[-1]
return str(folder_hash)
# Upload folder or file to ipfs, returns (folder_cid, file_cid)
def upload(path: str, ipfs: str = "ipfs", no_copy: bool = False) -> Tuple[str, str]:
if os.path.isdir(path):
cid = upload_folder(path, ipfs, no_copy)
return cid, cid
elif os.path.isfile(path):
return upload_file(path, ipfs, no_copy)
else:
raise ValueError(f"{path} is an invalid path to file or dir")
def copy_to_clipboard(string: str):
if Tk is not None:
r = Tk()
r.withdraw()
r.clipboard_clear()
r.clipboard_append(string)
r.update()
r.destroy()
def ipfs_share(options: ShareOptions):
if (ipfs := which("ipfs")) is None:
raise Exception("ipfs binary could not be found")
folder_cid, file_cid = upload(options.target, ipfs)
if options.pin:
options.pinner.pin(folder_cid)
urls = [urljoin(g, f"ipfs/{file_cid}") for g in options.gateways]
if options.enable_clipboard:
copy_to_clipboard(urls[-1])
print(f"CID: {folder_cid}")
for el in urls:
print(el)

51
ipfs_share/pinner.py

@ -0,0 +1,51 @@
from __future__ import annotations
import requests
from enum import Enum
from urllib.parse import urljoin
from abc import ABC
from typing import Union
class RemotePinnerType(Enum):
Node = "node"
Cluster = "cluster"
def __str__(self):
return self.value
class RemotePinner(ABC):
def pin(self, cid: str) -> None:
pass
@staticmethod
def from_type(kind: RemotePinnerType, host: str) -> Union[NodeAPI, ClusterAPI]:
if kind is RemotePinnerType.Node:
return NodeAPI(host)
elif kind is RemotePinnerType.Cluster:
return ClusterAPI(host)
else:
raise ValueError(f"Invalid RemotePinnerType: {kind}")
class NodeAPI(RemotePinner):
url: str
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
class ClusterAPI(RemotePinner):
url: str
def pin(self, cid: str) -> None:
raise NotImplementedError
def __init__(self, url: str):
self.url = url

3
requirements.txt

@ -0,0 +1,3 @@
setuptools~=54.1.1
requests~=2.25.1
future~=0.18.2

9
setup.py

@ -0,0 +1,9 @@
from setuptools import setup
setup(
name='ipfs_share',
packages=['ipfs_share'],
entry_points={
'console_scripts': ['ipfs-share=ipfs_share.cli:main']
}
)
Loading…
Cancel
Save