api/crud.py - main - killogram/E-Analyzer

api/crud.py
213 строк9 KB
Korabliov Kirill
Refactor add purchase feachure
01 июн 2025, 23:54
01 июн 2025, 23:549d93c35
from sqlalchemy.orm import Session
from sqlalchemy import text, UniqueConstraint
from typing import List, Optional
from datetime import datetime
from fastapi import HTTPException
from api.models import Purchase, Price, Date, Customer, TrackedPurchase, PurchaseCustomer
from api import schemas

# Существующие функции
def search_purchases(db: Session, query: str) -> List[Purchase]:
    normalized_query = query.replace("№", "").replace(" ", "").strip()
    ts_query = query.replace(" ", " & ") if query else "*:*"
    return db.query(Purchase).filter(
        (Purchase.registry_number.ilike(f"%{normalized_query}%")) |
        (Purchase.registry_number.ilike(f"%№{normalized_query}%")) |
        (text("to_tsvector('russian', name) @@ to_tsquery('russian', :ts_query)").bindparams(ts_query=ts_query))
    ).all()

def get_purchase_by_number(db: Session, registry_number: str) -> Optional[Purchase]:
    return db.query(Purchase).filter(Purchase.registry_number == registry_number).first()

def get_purchase_by_id(db: Session, purchase_id: int) -> Optional[Purchase]:
    return db.query(Purchase).filter(Purchase.id == purchase_id).first()

def get_prices(db: Session, purchase_id: int) -> List[Price]:
    return db.query(Price).filter(Price.purchase_id == purchase_id).all()

def get_dates(db: Session, purchase_id: int) -> List[Date]:
    return db.query(Date).filter(Date.purchase_id == purchase_id).all()

def get_customer_by_id(db: Session, customer_id: int) -> Optional[Customer]:
    return db.query(Customer).filter(Customer.id == customer_id).first()

def track_purchase(db: Session, purchase_id: int, status: str) -> Optional[int]:
    query = text("""
        INSERT INTO tracked_purchases (purchase_id, status, created_at)
        VALUES (:purchase_id, :status, CURRENT_TIMESTAMP)
        ON CONFLICT (purchase_id)
        DO UPDATE SET status = :status, created_at = CURRENT_TIMESTAMP
        RETURNING id
    """)
    result = db.execute(query, {"purchase_id": purchase_id, "status": status})
    db.commit()
    return result.scalar()

def get_tracked_purchases(db: Session) -> List[TrackedPurchase]:
    return db.query(TrackedPurchase).filter(TrackedPurchase.status != 'Скрыто').all()

def update_tracked_purchase_status(db: Session, purchase_id: int, status: str) -> bool:
    if status == 'Скрыто':
        result = db.query(TrackedPurchase).filter(TrackedPurchase.purchase_id == purchase_id).delete()
        db.commit()
        return result > 0
    result = db.query(TrackedPurchase).filter(TrackedPurchase.purchase_id == purchase_id).update({
        "status": status,
        "created_at": datetime.now()
    })
    db.commit()
    return result > 0

def get_purchase_by_registry_number(db: Session, registry_number: str):
    return db.query(Purchase).filter(Purchase.registry_number == registry_number).first()

def create_purchase(db: Session, purchase: schemas.PurchaseCreate):
    db_purchase = Purchase(
        registry_number=purchase.registry_number,
        stage=purchase.stage,
        method=purchase.method,
        name=purchase.name,
        law=purchase.law,
        okpd2=purchase.okpd2,
        extra_data=purchase.extra_data,
        created_at=purchase.created_at or datetime.now(),
        updated_at=purchase.updated_at or datetime.now()
    )
    db.add(db_purchase)
    db.commit()
    db.refresh(db_purchase)
    return db_purchase

def create_price(db: Session, price: schemas.PriceCreate):
    db_price = Price(
        purchase_id=price.purchase_id,
        initial_price=price.initial_price,
        currency=price.currency,
        contract_price=price.contract_price,
        contract_currency=price.contract_currency
    )
    db.add(db_price)
    db.commit()
    db.refresh(db_price)
    return db_price

def create_date(db: Session, date: schemas.DateCreate):
    db_date = Date(
        purchase_id=date.purchase_id,
        placement_date=date.placement_date,
        update_date=date.update_date,
        application_start_date=date.application_start_date,
        application_end_date=date.application_end_date
    )
    db.add(db_date)
    db.commit()
    db.refresh(db_date)
    return db_date

# Функция create_customer (без purchase_id)
def create_customer(db: Session, customer: schemas.CustomerCreate):
    db_customer = Customer(
        name=customer.name,
        extra_data=customer.extra_data or {}
    )
    db.add(db_customer)
    db.commit()
    db.refresh(db_customer)
    return db_customer

def create_tracked_purchase(db: Session, tracked_purchase: schemas.TrackedPurchaseCreate):
    db_tracked_purchase = TrackedPurchase(
        purchase_id=tracked_purchase.purchase_id,
        status=tracked_purchase.status,
        created_at=datetime.now()
    )
    db.add(db_tracked_purchase)
    db.commit()
    db.refresh(db_tracked_purchase)
    return db_tracked_purchase

# Функция для получения всех заказчиков
def get_all_customers(db: Session) -> List[Customer]:
    return db.query(Customer).all()

# Функция для связывания закупки и заказчика
def link_purchase_customer(db: Session, purchase_id: int, customer_name: str):
    db_customer = db.query(Customer).filter(Customer.name == customer_name).first()
    if not db_customer:
        db_customer = create_customer(db, schemas.CustomerCreate(name=customer_name))
    db_purchase = db.query(Purchase).filter(Purchase.id == purchase_id).first()
    if not db_purchase:
        raise HTTPException(status_code=404, detail="Закупка не найдена")
    db_link = db.query(PurchaseCustomer).filter(
        PurchaseCustomer.purchase_id == purchase_id,
        PurchaseCustomer.customer_id == db_customer.id
    ).first()
    if not db_link:
        db_link = PurchaseCustomer(purchase_id=purchase_id, customer_id=db_customer.id)
        db.add(db_link)
        db.commit()
        db.refresh(db_link)
    return db_link

# Новая функция для получения заказчиков по purchase_id
def get_customers(db: Session, purchase_id: int) -> List[Customer]:
    customer_ids = db.query(PurchaseCustomer.customer_id).filter(PurchaseCustomer.purchase_id == purchase_id).all()
    print(f"Purchase ID: {purchase_id}, Found customer IDs: {customer_ids}")  # Временное логирование
    if not customer_ids:
        return []
    customer_ids = [cid[0] for cid in customer_ids]
    customers = db.query(Customer).filter(Customer.id.in_(customer_ids)).all()
    print(f"Found customers: {customers}")  # Временное логирование
    return customers

# Функция create_purchase_with_customer
def create_purchase_with_customer(db: Session, purchase_data: schemas.AddPurchaseForm):
    # Проверяем, существует ли закупка с таким реестровым номером
    existing_purchase = get_purchase_by_registry_number(db, purchase_data.registry_number)
    if existing_purchase:
        raise HTTPException(status_code=400, detail="Purchase with this registry number already exists")

    # Создаём запись в таблице purchases
    purchase = schemas.PurchaseCreate(
        registry_number=purchase_data.registry_number,
        stage=purchase_data.stage,
        method=purchase_data.method,
        name=purchase_data.name,
        law=purchase_data.law,
        okpd2=purchase_data.okpd2,
        created_at=datetime.now(),
        updated_at=datetime.now()
    )
    db_purchase = create_purchase(db, purchase)

    # Создаём запись в таблице prices
    price = schemas.PriceCreate(
        purchase_id=db_purchase.id,
        initial_price=purchase_data.initial_price,
        currency=purchase_data.currency,
        contract_price=purchase_data.contract_price,
        contract_currency=purchase_data.contract_currency
    )
    create_price(db, price)

    # Создаём запись в таблице dates
    date = schemas.DateCreate(
        purchase_id=db_purchase.id,
        placement_date=purchase_data.placement_date,
        update_date=purchase_data.update_date,
        application_start_date=purchase_data.application_start_date,
        application_end_date=purchase_data.application_end_date
    )
    create_date(db, date)

    # Связь с заказчиком
    link_purchase_customer(db, db_purchase.id, purchase_data.customer_name)

    # Создаём запись в таблице tracked_purchases со статусом "Участвую"
    tracked = schemas.TrackedPurchaseCreate(
        purchase_id=db_purchase.id,
        status="Участвую"
    )
    create_tracked_purchase(db, tracked)

    return db_purchase