Add the ability to scheduled administration task inside Django admin

1 files changed, 123 insertions, 1 deletions

diff --git a/ishtar_common/models.py b/ishtar_common/models.py
index 5095619e0..59d1f9ebc 100644
--- a/ishtar_common/models.py
+++ b/ishtar_common/models.py
@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
-# Copyright (C) 2010-2016 Étienne Loks  <etienne.loks_AT_peacefrogsDOTnet>
+# Copyright (C) 2010-2017 Étienne Loks  <etienne.loks_AT_peacefrogsDOTnet>
 
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as
@@ -28,8 +28,10 @@ from PIL import Image
 from importlib import import_module
 import logging
 import os
+from os.path import isfile, join
 import re
 import shutil
+from subprocess import Popen, PIPE
 import tempfile
 import unicodecsv
 import zipfile
@@ -3161,3 +3163,123 @@ class SpatialReferenceSystem(GeneralType):
         ordering = ('label',)
 post_save.connect(post_save_cache, sender=SpatialReferenceSystem)
 post_delete.connect(post_save_cache, sender=SpatialReferenceSystem)
+
+
+class AdministrationScript(models.Model):
+    path = models.CharField(_(u"Filename"), max_length=30)
+    name = models.TextField(_(u"Name"),
+                            null=True, blank=True)
+
+    class Meta:
+        verbose_name = _(u"Administration script")
+        verbose_name_plural = _(u"Administration scripts")
+        ordering = ['name']
+
+    def __unicode__(self):
+        return unicode(self.name)
+
+
+SCRIPT_STATE = (("S", _(u"Scheduled")),
+                ("P", _(u"In progress")),
+                ("FE", _(u"Finished with errors")),
+                ("F", _(u"Finished")),
+                )
+
+SCRIPT_STATE_DCT = dict(SCRIPT_STATE)
+
+
+class AdministrationTask(models.Model):
+    script = models.ForeignKey(AdministrationScript)
+    state = models.CharField(_(u"State"), max_length=2, choices=SCRIPT_STATE,
+                             default='S')
+    creation_date = models.DateTimeField(default=datetime.datetime.now)
+    launch_date = models.DateTimeField(null=True, blank=True)
+    finished_date = models.DateTimeField(null=True, blank=True)
+    result = models.TextField(_(u"Result"), null=True, blank=True)
+
+    class Meta:
+        verbose_name = _(u"Administration task")
+        verbose_name_plural = _(u"Administration tasks")
+        ordering = ['script']
+
+    def __unicode__(self):
+        state = _(u"Unknown")
+        if self.state in SCRIPT_STATE_DCT:
+            state = unicode(SCRIPT_STATE_DCT[self.state])
+        return u"{} - {} - {}".format(self.script, self.creation_date,
+                                      state)
+
+    def execute(self):
+        if self.state != 'S':
+            return
+        self.launch_date = datetime.datetime.now()
+
+        script_dir = settings.ISHTAR_SCRIPT_DIR
+
+        if not script_dir:
+            self.result = unicode(
+                _(u"ISHTAR_SCRIPT_DIR is not set in your "
+                  u"local_settings. Contact your administrator."))
+            self.state = 'FE'
+            self.finished_date = datetime.datetime.now()
+            self.save()
+            return
+
+        if '..' in script_dir:
+            self.result = unicode(
+                _(u"Your ISHTAR_SCRIPT_DIR is containing "
+                  u"dots \"..\". As it can refer to relative "
+                  u"paths, it can be a security issue and this is "
+                  u"not allowed. Only put a full path."))
+            self.state = 'FE'
+            self.finished_date = datetime.datetime.now()
+            self.save()
+            return
+
+        if not os.path.isdir(script_dir):
+            self.result = unicode(
+                _(u"Your ISHTAR_SCRIPT_DIR: \"{}\" is not a valid directory.")
+            ).format(script_dir)
+            self.state = 'FE'
+            self.finished_date = datetime.datetime.now()
+            self.save()
+            return
+
+        script_name = None
+        # only script inside the script directory can be executed
+        for name in os.listdir(script_dir):
+            if name == self.script.path:
+                if isfile(join(script_dir, name)):
+                    script_name = join(script_dir, name)
+                break
+        if not script_name:
+            self.result = unicode(
+                _(u"Script \"{}\" is not available in your script directory. "
+                  u"Check your configuration.")
+            ).format(self.script.path)
+            self.state = 'FE'
+            self.finished_date = datetime.datetime.now()
+            self.save()
+            return
+        self.state = 'P'
+        self.save()
+
+        self.finished_date = datetime.datetime.now()
+        try:
+            session = Popen([script_name], stdout=PIPE, stderr=PIPE)
+            stdout, stderr = session.communicate()
+        except OSError as e:
+            self.state = 'FE'
+            self.result = u"Error executing \"{}\" script: {}".format(
+                self.script.path, e)
+            self.save()
+            return
+
+        self.finished_date = datetime.datetime.now()
+        if stderr:
+            self.state = 'FE'
+            self.result = u"Error: {}".format(stderr)
+        else:
+            self.state = 'F'
+            self.result = u"{}".format(stdout)
+        self.save()