database/operations.py - main - dbccx/VestnikChecker

database/operations.py
181 строка5 KB
t i a
сессии
15 июн 2026, 12:00
15 июн 2026, 12:00b0a93db
import hashlib
from typing import Optional
from sqlalchemy.orm import joinedload
from database.connection import SessionLocal
from database.models import User, Report, LLMSettings


# Пользователи

def authenticate_user(username: str, password: str) -> Optional[User]:
    session = SessionLocal()
    password_hash = hashlib.sha256(password.encode()).hexdigest()
    user = session.query(User).filter(
        User.username == username,
        User.password_hash == password_hash
    ).first()
    session.close()
    return user


def register_user(username: str, password: str, full_name: str, role: str = "editor") -> tuple:
    session = SessionLocal()
    existing = session.query(User).filter(User.username == username).first()
    if existing:
        session.close()
        return False, "Пользователь с таким логином уже существует"
    if role not in ("admin", "editor"):
        session.close()
        return False, "Недопустимая роль"
    password_hash = hashlib.sha256(password.encode()).hexdigest()
    new_user = User(
        username=username,
        password_hash=password_hash,
        role=role,
        full_name=full_name
    )
    session.add(new_user)
    session.commit()
    session.close()
    return True, "Регистрация успешна"


def get_user_by_id(user_id: int) -> Optional[User]:
    session = SessionLocal()
    user = session.query(User).filter(User.id == user_id).first()
    session.close()
    return user


#Отчеты
def get_file_hash(file_bytes: bytes) -> str:
    return hashlib.md5(file_bytes).hexdigest()


def find_report_by_hash(file_hash: str) -> Optional[Report]:
    session = SessionLocal()
    report = session.query(Report).filter(Report.file_hash == file_hash).first()
    session.close()
    return report


def save_report(article_title: str, file_name: str, file_hash: str,
                reviewer_id: int, content_category: str, content_percentage: float,
                format_category: str, format_percentage: float,
                result_json: str, original_pdf: bytes = None,
                report_pdf: bytes = None, llm_settings_id: int = None) -> Report:
    session = SessionLocal()
    report = Report(
        article_title=article_title,
        file_name=file_name,
        file_hash=file_hash,
        reviewer_id=reviewer_id,
        llm_settings_id=llm_settings_id,
        content_category=content_category,
        content_percentage=content_percentage,
        format_category=format_category,
        format_percentage=format_percentage,
        result_json=result_json,
        original_pdf=original_pdf,
        report_pdf=report_pdf,
    )
    session.add(report)
    session.commit()
    session.refresh(report)
    session.close()
    return report


def get_all_reports() -> list:
    session = SessionLocal()
    reports = (
        session.query(Report)
        .options(
            joinedload(Report.reviewer),
            joinedload(Report.llm_settings)
        )
        .order_by(Report.created_at.desc())
        .all()
    )
    session.close()
    return reports


def get_report_by_id(report_id: int) -> Optional[Report]:
    session = SessionLocal()
    report = (
        session.query(Report)
        .options(
            joinedload(Report.reviewer),
            joinedload(Report.llm_settings)
        )
        .filter(Report.id == report_id)
        .first()
    )
    session.close()
    return report


def delete_report(report_id: int) -> bool:
    session = SessionLocal()
    report = session.query(Report).filter(Report.id == report_id).first()
    if report:
        session.delete(report)
        session.commit()
        session.close()
        return True
    session.close()
    return False


#Настройки LLM

def get_llm_settings() -> Optional[LLMSettings]:
    session = SessionLocal()
    settings = session.query(LLMSettings).order_by(LLMSettings.id.desc()).first()
    session.close()
    return settings


def get_llm_settings_by_id(settings_id: int) -> Optional[LLMSettings]:
    session = SessionLocal()
    settings = session.query(LLMSettings).filter(LLMSettings.id == settings_id).first()
    session.close()
    return settings


def save_llm_settings(api_key: str, model_name: str,
                      system_prompt: str, temperature: float, max_tokens: int,
                      base_url: str = "https://proxy.gen-api.ru/v1",
                      provider: str = "openai") -> LLMSettings:
    session = SessionLocal()

    existing = session.query(LLMSettings).order_by(LLMSettings.id.desc()).first()

    if existing:
        # Обновляем существующую
        existing.provider = provider
        existing.api_key = api_key
        existing.model_name = model_name
        existing.base_url = base_url
        existing.system_prompt = system_prompt
        existing.temperature = temperature
        existing.max_tokens = max_tokens
        settings = existing
    else:
        # Создаём новую
        settings = LLMSettings(
            provider=provider,
            api_key=api_key,
            model_name=model_name,
            base_url=base_url,
            system_prompt=system_prompt,
            temperature=temperature,
            max_tokens=max_tokens
        )
        session.add(settings)

    session.commit()
    session.refresh(settings)
    session.close()
    return settings