Merge pull request #22 from fga-eps-mds/development

Development

README.md

@@ -71,6 +71,7 @@ EMAIL_ADDRESS= "YOUR-EMAIL-ADDRESS"
 EMAIL_APP_PASSWORD= "YOUR-EMAIL-PASSWORD-FOR-APPs"
 ```
 
+[how to generate email password for apps](https://support.google.com/mail/answer/185833?hl=en)  
 You can read more about BaseSettings class here: https://pydantic-docs.helpmanual.io/usage/settings/
 
 ## Pre-commit

gestao/tests/test_document.py

@@ -0,0 +1,41 @@
+from io import BytesIO
+from urllib import parse
+
+import pytest
+from fastapi import FastAPI
+from httpx import AsyncClient
+from starlette import status
+
+from gestao.web.api.document.utils import generate_report_users_file
+
+
+def test_generate_report_users_file_correct() -> None:
+    file_stream = generate_report_users_file([], [])
+    assert isinstance(file_stream, BytesIO)
+
+
+def test_generate_report_users_file_incorrect() -> None:
+    try:
+        file_stream = generate_report_users_file(1, 2)
+    except Exception as e:
+        assert isinstance(e, TypeError)
+
+
+@pytest.mark.anyio
+async def test_get_report_users_NoFilter(
+    client: AsyncClient, fastapi_app: FastAPI
+) -> None:
+    url = fastapi_app.url_path_for("get_report_users")
+    response = await client.get(url)
+    assert response.status_code == 200
+
+
+@pytest.mark.anyio
+async def test_get_report_users_Filter(
+    client: AsyncClient, fastapi_app: FastAPI
+) -> None:
+    url = fastapi_app.url_path_for("get_report_users")
+    filter = ["fullName", "registration", "bloodType"]
+    url += "?" + parse.urlencode({"filter_list": filter})
+    response = await client.get(url)
+    assert response.status_code == 200

gestao/tests/utils.py

@@ -25,7 +25,6 @@ def generate_fake_user():
         "scolarity": "null",
         "religion": "null",
         "bloodType": "null",
-        "function": "null",
         "actualWorkSituation": "null",
         "admissionDate": "2023-01-01",
         "systemRole": "null",

gestao/web/api/document/__init__.py

@@ -0,0 +1,3 @@
+from gestao.web.api.document.views import router
+
+__all__ = ["router"]

gestao/web/api/document/utils.py

@@ -0,0 +1,134 @@
+import io
+from datetime import datetime
+
+from docx import Document
+from openpyxl import Workbook
+from python_docx_replace import docx_replace
+
+from gestao.db.models.user import User
+
+
+def generate_affiliation_file(user: User):
+    input_path = "./files/affiliation_template.docx"
+    now = datetime.now()
+    month_switch = {
+        1: "janeiro",
+        2: "fevereiro",
+        3: "março",
+        4: "abril",
+        5: "maio",
+        6: "junho",
+        7: "julho",
+        8: "agosto",
+        9: "setembro",
+        10: "outubro",
+        11: "novembro",
+        12: "dezembro",
+    }
+    date_string = f"{now.day} de {month_switch.get(now.month)} de {now.year}"
+    doc = Document(input_path)
+    docx_replace(
+        doc,
+        **{
+            "cpf": user.cpf,
+            "name": user.fullName,
+            "date": date_string,
+        },
+    )
+
+    file_stream = io.BytesIO()
+    doc.save(file_stream)
+    file_stream.seek(0)
+
+    return file_stream
+
+
+def generate_report_users_file(all_users_list: list, filter_list: list):
+    template_data = {
+        "index": [],
+        "fullName": [],
+        "warName": [],
+        "registration": [],
+        "rg": [],
+        "cpf": [],
+        "placeOfBirth": [],
+        "ufNatural": [],
+        "civilState": [],
+        "cep": [],
+        "address": [],
+        "number": [],
+        "neighborhood": [],
+        "city": [],
+        "complement": [],
+        "uf": [],
+        "email": [],
+        "cellphone": [],
+        "phone": [],
+        "gender": [],
+        "motherName": [],
+        "fatherName": [],
+        "scolarity": [],
+        "religion": [],
+        "bloodType": [],
+        "actualWorkSituation": [],
+        "admissionDate": [],
+        "jobRole": [],
+        "bodyOfLaw": [],
+        "lotation": [],
+        "workPost": [],
+    }
+
+    for user in all_users_list:
+        template_data["index"].append(all_users_list.index(user) + 1)
+        template_data["fullName"].append(user.fullName)
+        template_data["warName"].append(user.warName)
+        template_data["registration"].append(user.registration)
+        template_data["rg"].append(user.rg)
+        template_data["cpf"].append(user.cpf)
+        template_data["placeOfBirth"].append(user.placeOfBirth)
+        template_data["ufNatural"].append(user.ufNatural)
+        template_data["civilState"].append(user.civilState)
+        template_data["cep"].append(user.cep)
+        template_data["address"].append(user.address)
+        template_data["number"].append(user.number)
+        template_data["neighborhood"].append(user.neighborhood)
+        template_data["city"].append(user.city)
+        template_data["complement"].append(user.complement)
+        template_data["uf"].append(user.uf)
+        template_data["email"].append(user.email)
+        template_data["cellphone"].append(user.cellphone)
+        template_data["phone"].append(user.phone)
+        template_data["gender"].append(user.gender)
+        template_data["motherName"].append(user.motherName)
+        template_data["fatherName"].append(user.fatherName)
+        template_data["scolarity"].append(user.scolarity)
+        template_data["religion"].append(user.religion)
+        template_data["bloodType"].append(user.bloodType)
+        template_data["actualWorkSituation"].append(user.actualWorkSituation)
+        template_data["admissionDate"].append(user.admissionDate)
+        template_data["jobRole"].append(user.jobRole)
+        template_data["bodyOfLaw"].append(user.bodyOfLaw)
+        template_data["lotation"].append(user.lotation)
+        template_data["workPost"].append(user.workPost)
+
+    if len(filter_list) == 0:
+        users_data = template_data
+    else:
+        users_data = {
+            key: value for key, value in template_data.items() if key in filter_list
+        }
+
+    wb = Workbook()
+    active_spreadsheet = wb.active
+
+    columns = list(users_data.keys())
+    active_spreadsheet.append(columns)
+
+    for line in zip(*users_data.values()):
+        active_spreadsheet.append(line)
+
+    file_stream = io.BytesIO()
+    wb.save(file_stream)
+    file_stream.seek(0)
+
+    return file_stream

gestao/web/api/document/views.py

@@ -0,0 +1,49 @@
+import logging
+from typing import List
+
+from fastapi import APIRouter, HTTPException
+from fastapi.responses import StreamingResponse
+
+from gestao.db.models.user import User
+from gestao.web.api.document.utils import (
+    generate_affiliation_file,
+    generate_report_users_file,
+)
+
+router = APIRouter()
+
+
+@router.get("/affiliation/{user_id}")
+async def get_user_affiliation(user_id: str) -> None:
+    try:
+        user = await User.objects.get(id=user_id)
+        file_stream = generate_affiliation_file(user)
+        file_name = "affiliation-doc.docx"
+        return StreamingResponse(
+            file_stream,
+            media_type=(
+                "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
+            ),
+            headers={
+                "Content-Disposition": f"attachment; filename={file_name}",
+            },
+        )
+    except Exception:
+        logging.error("Error occurred while get user", exc_info=True)
+        raise HTTPException(status_code=404, detail="User not found")
+
+
+@router.get("/report-users/")
+async def get_report_users(filter_list: List[str] = []):
+    users_list = await User.objects.all()
+    file_stream = generate_report_users_file(list(users_list), filter_list)
+    file_name = "report-users.xlsx"
+    return StreamingResponse(
+        file_stream,
+        media_type=(
+            "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
+        ),
+        headers={
+            "Content-Disposition": f"attachment; filename={file_name}",
+        },
+    )

gestao/web/api/login/utils.py

@@ -12,10 +12,7 @@ def generate_password(length: int = 8) -> str:
 
 
 def send_email(
-    user_name: str,
-    user_email: str,
-    new_password: str,
-    logo_path: str,
+    user_name: str, user_email: str, new_password: str, logo_path: str
 ) -> None:
     email_data = {
         "email_address": os.getenv("EMAIL_ADDRESS"),

gestao/web/api/router.py

@@ -1,9 +1,10 @@
 from fastapi.routing import APIRouter
 
-from gestao.web.api import echo, login, monitoring, user
+from gestao.web.api import document, echo, login, monitoring, user
 
 api_router = APIRouter()
 api_router.include_router(monitoring.router)
 api_router.include_router(echo.router, prefix="/echo", tags=["echo"])
 api_router.include_router(user.router, prefix="/users", tags=["users"])
 api_router.include_router(login.router, prefix="/login", tags=["login"])
+api_router.include_router(document.router, prefix="/documents", tags=["documents"])

gestao/web/api/user/views.py

@@ -19,10 +19,10 @@ async def get_users(
     offset: int = 0,
 ) -> List[User]:
     return (
-        await User.objects.limit(limit).offset(
+        await User.objects.limit(limit)
+        .offset(
             offset,
         )
-        # .filter(status=UserStatus.active)
         .all()
     )
 
@@ -44,9 +44,7 @@ async def create_user(create_user: CreateUserDTO) -> User:
         create_user_dict = create_user.dict()
         dependents = create_user_dict.pop("dependents", [])
         user_id = str(uuid4())
-
-        create_user_dict["password"] = create_user_dict["warName"]
-
+        create_user_dict["password"] = create_user_dict["cpf"]
         await User.objects.create(
             id=user_id, **create_user_dict, status=UserStatus.analyzing
         )