Refatorar mixin jobs

sigi/apps/casas/jobs/daily/usuario_contato.py

@@ -4,7 +4,7 @@ from django.utils.translation import gettext as _
 from django_extensions.management.jobs import DailyJob
 from sigi.apps.casas.models import Funcionario
 from sigi.apps.servidores.models import Servidor
-from sigi.apps.utils.mixins import JobReportMixin
+from sigi.apps.utils.management.jobs import JobReportMixin
 
 
 class Job(JobReportMixin, DailyJob):

sigi/apps/contatos/jobs/monthly/atualiza_ibge.py

@@ -12,7 +12,7 @@ from sigi.apps.contatos.models import (
     Municipio,
 )
 from sigi.apps.servidores.models import Servidor
-from sigi.apps.utils.mixins import JobReportMixin
+from sigi.apps.utils.management.jobs import JobReportMixin
 
 
 class Job(JobReportMixin, MonthlyJob):

sigi/apps/convenios/jobs/daily/importa_gescon.py

@@ -4,7 +4,7 @@ from django.core.mail import mail_admins
 from django.utils.translation import gettext as _
 from django_extensions.management.jobs import DailyJob
 from sigi.apps.convenios.models import Gescon
-from sigi.apps.utils.mixins import JobReportMixin
+from sigi.apps.utils.management.jobs import JobReportMixin
 
 
 class Job(JobReportMixin, DailyJob):

sigi/apps/servicos/jobs/daily/sincroniza_dns.py

@@ -9,7 +9,7 @@ from sigi.apps.servicos import generate_instance_name, nomeia_instancias
 from sigi.apps.servicos.models import Servico, TipoServico
 from sigi.apps.casas.models import Orgao
 from sigi.apps.contatos.models import UnidadeFederativa
-from sigi.apps.utils.mixins import JobReportMixin
+from sigi.apps.utils.management.jobs import JobReportMixin
 
 LOG_GERAL = _("Mensagens gerais")
 IGNORES = ["_psl", "k8s", "www.", "sapl.", "addr.arpa"]

sigi/apps/servicos/jobs/daily/sincroniza_rancher.py

@@ -7,7 +7,7 @@ from django_extensions.management.jobs import DailyJob
 from sigi.apps.servicos import generate_instance_name, nomeia_instancias
 from sigi.apps.servicos.models import Servico, TipoServico
 from sigi.apps.casas.models import Orgao
-from sigi.apps.utils.mixins import JobReportMixin
+from sigi.apps.utils.management.jobs import JobReportMixin
 
 
 class Job(JobReportMixin, DailyJob):

sigi/apps/servicos/jobs/daily/verifica_dominios.py

@@ -3,7 +3,7 @@ from django.utils import timezone
 from django.utils.translation import gettext as _
 from django_extensions.management.jobs import DailyJob
 from sigi.apps.servicos.models import Servico
-from sigi.apps.utils.mixins import JobReportMixin
+from sigi.apps.utils.management.jobs import JobReportMixin
 
 
 class Job(JobReportMixin, DailyJob):

sigi/apps/utils/management/jobs.py

@@ -0,0 +1,117 @@
+import datetime
+import docutils.core
+import traceback
+from django.contrib.admin.models import LogEntry, ADDITION, CHANGE
+from django.contrib.contenttypes.models import ContentType
+from django.core.mail import mail_admins
+from django.template.loader import render_to_string
+from django_extensions.management.jobs import BaseJob
+
+
+class MisconfiguredError(Exception):
+    pass
+
+
+class QuarterDailyJob(BaseJob):
+    when = "quarter_daily"
+
+
+class JobReportMixin:
+    error_report_template = "emails/report_error.rst"
+    report_template = "emails/base_report.rst"
+    report_data = None
+    sys_user = None
+
+    def execute(self):
+        start_time = datetime.datetime.now()
+
+        try:
+            from sigi.apps.servidores.models import Servidor
+
+            self.sys_user = Servidor.objects.get(sigi=True).user
+        except Exception:
+            pass
+
+        try:
+            self.do_job()
+        except Exception as e:
+            self.report_error(e)
+            return
+        end_time = datetime.datetime.now()
+        self.report(start_time, end_time)
+
+    def do_job(self):
+        raise NotImplementedError("Job needs to implement the 'do_job' method")
+
+    def _admin_log(self, object, action_flag, message=""):
+        if self.sys_user is None:
+            return  # No admin log
+        LogEntry.objects.log_action(
+            user_id=self.sys_user.id,
+            content_type_id=ContentType.objects.get_for_model(type(object)).pk,
+            object_id=object.id,
+            object_repr=str(object),
+            action_flag=action_flag,
+            change_message=message,
+        )
+
+    def admin_log_addition(self, object, message=""):
+        self._admin_log(object, ADDITION, message)
+
+    def admin_log_change(self, object, message=""):
+        self._admin_log(object, CHANGE, message)
+
+    def report_error(self, error):
+        rst = render_to_string(
+            self.error_report_template,
+            {
+                "title": self.help,
+                "traceback": traceback.format_exception(error),
+            },
+        )
+        html = docutils.core.publish_string(
+            rst,
+            writer_name="html5",
+            settings_overrides={
+                "input_encoding": "unicode",
+                "output_encoding": "unicode",
+            },
+        )
+        mail_admins(
+            subject=self.help,
+            message=rst,
+            html_message=html,
+            fail_silently=True,
+        )
+        print(rst)
+
+    def report(self, start_time, end_time):
+        if self.report_data is None:
+            raise MisconfiguredError(
+                "Job needs to define 'report_data' property"
+            )
+
+        rst = render_to_string(
+            self.report_template,
+            {
+                "title": self.help,
+                "start_time": start_time,
+                "end_time": end_time,
+                "report_data": self.report_data,
+            },
+        )
+        html = docutils.core.publish_string(
+            rst,
+            writer_name="html5",
+            settings_overrides={
+                "input_encoding": "unicode",
+                "output_encoding": "unicode",
+            },
+        )
+        mail_admins(
+            subject=f"JOB: {self.help}",
+            message=rst,
+            html_message=html,
+            fail_silently=True,
+        )
+        print(rst)

sigi/apps/utils/mixins.py

@@ -1,20 +1,13 @@
-import datetime
-import docutils.core
-import traceback
 from collections import OrderedDict
 from functools import update_wrapper
 from django import forms
 from django.contrib import admin
 from django.contrib.admin import helpers
-from django.contrib.admin.models import LogEntry, ADDITION, CHANGE
 from django.contrib.admin.options import csrf_protect_m
 from django.contrib.admin.utils import pretty_name
-from django.contrib.contenttypes.models import ContentType
 from django.core.exceptions import PermissionDenied, ImproperlyConfigured
-from django.core.mail import mail_admins
 from django.http import Http404
 from django.http.response import HttpResponse, HttpResponseRedirect
-from django.template.loader import render_to_string
 from django.template.response import TemplateResponse
 from django.urls import path
 from django.utils.translation import gettext as _, ngettext
@@ -26,10 +19,6 @@ from import_export.signals import post_export
 from sigi.apps.utils import field_label
 
 
-class MisconfiguredError(Exception):
-    pass
-
-
 class ValueField(Field):
     def get_value(self, obj):
         if self.attribute is None:
@@ -316,104 +305,3 @@ class ReturnMixin:
         if self._return_path:
             return HttpResponseRedirect(self._return_path)
         return response
-
-
-class JobReportMixin:
-    error_report_template = "emails/report_error.rst"
-    report_template = "emails/base_report.rst"
-    report_data = None
-    sys_user = None
-
-    def execute(self):
-        start_time = datetime.datetime.now()
-
-        try:
-            from sigi.apps.servidores.models import Servidor
-
-            self.sys_user = Servidor.objects.get(sigi=True).user
-        except Exception:
-            pass
-
-        try:
-            self.do_job()
-        except Exception as e:
-            self.report_error(e)
-            return
-        end_time = datetime.datetime.now()
-        self.report(start_time, end_time)
-
-    def do_job(self):
-        raise NotImplementedError("Job needs to implement the 'do_job' method")
-
-    def _admin_log(self, object, action_flag, message=""):
-        if self.sys_user is None:
-            return  # No admin log
-        LogEntry.objects.log_action(
-            user_id=self.sys_user.id,
-            content_type_id=ContentType.objects.get_for_model(type(object)).pk,
-            object_id=object.id,
-            object_repr=str(object),
-            action_flag=action_flag,
-            change_message=message,
-        )
-
-    def admin_log_addition(self, object, message=""):
-        self._admin_log(object, ADDITION, message)
-
-    def admin_log_change(self, object, message=""):
-        self._admin_log(object, CHANGE, message)
-
-    def report_error(self, error):
-        rst = render_to_string(
-            self.error_report_template,
-            {
-                "title": self.help,
-                "traceback": traceback.format_exception(error),
-            },
-        )
-        html = docutils.core.publish_string(
-            rst,
-            writer_name="html5",
-            settings_overrides={
-                "input_encoding": "unicode",
-                "output_encoding": "unicode",
-            },
-        )
-        mail_admins(
-            subject=self.help,
-            message=rst,
-            html_message=html,
-            fail_silently=True,
-        )
-        print(rst)
-
-    def report(self, start_time, end_time):
-        if self.report_data is None:
-            raise MisconfiguredError(
-                "Job needs to define 'report_data' property"
-            )
-
-        rst = render_to_string(
-            self.report_template,
-            {
-                "title": self.help,
-                "start_time": start_time,
-                "end_time": end_time,
-                "report_data": self.report_data,
-            },
-        )
-        html = docutils.core.publish_string(
-            rst,
-            writer_name="html5",
-            settings_overrides={
-                "input_encoding": "unicode",
-                "output_encoding": "unicode",
-            },
-        )
-        mail_admins(
-            subject=f"JOB: {self.help}",
-            message=rst,
-            html_message=html,
-            fail_silently=True,
-        )
-        print(rst)