1234

server/api-data-base/.env

@@ -0,0 +1,2 @@
+DOCS_USERNAME=your_username
+DOCS_PASSWORD=your_password

server/api-data-base/README.md

@@ -0,0 +1,2 @@
+# api-data-base
+ 

server/api-data-base/data_base/config.py

@@ -0,0 +1,32 @@
+from pydantic.generics import GenericModel  
+import re
+
+class Settings(GenericModel):
+    DB_HOST: str = "postgres"#"postgres"
+    DB_PORT: int = 5432
+    DB_USER: str = "tester"
+    DB_PASS: str = "tester"
+    DB_NAME: str = "test"
+
+    @classmethod
+    def validate_db_port(cls, v):
+        # Проверка и преобразование DB_PORT
+        if isinstance(v, str):
+            # Используем регулярное выражение для извлечения целого числа
+            match = re.search(r'\d+', v)
+            if match:
+                return int(match.group())
+            else:
+                raise ValueError(f"Invalid DB_PORT value: {v}")
+        return v
+
+    @property 
+    def DATABASE_URL_asyncpg(self) -> str:
+        return f"postgresql+asyncpg://{self.DB_USER}:{self.DB_PASS}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"
+    
+    @property 
+    def DATABASE_URL_syncpg(self) -> str:
+        return f"postgresql+psycopg2://{self.DB_USER}:{self.DB_PASS}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"
+
+# Загрузка настроек из файла .env
+settings = Settings()

server/api-data-base/data_base/data_base.py

@@ -0,0 +1,18 @@
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from sqlalchemy.orm import sessionmaker, DeclarativeBase
+from sqlalchemy import create_engine
+from data_base.config import settings
+
+# Создание асинхронного движка
+async_engine = create_async_engine(url=settings.DATABASE_URL_asyncpg, echo=True)
+
+# Создание синхронного движка
+sync_engine = create_engine(url=settings.DATABASE_URL_syncpg, echo=True)
+
+# Создание асинхронной и синхронной сессий
+async_session = async_sessionmaker(async_engine)
+sync_session = sessionmaker(sync_engine)
+
+# Базовый класс для декларативных моделей
+class Base(DeclarativeBase):
+    pass

server/api-data-base/data_base/models/main_tables.py

@@ -0,0 +1,94 @@
+from sqlalchemy import Integer, String,PrimaryKeyConstraint,UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+from data_base.data_base import Base
+
+
+
+class users(Base):
+    __tablename__ = "users"
+    id:Mapped[int]=mapped_column(Integer(), primary_key=True)
+    name_poo:Mapped[str]=mapped_column(String(50))
+    inn:Mapped[str]=mapped_column(String(50))#login
+    password:Mapped[str]=mapped_column(String(120))
+    mail:Mapped[str]=mapped_column(String(50))
+    __table_args__ = (
+        PrimaryKeyConstraint('id', name='user_pk'),
+        UniqueConstraint('inn')
+    )
+    
+class Classification_professions(Base):
+    """Классификация профессий"""
+    __tablename__ = "Classification_professions"
+    id:Mapped[int]=mapped_column(Integer(), primary_key=True)
+    name:Mapped[str]=mapped_column(String(),nullable=False)
+    __table_args__ = (
+        PrimaryKeyConstraint('id', name='id_pk_Classification_professions'),
+        UniqueConstraint('name')
+    )
+    
+class Professions(Base):
+    """Профессии"""
+    __tablename__ = "Professions"
+    id:Mapped[int]=mapped_column(Integer(), primary_key=True)
+    name:Mapped[str]=mapped_column(String(),nullable=False)
+    __table_args__ = (
+        PrimaryKeyConstraint('id', name='id_pk_Professions'),
+        UniqueConstraint('name')
+    )
+    
+class Positions(Base):
+    """Должность"""
+    __tablename__ = "Positions"
+    id:Mapped[int]=mapped_column(Integer(), primary_key=True)
+    name:Mapped[str]=mapped_column(String(),nullable=False)
+    __table_args__ = (
+        PrimaryKeyConstraint('id', name='id_pk_Positions'),
+        UniqueConstraint('name')
+    )
+    
+class Specialization(Base):
+    """Специализация"""
+    __tablename__ = "Specialization"
+    id:Mapped[int]=mapped_column(Integer(), primary_key=True)
+    name:Mapped[str]=mapped_column(String(),nullable=False)
+    __table_args__ = (
+        PrimaryKeyConstraint('id', name='id_pk_Specialization'),
+        UniqueConstraint('name')
+    )
+
+class Manufacturing_enterprises(Base):
+    """
+    производственные предприятия
+    (id:int,name:str,INN_company:str)
+    """
+    __tablename__ = "Manufacturing_enterprises"
+    id:Mapped[int]=mapped_column(Integer(),primary_key=True)
+    name:Mapped[str]=mapped_column(String(),nullable=False)
+    INN_company:Mapped[str]=mapped_column(String(), nullable=False)
+    
+
+class Enterprises(Base):
+    """Предприятия"""
+    __tablename__ = "Enterprises"
+    id:Mapped[int]=mapped_column(Integer(), primary_key=True)
+    name:Mapped[str]=mapped_column(String(),nullable=False)
+    __table_args__ = (
+        PrimaryKeyConstraint('id', name='id_pk_Enterprises'),
+        UniqueConstraint('name')
+    )
+
+class Сollection_form_opk(Base):
+    """Поо"""
+    __tablename__="Сollection_form_opk"
+    id:Mapped[int]=mapped_column(Integer(), primary_key=True)
+    время_создания:Mapped[str]=mapped_column(String(),nullable=False)
+    указаное_время:Mapped[str]=mapped_column(String(),nullable=False)
+    ПОО:Mapped[str]=mapped_column(String(),nullable=False)
+    Субъект_рф:Mapped[str]=mapped_column(String(),nullable=False)
+    ИНН_ОПК:Mapped[str]=mapped_column(String(),nullable=False)
+    ОПК:Mapped[str]=mapped_column(String(),nullable=False)
+    Группа_профессий_должностей:Mapped[str]=mapped_column(String(),nullable=False)
+    Наименование_профессии:Mapped[str]=mapped_column(String(),nullable=False)
+    Разряд:Mapped[str]=mapped_column(String(),nullable=False)
+    Профессии_специальности_СПО:Mapped[str]=mapped_column(String(),nullable=False)
+    Количество_вакансий:Mapped[str]=mapped_column(String(),nullable=False)

server/api-data-base/data_base/requests.py

@@ -0,0 +1,154 @@
+from data_base.data_base import async_session, Base, async_engine
+import data_base.models.main_tables as mt
+import json
+import model_messege as mm
+from sqlalchemy.future import select
+from sqlalchemy import MetaData, inspect, text, insert, Table
+
+class JsonJober:
+    data = []
+
+    @classmethod
+    def create_table(cls, table_name):
+        """"Создание объектов таблицы"""
+        temp_table = []
+        if mt.Professions.__tablename__ == table_name:
+            for row in JsonJober.data[table_name]:
+                temp_table.append(mt.Professions(name=row))
+                print(f"Added to Professions: {row}")
+        elif mt.Positions.__tablename__ == table_name:
+            for row in JsonJober.data[table_name]:
+                temp_table.append(mt.Positions(name=row))
+                print(f"Added to Positions: {row}")
+        elif mt.Classification_professions.__tablename__ == table_name:
+            for row in JsonJober.data[table_name]:
+                temp_table.append(mt.Classification_professions(name=row))
+                print(f"Added to Classification_professions: {row}")
+        elif mt.Manufacturing_enterprises.__tablename__ == table_name:
+            for i, name in enumerate(JsonJober.data[table_name]["name"]):
+                inn = JsonJober.data[table_name]["inn"][i]
+                temp_table.append(mt.Manufacturing_enterprises(name=name, INN_company=inn))
+                print(f"Added to Manufacturing_enterprises: {name}, {inn}")
+        else:
+            print(f"Unknown table name: {table_name}")
+        return temp_table
+
+    @classmethod
+    def ReadJson(cls, path_file: str = ""):
+        """Чтение json файла"""
+        JsonJober.data = []
+        mass_table = []
+        with open(path_file, "r", encoding="utf-8") as read_file:
+            JsonJober.data = json.load(read_file)
+            for table in JsonJober.data:
+                mass_table.append(JsonJober.create_table(table_name=table))
+        return mass_table
+
+class Request():
+    @classmethod
+    async def load_preset(cls):
+        """"Загрузка пресета"""
+        await cls.craete_all_table()
+        mass_table = JsonJober.ReadJson(path_file="./data_base/preset/test.json")
+        await cls.insert_mass_table(mass_table=mass_table)
+    
+    @classmethod
+    async def insert_mass_table(cls, mass_table):
+        """Добавление множества таблиц"""
+        async with async_session() as session:
+            async with session.begin():
+                for table in mass_table:
+                    session.add_all(table)
+            await session.commit()
+
+    @classmethod
+    async def select_drop_down_list(cls):
+        """Возвращает все выподающие списки"""
+        async with async_session() as session:
+            result = {}
+            tables = [mt.Professions, mt.Positions, mt.Classification_professions, mt.Manufacturing_enterprises]
+            for table in tables:
+                query = select(table)
+                result_proxy = await session.execute(query)
+                result[table.__tablename__] = [row._asdict() for row in result_proxy.fetchall()]
+            return result
+
+    @staticmethod
+    async def craete_all_table():
+        """"Создание всех таблиц"""
+        async with async_engine.begin() as conn:
+            await conn.run_sync(Base.metadata.drop_all)
+            await conn.run_sync(Base.metadata.create_all)
+    
+    @classmethod
+    async def login(cls, auth_request) -> mt.users:
+        """Авторизация"""
+        async with async_session() as session:
+            stmt = select(mt.users).where(
+                mt.users.inn == auth_request.login,
+            )
+            result = await session.execute(stmt)
+            return result.scalars().first()
+        
+    @classmethod
+    async def add_column(cls, table_name: str, column_name: str, data_type: str):
+        """Добавление столбца в таблицу"""
+        async with async_engine.connect() as connection:
+            if data_type == 'String':
+                data_type_sql = "VARCHAR"
+            elif data_type == 'Integer':
+                data_type_sql = "INTEGER"
+            else:
+                raise ValueError("Unsupported data type")
+            
+            alter_stmt = text(f'ALTER TABLE {table_name} ADD COLUMN {column_name} {data_type_sql}')
+            await connection.execute(alter_stmt)
+            await connection.commit()
+
+    @classmethod
+    async def drop_column(cls, table_name: str, column_name: str):
+        """Удаление столбца из таблицы"""
+        async with async_engine.connect() as connection:
+            alter_stmt = text(f'ALTER TABLE {table_name} DROP COLUMN {column_name}')
+            await connection.execute(alter_stmt)
+            await connection.commit()
+            
+    @classmethod
+    async def get_table_columns(cls, table_name: str):
+        """Получение заголовков (имен столбцов) указанной таблицы"""
+        async with async_engine.connect() as connection:
+            def sync_inspect(conn):
+                inspector = inspect(conn)
+                columns = inspector.get_columns(table_name)
+                return [column['name'] for column in columns]
+            
+            column_names = await connection.run_sync(sync_inspect)
+            return column_names
+
+    @classmethod
+    async def insert_table(cls, data: list, table_name:str = "Сollection_form_opk"):
+        data = data["data"]
+        metadata = MetaData()
+        async with async_engine.connect() as connection:
+            await connection.run_sync(metadata.reflect)
+            table = Table(table_name, metadata, autoload_with=connection)
+        async with async_session() as session:
+            async with session.begin():
+                for row in data:
+                    stmt = insert(table).values(**row)
+                    await session.execute(stmt)
+                await session.commit()
+
+    @classmethod
+    async def select_rows_with_filter(cls, value:str, table_name: str = "Сollection_form_opk", column_name: str = "ПОО"):
+        """Выгрузка строк с фильтром по указанному столбцу и значению"""
+        metadata = MetaData()
+        async with async_engine.connect() as connection:
+            await connection.run_sync(metadata.reflect)
+            table = Table(table_name, metadata, autoload_with=connection)
+        
+        async with async_session() as session:
+            query = select(table).where(table.c[column_name] == value)
+            result_proxy = await session.execute(query)
+            result = [row._asdict() for row in result_proxy.fetchall()]
+            return result

server/api-data-base/dockerfile

@@ -0,0 +1,15 @@
+# Используем официальный образ Python
+FROM python:3.11.8-slim AS build
+# Устанавливаем рабочую директорию
+WORKDIR /api-data-base
+# Копируем файл с зависимостями в контейнер
+COPY ./server/api-data-base/requirements.txt .
+# Устанавливаем необходимые зависимости
+RUN pip install -r requirements.txt
+# Копируем файл с кодом в контейнер
+COPY ./server/api-data-base/ .
+
+FROM build AS final
+WORKDIR /api-data-base
+# Команда для запуска приложения
+CMD ["python", "main.py"]

server/api-data-base/main.py

@@ -0,0 +1,57 @@
+import os
+from fastapi import FastAPI, Depends, HTTPException
+from fastapi.security import HTTPBasic, HTTPBasicCredentials
+from fastapi.openapi.docs import get_swagger_ui_html, get_redoc_html
+from fastapi.openapi.utils import get_openapi
+from router import router
+from dotenv import load_dotenv
+
+# Загрузите переменные окружения из файла .env
+load_dotenv()
+
+app = FastAPI(title="api-data-base")
+
+security = HTTPBasic()
+
+# Создайте функцию для проверки доступа
+def authenticate(credentials: HTTPBasicCredentials = Depends(security)):
+    correct_username = os.getenv("DOCS_USERNAME", "admin")  # Укажите имя пользователя
+    correct_password = os.getenv("DOCS_PASSWORD", "admin")  # Укажите пароль
+    if credentials.username == correct_username and credentials.password == correct_password:
+        return True
+    raise HTTPException(
+        status_code=401,
+        detail="Неверное имя пользователя или пароль",
+        headers={"WWW-Authenticate": "Basic"},
+    )
+
+# Переопределите маршруты для документации
+@app.get("/docs", include_in_schema=False)
+def custom_swagger_ui(credentials: HTTPBasicCredentials = Depends(authenticate)):
+    return get_swagger_ui_html(openapi_url=app.openapi_url, title="Документация API")
+
+@app.get("/redoc", include_in_schema=False)
+def custom_redoc_html(credentials: HTTPBasicCredentials = Depends(authenticate)):
+    return get_redoc_html(openapi_url=app.openapi_url, title="Документация API")
+
+# Настройте кастомный OpenAPI (если требуется)
+def custom_openapi():
+    if app.openapi_schema:
+        return app.openapi_schema
+    openapi_schema = get_openapi(
+        title="Пример API",
+        version="1.0.0",
+        description="Документация защищена паролем",
+        routes=app.routes,
+    )
+    app.openapi_schema = openapi_schema
+    return app.openapi_schema
+
+app.openapi = custom_openapi
+
+
+app.include_router(router)
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8001)

server/api-data-base/model_messege.py

@@ -0,0 +1,43 @@
+from pydantic import BaseModel
+from data_base.models.main_tables import users, Сollection_form_opk
+from enum import Enum
+
+class Message(BaseModel):
+    key: str
+    value: str
+
+class Login(BaseModel):
+    login:str
+    password:str
+
+class TableName(str,Enum):
+    users = "users"
+    Сollection_form_opk = "Сollection_form_opk"
+
+class DataType(str,Enum):
+    String="String"
+    str="String"
+    Str="String"
+    Integer="Integer"
+    int="Integer"
+    Int="Integer"
+
+class TableModel():
+    list={
+        "users":users
+    }
+
+class column_table(BaseModel):
+    table_name:TableName
+    column_name:str
+    data_type:DataType
+
+class Table(str, Enum):
+    Collection_form_opk = "Collection_form_opk"
+    
+class insert_table(BaseModel):
+    creation_time:str
+    inn:str
+    table_name:Table
+    data:dict
+    

server/api-data-base/requirements.txt

@@ -0,0 +1,8 @@
+fastapi
+uvicorn
+python-dotenv
+sqlalchemy
+asyncpg
+psycopg2-binary
+bcrypt
+orjson

server/api-data-base/router.py

@@ -0,0 +1,75 @@
+from fastapi import APIRouter, Query, HTTPException
+from data_base.requests import Request
+from fastapi.responses import FileResponse
+import model_messege as mm
+import bcrypt
+import asyncio
+
+router = APIRouter(prefix="/api-data-base")
+
+#table_name: mm.TableName = Query(...)
+
+def check_password(input_password, hashed_password) -> bool:
+    """Проверка пароля"""
+    return bcrypt.checkpw(input_password.encode('utf-8'), hashed_password.encode('utf-8'))
+
+@router.get("/load_preset")
+async def load_preset():
+    """Загрузка пресета"""
+    await Request.load_preset()
+    return {"messege":"пресет загружен"}
+
+@router.get("/start")
+async def create_tables():
+    """Создание таблиц"""
+    await Request.craete_all_table()
+    return {"message": f"все таблицы созданы"}
+
+@router.post("/login")
+async def login(auth_request: mm.Login):#auth_request: mm.Login
+    """Авторизация"""
+    user = await Request.login(auth_request=auth_request)
+    if user == None:#если пользователь не найден
+        return {
+            "id":0,
+            "name_poo":"",
+            "inn":"",
+            "mail":"",
+            "drop_table":""
+        }
+    if check_password(input_password=auth_request.password, hashed_password=user.password):#если пароль верный
+        #data = await Request.select_drop_down_list()
+        return {
+            "id":user.id,
+            "name_poo":user.name_poo,
+            "inn":user.inn,
+            "mail":user.mail
+            #"drop_table": data
+        }
+@router.post("/add_column")
+async def add_column(new_column: mm.column_table):
+    """Добавление колонки"""
+    await Request.add_column(table_name=new_column.table_name.name, column_name=new_column.column_name, data_type=new_column.data_type.name)
+    return {"message":"колонка добавлена"}
+
+@router.post("/drop_column")
+async def drop_column(new_column: mm.column_table):
+    """Удаление колонки"""
+    await Request.drop_column(table_name=new_column.table_name.name, column_name=new_column.column_name)
+    return {"message":"колонка удалена"}    
+
+@router.post("/get_name_column")
+async def get_name_column(table_name: mm.TableName):
+    """Получение имен колонок"""
+    return await Request.get_table_columns(table_name=table_name.name)
+
+
+@router.post("/insert_table")
+async def insert_table(data:dict):
+    """Добавление записи"""
+    return await Request.insert_table(data=data)
+
+@router.post("select_rows_with_filter")
+async def select_rows_with_filter(value:str):
+    """Выборка с фильтром"""
+    return await Request.select_rows_with_filter(value=value)

server/api/README.md

@@ -0,0 +1,2 @@
+# api
+ api-copp

server/api/config_kafka.py

@@ -0,0 +1,20 @@
+from kafka import KafkaProducer,KafkaConsumer
+import redis
+import json
+
+redis_client = redis.StrictRedis(host="redis", port=6379, db=0)
+
+consumer = KafkaConsumer(
+        'login-topic',
+        bootstrap_servers='kafka:9092',
+        value_deserializer=lambda x: json.loads(x.decode('utf-8')),
+        api_version = (0,10,2),
+        auto_offset_reset='earliest',
+        group_id='0'
+    )
+
+producer = KafkaProducer(
+    bootstrap_servers='kafka:9092',  
+    api_version = (0,10,2),
+    value_serializer=lambda v: json.dumps(v).encode('utf-8') 
+    )

server/api/dockerfile

@@ -0,0 +1,15 @@
+# Используем официальный образ Python
+FROM python:3.11.8-slim AS build
+# Устанавливаем рабочую директорию
+WORKDIR /api
+# Копируем файл с зависимостями в контейнер
+COPY ./server/api/requirements.txt .
+# Устанавливаем необходимые зависимости
+RUN pip install -r requirements.txt
+# Копируем файл с кодом в контейнер
+COPY ./server/api/ .
+
+FROM build AS final
+WORKDIR /api
+# Команда для запуска приложения
+CMD ["python", "-u", "main.py"]

server/api/main.py

@@ -0,0 +1,9 @@
+from fastapi import FastAPI
+from router import router
+
+app = FastAPI(title="api")
+app.include_router(router)
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)

server/api/model_messege.py

@@ -0,0 +1,13 @@
+from pydantic import BaseModel
+
+class collection_form(BaseModel):
+    login:str
+    data:list
+
+class Message(BaseModel):
+    key: str
+    value: str
+
+class Login(BaseModel):
+    login:str
+    password:str

server/api/requirements.txt

@@ -0,0 +1,6 @@
+fastapi
+uvicorn
+kafka-python
+python-dotenv
+redis
+httpx

server/api/router.py

@@ -0,0 +1,43 @@
+import model_messege as mm
+from config_kafka import producer, consumer, redis_client
+import threading
+import json
+from fastapi import APIRouter
+import httpx
+
+router = APIRouter(prefix="/api")
+
+@router.post("/api/validation_collection_form")
+async def validation_collection_form():
+    
+   
+    return {
+        "message": "Ошибка валидации формы"
+    }
+
+@router.post("/api/login")
+async def login(auth_request: mm.Login):
+    login = auth_request.login
+    password = auth_request.password
+    print(f"login: {login}, password: {password}")
+
+    # Отправляем данные в другую API
+    async with httpx.AsyncClient() as client:
+        response = await client.post("http://api-data-base:8001/api-data-base/login", json={'login': login, 'password': password})
+
+    # Проверяем статус ответа
+    if response.status_code == 200:
+        user_data = response.json()
+        return {
+            "message": "Авторизация успешна.",
+            "user": user_data
+        }
+    else:
+        return {
+            "message": "Ошибка авторизации",
+            "user": None
+        }
+    
+@router.on_event("shutdown")
+def shutdown_event():
+    producer.close()

server/service-login/README.md

@@ -0,0 +1,2 @@
+# service-login
+ service-copp

server/service-login/config_kafka.py

@@ -0,0 +1,18 @@
+from kafka import KafkaProducer,KafkaConsumer
+import json
+
+consumer = KafkaConsumer(
+    
+        'auth-topic',
+        bootstrap_servers='kafka:9092',
+        value_deserializer=lambda x: json.loads(x.decode('utf-8')),
+        auto_offset_reset='earliest',
+        api_version = (0,10,2),
+        group_id='0'
+    )
+#kafka
+producer = KafkaProducer(
+    bootstrap_servers='kafka:9092',  
+    api_version = (0,10,2),
+    value_serializer=lambda v: json.dumps(v).encode('utf-8') 
+    )

server/service-login/dockerfile

@@ -0,0 +1,16 @@
+# Используем официальный образ Python
+FROM python:3.11.8-slim AS build
+# Устанавливаем рабочую директорию
+WORKDIR /service-login
+# Копируем файл с зависимостями в контейнер
+COPY ./server/service-login/requirements.txt .
+# Устанавливаем необходимые зависимости
+RUN pip install -r requirements.txt
+# Копируем файл с кодом в контейнер
+COPY ./server/service-login/ .
+
+FROM build AS final
+WORKDIR /service-login
+# Команда для запуска приложения
+CMD ["python", "-u", "./main.py"]
+

server/service-login/main.py

@@ -0,0 +1,23 @@
+from config_kafka import consumer,producer
+import requests
+
+def process_message(message):
+    url = 'http://api-data-base:8001/api-data-base/login'
+    auth_request = {
+        'login': message.value['login'],
+        'password': message.value['password']
+    }
+    response = requests.post(url, json=auth_request)
+    if response.status_code == 200:
+        return response.json()
+        
+
+
+def main():
+    for message in consumer:
+        result = process_message(message)
+        producer.send("login-topic", {'login': message.value['login'], 'success': result})
+        producer.flush()
+        
+if __name__ == "__main__":
+    main()

server/service-login/requirements.txt

@@ -0,0 +1,3 @@
+kafka-python
+python-dotenv
+requests

server/validation_collection_form/main.py

@@ -0,0 +1,47 @@
+import threading
+from confluent_kafka import Consumer, KafkaException, KafkaError
+
+def consume_messages():
+    # Настройка конфигурации потребителя
+    conf = {
+        'bootstrap.servers': 'localhost:9092',
+        'group.id': 'my_group',
+        'auto.offset.reset': 'earliest'
+    }
+
+    # Создание потребителя
+    consumer = Consumer(conf)
+
+    # Подписка на топик
+    consumer.subscribe(['my_topic'])
+
+    try:
+        while True:
+            # Получение сообщения
+            msg = consumer.poll(timeout=1.0)
+            if msg is None:
+                continue
+            if msg.error():
+                if msg.error().code() == KafkaError._PARTITION_EOF:
+                    # Достигнут конец раздела
+                    continue
+                else:
+                    raise KafkaException(msg.error())
+            # Обработка сообщения
+            print(f'Received message: {msg.value().decode("utf-8")}')
+    except KeyboardInterrupt:
+        pass
+    finally:
+        # Закрытие потребителя
+        consumer.close()
+
+def start_consumer_thread():
+    consumer_thread = threading.Thread(target=consume_messages)
+    consumer_thread.daemon = True
+    consumer_thread.start()
+
+if __name__ == "__main__":
+    start_consumer_thread()
+    # Основной поток может выполнять другие задачи
+    while True:
+        pass