Team Project GoIt. Data Sciense
# import redis.asyncio as redis
import uvicorn
from dotenv import load_dotenv

from fastapi import FastAPI, Request, Form, HTTPException, UploadFile, File, status
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
from starlette.middleware.sessions import SessionMiddleware

# from fastapi_limiter import FastAPILimiter
from src.conf.config import settings
from sqladmin import Admin

from src.database.db import create_async_engine
from src.services.admin_panel.admin_panel import UserAdmin
from src.routes import auth, users, pdffile, chat, chat_history


engine = create_async_engine(settings.sqlalchemy_database_url)

load_dotenv()

# Створюємо екземпляр FastApi, встановлюємо назву додатка у swagger та відсортуємо роути по методах:
app = FastAPI(swagger_ui_parameters={"operationsSorter": "method"}, title='FreeChatAI app')
app.add_middleware(SessionMiddleware, secret_key="some-random-secret-key")

# підключаємо адмін-панель
# http://localhost:8001/admin/
admin = Admin(app, engine)
admin.add_view(UserAdmin)

templates = Jinja2Templates(directory='templates')
app.mount('/static', StaticFiles(directory='static'), name='static')

app.include_router(auth.router, prefix='/api')
app.include_router(users.router, prefix='/api')
app.include_router(pdffile.router)
app.include_router(chat.router)
app.include_router(chat_history.router)


@app.get("/", response_class=HTMLResponse, tags=["Main index.html"])
async def root(request: Request):
    return templates.TemplateResponse('index.html', {'request': request, "response": None})


# @app.on_event("startup")
# async def startup():
#     r = await redis.Redis(host=settings.redis_host, port=settings.redis_port, password=settings.redis_password,
#     encoding="utf-8", db=0)
#     await FastAPILimiter.init(r)

if __name__ == '__main__':
    uvicorn.run('main:app', port=8080, reload=True)

main.py
DenBilokon/FreeChatAi
import asyncio
from logging.config import fileConfig

from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config

from alembic import context

from src.conf.config import settings as app_config
from src.database.models import Base

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
target_metadata = Base.metadata
config.set_main_option("sqlalchemy.url", app_config.sqlalchemy_database_url)
# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.


def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode.

    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.

    Calls to context.execute() here emit the given string to the
    script output.

    """
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def do_run_migrations(connection: Connection) -> None:
    context.configure(connection=connection, target_metadata=target_metadata)

    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations() -> None:
    """In this scenario we need to create an Engine
    and associate a connection with the context.

    """

    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()


def run_migrations_online() -> None:
    """Run migrations in 'online' mode."""

    asyncio.run(run_async_migrations())


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

migrations/env.py
DenBilokon/FreeChatAi
"""Init

Revision ID: 23b5f17127cd
Revises: 
Create Date: 2023-10-02 20:40:41.037072

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision: str = '23b5f17127cd'
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('users',
    sa.Column('user_id', sa.Integer(), nullable=False),
    sa.Column('roles', sa.Enum('user', 'admin', name='role'), nullable=True),
    sa.Column('password', sa.String(length=255), nullable=False),
    sa.Column('username', sa.String(length=50), nullable=True),
    sa.Column('email', sa.String(length=250), nullable=False),
    sa.Column('phone', sa.String(length=255), nullable=True),
    sa.Column('avatar', sa.String(length=255), nullable=True),
    sa.Column('refresh_token', sa.String(length=255), nullable=True),
    sa.Column('confirmed', sa.Boolean(), nullable=True),
    sa.Column('banned', sa.Boolean(), nullable=True),
    sa.Column('created_at', sa.DateTime(), nullable=True),
    sa.Column('updated_at', sa.DateTime(), nullable=True),
    sa.PrimaryKeyConstraint('user_id'),
    sa.UniqueConstraint('email')
    )
    op.create_table('admin',
    sa.Column('admin_id', sa.Integer(), nullable=False),
    sa.Column('user_id', sa.Integer(), nullable=False),
    sa.Column('is_active', sa.Boolean(), nullable=True),
    sa.Column('last_visit', sa.DateTime(), nullable=True),
    sa.ForeignKeyConstraint(['user_id'], ['users.user_id'], ),
    sa.PrimaryKeyConstraint('admin_id')
    )
    op.create_index(op.f('ix_admin_admin_id'), 'admin', ['admin_id'], unique=False)
    op.create_table('chats',
    sa.Column('chat_id', sa.Integer(), nullable=False),
    sa.Column('title_chat', sa.String(), nullable=False),
    sa.Column('file_url', sa.String(), nullable=True),
    sa.Column('chat_data', sa.String(), nullable=True),
    sa.Column('created_at', sa.DateTime(), nullable=True),
    sa.Column('updated_at', sa.DateTime(), nullable=True),
    sa.Column('user_id', sa.Integer(), nullable=True),
    sa.ForeignKeyConstraint(['user_id'], ['users.user_id'], ),
    sa.PrimaryKeyConstraint('chat_id')
    )
    op.create_index(op.f('ix_chats_chat_id'), 'chats', ['chat_id'], unique=False)
    op.create_table('chathistories',
    sa.Column('chat_history_id', sa.Integer(), nullable=False),
    sa.Column('message', sa.String(), nullable=False),
    sa.Column('created_at', sa.DateTime(), nullable=True),
    sa.Column('user_id', sa.Integer(), nullable=True),
    sa.Column('chat_id', sa.Integer(), nullable=True),
    sa.ForeignKeyConstraint(['chat_id'], ['chats.chat_id'], ),
    sa.ForeignKeyConstraint(['user_id'], ['users.user_id'], ),
    sa.PrimaryKeyConstraint('chat_history_id')
    )
    op.create_index(op.f('ix_chathistories_chat_history_id'), 'chathistories', ['chat_history_id'], unique=False)
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_index(op.f('ix_chathistories_chat_history_id'), table_name='chathistories')
    op.drop_table('chathistories')
    op.drop_index(op.f('ix_chats_chat_id'), table_name='chats')
    op.drop_table('chats')
    op.drop_index(op.f('ix_admin_admin_id'), table_name='admin')
    op.drop_table('admin')
    op.drop_table('users')
    # ### end Alembic commands ###

migrations/versions/23b5f17127cd_init.py
DenBilokon/FreeChatAi
from pydantic import ConfigDict
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    sqlalchemy_database_url: str = "postgresql+asyncpg://postgres:password@localhost:5432/db_chat_ai"
    secret_key: str = "secret key"
    algorithm: str = "HS256"
    mail_username: str = "example@meta.ua"
    mail_password: str = "qwerty"
    mail_from: str = "example@meta.ua"
    mail_port: int = 465
    mail_server: str = "smtp.meta.ua"
    redis_host: str = 'localhost'
    redis_port: int = 6379
    cloudinary_name: str = "cloudinary_name"
    cloudinary_api_key: str = "1234"
    cloudinary_api_secret: str = "213213"
    openai_api_key: str = 'key'

    openai_key: str = "openai key"

    model_config = ConfigDict(extra='ignore', env_file=".env", env_file_encoding="utf-8")


settings = Settings()

src/conf/config.py
DenBilokon/FreeChatAi
class AuthMessages:
    account_already_exists = "Account already exists"
    invalid_email = "Invalid email"
    email_not_confirmed = "Email not confirmed"
    invalid_password = "Invalid password"
    invalid_refresh_token = "Invalid refresh token"
    verification_error = "Verification error"
    your_email_is_already_confirmed = "Your email is already confirmed"
    check_your_email_for_confirmation = "Check your email for confirmation"
    invalid_scope_for_token = "Invalid scope for token"
    could_not_validate_credentials = "Could not validate credentials"
    invalid_token_for_email_verification = "Invalid token for email verification"
    banned = "You are banned from this app!"


class RolesMessages:
    operation_forbidden = "Operation forbidden"


class ChatMessages:
    CHAT_NOT_FOUND = "Chat not found"
    NOT_FOUND = "Not Found"

src/conf/messages.py
DenBilokon/FreeChatAi
import contextlib
from typing import AsyncIterator

from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

from src.conf.config import settings


class Base(DeclarativeBase):
    pass


class DatabaseSessionManager:
    def __init__(self, url: str):
        self._engine: AsyncEngine | None = create_async_engine(url)
        self._session_maker: async_sessionmaker | None = async_sessionmaker(autocommit=False, autoflush=False,
                                                                            bind=self._engine)

    @contextlib.asynccontextmanager
    async def session(self) -> AsyncIterator[AsyncSession]:
        if self._session_maker is None:
            raise Exception("DatabaseSessionManager is not initialized")
        session = self._session_maker()
        try:
            yield session
        except Exception as err:
            print(err)
            await session.rollback()
        finally:
            await session.close()


sessionmanager = DatabaseSessionManager(settings.sqlalchemy_database_url)


# Dependency
async def get_db():
    async with sessionmanager.session() as session:
        yield session

src/database/db.py
DenBilokon/FreeChatAi
from datetime import datetime
import enum

import uuid
from sqlalchemy.dialects.postgresql import UUID

from sqlalchemy import Column, String, DateTime, ForeignKey, Boolean, Enum, Integer, func, Float, Table
from sqlalchemy.orm import declarative_base, relationship
from sqlalchemy.sql.schema import Table


Base = declarative_base()


class Role(enum.Enum):
    user = 'user'
    admin = 'admin'


class User(Base):
    __tablename__ = "users"

    user_id = Column(Integer, primary_key=True)
    user_role = Column('roles', Enum(Role), default=Role.user)
    password = Column(String(255), nullable=False)
    username = Column(String(50))
    email = Column(String(250), nullable=False, unique=True)
    phone = Column(String(255))
    avatar = Column(String(255), nullable=True)
    refresh_token = Column(String(255), nullable=True)
    confirmed = Column(Boolean, default=False)
    banned = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.now)
    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)

    def __str__(self):
        return f"{self.username}"


class Admin(Base):
    __tablename__ = 'admin'
    admin_id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey('users.user_id'), nullable=False)
    is_active = Column(Boolean, default=False)
    last_visit = Column(DateTime, default=func.now())


class Chat(Base):
    __tablename__ = "chats"
    chat_id = Column(Integer, primary_key=True, index=True)
    title_chat = Column(String, nullable=False)
    file_url = Column(String, nullable=True)
    chat_data = Column(String, nullable=True)
    created_at = Column(DateTime, default=func.now())
    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())
    user_id = Column(Integer, ForeignKey("users.user_id"), nullable=True)
    user = relationship('User', backref="chats")


class ChatHistory(Base):
    __tablename__ = "chathistories"
    chat_history_id = Column(Integer, primary_key=True, index=True)
    message = Column(String, nullable=False)
    created_at = Column(DateTime, default=func.now())
    user_id = Column(Integer, ForeignKey("users.user_id"), nullable=True)
    chat_id = Column(Integer, ForeignKey("chats.chat_id"), nullable=True)
    user = relationship('User', backref="chathistories")
    chat = relationship('Chat', backref="chathistories")

src/database/models.py
DenBilokon/FreeChatAi
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import and_
import ast

from src.database.models import User, ChatHistory
from src.schemas.chat_history_schemas import ChatHistoryBase
from src.conf.messages import ChatMessages
from src.repository.chats import get_chat_by_id

from fastapi import HTTPException, status
from typing import List
from dotenv import load_dotenv

from langchain.chains import ConversationalRetrievalChain
from langchain.chat_models import ChatOpenAI
from langchain.embeddings import OpenAIEmbeddings
from langchain.memory import ConversationBufferMemory
from langchain.text_splitter import CharacterTextSplitter
from langchain.vectorstores import FAISS

load_dotenv()


async def get_context(chat_id: int, db: AsyncSession, user: User) -> List:
    chat = await get_chat_by_id(chat_id, db, user)
    context = chat.file_url
    if context is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ChatMessages.NOT_FOUND)

    with open(context, "r", encoding="utf-8") as file:
        text = file.read()

    text_splitter = CharacterTextSplitter(
        separator="\n",
        chunk_size=1000,
        chunk_overlap=200,
        length_function=len
    )
    chunks = text_splitter.split_text(text)
    return chunks


def get_conversation_chain(user_question, context) -> str:
    embeddings = OpenAIEmbeddings()
    vectorstore = FAISS.from_texts(texts=context, embedding=embeddings)

    llm = ChatOpenAI(model='gpt-3.5-turbo')
    memory = ConversationBufferMemory(memory_key='chat_history', return_messages=True)
    conversation_chain = ConversationalRetrievalChain.from_llm(
        llm=llm,
        retriever=vectorstore.as_retriever(),
        memory=memory,
    )

    response = conversation_chain({'question': user_question})
    chat_history = response['chat_history']

    # for separate question/answer items
    response_dict = {'user_messages': [], 'bot_messages': []}

    for i, message in enumerate(chat_history):
        if i % 2 == 0:
            response_dict['user_messages'].append(message.content)
        else:
            response_dict['bot_messages'].append(message.content)

    return str(response_dict)


async def create_message(chat_id: int, body: ChatHistoryBase, db: AsyncSession, user: User) -> ChatHistory:
    context = await get_context(chat_id, db, user)
    answer = get_conversation_chain(body.message, context)

    response_dict = ast.literal_eval(answer)
    user_question = 'Q: ' + response_dict['user_messages'][0]
    bot_response = 'A: ' + response_dict['bot_messages'][0]

    question = ChatHistory(message=user_question, user_id=user.id, chat_id=chat_id)
    response = ChatHistory(message=bot_response, user_id=user.id, chat_id=chat_id)
    db.add(question)
    db.add(response)

    await db.commit()
    await db.refresh(response)
    return response


async def get_history_by_chat(chat_id: int, db: AsyncSession) -> ChatHistory:
    return db.query(ChatHistory).filter(and_(ChatHistory.chat_id == chat_id)).order_by(ChatHistory.created_at).all()

src/repository/chat_history.py
DenBilokon/FreeChatAi
import os

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import and_, desc

from src.database.models import User, Chat, Role
from src.schemas.chat_schemas import ChatBase, ChatModel, ChatUpdate


async def delete_source_file(path):
    file_path = path
    os.remove(file_path)


async def create_chat(body: ChatBase, db: AsyncSession, user: User) -> Chat:
    context = True

    new_chat = Chat(title_chat=body.title_chat, chat_data=context, file_url=body.file_url, user_id=user.id)
    db.add(new_chat)
    await db.commit()
    await db.refresh(new_chat)
    return new_chat


async def delete_chat(chat_id: int, db: AsyncSession, user: User) -> Chat | None:
    chat = db.query(Chat).filter(Chat.id == chat_id).first()
    if chat:
        if user.roles in [Role.admin] or chat.user_id == user.id:
            await delete_source_file(chat.file_url)
            await db.delete(chat)
            await db.commit()
    return chat


async def get_chat_by_id(chat_id: int, db: AsyncSession, user: User) -> Chat | None:
    return db.query(Chat).filter(and_(Chat.id == chat_id, Chat.user_id == user.id)).first()


async def get_chats(limit: int, offset: int, user: User, db: AsyncSession) -> list | None:
    chats = db.query(Chat).filter(and_(Chat.user_id == user.id)).order_by(desc(Chat.created_at)).limit(limit).offset(
        offset).all()
    return chats

src/repository/chats.py
DenBilokon/FreeChatAi
import logging

from fastapi import HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from starlette import status

from src.database.models import User
from src.schemas.user_schemas import UserModel


async def get_user_by_username(username: str, db: AsyncSession) -> User | None:
    sq = select(User).filter_by(username=username)
    result = await db.execute(sq)
    user = result.scalar_one_or_none()
    logging.info(user)
    return user


async def get_user_by_email(email: str, db: AsyncSession) -> User | None:
    sq = select(User).filter_by(email=email)
    result = await db.execute(sq)
    user = result.scalar_one_or_none()
    logging.info(user)
    return user


async def create_user(body: UserModel, db: AsyncSession) -> User:
    new_user = User(**body.model_dump())
    db.add(new_user)
    await db.commit()
    await db.refresh(new_user)
    return new_user


async def update_token(user: User, token: str | None, db: AsyncSession) -> None:
    user.refresh_token = token
    await db.commit()


async def confirmed_email(email: str, db: AsyncSession) -> None:
    user = await get_user_by_email(email, db)
    user.confirmed = True
    await db.commit()


async def get_user_info(name: str, db: AsyncSession):
    sq = select(User).filter_by(name=name)
    result = await db.execute(sq)
    user = result.scalar_one_or_none()
    return user


async def update_user_info(body: UserModel, name: str, db: AsyncSession):
    sq = select(User).filter_by(name=name)
    result = await db.execute(sq)
    updated_user = result.scalar_one_or_none()
    if not updated_user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
    updated_user.name = body.name
    updated_user.email = body.email
    await db.commit()
    return updated_user


async def block(email: str, body: UserModel, db: AsyncSession):
    user = await get_user_by_email(email, db)
    if user:
        user.banned = body.banned
    await db.commit()
    return user

src/repository/users.py
DenBilokon/FreeChatAi
from fastapi import APIRouter, HTTPException, Depends, status, Security, BackgroundTasks, Request
from fastapi.security import OAuth2PasswordRequestForm, HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi.templating import Jinja2Templates

from src.database.db import get_db
from src.repository import users as repository_users
from src.schemas.user_schemas import UserModel, UserResponse
from src.schemas.auth_schemas import TokenModel, RequestEmail
from src.services.auth import auth_service
from src.services.email1 import send_email
from src.conf.messages import AuthMessages

templates = Jinja2Templates(directory='templates')

router = APIRouter(prefix='/auth', tags=["auth"])
security = HTTPBearer()


@router.get("/signup")
async def signup_form(request: Request):
    return templates.TemplateResponse('signup.html', {'request': request})


@router.post("/signup", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def signup(body: UserModel,
                 background_tasks: BackgroundTasks,
                 request: Request,
                 db: AsyncSession = Depends(get_db)):

    exist_user = await repository_users.get_user_by_email(body.email, db)
    if exist_user:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=AuthMessages.account_already_exists)
    body.password = auth_service.get_password_hash(body.password)
    new_user = await repository_users.create_user(body, db)
    background_tasks.add_task(send_email, new_user.email, new_user.username, str(request.base_url))
    return new_user


@router.get("/login")
async def login_form(request: Request):
    return templates.TemplateResponse('login.html', {'request': request})


@router.post("/login", response_model=TokenModel)
async def login(body: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db)):

    user = await repository_users.get_user_by_username(body.username, db)
    if user is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=AuthMessages.invalid_email)
    if not user.confirmed:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=AuthMessages.email_not_confirmed)
    if not auth_service.verify_password(body.password, user.password):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=AuthMessages.invalid_password)
    if user.banned:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=AuthMessages.banned)
    # Generate JWT
    access_token = await auth_service.create_access_token(data={"sub": user.email})
    refresh_token = await auth_service.create_refresh_token(data={"sub": user.email})
    await repository_users.update_token(user, refresh_token, db)
    return {"access_token": access_token, "refresh_token": refresh_token, "token_type": "bearer"}


@router.get('/refresh_token', response_model=TokenModel)
async def refresh_token(credentials: HTTPAuthorizationCredentials = Security(security),
                        db: AsyncSession = Depends(get_db)):

    token = credentials.credentials
    email = await auth_service.decode_refresh_token(token)
    user = await repository_users.get_user_by_email(email, db)
    if user.refresh_token != token:
        await repository_users.update_token(user, None, db)
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=AuthMessages.invalid_refresh_token)
    if user.banned:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=AuthMessages.banned)

    access_token = await auth_service.create_access_token(data={"sub": email})
    refresh_token = await auth_service.create_refresh_token(data={"sub": email})
    await repository_users.update_token(user, refresh_token, db)
    return {"access_token": access_token, "refresh_token": refresh_token, "token_type": "bearer"}


@router.get('/confirmed_email/{token}')
async def confirmed_email(token: str, db: AsyncSession = Depends(get_db)):

    email = await auth_service.get_email_from_token(token)
    user = await repository_users.get_user_by_email(email, db)
    if user is None:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=AuthMessages.verification_error)
    if user.confirmed:
        return {"message": "Your email is already confirmed"}
    await repository_users.confirmed_email(email, db)
    return {"message": "Email confirmed"}


@router.post('/request_email')
async def request_email(body: RequestEmail, background_tasks: BackgroundTasks, request: Request,
                        db: AsyncSession = Depends(get_db)):
    user = await repository_users.get_user_by_email(body.email, db)
    if user.confirmed:
        return {"message": AuthMessages.your_email_is_already_confirmed}
    if user:
        background_tasks.add_task(send_email, user.email, user.name, str(request.base_url))
    return {"message": AuthMessages.check_your_email_for_confirmation}

src/routes/auth.py
DenBilokon/FreeChatAi
from typing import List

from fastapi import APIRouter, HTTPException, Depends, status, Query, Request, Form
from sqlalchemy.ext.asyncio import AsyncSession

from src.conf.messages import ChatMessages
from src.database.db import get_db
from src.database.models import User, Role
from src.repository import chats as repository_chats
from src.schemas.chat_schemas import ChatBase, ChatModel
from src.services.auth import auth_service
from src.services.roles import RolesAccess
from src.services.vector_doc_ai import vector_func

router = APIRouter(prefix='/chats', tags=["chats"])

allowed_get_chats = RolesAccess([Role.admin, Role.user])  # noqa
allowed_add_chats = RolesAccess([Role.admin, Role.user])  # noqa
allowed_delete_chats = RolesAccess([Role.admin, Role.user])  # noqa


@router.post("/ask", status_code=status.HTTP_201_CREATED)
async def post_question(request: Request, question: str = Form(...)):
    path_to_file = request.session.get("current_path_document")
    print(path_to_file)

    if not path_to_file:
        raise HTTPException(status_code=400, detail="Error path_to_file.")
    response = await vector_func(path_to_file, question)
    # print(response)
    return {"response": response}


@router.post("/", response_model=ChatModel, status_code=status.HTTP_201_CREATED,
             dependencies=[Depends(allowed_add_chats)])
async def create_chat(body: ChatBase, db: AsyncSession = Depends(get_db),
                      current_user: User = Depends(auth_service.get_current_user)):
    new_chat = await repository_chats.create_chat(body, db, current_user)
    return new_chat


@router.get("/", response_model=List[ChatModel], status_code=status.HTTP_200_OK,
            dependencies=[Depends(allowed_get_chats)])
async def get_chats(limit: int = Query(10, le=50), offset: int = 0,
                    current_user: User = Depends(auth_service.get_current_user),
                    db: AsyncSession = Depends(get_db)):
    chats = await repository_chats.get_chats(limit, offset, current_user, db)
    if chats is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ChatMessages.NOT_FOUND)
    return chats


@router.get("/{chat_id}", response_model=ChatModel, dependencies=[Depends(allowed_get_chats)])
async def get_chat(chat_id: int, db: AsyncSession = Depends(get_db),
                   current_user: User = Depends(auth_service.get_current_user)):
    chat = await repository_chats.get_chat_by_id(chat_id, db, current_user)
    if chat is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ChatMessages.CHAT_NOT_FOUND)
    return chat


@router.delete("/{chat_id}", response_model=ChatModel,
               dependencies=[Depends(allowed_delete_chats)])
async def delete_chat(chat_id: int, db: AsyncSession = Depends(get_db),
                      current_user: User = Depends(auth_service.get_current_user)):
    deleted_chat = await repository_chats.delete_chat(chat_id, db, current_user)
    if deleted_chat is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ChatMessages.CHAT_NOT_FOUND)
    return deleted_chat


src/routes/chat.py
DenBilokon/FreeChatAi
from typing import List

from fastapi import APIRouter, HTTPException, Depends, status
from sqlalchemy.ext.asyncio import AsyncSession

from src.conf.messages import ChatMessages
from src.database.db import get_db
from src.database.models import User, Role
from src.repository import chat_history as repository_history
from src.schemas.chat_history_schemas import ChatHistoryBase, ChatHistoryModel
from src.services.auth import auth_service
from src.services.roles import RolesAccess

router = APIRouter(prefix='/history', tags=["history"])

allowed_get_history = RolesAccess([Role.admin, Role.user])  # noqa
allowed_add_messages = RolesAccess([Role.user, Role.admin])  # noqa


@router.post("/{chat_id}", response_model=ChatHistoryModel, status_code=status.HTTP_201_CREATED,
             dependencies=[Depends(allowed_add_messages)])
async def create_message(chat_id: int, body: ChatHistoryBase, db: AsyncSession = Depends(get_db),
                         current_user: User = Depends(auth_service.get_current_user)):
    new_message = await repository_history.create_message(chat_id, body, db, current_user)
    return new_message


@router.get("/{chat_id}", response_model=List[ChatHistoryModel], dependencies=[Depends(allowed_get_history)])
async def get_history(chat_id: int, db: AsyncSession = Depends(get_db),
                      current_user: User = Depends(auth_service.get_current_user)):
    history = await repository_history.get_history_by_chat(chat_id, db)
    if history is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ChatMessages.CHAT_NOT_FOUND)
    return history

src/routes/chat_history.py
DenBilokon/FreeChatAi
from pathlib import Path
from typing import List
import logging

from fastapi import UploadFile, HTTPException, status, APIRouter, File, Request
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

router = APIRouter(prefix='/pdffile', tags=['PDFfile'])


@router.post("/upload", status_code=status.HTTP_201_CREATED)
async def upload_pdf(request: Request, file: UploadFile = File(...)):
    try:
        file_name, extension = file.filename.split('.')
        if extension != 'pdf':
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="підтримує тільки PDF")

        save_path = Path("uploaded_files") / file.filename

        save_path.parent.mkdir(parents=True, exist_ok=True)

        with save_path.open("wb") as buffer:
            buffer.write(await file.read())

        # Зберігаємо шлях до файлу у сесії
        request.session["current_path_document"] = str(save_path.absolute())
        return {"success": True, 'filename': file_name, 'path': str(save_path.absolute())}
    except Exception as e:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"Помилка при завантаженні файлу: {e}")

src/routes/pdffile.py
DenBilokon/FreeChatAi
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from starlette import status

from src.database.db import get_db
from src.database.models import User, Role
from src.repository.users import block, update_user_info, get_user_info

from src.schemas.user_schemas import UserResponse, UserModel
from src.services.auth import auth_service
from src.services.roles import RolesAccess

access_get = RolesAccess([Role.admin, Role.user])
access_create = RolesAccess([Role.admin, Role.user])
access_update = RolesAccess([Role.admin, Role.user])
access_delete = RolesAccess([Role.admin])
access_block = RolesAccess([Role.admin])

router = APIRouter(prefix="/users", tags=["Users profile"])


@router.get("/me/", response_model=UserResponse)
async def read_users_me(current_user: User = Depends(auth_service.get_current_user)):

    return current_user


@router.get("/{username}/", response_model=UserResponse)
async def profile_info(username: str, db: AsyncSession = Depends(get_db)):

    user_info = await get_user_info(username, db)
    if user_info is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user_info


@router.put('/{username}', response_model=UserResponse, dependencies=[Depends(access_update)])
async def profile_update(username: str, body: UserModel, db: AsyncSession = Depends(get_db),
                         current_user: User = Depends(auth_service.get_current_user)):

    if current_user.username != username and current_user.role in ('client', 'master'):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="You can only update your own profile")

    updated_user = await update_user_info(body, username, db)

    return updated_user


@router.patch("/{email}/blacklist", response_model=UserResponse, dependencies=[Depends(access_block)])
async def block_user(email: str, body: UserModel, db: AsyncSession = Depends(get_db),
                        _: User = Depends(auth_service.get_current_user)):
    blocked_user = await block(email, body, db)
    if blocked_user is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found")
    return blocked_user

src/routes/users.py
DenBilokon/FreeChatAi
from pydantic import BaseModel, EmailStr, Field


class TokenModel(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str = "bearer"


class RequestEmail(BaseModel):
    email: EmailStr

src/schemas/auth_schemas.py
DenBilokon/FreeChatAi
import datetime

from pydantic import BaseModel, Field


class ChatHistoryBase(BaseModel):
    message: str = Field(max_length=5000)


class ChatHistoryModel(ChatHistoryBase):
    id: int
    created_at: datetime.datetime | None
    user_id: int
    chat_id: int

    class Config:
        from_attributes = True

src/schemas/chat_history_schemas.py
DenBilokon/FreeChatAi
import datetime

from pydantic import BaseModel, Field


class ChatBase(BaseModel):
    title_chat: str = Field(max_length=500)
    file_url: str | None
    chat_data: str | None


class ChatModel(ChatBase):
    id: int
    # file_url: str | None
    # chat_data: str = Field()
    created_at: datetime.datetime | None
    updated_at: datetime.datetime | None
    user_id: int

    class Config:
        from_attributes = True


class ChatUpdate(ChatModel):
    updated_at: datetime.datetime

    class Config:
        from_attributes = True

src/schemas/chat_schemas.py
DenBilokon/FreeChatAi
from typing import Optional
from pydantic import BaseModel, EmailStr, Field, HttpUrl
from sqlalchemy.dialects.postgresql import UUID
from uuid import uuid4
from src.database.models import Role


class UserModel(BaseModel):
    user_role: Role = 'user'
    password: str = Field(min_length=8)
    username: str = Field(min_length=5, max_length=30)
    email: EmailStr
    phone: str
    avatar: Optional[str] = Field(None)


class UserResponse(BaseModel):
    user_id: int = Field(default_factory=lambda: uuid4().hex)
    user_role: Role = 'user'
    password: str = Field(min_length=6)
    username: str = "Oksana"
    email: EmailStr = "oksana@gmail.com"
    phone: str
    avatar: Optional[str] = Field(None)

    class Config:
        from_attributes = True


class AdminModel(UserModel):
    user_id: int = Field(default_factory=lambda: uuid4().hex)
    user_role: Role = 'admin'
    is_active: bool = True


class AdminResponse(UserResponse):
    admin_id: int = Field(default_factory=lambda: uuid4().hex)
    user_role: Role = 'admin'
    is_active: bool

    class Config:
        from_attributes = True


class UserUpdate(BaseModel):
    username: str
    email: EmailStr


class UserBlackList(BaseModel):
    banned: Optional[bool] = False

    class Config:
        from_attributes = True


class UserBlacklistResponse(BaseModel):
    user_id: int = Field(default_factory=lambda: uuid4().hex)
    username: str = "Oksana"
    email: EmailStr = "oksana@gmail.com"
    role: Role = "client"
    banned: Optional[bool] = False

    class Config:
        from_attributes = True

src/schemas/user_schemas.py
DenBilokon/FreeChatAi
from sqladmin import ModelView

from src.database.models import User


# визначаємо зміст адмін-панелі: якими моделями бази даних і якими полями хочемо керувати через адмін-панель:
class UserAdmin(ModelView, model=User):
    column_list = [User.username, User.email, User.phone, User.user_role, User.confirmed, User.created_at,
                   User.updated_at]
    column_searchable_list = [User.username]
    column_sortable_list = [User.user_id, User.user_role]
    column_default_sort = [(User.email, True), (User.username, False)]
    can_create = True
    can_edit = True
    can_delete = True
    can_view_details = True
    can_export = True
    name = "Користувач"
    name_plural = "Користувачі"
    icon = "fa-solid fa-user"

src/services/admin_panel/admin_panel.py
DenBilokon/FreeChatAi
from sqladmin import Admin
from sqladmin.authentication import AuthenticationBackend
from starlette.requests import Request
from starlette.responses import RedirectResponse


class AdminAuth(AuthenticationBackend):
    async def login(self, request: Request) -> bool:
        form = await request.form()
        username, password = form["username"], form["password"]

        # Validate username/password credentials
        # And update session
        request.session.update({"token": "..."})

        return True

    async def logout(self, request: Request) -> bool:
        # Usually you'd want to just clear the session
        request.session.clear()
        return True

    async def authenticate(self, request: Request) -> bool:
        token = request.session.get("token")

        if not token:
            return False

        # Check the token in depth
        return True


authentication_backend = AdminAuth(secret_key="...")
admin = Admin(app=..., authentication_backend=authentication_backend, ...)


src/services/admin_panel/admin_panel_auth.py
DenBilokon/FreeChatAi
# import pickle
from datetime import datetime, timedelta
from typing import Optional

# import redis
from fastapi import HTTPException, status, Depends
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from passlib.context import CryptContext
from sqlalchemy.orm import Session

from src.conf.config import settings
from src.conf.messages import AuthMessages
from src.database.db import get_db
from src.repository import users as repository_users
# from main import startup


class Auth:
    pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
    SECRET_KEY = settings.secret_key    #.jwt_secret_key
    ALGORITHM = settings.algorithm  #jwt_algorithm
    oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")
    # r = redis.Redis(host=settings.redis_host, port=settings.redis_port, password=settings.redis_password, encoding="utf-8", db=0)

    def verify_password(self, plain_password, hashed_password):
        """
        The verify_password function takes a plain-text password and hashed
        password as arguments. It then uses the pwd_context object to verify that the
        plain-text password matches the hashed one.

        :param plain_password: Check if the password entered by the user matches with what is stored in the database
        :param hashed_password: Check the password that is being passed in against the hashed password stored
        in the database
        :return: True if the plain_password is correct, and false otherwise
        """
        return self.pwd_context.verify(plain_password, hashed_password)

    def get_password_hash(self, password: str):
        """
        The get_password_hash function takes a password as input and returns the hash of that password.
        The hash is generated using the pwd_context object, which is an instance of Flask-Bcrypt's Bcrypt class.

        :param password: str: Specify the password that will be hashed
        :return: A string that is a hash of the password
        """
        return self.pwd_context.hash(password)

    # define a function to generate a new access token
    async def create_access_token(self, data: dict, expires_delta: Optional[float] = None):
        """
        The create_access_token function creates a new access token.

        :param data: dict: Pass the data that will be encoded in the access token
        :param expires_delta: Optional[float]: Set the expiration time of the token
        :return: An encoded access token
        """
        to_encode = data.copy()
        if expires_delta:
            expire = datetime.utcnow() + timedelta(seconds=expires_delta)
        else:
            expire = datetime.utcnow() + timedelta(minutes=60)
        to_encode.update({"iat": datetime.utcnow(), "exp": expire, "scope": "access_token"})
        encoded_access_token = jwt.encode(to_encode, self.SECRET_KEY, algorithm=self.ALGORITHM)
        return encoded_access_token

    # define a function to generate a new refresh token
    async def create_refresh_token(self, data: dict, expires_delta: Optional[float] = None):
        """
        The create_refresh_token function creates a refresh token for the user.

        :param data: dict: Pass in the user's information, such as their username and email
        :param expires_delta: Optional[float]: Set the expiry time of the refresh token
        :return: An encoded token that contains the data passed to it as well as a timestamp
        when the token was created and an expiration date
        """
        to_encode = data.copy()
        if expires_delta:
            expire = datetime.utcnow() + timedelta(seconds=expires_delta)
        else:
            expire = datetime.utcnow() + timedelta(days=7)
        to_encode.update({"iat": datetime.utcnow(), "exp": expire, "scope": "refresh_token"})
        encoded_refresh_token = jwt.encode(to_encode, self.SECRET_KEY, algorithm=self.ALGORITHM)
        return encoded_refresh_token

    async def decode_refresh_token(self, refresh_token: str):
        """
        The decode_refresh_token function is used to decode the refresh token.
        It takes a refresh_token as an argument and returns the email of the user if it's valid.
        If not, it raises an HTTPException with status code 401 (UNAUTHORIZED)
        and detail 'Could not validate credentials'.

        :param refresh_token: str: Pass the refresh token to the function
        :return: The email of the user associated with the refresh token
        """
        try:
            payload = jwt.decode(refresh_token, self.SECRET_KEY, algorithms=[self.ALGORITHM])
            if payload['scope'] == 'refresh_token':
                email = payload['sub']
                return email
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=AuthMessages.invalid_scope_for_token)
        except JWTError:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
                                detail=AuthMessages.could_not_validate_credentials)

    async def get_current_user(self, token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
        """
        The get_current_user function is a dependency that will be used in the
        protected endpoints. It takes a token as an argument and returns the user
        if it's valid, otherwise raises an HTTPException with status code 401.

        :param token: str: Get the token from the authorization header
        :param db: Session: Pass the database session to the function
        :return: A user object
        """
        credentials_exception = HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=AuthMessages.could_not_validate_credentials,
            headers={"WWW-Authenticate": "Bearer"},
        )
        try:
            # Decode JWT
            payload = jwt.decode(token, self.SECRET_KEY, algorithms=[self.ALGORITHM])
            if payload['scope'] == 'access_token':
                email = payload["sub"]
                if email is None:
                    raise credentials_exception
            else:
                raise credentials_exception
        except JWTError as e:
            raise credentials_exception
        # user = await startup.get(f"user:{email}")
        # if user is None:
        user = await repository_users.get_user_by_email(email, db)
        #     if user is None:
        #         raise credentials_exception
        #     await startup.set(f"user:{email}", pickle.dumps(user))
        #     await startup.expire(f"user:{email}", 900)
        # else:
        #     user = pickle.loads(user)
        return user

    def create_email_token(self, data: dict):
        """
        The create_email_token function takes a dictionary of data and returns a token.
        The token is created by encoding the data with the SECRET_KEY and ALGORITHM,
        and adding an iat (issued at) timestamp and exp (expiration) timestamp to it.

        :param data: dict: Pass in the data that will be encoded into the token
        :return: A token
        """
        to_encode = data.copy()
        expire = datetime.utcnow() + timedelta(days=1)
        to_encode.update({"iat": datetime.utcnow(), "exp": expire})
        token = jwt.encode(to_encode, self.SECRET_KEY, algorithm=self.ALGORITHM)
        return token

    async def get_email_from_token(self, token: str):
        """
        The get_email_from_token function takes a token as an argument
        and returns the email address associated with that token.
        The function uses the jwt library to decode the token, which is then used to return the email address.

        :param token: str: Pass in the token that is sent to the user's email address
        :return: The email address of the user who is currently logged in
        """
        try:
            payload = jwt.decode(token, self.SECRET_KEY, algorithms=[self.ALGORITHM])
            email = payload["sub"]
            return email
        except JWTError as e:
            print(e)
            raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                                detail=AuthMessages.invalid_token_for_email_verification)


auth_service = Auth()

src/services/auth.py
DenBilokon/FreeChatAi
# import streamlit as st
# from dotenv import load_dotenv
# from PyPDF2 import PdfReader
# from langchain.text_splitter import CharacterTextSplitter
# from langchain.embeddings import OpenAIEmbeddings, HuggingFaceEmbeddings
# from langchain.vectorstores import FAISS
# from langchain.memory import ConversationBufferMemory
# from langchain.chains import ConversationalRetrievalChain
# from langchain.chat_models import ChatOpenAI
# from langchain.llms import HuggingFaceHub
# from htmlTemplates import bot_template, user_template, css
#
#
# def get_pdf_text(pdf_files):
#     text = ""
#     for pdf_file in pdf_files:
#         reader = PdfReader(pdf_file)
#         for page in reader.pages:
#             text += page.extract_text()
#     return text
#
#
# def get_chunk_text(text):
#     text_splitter = CharacterTextSplitter(
#         separator="\n",
#         chunk_size=1000,
#         chunk_overlap=200,
#         length_function=len
#     )
#
#     chunks = text_splitter.split_text(text)
#
#     return chunks
#
#
# def get_vector_store(text_chunks):
#     # For OpenAI Embeddings
#
#     embeddings = OpenAIEmbeddings()
#
#     # For Huggingface Embeddings
#
#     # embeddings = HuggingFaceEmbeddings(model_name='sentence-transformers/all-mpnet-base-v2')
#
#     vectorstore = FAISS.from_texts(texts=text_chunks, embedding=embeddings)
#
#     return vectorstore
#
#
# def get_conversation_chain(vector_store):
#     # OpenAI Model
#
#     llm = ChatOpenAI()
#
#     # HuggingFace Model
#
#     # llm = HuggingFaceHub(repo_id="tiiuae/falcon-7b-instruct", model_kwargs={"temperature":0.6, "max_new_tokens": 100})
#
#     memory = ConversationBufferMemory(memory_key='chat_history', return_messages=True)
#
#     conversation_chain = ConversationalRetrievalChain.from_llm(
#         llm=llm,
#         retriever=vector_store.as_retriever(search_kwargs={'k': 6}),
#         memory=memory
#     )
#
#     return conversation_chain
#
#
# def handle_user_input(question):
#     response = st.session_state.conversation({'question': question})
#     st.session_state.chat_history = response['chat_history']
#
#     for i, message in enumerate(st.session_state.chat_history):
#         if i % 2 == 0:
#             st.write(user_template.replace("{{MSG}}", message.content), unsafe_allow_html=True)
#         else:
#             st.write(bot_template.replace("{{MSG}}", message.content), unsafe_allow_html=True)
#
#
# def main():
#     load_dotenv()
#     st.set_page_config(page_title='Chat with Your own PDFs', page_icon=':books:')
#
#     st.write(css, unsafe_allow_html=True)
#
#     if "conversation" not in st.session_state:
#         st.session_state.conversation = None
#
#     if "chat_history" not in st.session_state:
#         st.session_state.chat_history = None
#
#     st.header('Chat with Your own PDFs :books:')
#     question = st.text_input("Ask anything to your PDF: ")
#
#     if question:
#         handle_user_input(question)
#
#     with st.sidebar:
#         st.subheader("Upload your Documents Here: ")
#         pdf_files = st.file_uploader("Choose your PDF Files and Press OK", type=['pdf'], accept_multiple_files=True)
#
#         if st.button("OK"):
#             with st.spinner("Processing your PDFs..."):
#                 # Get PDF Text
#                 raw_text = get_pdf_text(pdf_files)
#
#                 # Get Text Chunks
#                 text_chunks = get_chunk_text(raw_text)
#
#                 # Create Vector Store
#
#                 vector_store = get_vector_store(text_chunks)
#                 st.write("DONE")
#
#                 # Create conversation chain
#
#                 st.session_state.conversation = get_conversation_chain(vector_store)
#
#
# if __name__ == '__main__':
#     main()

src/services/doc_handler.py
DenBilokon/FreeChatAi
from pathlib import Path

import logging
from fastapi_mail import FastMail, MessageSchema, ConnectionConfig, MessageType
from fastapi_mail.errors import ConnectionErrors
from pydantic import EmailStr

from src.services.auth import auth_service
from src.conf.config import settings

conf = ConnectionConfig(
    MAIL_USERNAME=settings.mail_username,
    MAIL_PASSWORD=settings.mail_password,
    MAIL_FROM=settings.mail_from,
    MAIL_PORT=settings.mail_port,
    MAIL_SERVER=settings.mail_server,
    MAIL_FROM_NAME="App Name",  # TODO Change name
    MAIL_STARTTLS=False,
    MAIL_SSL_TLS=True,
    USE_CREDENTIALS=True,
    VALIDATE_CERTS=True,
    TEMPLATE_FOLDER=Path(__file__).parent / 'templates',
)


async def send_email(email: EmailStr, username: str, host: str):
    """
    The send_email function sends an email to the user with a link to confirm their email address.
    The function takes in three parameters:
    -email: EmailStr, the user's email address that they entered when signing up for an account.
    -username: str, the username of the user who is trying to sign up for an account. This will be used in
    conjunction with host (see below) and token_verification (see below) as part of a URL that will be sent
    via email to verify their identity and allow them to access into our system.

    :param email: EmailStr: Specify the email address of the recipient
    :param username: str: Pass the username to the template
    :param host: str: Create the link to the frontend
    :return: A coroutine object
    """
    try:
        token_verification = auth_service.create_email_token({"sub": email})
        message = MessageSchema(
            subject="Confirm your email",
            recipients=[email],
            template_body={"host": host, "username": username, "token": token_verification},
            subtype=MessageType.html
        )

        fm = FastMail(conf)
        await fm.send_message(message, template_name="email_template.html")
    except ConnectionErrors as err:
        logging.error(err)

src/services/email1.py
DenBilokon/FreeChatAi
# from pathlib import Path
#
#
# BASE_DIR = Path(__file__).resolve().parent
# filename = 'stepik-certificate.pdf'
#
# path_to_file = BASE_DIR / "docs" / filename
#
# print(path_to_file)

a = "hello world"


b = a.split()

print(b)
src/services/for_test.py
DenBilokon/FreeChatAi
css = '''
<style>
.chat-message {
    padding: 1.5rem; border-radius: 0.5rem; margin-bottom: 1rem; display: flex
}
.chat-message.user {
    background-color: #2b313e
}
.chat-message.bot {
    background-color: #475063
}
.chat-message .avatar {
  width: 20%;
}
.chat-message .avatar img {
  max-width: 78px;
  max-height: 78px;
  border-radius: 50%;
  object-fit: cover;
}
.chat-message .message {
  width: 80%;
  padding: 0 1.5rem;
  color: #fff;
}
'''

bot_template = '''
<div class="chat-message bot">
    <div class="avatar">
        <img src="https://i.ibb.co/cN0nmSj/Screenshot-2023-05-28-at-02-37-21.png">
    </div>
    <div class="message">{{MSG}}</div>
</div>
'''

user_template = '''
<div class="chat-message user">
    <div class="avatar">
        <img src="https://i.ibb.co/rdZC7LZ/Photo-logo-1.png">
    </div>    
    <div class="message">{{MSG}}</div>
</div>
'''
src/services/htmlTemplates.py
DenBilokon/FreeChatAi
import logging

import openai

from src.conf.config import settings

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


async def question_to_ai(question, pdf_text: str):
    openai.api_key = settings.openai_key

    worker = (
        f"Document Text: {pdf_text}\n\n"
        "Your main task is to give answers with Document Text. "
        "Your name is BOB. "
        "Always strive to give concise answers. "
        "Answer in Ukrainian, unless otherwise indicated. "
    )

    messages = [{"role": "system", "content": worker}]

    if len(question) > 4000:  # Перевіряємо обмеження на довжину запиту

        return "Ваше питання занадто довге. Спробуйте скоротити його."
    messages.append({"role": "user", "content": question})

    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        max_tokens=500,
        messages=messages
    )

    total_tokens = response.get("usage").get("total_tokens")
    logger.info(f"Start request to GPT with messages: {messages}")
    logger.info(f'Total tokens: {total_tokens}')

    response_html = f'I: {question}\nFreeDoc-AI: {response.choices[0].message.content.strip()}'

    return response_html

src/services/question_to_chatGPT.py
DenBilokon/FreeChatAi
from typing import List

from fastapi import Request, Depends, HTTPException, status

from src.conf.messages import RolesMessages
from src.database.models import User, Role
from src.services.auth import auth_service


class RolesAccess:
    def __init__(self, allowed_role: List[Role]):
        self.allowed_roles = allowed_role

    async def __call__(self, request: Request, current_user: User = Depends(auth_service.get_current_user)):
        if current_user.user_role not in self.allowed_roles:
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=RolesMessages.operation_forbidden)

src/services/roles.py
DenBilokon/FreeChatAi
from fastapi import HTTPException, status
from typing import List
from langchain.chains import ConversationalRetrievalChain
from langchain.chat_models import ChatOpenAI
from langchain.embeddings import OpenAIEmbeddings
from langchain.memory import ConversationBufferMemory
from langchain.text_splitter import CharacterTextSplitter
from langchain.vectorstores import FAISS
from sqlalchemy.orm import Session

from src.conf import messages
from src.database.models import User
from src.repository.chats import get_chat_by_id
from dotenv import load_dotenv

load_dotenv()


async def get_context(chat_id: int, db: Session, user: User) -> List:

    chat = await get_chat_by_id(chat_id, db, user)
    context = chat.file_url
    if context is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=messages.NOT_FOUND)

    with open(context, "r", encoding="utf-8") as file:
        text = file.read()

    text_splitter = CharacterTextSplitter(
        separator="\n",
        chunk_size=1000,
        chunk_overlap=200,
        length_function=len
    )
    chunks = text_splitter.split_text(text)

    return chunks


def get_conversation_chain(user_question, context) -> str:

    embeddings = OpenAIEmbeddings()
    vectorstore = FAISS.from_texts(texts=context, embedding=embeddings)

    llm = ChatOpenAI(model='gpt-3.5-turbo')
    memory = ConversationBufferMemory(
        memory_key='chat_history', return_messages=True)
    conversation_chain = ConversationalRetrievalChain.from_llm(
        llm=llm,
        retriever=vectorstore.as_retriever(),
        memory=memory,
    )

    response = conversation_chain({'question': user_question})
    chat_history = response['chat_history']

    # for separate question/answer items
    response_dict = {'user_messages': [], 'bot_messages': []}

    for i, message in enumerate(chat_history):
        if i % 2 == 0:
            response_dict['user_messages'].append(message.content)
        else:
            response_dict['bot_messages'].append(message.content)

    return str(response_dict)

src/services/single_long_doc.py
DenBilokon/FreeChatAi
import asyncio
import logging
import os
from pathlib import Path

from langchain.chains import RetrievalQA
from langchain.document_loaders import PyPDFLoader
from langchain.embeddings import OpenAIEmbeddings
from langchain.chat_models import ChatOpenAI
from langchain.text_splitter import CharacterTextSplitter
from langchain.vectorstores import Chroma

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# BASE_DIR = Path(__file__).parent.parent.resolve().parent
# filename = 'stepik-certificate(2).pdf'
#
# path_to_file = str(BASE_DIR / "uploaded_files" / filename)
#
#
# print(path_to_file)


async def vector_func(path_file, question):
    loader = PyPDFLoader(path_file)
    documents = loader.load()
    print(f"vector_func{documents}")
    vectordb = Chroma.from_documents(
        documents,
        embedding=OpenAIEmbeddings(),
        persist_directory='./data'
    )
    vectordb.persist()

    qa_chain = RetrievalQA.from_chain_type(
        llm=ChatOpenAI(model='gpt-3.5-turbo'),
        retriever=vectordb.as_retriever(search_kwargs={'k': 7}),
        return_source_documents=True
    )

    # we can now execute queries against our Q&A chain
    result = qa_chain({'query': question})
    print(f'result query: {result}')
    print(result['result'])

    return f'I: {question}\nFreeDoc-AI: {result["result"]}'

# while True:
#     a = input(f'>>>')
#     loop = asyncio.get_event_loop()
#     loop.run_until_complete(vector_func(path_to_file, a))
#     if a == '0':
#         break

src/services/vector_doc_ai.py
DenBilokon/FreeChatAi