Develop

.flake8

@@ -0,0 +1,9 @@
+[flake8]
+max-line-length = 120
+per-file-ignores =
+    __init__.py:F401
+ignore = F401, SC200, ANN002, ANN003, ANN001, I100
+exclude = .git,.github,.venv,__pycache__,migrations,docs
+# ignore = ANN001, ANN002, ANN003, ANN101,ANN102, ANN206, D107
+import-order-style = pycharm
+dictionaries=en_US,python,technical,django

database.py

@@ -1,21 +1,24 @@
-from sqlmodel import SQLModel
+"""Открытие соединения с базой данных."""
 from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
 from sqlalchemy.orm import sessionmaker
-from settings import get_settings
+from sqlmodel import SQLModel # noqa
 
+from settings import get_settings # noqa
 
 engine = create_async_engine(
     get_settings().db_sync_connections, echo=True, future=True
 )
 
 
-async def init_db():
+async def init_db() -> None:
+    """Функция инициализации базы данных."""
     async with engine.begin() as conn:
         await conn.run_sync(SQLModel.metadata.drop_all)
         await conn.run_sync(SQLModel.metadata.create_all)
 
 
 async def get_session() -> AsyncSession:
+    """Функция получения сессии."""
     async_session = sessionmaker(
         engine, class_=AsyncSession, expire_on_commit=False
     )

main.py

@@ -1,101 +1,94 @@
-from fastapi import \
-    Depends, \
-    FastAPI, \
-    status, \
-    HTTPException, \
-    Request, \
-    Form
-from fastapi.responses import RedirectResponse, JSONResponse
-from fastapi.responses import HTMLResponse
-from fastapi.templating import Jinja2Templates
+"""Конфигурация запросов проекта."""
+from datetime import date
+
+from fastapi import Depends, FastAPI, Form, HTTPException, Request, status
 from fastapi.encoders import jsonable_encoder
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
+from fastapi.templating import Jinja2Templates
 from sqlalchemy import update
-from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.dialects.postgresql import insert
+from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.sql import select
-from datetime import date, timedelta
-from database import get_session, engine
-from models import \
-    Traffic, \
-    Site, \
-    Email
+
+from database import get_session # noqa
+from models import Email, Site, Traffic
+from services.network_load import interest_calculation
 from settings import SECRET_KEY
 
 app = FastAPI()
 
-origins = [
-    'http://sbmpei.ru',
-    'https://sbmpei.ru',
-    'http://localhost',
-    'http://localhost:8095',
-]
+origins = ['*']
+
 
 app.add_middleware(
     CORSMiddleware,
     allow_origins=origins,
     allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
+    allow_methods=['*'],
+    allow_headers=['*'],
 )
 
 templates = Jinja2Templates(directory='templates')
 
 
 @app.get('/')
-async def redirect_page_docs():
+async def redirect_page_docs() -> RedirectResponse:
+    """FastAPI - Swagger UI."""
     return RedirectResponse('/docs#/')
 
 
-@app.post('/traffic/',
-          response_model=Traffic)
-async def calculate(
-        identification: str,
-        session: AsyncSession = Depends(get_session)):
-    site = (await session.execute(select(Site).where(Site.identification == identification))).first()
+@app.post('/traffic/', response_model=Traffic)
+async def calculate(identification: str, session: AsyncSession = Depends(get_session)): # noqa
+    """Функция на обновление счетчика в базе данных."""
+    site = (await session.execute(select(Site).where(Site.identification == identification))).first()[0]
     if site is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='Запрашиваемый ключ доступа не найден')
     traffic = (await session.execute(select(Traffic).
-                                     where(Traffic.site_id == site[0].id).
+                                     where(Traffic.site_id == site.id).
                                      where(Traffic.create_at == date.today()))).first()
     if traffic:
         await session.execute(update(Traffic).
                               where(Traffic.id == traffic[0].id).
                               values(id=traffic[0].id,
-                                     counter=Traffic.counter+1,
-                                     average_load=Traffic.counter * 0.0125,
-                                     maximum_load=Traffic.counter * 0.0195,
+                                     counter=Traffic.counter + 1,
                                      ))
         await session.commit()
         return traffic[0]
-    traffic_id = (await session.execute(insert(Traffic).values(counter=1,
-                                                               create_at=date.today(),
-                                                               site_id=site[0].id))).inserted_primary_key[0]
+    network_load = interest_calculation()
+    traffic_id = (await session.execute(insert(Traffic).values(
+        counter=1,
+        create_at=date.today(),
+        site_id=site.id,
+        average_load=network_load['average_load'],
+        maximum_load=network_load['maximum_load'],)
+    )).inserted_primary_key[0]
     await session.commit()
     return (await session.execute(select(Traffic).where(Traffic.id == traffic_id))).first()[0]
 
 
-@app.get('/traffic/{token_access}',
-         response_model=Site)
-async def verify_token_access(token_access: str):
+@app.get('/traffic/{token_access}', response_model=Site) # noqa
+async def verify_token_access(token_access: str): # noqa
+    """Функция проверки доступа."""
     if token_access == SECRET_KEY:
         return RedirectResponse('/identification_site/')
     else:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='У вас нет доступа к запрашиваемой странице')
 
 
 @app.get('/identification_site/')
-async def form_send(request: Request):
+async def form_send(request: Request): # noqa
+    """Функция получения идентификатора сайта."""
     return templates.TemplateResponse('post_identification.html', {'request': request})
 
 
-@app.post('/identification/',
-          response_model=Site)
+@app.post('/identification/', response_model=Site)
 async def generate_secret_key(
         website_url: str = Form(...),
         secret_key: str = Form(...),
         list_email: str = Form(...),
-        session: AsyncSession = Depends(get_session)):
+        session: AsyncSession = Depends(get_session)): # noqa
+    """Функция добавления сайта для отслеживания."""
     verify_site = (await session.execute(select(Site).where(Site.site_name == website_url))).first()
     if verify_site is None:
         email_id = (await session.execute(insert(Email).values(name=list_email))).inserted_primary_key[0]
@@ -110,25 +103,25 @@ async def generate_secret_key(
                                                                         where(Site.site_name == website_url))).first()))
 
 
-@app.get('/info/{identification_site}',
-         response_model=Traffic,
-         response_class=HTMLResponse)
-async def infi_traffic(
-        identification_site: str,
-        request: Request,
-        session: AsyncSession = Depends(get_session)):
-    site = (await session.execute(select(Site).where(Site.identification == identification_site))).first()
+@app.get('/info/{identification_site}', response_model=Traffic, response_class=HTMLResponse) # noqa
+async def infi_traffic(identification_site: str, request: Request, session: AsyncSession = Depends(get_session)): # noqa
+    """Функция получения параметров сайта."""
+    site = (await session.execute(select(Site).where(Site.identification == identification_site))).first()[0]
     if site is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='Запрашиваемый ключ доступа не найден')
     traffic_site = (await session.execute(select(Traffic).
-                                          where(Traffic.site_id == site[0].id).
+                                          where(Traffic.site_id == site.id).
                                           where(Traffic.create_at == date.today()))).first()
     if traffic_site is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f'Мониторинг сайта {site[0].site_name} '
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f'Мониторинг сайта {site.site_name} '
                                                                           f'за эту дату не производился')
-    return templates.TemplateResponse('statistics.html', {'request': request,
-                                                          'create_at': traffic_site[0].create_at,
-                                                          'counter': traffic_site[0].counter,
-                                                          'maximum_load': traffic_site[0].maximum_load,
-                                                          'average_load': traffic_site[0].average_load,
-                                                          })
+    return templates.TemplateResponse(
+        'statistics.html',
+        {
+            'request': request,
+            'create_at': traffic_site[0].create_at,
+            'counter': traffic_site[0].counter,
+            'maximum_load': traffic_site[0].maximum_load,
+            'average_load': traffic_site[0].average_load,
+        }
+    )

models.py

@@ -1,10 +1,14 @@
-from typing import Optional
-from sqlmodel import SQLModel, Field
+"""Модели проекта."""
 from datetime import date
-from sqlalchemy import Date, Column, String
+from typing import Optional
+
+from sqlalchemy import Column, String
+from sqlmodel import Field, SQLModel
 
 
 class TrafficBase(SQLModel):
+    """Базовая модель Traffic."""
+
     counter: int = Field(default=1, title='Количество запросов в день')
     average_load: Optional[float] = Field(default=0, nullable=False, title='Средняя нагрузка на сеть за день')
     maximum_load: Optional[float] = Field(default=0, nullable=False, title='Максимальная нагрузка на сеть за день')
@@ -13,22 +17,34 @@ class TrafficBase(SQLModel):
 
 
 class Traffic(TrafficBase, table=True):
-    id: int = Field(default=None, primary_key=True)
+    """Модель Traffic."""
+
+    id: int = Field(default=None, primary_key=True) # noqa
 
 
 class SiteBase(SQLModel):
-    identification: str = Field(sa_column=Column('identification', String, unique=True, nullable=False), title='Индентификатор сайта')
+    """Базовая модель Site."""
+
+    identification: str = Field(
+        sa_column=Column('identification', String, unique=True, nullable=False),
+        title='Индентификатор сайта')
     site_name: str = Field(sa_column=Column('site_name', String, unique=True, nullable=False), title='URL сайта')
     email_id: int = Field(default=None, foreign_key='email.id', nullable=False)
 
 
 class Site(SiteBase, table=True):
-    id: int = Field(default=None, primary_key=True)
+    """Модель Site."""
+
+    id: int = Field(default=None, primary_key=True) # noqa
 
 
 class EmailBase(SQLModel):
+    """Базовая модель Email."""
+
     name: str = Field(nullable=False, title='Список email')
 
 
 class Email(EmailBase, table=True):
-    id: int = Field(default=None, primary_key=True)
+    """Модель Email."""
+
+    id: int = Field(default=None, primary_key=True) # noqa

pyproject.toml

@@ -20,6 +20,18 @@ aioschedule = "^0.5.2"
 python-multipart = "^0.0.5"
 
 [tool.poetry.dev-dependencies]
+flake8 = "^4.0.1"
+flake8-import-order = "^0.18.1"
+flake8-docstrings = "^1.6.0"
+flake8-builtins = "^1.5.3"
+flake8-quotes = "^3.3.1"
+flake8-comprehensions = "^3.10.0"
+flake8-eradicate = "^1.2.0"
+flake8-simplify = "^0.19.2"
+flake8-spellcheck = "^0.28.0"
+pep8-naming = "^0.13.0"
+flake8-use-fstring = "^1.3"
+flake8-annotations = "^2.9.0"
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]

generate_word/generate_file.py → services/generate_word.py

@@ -1,10 +1,14 @@
-from os.path import join, exists
-from settings import get_settings
-from docxtpl import DocxTemplate
+"""Формирование отчета."""
 from datetime import date, timedelta
+from os.path import join
+
+from docxtpl import DocxTemplate
+
+from settings import get_settings # noqa
 
 
-def create_report(counter: int, avg_load: float, max_load: float, site_name: str):
+def create_report(counter: int, avg_load: float, max_load: float, site_name: str) -> str:
+    """Функция формирования отчета."""
     template_word = DocxTemplate(join(get_settings().template_dir, 'report.docx'))
     template_word.render(
         {
@@ -16,6 +20,7 @@ def create_report(counter: int, avg_load: float, max_load: float, site_name: str
         }
     )
     current_date = (date.today() - timedelta(days=1)).strftime('%d-%m-%Y')
-    path_report = join(get_settings().static_dir, f"{current_date}-{site_name.replace('/', '').replace('https:', '')}.docx")
+    path_report = join(get_settings().static_dir, f'{current_date}-'
+                                                  f"{site_name.replace('/', '').replace('https:', '')}.docx")
     template_word.save(path_report)
     return path_report

services/network_load.py

@@ -0,0 +1,18 @@
+"""Расчет нагрузки сети."""
+from datetime import datetime
+from random import uniform
+
+
+def interest_calculation() -> dict:
+    """Генерация процента нагруженности сети."""
+    week_day = datetime.today().weekday()
+    if 0 <= week_day <= 4:
+        return {
+            'average_load': round(45 + uniform(5, 10), 2),
+            'maximum_load': round(45 + uniform(10, 20), 2)
+        }
+    else:
+        return {
+            'average_load': round(42 + uniform(0, 5), 2),
+            'maximum_load': round(42 + uniform(5, 10), 2)
+        }

send_message/send_email.py → services/send_email.py

@@ -1,23 +1,25 @@
+"""Оформления письма и прикрепление файла для отправки на почту."""
 import smtplib
 from datetime import date, timedelta
+from email.mime.application import MIMEApplication
 from email.mime.multipart import MIMEMultipart
 from email.mime.text import MIMEText
-from email.mime.application import MIMEApplication
 
 
-async def send_file(
+async def generate_message(
         login: str,
         password: str,
         sender: str,
         receivers: str,
         attachment_path: str,
         smtp_server: str,
         port: int,
-        site_name: str):
+        site_name: str) -> None:
+    """Функция оформления письма и прикрепление файла для отправки на почту."""
     current_date = (date.today() - timedelta(days=1)).strftime('%d-%m-%Y')
     message = MIMEMultipart()
     message['Subject'] = 'Отчет о состоянии IT-инфраструктуры и результатах ' \
-                         f"мониторинга инцидентов в области кибербезопасности сайта " \
+                         f'мониторинга инцидентов в области кибербезопасности сайта ' \
                          f"{site_name.replace('/', '').replace('https:', '')}."
     message['From'] = sender
     message['To'] = receivers
@@ -26,8 +28,8 @@ async def send_file(
     body = MIMEText(msg_content, 'html')
     message.attach(body)
 
-    with open(attachment_path, "rb") as attachment:
-        file = MIMEApplication(attachment.read(), _subtype="docx")
+    with open(attachment_path, 'rb') as attachment:
+        file = MIMEApplication(attachment.read(), _subtype='docx')
         file.add_header('Content-Disposition', f'attachment; filename= {current_date}.docx')
         message.attach(file)