Merge pull request #60 from ISSResearch/feat-stat-export

Feat stat export

backend-app/file/export.py

@@ -0,0 +1,161 @@
+from abc import ABC, abstractmethod
+from typing import Any, List, Dict
+from io import BytesIO
+from json import dumps
+from xlsxwriter import Workbook
+from xlsxwriter.worksheet import Worksheet
+
+ROW = Dict[str, Any]
+DATA = List[ROW]
+ENCODING = "utf-8"
+IMPLEMENTED = {"json", "csv", "xlsx"}
+
+
+class Export(ABC):
+    ATTRIBUTE_HEADERS = "Attribute,Level,Validation Images,Validation Videos,Accepted Images, Accepted Videos,Declined Images, Declined Videos,Total\n"
+    USER_HEADERS = "User,Validation Images,Validation Videos,Accepted Images, Accepted Videos,Declined Images, Declined Videos,Total\n"
+
+    t_val = lambda _, x: (x.get('image', 0), x.get('video', 0))
+    sm = lambda _, x, y, z: sum(x + y + z)
+
+    def __init__(self, data: DATA, *args):
+        self._data = data
+        self._type = args[0]
+
+    @abstractmethod
+    def into_response(self) -> BytesIO: ...
+
+    @property
+    def _data(self) -> DATA: return self.__data
+
+    @_data.setter
+    def _data(self, data: DATA): self.__data = data
+
+
+class JSON(Export):
+    def into_response(self) -> BytesIO:
+        file = BytesIO()
+
+        prepared_data = bytes(dumps(self._data), encoding=ENCODING)
+
+        file.write(prepared_data)
+        file.seek(0)
+
+        return file
+
+
+class CSV(Export):
+    def _write_attribute(self, dest: BytesIO, data: ROW):
+        name = data.get("name")
+        level_name = data.get("levelName")
+        children = data.get("children", [])
+
+        val = self.t_val(data.get("v", {}))
+        acc = self.t_val(data.get("a", {}))
+        dec = self.t_val(data.get("d", {}))
+        total = self.sm(val, acc, dec)
+
+        dest.write(bytes(
+            f"{name},{level_name},{val[0]},{val[1]},{acc[0]},{acc[1]},{dec[0]},{dec[1]},{total}\n",
+            encoding=ENCODING
+        ))
+
+        for child in children: self._write_attribute(dest, child)
+
+    def _write_user(self, dest: BytesIO, data: ROW):
+        name = data.get("name")
+
+        val = self.t_val(data.get("v", {}))
+        acc = self.t_val(data.get("a", {}))
+        dec = self.t_val(data.get("d", {}))
+        total = self.sm(val, acc, dec)
+
+        dest.write(bytes(
+            f"{name},{val[0]},{val[1]},{acc[0]},{acc[1]},{dec[0]},{dec[1]},{total}\n",
+            encoding=ENCODING
+        ))
+
+    def into_response(self) -> BytesIO:
+        file = BytesIO()
+
+        match self._type:
+            case "attribute":
+                headers = self.ATTRIBUTE_HEADERS
+                write = self._write_attribute
+            case "user":
+                headers = self.USER_HEADERS
+                write = self._write_user
+            case _: raise AttributeError
+
+        file.write(bytes(headers, encoding=ENCODING))
+        for row in self._data: write(file, row)
+
+        file.seek(0)
+
+        return file
+
+
+class XLS(Export):
+    __row_n = 0
+
+    @property
+    def _row_n(self) -> int: return self.__row_n
+
+    @_row_n.setter
+    def _row_n(self, new: int): self.__row_n = new
+
+    def _write_attribute(self, dest: Worksheet, data: ROW):
+        name = data.get("name")
+        level_name = data.get("levelName")
+        children = data.get("children", [])
+
+        val = self.t_val(data.get("v", {}))
+        acc = self.t_val(data.get("a", {}))
+        dec = self.t_val(data.get("d", {}))
+        total = self.sm(val, acc, dec)
+
+        row = (name, level_name, val[0], val[1], acc[0], acc[1], dec[0], dec[1], total)
+
+        for i, item in enumerate(row): dest.write(self._row_n, i, item)
+        self._row_n += 1
+
+        for child in children: self._write_attribute(dest, child)
+
+    def _write_user(self, dest: Worksheet, data: ROW):
+        name = data.get("name")
+
+        val = self.t_val(data.get("v", {}))
+        acc = self.t_val(data.get("a", {}))
+        dec = self.t_val(data.get("d", {}))
+        total = self.sm(val, acc, dec)
+
+        row = (name, val[0], val[1], acc[0], acc[1], dec[0], dec[1], total)
+
+        for i, item in enumerate(row): dest.write(self._row_n, i, item)
+        self._row_n += 1
+
+    def into_response(self) -> BytesIO:
+        file = BytesIO()
+
+        match self._type:
+            case "attribute":
+                headers = self.ATTRIBUTE_HEADERS
+                write = self._write_attribute
+            case "user":
+                headers = self.USER_HEADERS
+                write = self._write_user
+            case _: raise AttributeError
+
+        xl = Workbook(file)
+        sheet = xl.add_worksheet()
+
+        headers = headers.split(",")
+
+        for i, header in enumerate(headers): sheet.write(self._row_n, i, header)
+        self._row_n += 1
+        for row in self._data: write(sheet, row)
+
+        xl.close()
+        file.seek(0)
+
+        return file

backend-app/file/file_tests/export_test.py

@@ -0,0 +1,33 @@
+from file.export import JSON, CSV
+from file.services import StatsServices
+from django.test import TestCase
+from attribute.attribute_tests.mock_attribute import MockCase
+from json import loads
+
+
+class ExportTest(TestCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.case = MockCase()
+        cls.attr_stat, _ = StatsServices.from_attribute(cls.case.project.id)
+        cls.user_stat, _ = StatsServices.from_user(cls.case.project.id)
+
+    # TODO:
+    def test_xls(self): ...
+
+    def test_csv(self):
+        attr_res = CSV(self.attr_stat, "attribute").into_response()
+        user_res = CSV(self.user_stat, "user").into_response()
+
+        attributes = attr_res.read().decode().split("\n")
+        users = user_res.read().decode().split("\n")
+
+        self.assertTrue(len(attributes) == len(users) == 3)
+        # TODO:
+
+    def test_json(self):
+        attr_res = JSON(self.attr_stat, 0).into_response()
+        user_res = JSON(self.user_stat, 0).into_response()
+        self.assertEqual(self.attr_stat, loads(attr_res.read().decode()))
+        self.assertEqual(self.user_stat, loads(user_res.read().decode()))

backend-app/file/file_tests/services_test.py

@@ -4,7 +4,8 @@
     FileUploader,
     StatsServices,
     ViewSetServices,
-    _annotate_files
+    _annotate_files,
+    form_export_file
 )
 from json import dumps
 from attribute.models import AttributeGroup, Attribute
@@ -360,7 +361,6 @@ class StatsServiceTest(TestCase):
     def setUpClass(cls):
         super().setUpClass()
         cls.case = MockCase()
-        cls.case = MockCase()
         cls.empty_project = Project.objects.create(name="some")
 
     def test_from_attribute(self):
@@ -369,47 +369,45 @@ def test_from_attribute(self):
         res, code = StatsServices.from_attribute(self.case.project.id)
 
         check_against = {
-            'attribute__attributegroup__file__file_type': self.case.file_.file_type,
-            'attribute__attributegroup__file__status': self.case.file_.status,
-            'attribute__id': self.case.attribute.id,
-            'attribute__name': self.case.attribute.name,
-            'attribute__parent': self.case.attribute.parent,
-            'count': self.case.file_.attributegroup_set.first().attribute.count(),
-            'name': self.case.level.name,
-            'order': self.case.level.order
+            "id": self.case.attribute.id,
+            "levelName": self.case.level.name,
+            "order": self.case.level.order,
+            "name": self.case.attribute.name,
+            "parent": self.case.attribute.parent,
+            self.case.file_.status or "v": {
+                self.case.file_.file_type or "no data":
+                    self.case.file_.attributegroup_set.first().attribute.count()
+            },
         }
 
         self.assertTrue(empty_code == code == no_proj_code == 200)
         self.assertTrue(empty_res == no_proj_res == [])
 
         self.assertEqual(len(res), self.case.project.file_set.count())
-        self.assertEqual(
-            set(res[0].keys()),
-            set(StatsServices._ATTRIUBE_QUERY_VALUES).union(["count"])
-        )
         self.assertEqual(res[0], check_against)
 
     def test_from_user(self):
         empty_res, empty_code = StatsServices.from_user(self.empty_project.id)
         no_proj_res, no_proj_code = StatsServices.from_user(9999)
         res, code = StatsServices.from_user(self.case.project.id)
 
+        empty_res = list(empty_res)
+        no_proj_res = list(no_proj_res)
+        res = list(res)
+
         check_against = {
-            "author_id": self.case.user.id,
-            "author__username": self.case.user.username,
-            "status": self.case.file_.status,
-            "file_type": self.case.file_.file_type,
-            "count": self.case.project.file_set.count()
+            "id": self.case.user.id,
+            "name": self.case.user.username,
+            self.case.file_.status or "v": {
+                self.case.file_.file_type or "no data":
+                    self.case.project.file_set.count()
+            },
         }
 
         self.assertTrue(empty_code == code == no_proj_code == 200)
         self.assertTrue(empty_res == no_proj_res == [])
 
         self.assertEqual(len(res), self.case.project.file_set.count())
-        self.assertEqual(
-            set(res[0].keys()),
-            set(StatsServices._USER_QUERY_VALUES).union(["count"])
-        )
         self.assertEqual(res[0], check_against)
 
 
@@ -481,3 +479,39 @@ def test_assign_attributes(self):
             set(groups[0].attribute.values_list("id", flat=True)),
             set(meta_groups[0])
         )
+
+
+class ExportServicesTest(TestCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.case = MockCase()
+
+    def test_export(self):
+        self._assert_query_fail({})
+        self._assert_query_fail({"type": 1})
+        self._assert_query_fail({"type": 1, "project_id": 1})
+        self._assert_query_fail({"type": "json", "project_id": 1, "choice": "user"}, True)
+
+        try:
+            form_export_file({"type": "asd", "project_id": 1, "choice": "zxc"})
+            self.assertTrue(False)
+        except Exception as e: self.assertEqual(str(e), "asd not implemented")
+
+        try:
+            form_export_file({"type": "json", "project_id": 1, "choice": "zxc"})
+            self.assertTrue(False)
+        except Exception as e: self.assertEqual(str(e), "export for zxc is not implemented")
+
+    def _assert_query_fail(self, data, intential=False):
+        queries = {"type", "project_id", "choice"}
+
+        string = " must be provided"
+        try:
+            form_export_file(data)
+            self.assertTrue(intential, "not suppose to go there")
+        except Exception as e:
+            self.assertEqual(
+                set(str(e)[:-len(string)].split(", ")),
+                queries - set(data)
+            )