---
title: "Система сбора логов и обработки ошибок"
---
flowchart TB
    classDef person fill:#08427b,color:#ffffff,stroke:#052456
    classDef system fill:#1168bd,color:#ffffff,stroke:#0b4883
    classDef container fill:#438dd5,color:#ffffff,stroke:#2a6ea8
    classDef db fill:#438dd5,color:#ffffff,stroke:#2a6ea8

    subgraph External["Внешние системы"]
        App1["Приложение 1<br/>(Генератор событий)"]:::system
        App2["Приложение 2<br/>(Генератор отчетов)"]:::system
        App3["Приложение 3<br/>(Генератор событий)"]:::system
    end

    subgraph MainService["Сервис сбора логов и обработки ошибок"]
        APILayer["API<br/>- /error<br/>- /upload_log<br/>- /report/send"]:::container
        Scheduler["Scheduler<br/>- Обработка файлов<br/>- Пакетная вставка в БД<br/>- Удаление обработанных файлов"]:::container
        ReportGen["Report Generator<br/>- Агрегация данных по сервисам<br/>- Генерация HTML-отчетов"]:::container
        MailService["Mail Service<br/>- Отправка отчета<br/>- Отправка уведомлений об ошибках"]:::container
    end

    DB[("БД")]:::db
    User["Email Client<br/>(Пользователь)<br/>- Получение отчетов<br/>- Получение уведомлений об ошибках"]:::person

    App1 -->|<span style="color:yellow">POST /error</span>| APILayer
    App2 -->|<span style="color:orange">POST /report/send</span>| APILayer
    App3 -->|"POST /upload_log"| APILayer
    APILayer -->|"Передача файлов"| Scheduler
    APILayer -->|<span style="color:yellow">Передача ошибки</span>| MailService
    APILayer -->|<span style="color:orange">Формирование отчета</span>| ReportGen
    ReportGen -->|<span style="color:orange">Запрос данных</span>| DB
    Scheduler -->|"Сохранение данных"| DB
    
    ReportGen -->|<span style="color:orange">Отправка отчета</span>| MailService
    MailService -->|"Email уведомления"| User

from fastapi import FastAPI, BackgroundTasks, UploadFile, File, HTTPException, Request
from api_contract.contract import error
from fastapi.responses import JSONResponse
from mail_sender.mail_sender import MailSend 
from core.utils import HandlerJsonData 
import asyncio
import os
import aiofiles
from dotenv import load_dotenv
from contextlib import asynccontextmanager
from core.handler_file import HandlerLogFile

# ---------- Настройка обработки лог-файлов ----------

load_dotenv()
PROCESSING_DIR = os.getenv("LOG_PROCESSING_DIR")
os.makedirs(PROCESSING_DIR, exist_ok=True)

MessageSend = MailSend()
HandlerData = HandlerJsonData()

# ---------- Функция получения реального IP клиента ----------
def GetClientIp(request: Request) -> str:
    """Пытается определить реальный IP клиента, учитывая прокси-заголовки."""
    forwarded = request.headers.get("X-Forwarded-For")
    if forwarded:
        return forwarded.split(",")[0].strip()# берём первый IP в цепочке
    realIp = request.headers.get("X-Real-IP")
    if realIp:
        return realIp.strip()
    client = request.client # fallback на прямой IP соединения
    return client.host if client else "unknown"

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Запускаем синхронный шедулер в фоновом потоке
    HandlerLogFile.startScheduler()
    yield
    

app = FastAPI(lifespan=lifespan)

# ---------- Глобальная проверка IP (middleware) ----------
ALLOWED_IPS = {"твое IP","твое IP"}

@app.middleware("http")
async def ip_whitelist_middleware(request: Request, call_next):
    # Эндпоинт /health доступен всем (для мониторинга)
    if request.url.path == "/health":
        return await call_next(request)

    clientIp = GetClientIp(request)
    if clientIp not in ALLOWED_IPS:
        return JSONResponse(
            status_code=403,
            content={"detail": "Access denied"}
        )
    return await call_next(request)

# ---------- API эндпоинты ----------

@app.post("/error")
async def error_sender(input: error, background_tasks: BackgroundTasks):
    # асинхронная обработка запросов
    async def task_send():
        await asyncio.to_thread(MessageSend.ErrorMessage, input)

    background_tasks.add_task(task_send)
    return JSONResponse(content={"message": "success"}, status_code=200)

@app.get("/report/send")
def report_data():
    htmlReport = HandlerData.SendReport()
    MessageSend.ReportMessage(htmlReport)
    return JSONResponse(content={"message": "success"}, status_code=200)

@app.get("/health")
async def health_answer():
    return JSONResponse(content={"message": "I am alive"}, status_code=200)

@app.post("/upload_log")
async def upload_log(file: UploadFile = File(...)):
    # Валидация расширения
    if not file.filename.endswith('.log'):
        raise HTTPException(status_code=400, detail="File not allowed")
    
    # Проверка, что файл не пустой
    content = await file.read(1024)  # читаем первые 1KB для валидации
    if not content:
        raise HTTPException(status_code=400, detail="Empty file")
    
    
    # Сохраняем файл в директорию для обработки
    file_path = os.path.join(PROCESSING_DIR, file.filename)
    await file.seek(0)  # возвращаем курсор в начало после чтения
    async with aiofiles.open(file_path, 'wb') as out_file:
        content = await file.read()
        await out_file.write(content)
    
    return JSONResponse(content={"message": f"File {file.filename} uploaded successfully"}, status_code=200)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app=app, host="127.0.0.1", port=8000)

from pydantic import BaseModel, Field
from typing import Dict, Any

class error(BaseModel):
    error_message: str = Field(default=..., min_length=1, max_length=2000, description="Текст ошибки")
    sender_service: str = Field(default=..., min_length=1, max_length=150, description="Сервис отправитель")

class data_report(BaseModel):
    type_report: str = Field(default=..., min_length=1, max_length=10, description="Тип отчета")
    data: Dict[str, Any] = Field(default=..., min_length=1, max_length=1000, description="Данные для отчета")

import os
import json
import time
import threading
from typing import List, Dict, Any
from db.db import GetSession
from db.model import LogMessages
from api_contract.contract import error
from mail_sender.mail_sender import MailSend
from datetime import datetime

MessageSend = MailSend()
PROCESSING_DIR = os.getenv("LOG_PROCESSING_DIR", "./logs_to_process")

class HandlerLogFile:
    @staticmethod
    def errorSend(error_text: str, sender_service: str = "Log Handler") -> None:
        """Отправляет ошибку (синхронно)."""
        error_msg = error(error_message=error_text, sender_service=sender_service)
        MessageSend.ErrorMessage(error_msg)  # предполагаем, что метод синхронный

    @staticmethod
    def insertBatch(batch: List[Dict[str, Any]], maxRetries: int = 2) -> bool:
        for attempt in range(1, maxRetries + 1):
            try:
                with GetSession() as session:
                    objects = []
                    for item in batch:
                        # 1. Преобразуем поле event -> type_event, если нужно
                        if 'event' in item:
                            item['event'] = item.pop('event')
                        
                        # 2. Преобразуем строку date в объект datetime
                        if 'date' in item and isinstance(item['date'], str):
                            # Заменяем 'Z' на '+00:00' для корректного парсинга UTC
                            date_string = item['date'].replace('Z', '+00:00')
                            item['date'] = datetime.fromisoformat(date_string)
                        
                        # 3. Отбираем только нужные поля
                        allowed_fields = {'service', 'event', 'status', 'status_description', 'message', 'date', 'created_at'}
                        filtered_item = {k: v for k, v in item.items() if k in allowed_fields}
                        objects.append(LogMessages(**filtered_item))
                    
                    session.add_all(objects)
                    session.commit()
                    return True
            except Exception as e:
                if attempt == maxRetries:
                    print(f"Insert failed after {maxRetries} attempts: {e}")
                    return False
                time.sleep(2 ** attempt)
        return False

    @staticmethod
    def processLogFile(filePath: str) -> None:
        """Синхронно читает файл, валидирует JSON и вставляет батчами."""
        batch = []
        batchSize = 30

        try:
            with open(filePath, 'r', encoding='utf-8') as f:
                for line in f:
                    line = line.strip()
                    if not line:
                        continue
                    try:
                        data = json.loads(line)
                        batch.append(data)
                    except json.JSONDecodeError:
                        continue  # пропускаем битые строки

                    if len(batch) >= batchSize:
                        if not HandlerLogFile.insertBatch(batch):
                            print(f"Failed to insert batch from {filePath}, stopping")
                            return  # файл не удаляем
                        batch = []

            # Оставшиеся записи
            if batch:
                if not HandlerLogFile.insertBatch(batch):
                    print(f"Failed to insert final batch from {filePath}")
                    return

            # Удаляем файл только после полной успешной обработки
            os.remove(filePath)
            print(f"Processed and deleted: {filePath}")

        except Exception as e:
            HandlerLogFile.errorSend(f"Failed to process file {os.path.basename(filePath)}: {e}")

    @staticmethod
    def schedulerLoop() -> None:
        """Синхронный цикл шедулера (запускается в отдельном потоке)."""
        while True:
            try:
                files = []
                if not os.path.exists(PROCESSING_DIR):
                    os.makedirs(PROCESSING_DIR, exist_ok=True)
                for f in os.listdir(PROCESSING_DIR):
                    filePath = os.path.join(PROCESSING_DIR, f)
                    if os.path.isfile(filePath) and f.endswith('.log'):
                        files.append((filePath, os.path.getctime(filePath)))
                # Сортируем по времени создания (старые первыми)
                files.sort(key=lambda x: x[1])
                for filePath, _ in files:
                    HandlerLogFile.processLogFile(filePath)
            except Exception as e:
                print(f"Scheduler error: {e}")
            time.sleep(300)  # 5 минут

    @staticmethod
    def startScheduler() -> None:
        """Запускает шедулер в фоновом демон-потоке."""
        thread = threading.Thread(target=HandlerLogFile.schedulerLoop, daemon=True)
        thread.start()
        print("Scheduler thread started")

from sqlalchemy import DateTime, String, Text, Integer, Identity, PrimaryKeyConstraint, Index, text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime
from typing import Optional

class Base(DeclarativeBase):
    pass

class LogMessages(Base):
    __tablename__ = 'log_messages'
    __table_args__ = (
        PrimaryKeyConstraint('id', name='log_messages_pkey'),
        Index('date_index', 'date'),
        Index('service_index', 'service'),
        {'schema': 'log'}
    )

    id: Mapped[int] = mapped_column(Integer, Identity(always=True, start=1, increment=1, minvalue=1, maxvalue=2147483647, cycle=False, cache=1), primary_key=True)
    service: Mapped[Optional[str]] = mapped_column(String(100))
    event: Mapped[Optional[str]] = mapped_column(String(100))
    status: Mapped[Optional[str]] = mapped_column(String(100))
    status_description: Mapped[Optional[str]] = mapped_column(String(100))
    message: Mapped[Optional[str]] = mapped_column(Text)
    date: Mapped[Optional[datetime]] = mapped_column(DateTime(True))
    created_at: Mapped[Optional[datetime]] = mapped_column(DateTime(True), server_default=text('now()'))

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from contextlib import contextmanager
import os
from urllib.parse import quote_plus

dbname = os.getenv("DBNAME")
user = os.getenv("USER")
password = quote_plus(os.getenv("PASSWORD_DB", ""))
host = os.getenv("HOST")

SYNC_DB_URL = f"postgresql+psycopg2://{user}:{password}@{host}/{dbname}"
sync_engine = create_engine(SYNC_DB_URL, echo=False)

SyncSessionLocal = sessionmaker(bind=sync_engine)

@contextmanager
def GetSession():
    session = SyncSessionLocal()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

from db.db import GetSession
from db.model import LogMessages
from decorators.decorators import catch_all_exceptions
import json
import os
from sqlalchemy import func, select


class HandlerJsonData():
    """класс для управления загрузки данных в LocalDB"""

    def __init__(self):
        pass

    @catch_all_exceptions
    def SendReport(self):
        
        with open(os.path.join(os.getcwd(), "config","name_services.json"), 'r', encoding='utf-8') as f:
            config = json.load(f)
            services = config.get("NAME_SERVICE")
       
        # 2. Используем контекстный менеджер для безопасной работы с БД
        with GetSession() as session:
    
            tables = []   # здесь будем собирать все таблицы для отчёта

            for service_name in services:
                # Формируем запрос для конкретного сервиса
                stmt = select(
                    LogMessages.service,
                    LogMessages.event,
                    LogMessages.status,
                    LogMessages.status_description,
                    func.count().label("count")
                ).where(
                    LogMessages.service == service_name,
                    # Вчерашний день: данные за 24 часа
                    LogMessages.date >= func.current_date() - 1,
                    LogMessages.date < func.current_date()
                ).group_by(
                    LogMessages.service,
                    LogMessages.event,
                    LogMessages.status,
                    LogMessages.status_description
                ).order_by(
                    func.count().desc()
                )

                results = session.execute(stmt).all()

                # Преобразуем результат в список списков для таблицы
                data_rows = []
                for row in results:
                    data_rows.append([
                        row.service,
                        row.event,
                        row.status,
                        row.status_description,
                        row.count
                    ])

                # Формируем структуру таблицы для этого сервиса
                table = {
                    "title": service_name,   # заголовок таблицы – имя сервиса
                    "headers": ["service", "event", "status", "status_description", "count"],
                    "data": data_rows
                }
                
                tables.append(table)
                
        return self.PrepareHtml("Отчеты по работе сервисов за последние 24 часа", tables)    
    
    def PrepareHtml(self, body: str, tables: list = None):
        html_body = f"<html><body><p>{body}</p>"
        
        # Если есть таблицы
        if tables:
            for i, table in enumerate(tables):
                table_data = table.get('data')
                table_headers = table.get('headers')
                table_title = table.get('title', f'Таблица {i+1}')
                
                # Добавляем заголовок таблицы
                html_body += f"<h4 style='margin-bottom: 2px;'>{table_title}</h4>"
                
                # Создаем таблицу
                html_body += "<table border='1' cellpadding='5' style='border-collapse: collapse; margin-bottom: 5px;'>"
                
                # Добавляем заголовки столбцов
                if table_headers:
                    html_body += "<tr>"
                    for header in table_headers:
                        html_body += f"<th style='background-color: #f2f2f2;'>{header}</th>"
                    html_body += "</tr>"
                
                # Добавляем данные
                if table_data:
                    for row in table_data:
                        html_body += "<tr>"
                        for cell in row:
                            html_body += f"<td>{cell}</td>"
                        html_body += "</tr>"
                
                html_body += "</table>"
        
        html_body += "</body></html>"
        return html_body 

import smtplib
from email.mime.text import MIMEText
from email.header import Header
from email.mime.multipart import MIMEMultipart
import os
from dotenv import load_dotenv
from api_contract.contract import error


load_dotenv()

class MailSend():

        def __init__(self):
                self.server = os.getenv('SERVER')
                self.port = os.getenv('PORT')
                self.username = os.getenv('USERNAME_MAIL')
                self.password = os.getenv('PASSWORD')
                self.from_address = os.getenv('FROM_ADDRESS')
                self.to_address = os.getenv('TO_ADDRESS')

        def ErrorMessage(self, errorMessage: error):
                # Тема и тело письма
                subject = f"Ошибка от сервиса {errorMessage.sender_service}"
                body = f"Сервис {errorMessage.sender_service}.\n\nРасшифровка:\n {errorMessage.error_message}"
                
                msg = MIMEText(body.encode('utf-8'), _charset='utf-8')
                msg['Subject'] = Header(subject, 'utf-8')
                msg['From'] = self.from_address
                msg['To'] = self.to_address
                
                self.SenderMessage(msg.as_string())

        def ReportMessage(self, report: str):
                # Тема и тело письма
                subject = f"Отчет за сутки"

                msg = MIMEMultipart('alternative')
                msg['Subject'] = Header(subject, 'utf-8')
                msg['From'] = self.from_address
                msg['To'] = self.to_address

                # Создаем HTML-часть с правильным Content-Type
                html_part = MIMEText(report.encode('utf-8'), 'html', 'utf-8')
                
                # Добавляем HTML-часть в письмо
                msg.attach(html_part)

                # отправляем по почте
                self.SenderMessage(msg.as_string())

        def SenderMessage(self, msg: str):
                
                with smtplib.SMTP_SSL(self.server, self.port) as server:
                        # Авторизация на сервере
                        server.login(self.username, self.password)
                        
                        # Отправка письма
                        server.sendmail(self.from_address, self.to_address, msg)

                        server.quit()

from functools import wraps
from mail_sender.mail_sender import MailSend 
import traceback
from api_contract.contract import error

# если при обработки произойдет ошибка
def catch_all_exceptions(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as ex:
                errorMessage = (
                f"Ошибка: {type(ex).__name__}\n"
                f"Сообщение: {ex}\n"
                f"Трейсбэк: {''.join(traceback.format_tb(ex.__traceback__))}"
                )
                service_name = "service msgbox"
                # Создаем объект ошибки согласно модели
                error_data = error(
                    error_message=errorMessage[:1000],  # Обрезаем до максимальной длины
                    sender_service=service_name[:150]  # Обрезаем до максимальной длины
                )
        
                mail = MailSend()
                mail.ErrorMessage(error_data)
                raise
    return wrapper

{
    "NAME_SERVICE": ["app1", "app2"]
}

SERVER =
PORT =
USERNAME_MAIL =
PASSWORD =
FROM_ADDRESS =
TO_ADDRESS =
LOG_PROCESSING_DIR =
DBNAME =
USER =
PASSWORD_DB = 
HOST =
SCHEMA =

FROM python:3.11.4-slim

# Создаем рабочую директорию
RUN mkdir -p /usr/src/app/
WORKDIR /usr/src/app/

# Копируем только папку src в рабочую директорию
COPY app/. /usr/src/app/

# Устанавливаем pip последней версии и необходимые зависимости
RUN apt-get update && \
    apt-get install -y libpq-dev postgresql-client && \
    python -m pip install --upgrade pip && \
    pip install --no-cache-dir -r requirements.txt && \
    rm -rf /root/.cache/pip

# Устанавливаем часовой пояс
ENV TZ Europe/Moscow

EXPOSE 8000

# Указываем точку входа приложения
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

import requests
import os
from dotenv import load_dotenv
from pathlib import Path
import yaml
from yaml.loader import SafeLoader

load_dotenv() 

def SendError(nameService: str, descriptionError: str):
    """отправка ошибки"""
    
    param = {
             "error_message": descriptionError,
             "sender_service": nameService
            }
        
    EVENT_URL = os.getenv("EVENT_URL") # забираем адрес

    requests.post(f"{EVENT_URL}/error", json=param)

def SendLogs():
    """Отправляет все .log файлы из директории LOG_PATH на сервер."""
  
    with open(os.path.join(os.getcwd(),"config","config.yaml"), "r") as stream:
        configSystem = yaml.load(stream, Loader=SafeLoader)

    
    EVENT_URL = os.getenv("EVENT_URL").rstrip('/')
    logPath = Path(configSystem['LogPath']['LOG_PATH'])
    # Находим все .log файлы в корне папки
    logFiles = list(logPath.glob("*.log"))
    if not logFiles:
        return
    
    for logFile in logFiles:

        with open(logFile, 'rb') as f:
            files = {'file': (logFile.name, f, 'application/octet-stream')}
            response = requests.post(f"{EVENT_URL}/upload_log", files=files)
        
        if response.status_code == 200:
            logFile.unlink()# Удаляем файл только при успешной отправке
                
    remaining = list(logPath.glob("*.log"))# После обработки всех файлов проверяем, остались ли ещё .log файлы
    
    if not remaining:
        os.environ["LOG_LATEST_FILE"] = ""