import model import subprocess import os def run_scripts(script_ids: list[int] | None = None): db = model.SessionLocal() if script_ids: scripts = db.query(model.Script).filter(model.Script.id.in_(script_ids)).all() else: scripts = db.query(model.Script).filter(model.Script.enabled).all() for script in scripts: print(f"Running script: {script.name}") dump_script_to_file(script, f"exec_folder/{script.name}.py") result = execute_script(f"exec_folder/{script.name}.py") db.add( model.Log( script_id=script.id, error_code=result.returncode, message=result.stdout, error_message=result.stderr, ) ) db.commit() delete_script(f"exec_folder/{script.name}.py") db.close() def dump_script_to_file(script, filename): with open(filename, "w") as file: file.write("from dotenv import load_dotenv\nload_dotenv()\n") file.write(script.script_content) def execute_script(filename) -> subprocess.CompletedProcess: result = subprocess.run( ["exec_folder/venv/bin/python", filename], capture_output=True, text=True ) return result def update_requirements(settings): if settings is None: raise ValueError("No default settings found") # create requirements.txt with open("exec_folder/requirements.txt", "w") as file: file.write("dotenv\n") file.write(settings.requirements) # install requirements subprocess.run( ["exec_folder/venv/bin/pip", "install", "-r", "exec_folder/requirements.txt"], check=True, ) def update_environment(settings): if settings is None: raise ValueError("No default settings found") # create .env file with open("exec_folder/.env", "w") as file: file.write(settings.environment) def delete_script(filename): try: os.remove(filename) except FileNotFoundError: pass def main(): run_scripts() if __name__ == "__main__": main()