60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
|
from fastapi import FastAPI, staticfiles, UploadFile
|
||
|
|
from starlette.responses import FileResponse
|
||
|
|
|
||
|
|
import aiofile
|
||
|
|
from asyncio import subprocess
|
||
|
|
|
||
|
|
import tempfile
|
||
|
|
import os
|
||
|
|
import glob
|
||
|
|
import enum
|
||
|
|
|
||
|
|
class OutputFormat(enum.Enum):
|
||
|
|
pdf = "pdf"
|
||
|
|
|
||
|
|
data_dir = os.path.realpath(os.environ.get("DATA_DIR", "./data"))
|
||
|
|
soffice_data_dir = os.path.realpath(os.environ.get("UNO_DATA_DIR", data_dir))
|
||
|
|
os.makedirs(data_dir, exist_ok=True)
|
||
|
|
|
||
|
|
soffice_command = os.environ.get("UNO_COMMAND", "soffice --convert-to {outputfmt} --outdir {outputdir} {inputfilename}")
|
||
|
|
|
||
|
|
app = FastAPI()
|
||
|
|
app.mount("/static", staticfiles.StaticFiles(directory="static"), name="static")
|
||
|
|
|
||
|
|
@app.get("/")
|
||
|
|
async def get_index():
|
||
|
|
return FileResponse('static/index.html')
|
||
|
|
|
||
|
|
@app.post("/")
|
||
|
|
async def convert_file(file: UploadFile, output_format: OutputFormat | None = None):
|
||
|
|
output_format = output_format or OutputFormat.pdf
|
||
|
|
|
||
|
|
outpath = tempfile.mkdtemp(dir=data_dir)
|
||
|
|
inputfile = os.path.join(outpath, file.filename)
|
||
|
|
|
||
|
|
uno_path = os.path.join(soffice_data_dir, os.path.relpath(outpath, data_dir))
|
||
|
|
uno_inputfile = os.path.join(uno_path, file.filename)
|
||
|
|
|
||
|
|
async with aiofile.AIOFile(inputfile, mode="wb+") as fp:
|
||
|
|
try:
|
||
|
|
while True:
|
||
|
|
content = await file.read(1024*1024*4)
|
||
|
|
if not content:
|
||
|
|
break
|
||
|
|
await fp.write(content)
|
||
|
|
except EOFError:
|
||
|
|
pass
|
||
|
|
else:
|
||
|
|
print(f"file written to {inputfile}")
|
||
|
|
|
||
|
|
command = soffice_command.format(outputfmt=output_format.value, outputdir=uno_path, inputfilename=uno_inputfile)
|
||
|
|
|
||
|
|
result_process = await subprocess.create_subprocess_shell(command)
|
||
|
|
assert await result_process.wait() == 0, f"expect return code 0, got {result_process.returncode}"
|
||
|
|
outfiles = glob.glob(f"{uno_path}/*.{output_format.value}")
|
||
|
|
assert len(outfiles) == 1, f"expect one output file, got {outfiles}"
|
||
|
|
|
||
|
|
return FileResponse(outfiles[0])
|
||
|
|
|
||
|
|
|