86 lines
2.0 KiB
Python
86 lines
2.0 KiB
Python
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()
|