import os import glob from subprocess import check_call, CalledProcessError from flask import Flask from flask import request, render_template, make_response from threading import Thread app = Flask( __name__, # static_url_path="./", static_folder="static", template_folder="templates", ) os.makedirs("uploads", exist_ok=True) def convert_to_wav(file_name: str): output, extension = file_name.rsplit(".", maxsplit=1) assert extension != "wav" print(f"{extension=}") command = [ "tape2wav", os.path.join("uploads", file_name), os.path.join("uploads", f"${output}.wav"), ] try: check_call(command) except CalledProcessError as exc: print(f"tape2wav failed: {exc.output=}\n{exc.stderr=}") raise def run_cassetta(cassetta: str): command = [ "aplay", os.path.join("uploads", cassetta), ] try: check_call(command) except CalledProcessError as exc: print(f"aplay failed: {exc.output=}\n{exc.stderr=}") def uploads(): return map( os.path.basename, glob.glob("uploads/*wav"), ) @app.route("/") def upload_form(): return render_template( "upload.html", title="Upload your shit here", files=uploads(), ) @app.route("/upload", methods=["POST"]) def upload_file(): if "file" not in request.files: return "No file part" file = request.files["file"] if file.filename == "": return "No selected file" if not file: return "No file?" save_path = os.path.join("uploads", file.filename) if os.path.exists(save_path): response = make_response(make_response("file already exists", 400)) response.headers["Location"] = "/" response.status_code = 400 try: file.save(save_path) except Exception as exc: return f"fail: {exc=}" name, extension = file.filename.rsplit(".", maxsplit=1) if extension != "wav": thread = Thread(target=convert_to_wav, args=(file.filename,)) thread.start() print(f"thread started for {file.filename=}") response = make_response(make_response("file uploaded succesfully", 301)) response.headers["Location"] = "/" response.status_code = 301 return response @app.route("/run/", methods=["GET"]) def run(cassetta: str): thread = Thread(target=run_cassetta, args=(cassetta,)) thread.start() response = make_response(make_response("", 301)) response.headers["Location"] = "/" response.status_code = 301 return response if __name__ == "__main__": app.run()