dece/Shrlok
Archived
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

init

Browse source
This commit is contained in:
dece 2022-07-04 21:52:48 +02:00
commit 794658b1ab
5 changed files with 118 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

0
README.md Normal file
View file

3
pyproject.toml Normal file
View file

@ -0,0 +1,3 @@
[build-system]
requires = ["setuptools", "wheel"]
build-backend = "setuptools.build_meta"

22
setup.cfg Normal file
View file

@ -0,0 +1,22 @@
[metadata]
name = shrlok
version = 0.0.1
description = Publish stuff on the Web by piping it to a Unix domain socket.
long_description = file: README.md
long_description_content_type = text/markdown
license = GPLv3
author = dece
author_email = shgck@pistache.land
home_page = https://git.dece.space/Dece/Shrlok
classifiers =
Environment :: Console
Programming Language :: Python :: 3
[options]
packages = shrlok
python_requires = >= 3.7
setup_requires = setuptools >= 38.3.0
[options.entry_points]
console_scripts =
shrlok = shrlok.shrlok:main

2
shrlok/shr-txt.py Normal file
View file

@ -0,0 +1,2 @@
#!/usr/bin/env python3

91
shrlok/shrlok.py Normal file
View file

@ -0,0 +1,91 @@
#!/usr/bin/env python3
import argparse
import json
import os
import socketserver
import tempfile
from pathlib import Path
ARGS = None
HTML_TEMPLATE = """\
<!doctype html>
<html>
<head>
<meta charset="UTF-8">
<title>{title}</title>
<style>
body {{ max-width: 40em; }}
</style>
</head>
<body>
{content}
</body>
</html>
"""
class Handler(socketserver.StreamRequestHandler):
def handle(self):
fragments = []
while (chunk := self.request.recv(4096)):
fragments.append(chunk)
data = b"".join(fragments)
try:
first_zero = data.index(b"\0")
header_data, data = data[:first_zero], data[first_zero + 1:]
header = json.loads(header_data.decode())
except (ValueError, IndexError):
print("Bad header.")
return
if header.get("type") == "txt":
file_name = write_text(data, title=header.get("title"))
else:
print("Unknown type.")
return
print(f"{len(data)} bytes — {header} — '{file_name}'.")
self.request.sendall(file_name.encode())
def write_text(data: bytes, title=None):
content = "<pre>{}</pre>".format(data.decode(errors="replace"))
html = HTML_TEMPLATE.format(title=title or "?", content=content)
path = str(Path(ARGS.root) / "txt")
if not os.path.isdir(path):
os.makedirs(path)
with tempfile.NamedTemporaryFile(
mode="wt",
dir=path,
suffix=".html",
delete=False
) as output_file:
output_file.write(html)
return output_file.name
def main():
argp = argparse.ArgumentParser()
argp.add_argument("root", help="root path where to put files")
global ARGS
ARGS = argp.parse_args()
uid = os.getuid()
socket_path = f"/run/user/{uid}/shr.sock"
if os.path.exists(socket_path):
os.unlink(socket_path)
try:
with socketserver.UnixStreamServer(socket_path, Handler) as server:
server.serve_forever()
except KeyboardInterrupt:
print("Stopping server.")
finally:
os.unlink(socket_path)
if __name__ == "__main__":
main()