Browse Source

Add CLI

master
dave 2 years ago
parent
commit
791e457f45
  1. 6
      Dockerfile
  2. 6
      requirements.txt
  3. 3
      setup.py
  4. 95
      wastebin/cli.py
  5. 41
      wastebin/daemon.py

6
Dockerfile

@ -1,15 +1,15 @@
FROM ubuntu:bionic
ADD . /tmp/code/
RUN apt-get update && \
apt-get install -y python3-pip
ADD . /tmp/code/
RUN pip3 install -U pip && \
cd /tmp/code && \
python3 setup.py install && \
useradd --uid 1000 app
VOLUME /data/
USER app
ENTRYPOINT ["wastebind", "-d", "/data/"]

6
requirements.txt

@ -1,10 +1,16 @@
appdirs==1.4.3
backports.functools-lru-cache==1.5
certifi==2018.11.29
chardet==3.0.4
cheroot==6.5.4
CherryPy==18.1.0
idna==2.8
jaraco.functools==2.0
more-itertools==5.0.0
portend==2.3
pytz==2018.9
requests==2.21.0
six==1.12.0
tempora==1.14
urllib3==1.24.1
zc.lockfile==1.4

3
setup.py

@ -19,7 +19,8 @@ setup(name='wastebin',
install_requires=__requirements__,
entry_points={
"console_scripts": [
"wastebind = wastebin.daemon:main"
"wastebind = wastebin.daemon:main",
"wpaste = wastebin.cli:main"
]
},
zip_safe=False)

95
wastebin/cli.py

@ -0,0 +1,95 @@
from appdirs import user_config_dir
import os
import json
import argparse
import requests
import tempfile
import subprocess
APPNAME = "wpaste"
CONFDIR = user_config_dir(APPNAME)
CONFPATH = os.path.join(CONFDIR, "conf.json")
def editor(fpath):
"""
Open the editor
"""
subprocess.check_call([os.environ["EDITOR"], fpath]) # XXX commented for testing
with open(fpath) as f:
content = f.read()
return content
def main():
conf = {"host": "", "username": "", "password": ""}
if os.path.exists(CONFPATH):
with open(CONFPATH) as cf:
conf = json.load(cf)
else:
os.makedirs(CONFDIR, exist_ok=True)
with open(CONFPATH, "w") as cf:
json.dump(conf, cf)
parser = argparse.ArgumentParser(description="Wastebin cli",
epilog="host/username/password will be saved to {} "
"after first use.".format(CONFPATH))
parser.add_argument("--host", default=conf["host"], help="http/s host to connect to")
# parser.add_argument("-u", "--username", help="username")
# parser.add_argument("-p", "--password", help="password")
spr_action = parser.add_subparsers(dest="action", help="action to take")
spr_action.add_parser("list", help="show list of pastes")
spr_new = spr_action.add_parser("new", help="create a paste")
spr_new.add_argument("name", nargs="?", default="", help="name of paste to create")
spr_get = spr_action.add_parser("get", help="get a paste")
spr_get.add_argument("name", help="name of paste to get")
spr_edit = spr_action.add_parser("edit", help="edit a paste")
spr_edit.add_argument("name", help="name of paste to edit")
spr_del = spr_action.add_parser("del", help="delete a paste")
spr_del.add_argument("name", help="name of paste to delete")
args = parser.parse_args()
r = requests.session()
host = args.host.rstrip("/") + "/"
def getpaste(name):
req = r.get(host + name)
req.raise_for_status()
return req.text
def putpaste(name, body):
return r.post(host + "make", data={"name": name, "contents": body})
if args.action in ("new", "edit", "get"):
if args.action in ("edit", "get"):
content = getpaste(args.name)
if args.action == "get":
print(content, end="")
return
with tempfile.NamedTemporaryFile() as f:
if args.action == "edit":
f.write(content.encode("utf-8"))
f.flush()
content = editor(f.name)
if not content:
print("Blank paste, exiting")
r = putpaste(args.name, content)
r.raise_for_status()
print(r.url)
elif args.action == "del":
r.delete(host + args.name).raise_for_status()
elif args.action == "list":
print(r.get(host + "search").text, end="")
if __name__ == "__main__":
main()

41
wastebin/daemon.py

@ -3,7 +3,7 @@ import cherrypy
import logging
import hashlib
import re
from threading import Thread
PAGE = """<!DOCTYPE html>
<html lang="en">
@ -34,6 +34,18 @@ def sha256(data):
class WasteWeb(object):
def __init__(self, datadir):
self.datadir = datadir
self.namecache = set()
t = Thread(target=self.prep_cache)
t.daemon = True
t.start()
def prep_cache(self):
print("Populating index cache....")
for dirpath, dirnames, filenames in os.walk(self.datadir):
for fname in filenames:
with open(os.path.join(dirpath, fname)) as f:
self.namecache.update([f.readline().strip()])
print("Indexed {} items".format(len(self.namecache)))
@cherrypy.expose
def index(self, load=None):
@ -45,15 +57,24 @@ class WasteWeb(object):
@cherrypy.expose
def make(self, name, contents):
assert RE_NAME.match(name)
pname = name or sha256(contents)
assert RE_NAME.match(pname)
self.writepaste(pname, contents)
raise cherrypy.HTTPRedirect("/" + pname)
@cherrypy.expose
def default(self, *args):
data = self.loadpaste(args[0])
yield data
if cherrypy.request.method == "DELETE":
self.delpaste(args[0])
return "OK"
else:
cherrypy.response.headers['Content-Type'] = 'text/plain'
return self.loadpaste(args[0]).encode("utf-8")
@cherrypy.expose
def search(self):
for entry in self.namecache:
yield entry + "\n"
def loadpaste(self, name):
path = self.pastepath(sha256(name))
@ -69,6 +90,18 @@ class WasteWeb(object):
f.write(name)
f.write("\n")
f.write(contents)
self.namecache.update({name})
def delpaste(self, name):
self.namecache.remove(name)
path = self.pastepath(sha256(name))
os.unlink(path)
pdir = os.path.dirname(path)
try:
os.rmdir(os.path.normpath(pdir))
os.rmdir(os.path.normpath(os.path.join(pdir, "../")))
except:
pass
def pastepath(self, hashedname):
return os.path.join(self.datadir, hashedname[0], hashedname[1], hashedname + ".txt")

Loading…
Cancel
Save