920 lines
36 KiB
Python
920 lines
36 KiB
Python
# services/inventory/app/repositories/stock_repository.py
|
|
"""
|
|
Stock Repository using Repository Pattern
|
|
"""
|
|
|
|
from typing import List, Optional, Dict, Any, Tuple
|
|
from uuid import UUID
|
|
from datetime import datetime, timedelta
|
|
from decimal import Decimal
|
|
from sqlalchemy import select, func, and_, or_, desc, asc, update, exists
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
import structlog
|
|
|
|
from app.models.inventory import Stock, Ingredient
|
|
from app.schemas.inventory import StockCreate, StockUpdate
|
|
from shared.database.repository import BaseRepository
|
|
from shared.utils.batch_generator import BatchCountProvider
|
|
|
|
logger = structlog.get_logger()
|
|
|
|
|
|
class StockRepository(BaseRepository[Stock, StockCreate, StockUpdate], BatchCountProvider):
|
|
"""Repository for stock operations"""
|
|
|
|
def __init__(self, session: AsyncSession):
|
|
super().__init__(Stock, session)
|
|
|
|
async def create_stock_entry(self, stock_data: StockCreate, tenant_id: UUID) -> Stock:
|
|
"""Create a new stock entry"""
|
|
try:
|
|
# Prepare data
|
|
create_data = stock_data.model_dump()
|
|
create_data['tenant_id'] = tenant_id
|
|
|
|
# Ensure production_stage is properly converted to enum value
|
|
if 'production_stage' in create_data:
|
|
if hasattr(create_data['production_stage'], 'value'):
|
|
create_data['production_stage'] = create_data['production_stage'].value
|
|
elif isinstance(create_data['production_stage'], str):
|
|
# If it's a string, ensure it's the correct enum value
|
|
from app.models.inventory import ProductionStage
|
|
try:
|
|
enum_obj = ProductionStage[create_data['production_stage']]
|
|
create_data['production_stage'] = enum_obj.value
|
|
except KeyError:
|
|
# If it's already the value, keep it as is
|
|
pass
|
|
|
|
# Calculate available quantity
|
|
available_qty = create_data['current_quantity'] - create_data.get('reserved_quantity', 0)
|
|
create_data['available_quantity'] = max(0, available_qty)
|
|
|
|
# Calculate total cost if unit cost provided
|
|
if create_data.get('unit_cost') and create_data.get('current_quantity'):
|
|
unit_cost = create_data['unit_cost']
|
|
current_quantity = Decimal(str(create_data['current_quantity']))
|
|
create_data['total_cost'] = unit_cost * current_quantity
|
|
|
|
# Create record
|
|
record = await self.create(create_data)
|
|
logger.info(
|
|
"Created stock entry",
|
|
stock_id=record.id,
|
|
ingredient_id=record.ingredient_id,
|
|
quantity=record.current_quantity,
|
|
tenant_id=tenant_id
|
|
)
|
|
return record
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to create stock entry", error=str(e), tenant_id=tenant_id)
|
|
raise
|
|
|
|
async def get_stock_by_ingredient(
|
|
self,
|
|
tenant_id: UUID,
|
|
ingredient_id: UUID,
|
|
include_unavailable: bool = False
|
|
) -> List[Stock]:
|
|
"""Get all stock entries for a specific ingredient"""
|
|
try:
|
|
query = select(self.model).where(
|
|
and_(
|
|
self.model.tenant_id == tenant_id,
|
|
self.model.ingredient_id == ingredient_id
|
|
)
|
|
)
|
|
|
|
if not include_unavailable:
|
|
query = query.where(self.model.is_available == True)
|
|
|
|
query = query.order_by(asc(self.model.expiration_date))
|
|
|
|
result = await self.session.execute(query)
|
|
return result.scalars().all()
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get stock by ingredient", error=str(e), ingredient_id=ingredient_id)
|
|
raise
|
|
|
|
async def get_stock_by_product(
|
|
self,
|
|
tenant_id: UUID,
|
|
inventory_product_id: UUID,
|
|
include_unavailable: bool = False
|
|
) -> List[Stock]:
|
|
"""
|
|
Get all stock entries for a specific product.
|
|
|
|
Note: inventory_product_id and ingredient_id refer to the same entity.
|
|
The 'ingredients' table is used as a unified catalog for both raw ingredients
|
|
and finished products, distinguished by the product_type field.
|
|
|
|
This method is an alias for get_stock_by_ingredient for clarity when called
|
|
from contexts that use 'product' terminology (e.g., procurement service).
|
|
"""
|
|
return await self.get_stock_by_ingredient(
|
|
tenant_id=tenant_id,
|
|
ingredient_id=inventory_product_id,
|
|
include_unavailable=include_unavailable
|
|
)
|
|
|
|
async def get_total_stock_by_ingredient(self, tenant_id: UUID, ingredient_id: UUID) -> Dict[str, float]:
|
|
"""Get total stock quantities for an ingredient"""
|
|
try:
|
|
result = await self.session.execute(
|
|
select(
|
|
func.coalesce(func.sum(Stock.current_quantity), 0).label('total_quantity'),
|
|
func.coalesce(func.sum(Stock.reserved_quantity), 0).label('total_reserved'),
|
|
func.coalesce(func.sum(Stock.available_quantity), 0).label('total_available')
|
|
).where(
|
|
and_(
|
|
Stock.tenant_id == tenant_id,
|
|
Stock.ingredient_id == ingredient_id,
|
|
Stock.is_available == True
|
|
)
|
|
)
|
|
)
|
|
|
|
row = result.first()
|
|
return {
|
|
'total_quantity': float(row.total_quantity) if row.total_quantity else 0.0,
|
|
'total_reserved': float(row.total_reserved) if row.total_reserved else 0.0,
|
|
'total_available': float(row.total_available) if row.total_available else 0.0
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get total stock", error=str(e), ingredient_id=ingredient_id)
|
|
raise
|
|
|
|
async def get_expiring_stock(
|
|
self,
|
|
tenant_id: UUID,
|
|
days_ahead: int = 7
|
|
) -> List[Tuple[Stock, Ingredient]]:
|
|
"""Get stock items expiring within specified days using state-dependent expiration logic"""
|
|
try:
|
|
expiry_date = datetime.now() + timedelta(days=days_ahead)
|
|
|
|
# Use final_expiration_date if available (for transformed products),
|
|
# otherwise use regular expiration_date
|
|
result = await self.session.execute(
|
|
select(Stock, Ingredient)
|
|
.join(Ingredient, Stock.ingredient_id == Ingredient.id)
|
|
.where(
|
|
and_(
|
|
Stock.tenant_id == tenant_id,
|
|
Stock.is_available == True,
|
|
or_(
|
|
and_(
|
|
Stock.final_expiration_date.isnot(None),
|
|
Stock.final_expiration_date <= expiry_date
|
|
),
|
|
and_(
|
|
Stock.final_expiration_date.is_(None),
|
|
Stock.expiration_date.isnot(None),
|
|
Stock.expiration_date <= expiry_date
|
|
)
|
|
)
|
|
)
|
|
)
|
|
.order_by(
|
|
asc(
|
|
func.coalesce(Stock.final_expiration_date, Stock.expiration_date)
|
|
)
|
|
)
|
|
)
|
|
|
|
return result.all()
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get expiring stock", error=str(e), tenant_id=tenant_id)
|
|
raise
|
|
|
|
async def get_expired_stock(self, tenant_id: UUID) -> List[Tuple[Stock, Ingredient]]:
|
|
"""Get stock items that have expired using state-dependent expiration logic"""
|
|
try:
|
|
current_date = datetime.now()
|
|
|
|
# Use final_expiration_date if available (for transformed products),
|
|
# otherwise use regular expiration_date
|
|
result = await self.session.execute(
|
|
select(Stock, Ingredient)
|
|
.join(Ingredient, Stock.ingredient_id == Ingredient.id)
|
|
.where(
|
|
and_(
|
|
Stock.tenant_id == tenant_id,
|
|
Stock.is_available == True,
|
|
or_(
|
|
and_(
|
|
Stock.final_expiration_date.isnot(None),
|
|
Stock.final_expiration_date < current_date
|
|
),
|
|
and_(
|
|
Stock.final_expiration_date.is_(None),
|
|
Stock.expiration_date.isnot(None),
|
|
Stock.expiration_date < current_date
|
|
)
|
|
)
|
|
)
|
|
)
|
|
.order_by(
|
|
desc(
|
|
func.coalesce(Stock.final_expiration_date, Stock.expiration_date)
|
|
)
|
|
)
|
|
)
|
|
|
|
return result.all()
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get expired stock", error=str(e), tenant_id=tenant_id)
|
|
raise
|
|
|
|
async def reserve_stock(
|
|
self,
|
|
tenant_id: UUID,
|
|
ingredient_id: UUID,
|
|
quantity: float,
|
|
fifo: bool = True
|
|
) -> List[Dict[str, Any]]:
|
|
"""Reserve stock using FIFO/LIFO method with state-dependent expiration"""
|
|
try:
|
|
# Order by appropriate expiration date based on transformation status
|
|
effective_expiration = func.coalesce(Stock.final_expiration_date, Stock.expiration_date)
|
|
order_clause = asc(effective_expiration) if fifo else desc(effective_expiration)
|
|
|
|
result = await self.session.execute(
|
|
select(Stock).where(
|
|
and_(
|
|
Stock.tenant_id == tenant_id,
|
|
Stock.ingredient_id == ingredient_id,
|
|
Stock.is_available == True,
|
|
Stock.available_quantity > 0
|
|
)
|
|
).order_by(order_clause)
|
|
)
|
|
|
|
stock_items = result.scalars().all()
|
|
reservations = []
|
|
remaining_qty = quantity
|
|
|
|
for stock_item in stock_items:
|
|
if remaining_qty <= 0:
|
|
break
|
|
|
|
available = stock_item.available_quantity
|
|
to_reserve = min(remaining_qty, available)
|
|
|
|
# Update stock reservation
|
|
new_reserved = stock_item.reserved_quantity + to_reserve
|
|
new_available = stock_item.current_quantity - new_reserved
|
|
|
|
await self.session.execute(
|
|
update(Stock)
|
|
.where(Stock.id == stock_item.id)
|
|
.values(
|
|
reserved_quantity=new_reserved,
|
|
available_quantity=new_available
|
|
)
|
|
)
|
|
|
|
reservations.append({
|
|
'stock_id': stock_item.id,
|
|
'reserved_quantity': to_reserve,
|
|
'batch_number': stock_item.batch_number,
|
|
'expiration_date': stock_item.expiration_date
|
|
})
|
|
|
|
remaining_qty -= to_reserve
|
|
|
|
if remaining_qty > 0:
|
|
logger.warning(
|
|
"Insufficient stock for reservation",
|
|
ingredient_id=ingredient_id,
|
|
requested=quantity,
|
|
unfulfilled=remaining_qty
|
|
)
|
|
|
|
return reservations
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to reserve stock", error=str(e), ingredient_id=ingredient_id)
|
|
raise
|
|
|
|
async def release_stock_reservation(
|
|
self,
|
|
stock_id: UUID,
|
|
quantity: float
|
|
) -> Optional[Stock]:
|
|
"""Release reserved stock"""
|
|
try:
|
|
stock_item = await self.get_by_id(stock_id)
|
|
if not stock_item:
|
|
return None
|
|
|
|
# Calculate new quantities
|
|
new_reserved = max(0, stock_item.reserved_quantity - quantity)
|
|
new_available = stock_item.current_quantity - new_reserved
|
|
|
|
# Update stock
|
|
await self.session.execute(
|
|
update(Stock)
|
|
.where(Stock.id == stock_id)
|
|
.values(
|
|
reserved_quantity=new_reserved,
|
|
available_quantity=new_available
|
|
)
|
|
)
|
|
|
|
# Refresh and return updated stock
|
|
await self.session.refresh(stock_item)
|
|
return stock_item
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to release stock reservation", error=str(e), stock_id=stock_id)
|
|
raise
|
|
|
|
async def consume_stock(
|
|
self,
|
|
stock_id: UUID,
|
|
quantity: float,
|
|
from_reserved: bool = True
|
|
) -> Optional[Stock]:
|
|
"""Consume stock (reduce current quantity)"""
|
|
try:
|
|
stock_item = await self.get_by_id(stock_id)
|
|
if not stock_item:
|
|
return None
|
|
|
|
if from_reserved:
|
|
# Reduce from reserved quantity
|
|
new_reserved = max(0, stock_item.reserved_quantity - quantity)
|
|
new_current = max(0, stock_item.current_quantity - quantity)
|
|
new_available = new_current - new_reserved
|
|
else:
|
|
# Reduce from available quantity
|
|
new_current = max(0, stock_item.current_quantity - quantity)
|
|
new_available = max(0, stock_item.available_quantity - quantity)
|
|
new_reserved = stock_item.reserved_quantity
|
|
|
|
# Update stock
|
|
await self.session.execute(
|
|
update(Stock)
|
|
.where(Stock.id == stock_id)
|
|
.values(
|
|
current_quantity=new_current,
|
|
reserved_quantity=new_reserved,
|
|
available_quantity=new_available,
|
|
is_available=new_current > 0
|
|
)
|
|
)
|
|
|
|
# Refresh and return updated stock
|
|
await self.session.refresh(stock_item)
|
|
return stock_item
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to consume stock", error=str(e), stock_id=stock_id)
|
|
raise
|
|
|
|
async def get_stock_summary_by_tenant(self, tenant_id: UUID) -> Dict[str, Any]:
|
|
"""Get stock summary for tenant dashboard"""
|
|
try:
|
|
# Basic stock summary
|
|
basic_result = await self.session.execute(
|
|
select(
|
|
func.count(Stock.id).label('total_stock_items'),
|
|
func.coalesce(func.sum(Stock.total_cost), 0).label('total_stock_value'),
|
|
func.count(func.distinct(Stock.ingredient_id)).label('unique_ingredients')
|
|
).where(
|
|
and_(
|
|
Stock.tenant_id == tenant_id,
|
|
Stock.is_available == True
|
|
)
|
|
)
|
|
)
|
|
basic_summary = basic_result.first()
|
|
|
|
# Count expired items
|
|
expired_result = await self.session.execute(
|
|
select(func.count(Stock.id)).where(
|
|
and_(
|
|
Stock.tenant_id == tenant_id,
|
|
Stock.is_available == True,
|
|
Stock.expiration_date < datetime.now()
|
|
)
|
|
)
|
|
)
|
|
expired_count = expired_result.scalar() or 0
|
|
|
|
# Count expiring soon items
|
|
expiring_result = await self.session.execute(
|
|
select(func.count(Stock.id)).where(
|
|
and_(
|
|
Stock.tenant_id == tenant_id,
|
|
Stock.is_available == True,
|
|
Stock.expiration_date.isnot(None),
|
|
Stock.expiration_date <= datetime.now() + timedelta(days=7)
|
|
)
|
|
)
|
|
)
|
|
expiring_count = expiring_result.scalar() or 0
|
|
|
|
# Count out of stock items (ingredients with no available stock)
|
|
out_of_stock_result = await self.session.execute(
|
|
select(func.count(Ingredient.id)).where(
|
|
and_(
|
|
Ingredient.tenant_id == tenant_id,
|
|
~exists(
|
|
select(1).where(
|
|
and_(
|
|
Stock.ingredient_id == Ingredient.id,
|
|
Stock.tenant_id == tenant_id,
|
|
Stock.is_available == True,
|
|
Stock.available_quantity > 0
|
|
)
|
|
)
|
|
)
|
|
)
|
|
)
|
|
)
|
|
out_of_stock_count = out_of_stock_result.scalar() or 0
|
|
|
|
return {
|
|
'total_stock_items': basic_summary.total_stock_items or 0,
|
|
'total_stock_value': float(basic_summary.total_stock_value) if basic_summary.total_stock_value else 0.0,
|
|
'unique_ingredients': basic_summary.unique_ingredients or 0,
|
|
'expired_items': expired_count,
|
|
'expiring_soon_items': expiring_count,
|
|
'out_of_stock_count': out_of_stock_count
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get stock summary", error=str(e), tenant_id=tenant_id)
|
|
raise
|
|
|
|
async def mark_expired_stock(self, tenant_id: UUID) -> int:
|
|
"""Mark expired stock items as expired using state-dependent expiration logic"""
|
|
try:
|
|
current_date = datetime.now()
|
|
|
|
# Mark items as expired based on final_expiration_date or expiration_date
|
|
result = await self.session.execute(
|
|
update(Stock)
|
|
.where(
|
|
and_(
|
|
Stock.tenant_id == tenant_id,
|
|
Stock.is_expired == False,
|
|
or_(
|
|
and_(
|
|
Stock.final_expiration_date.isnot(None),
|
|
Stock.final_expiration_date < current_date
|
|
),
|
|
and_(
|
|
Stock.final_expiration_date.is_(None),
|
|
Stock.expiration_date.isnot(None),
|
|
Stock.expiration_date < current_date
|
|
)
|
|
)
|
|
)
|
|
)
|
|
.values(is_expired=True, quality_status="expired")
|
|
)
|
|
|
|
expired_count = result.rowcount
|
|
logger.info(f"Marked {expired_count} stock items as expired using state-dependent logic", tenant_id=tenant_id)
|
|
|
|
return expired_count
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to mark expired stock", error=str(e), tenant_id=tenant_id)
|
|
raise
|
|
|
|
async def get_stock_by_production_stage(
|
|
self,
|
|
tenant_id: UUID,
|
|
production_stage: 'ProductionStage',
|
|
ingredient_id: Optional[UUID] = None
|
|
) -> List['Stock']:
|
|
"""Get stock items by production stage"""
|
|
try:
|
|
conditions = [
|
|
Stock.tenant_id == tenant_id,
|
|
Stock.production_stage == production_stage,
|
|
Stock.is_available == True
|
|
]
|
|
|
|
if ingredient_id:
|
|
conditions.append(Stock.ingredient_id == ingredient_id)
|
|
|
|
result = await self.session.execute(
|
|
select(Stock)
|
|
.where(and_(*conditions))
|
|
.order_by(asc(func.coalesce(Stock.final_expiration_date, Stock.expiration_date)))
|
|
)
|
|
|
|
return result.scalars().all()
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get stock by production stage", error=str(e), production_stage=production_stage)
|
|
raise
|
|
|
|
async def get_stock_entries(
|
|
self,
|
|
tenant_id: UUID,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
ingredient_id: Optional[UUID] = None,
|
|
available_only: bool = True
|
|
) -> List[Stock]:
|
|
"""Get stock entries with filtering and pagination"""
|
|
try:
|
|
conditions = [Stock.tenant_id == tenant_id]
|
|
|
|
if available_only:
|
|
conditions.append(Stock.is_available == True)
|
|
|
|
if ingredient_id:
|
|
conditions.append(Stock.ingredient_id == ingredient_id)
|
|
|
|
query = (
|
|
select(Stock)
|
|
.where(and_(*conditions))
|
|
.order_by(desc(Stock.created_at))
|
|
.offset(skip)
|
|
.limit(limit)
|
|
)
|
|
|
|
result = await self.session.execute(query)
|
|
return result.scalars().all()
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get stock entries", error=str(e), tenant_id=tenant_id)
|
|
raise
|
|
|
|
async def delete_by_ingredient(self, ingredient_id: UUID, tenant_id: UUID) -> int:
|
|
"""Delete all stock entries for a specific ingredient"""
|
|
try:
|
|
from sqlalchemy import delete
|
|
|
|
stmt = delete(Stock).where(
|
|
and_(
|
|
Stock.ingredient_id == ingredient_id,
|
|
Stock.tenant_id == tenant_id
|
|
)
|
|
)
|
|
|
|
result = await self.session.execute(stmt)
|
|
deleted_count = result.rowcount
|
|
|
|
logger.info(
|
|
"Deleted stock entries for ingredient",
|
|
ingredient_id=str(ingredient_id),
|
|
tenant_id=str(tenant_id),
|
|
deleted_count=deleted_count
|
|
)
|
|
|
|
return deleted_count
|
|
|
|
except Exception as e:
|
|
logger.error(
|
|
"Failed to delete stock entries for ingredient",
|
|
error=str(e),
|
|
ingredient_id=str(ingredient_id),
|
|
tenant_id=str(tenant_id)
|
|
)
|
|
raise
|
|
|
|
async def get_daily_batch_count(
|
|
self,
|
|
tenant_id: str,
|
|
date_start: datetime,
|
|
date_end: datetime,
|
|
prefix: Optional[str] = None
|
|
) -> int:
|
|
"""Get the count of batches created today for the given tenant"""
|
|
try:
|
|
conditions = [
|
|
Stock.tenant_id == tenant_id,
|
|
Stock.created_at >= date_start,
|
|
Stock.created_at <= date_end
|
|
]
|
|
|
|
if prefix:
|
|
conditions.append(Stock.batch_number.like(f"{prefix}-%"))
|
|
|
|
stmt = select(func.count(Stock.id)).where(and_(*conditions))
|
|
result = await self.session.execute(stmt)
|
|
count = result.scalar() or 0
|
|
|
|
logger.debug(
|
|
"Retrieved daily batch count",
|
|
tenant_id=tenant_id,
|
|
prefix=prefix,
|
|
count=count,
|
|
date_start=date_start,
|
|
date_end=date_end
|
|
)
|
|
|
|
return count
|
|
|
|
except Exception as e:
|
|
logger.error(
|
|
"Failed to get daily batch count",
|
|
error=str(e),
|
|
tenant_id=tenant_id,
|
|
prefix=prefix
|
|
)
|
|
raise
|
|
|
|
async def get_expired_batches_for_processing(self, tenant_id: UUID) -> List[Tuple[Stock, Ingredient]]:
|
|
"""Get expired batches that haven't been processed yet (for automatic processing)"""
|
|
try:
|
|
current_date = datetime.now()
|
|
|
|
# Find expired batches that are still available and not yet marked as expired
|
|
result = await self.session.execute(
|
|
select(Stock, Ingredient)
|
|
.join(Ingredient, Stock.ingredient_id == Ingredient.id)
|
|
.where(
|
|
and_(
|
|
Stock.tenant_id == tenant_id,
|
|
Stock.is_available == True,
|
|
Stock.is_expired == False,
|
|
Stock.current_quantity > 0,
|
|
or_(
|
|
and_(
|
|
Stock.final_expiration_date.isnot(None),
|
|
Stock.final_expiration_date <= current_date
|
|
),
|
|
and_(
|
|
Stock.final_expiration_date.is_(None),
|
|
Stock.expiration_date.isnot(None),
|
|
Stock.expiration_date <= current_date
|
|
)
|
|
)
|
|
)
|
|
)
|
|
.order_by(
|
|
asc(func.coalesce(Stock.final_expiration_date, Stock.expiration_date))
|
|
)
|
|
)
|
|
|
|
expired_batches = result.all()
|
|
logger.info("Found expired batches for processing",
|
|
tenant_id=str(tenant_id),
|
|
count=len(expired_batches))
|
|
|
|
return expired_batches
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get expired batches for processing",
|
|
error=str(e), tenant_id=tenant_id)
|
|
raise
|
|
|
|
async def mark_batch_as_expired(self, stock_id: UUID, tenant_id: UUID) -> bool:
|
|
"""Mark a specific batch as expired and unavailable"""
|
|
try:
|
|
result = await self.session.execute(
|
|
update(Stock)
|
|
.where(
|
|
and_(
|
|
Stock.id == stock_id,
|
|
Stock.tenant_id == tenant_id
|
|
)
|
|
)
|
|
.values(
|
|
is_expired=True,
|
|
is_available=False,
|
|
quality_status="expired",
|
|
updated_at=datetime.now()
|
|
)
|
|
)
|
|
|
|
if result.rowcount > 0:
|
|
logger.info("Marked batch as expired",
|
|
stock_id=str(stock_id),
|
|
tenant_id=str(tenant_id))
|
|
return True
|
|
else:
|
|
logger.warning("No batch found to mark as expired",
|
|
stock_id=str(stock_id),
|
|
tenant_id=str(tenant_id))
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to mark batch as expired",
|
|
error=str(e),
|
|
stock_id=str(stock_id),
|
|
tenant_id=str(tenant_id))
|
|
raise
|
|
|
|
async def update_stock_to_zero(self, stock_id: UUID, tenant_id: UUID) -> bool:
|
|
"""Update stock quantities to zero after moving to waste"""
|
|
try:
|
|
result = await self.session.execute(
|
|
update(Stock)
|
|
.where(
|
|
and_(
|
|
Stock.id == stock_id,
|
|
Stock.tenant_id == tenant_id
|
|
)
|
|
)
|
|
.values(
|
|
current_quantity=0,
|
|
available_quantity=0,
|
|
updated_at=datetime.now()
|
|
)
|
|
)
|
|
|
|
if result.rowcount > 0:
|
|
logger.info("Updated stock quantities to zero",
|
|
stock_id=str(stock_id),
|
|
tenant_id=str(tenant_id))
|
|
return True
|
|
else:
|
|
logger.warning("No stock found to update to zero",
|
|
stock_id=str(stock_id),
|
|
tenant_id=str(tenant_id))
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to update stock to zero",
|
|
error=str(e),
|
|
stock_id=str(stock_id),
|
|
tenant_id=str(tenant_id))
|
|
raise
|
|
|
|
async def get_expiring_products(self, tenant_id: UUID, days_threshold: int = 7) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get products expiring soon or already expired
|
|
"""
|
|
try:
|
|
from sqlalchemy import text
|
|
query = text("""
|
|
SELECT
|
|
i.id as ingredient_id,
|
|
i.name as ingredient_name,
|
|
s.id as stock_id,
|
|
s.batch_number,
|
|
s.expiration_date,
|
|
s.current_quantity,
|
|
i.unit_of_measure,
|
|
s.unit_cost,
|
|
(s.current_quantity * s.unit_cost) as total_value,
|
|
CASE
|
|
WHEN s.expiration_date < CURRENT_DATE THEN 'expired'
|
|
WHEN s.expiration_date <= CURRENT_DATE + INTERVAL '1 day' THEN 'expires_today'
|
|
WHEN s.expiration_date <= CURRENT_DATE + INTERVAL '3 days' THEN 'expires_soon'
|
|
ELSE 'warning'
|
|
END as urgency,
|
|
EXTRACT(DAY FROM (s.expiration_date - CURRENT_DATE)) as days_until_expiry
|
|
FROM stock s
|
|
JOIN ingredients i ON s.ingredient_id = i.id
|
|
WHERE i.tenant_id = :tenant_id
|
|
AND s.is_available = true
|
|
AND s.expiration_date <= CURRENT_DATE + (INTERVAL '1 day' * :days_threshold)
|
|
ORDER BY s.expiration_date ASC, total_value DESC
|
|
""")
|
|
|
|
result = await self.session.execute(query, {
|
|
"tenant_id": tenant_id,
|
|
"days_threshold": days_threshold
|
|
})
|
|
return [dict(row._mapping) for row in result.fetchall()]
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get expiring products", error=str(e), tenant_id=str(tenant_id))
|
|
raise
|
|
|
|
async def get_temperature_breaches(self, tenant_id: UUID, hours_back: int = 24) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get temperature monitoring breaches
|
|
"""
|
|
try:
|
|
from sqlalchemy import text
|
|
query = text("""
|
|
SELECT
|
|
tl.id,
|
|
tl.equipment_id,
|
|
tl.equipment_name,
|
|
tl.storage_type,
|
|
tl.temperature_celsius,
|
|
tl.min_threshold,
|
|
tl.max_threshold,
|
|
tl.is_within_range,
|
|
tl.recorded_at,
|
|
tl.alert_triggered,
|
|
EXTRACT(EPOCH FROM (NOW() - tl.recorded_at))/3600 as hours_ago,
|
|
CASE
|
|
WHEN tl.temperature_celsius < tl.min_threshold
|
|
THEN tl.min_threshold - tl.temperature_celsius
|
|
WHEN tl.temperature_celsius > tl.max_threshold
|
|
THEN tl.temperature_celsius - tl.max_threshold
|
|
ELSE 0
|
|
END as deviation
|
|
FROM temperature_logs tl
|
|
WHERE tl.tenant_id = :tenant_id
|
|
AND tl.is_within_range = false
|
|
AND tl.recorded_at > NOW() - (INTERVAL '1 hour' * :hours_back)
|
|
AND tl.alert_triggered = false
|
|
ORDER BY deviation DESC, tl.recorded_at DESC
|
|
""")
|
|
|
|
result = await self.session.execute(query, {
|
|
"tenant_id": tenant_id,
|
|
"hours_back": hours_back
|
|
})
|
|
return [dict(row._mapping) for row in result.fetchall()]
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get temperature breaches", error=str(e), tenant_id=str(tenant_id))
|
|
raise
|
|
|
|
async def get_waste_opportunities(self, tenant_id: UUID) -> List[Dict[str, Any]]:
|
|
"""
|
|
Identify waste reduction opportunities
|
|
"""
|
|
try:
|
|
from sqlalchemy import text
|
|
query = text("""
|
|
WITH waste_analysis AS (
|
|
SELECT
|
|
i.id as ingredient_id,
|
|
i.name as ingredient_name,
|
|
i.ingredient_category,
|
|
COUNT(sm.id) as waste_incidents,
|
|
SUM(sm.quantity) as total_waste_quantity,
|
|
SUM(sm.total_cost) as total_waste_cost,
|
|
AVG(sm.quantity) as avg_waste_per_incident,
|
|
MAX(sm.movement_date) as last_waste_date
|
|
FROM stock_movements sm
|
|
JOIN ingredients i ON sm.ingredient_id = i.id
|
|
WHERE i.tenant_id = :tenant_id
|
|
AND sm.movement_type = 'WASTE'
|
|
AND sm.movement_date > NOW() - INTERVAL '30 days'
|
|
GROUP BY i.id, i.name, i.ingredient_category
|
|
HAVING COUNT(sm.id) >= 3 OR SUM(sm.total_cost) > 50
|
|
)
|
|
SELECT * FROM waste_analysis
|
|
ORDER BY total_waste_cost DESC, waste_incidents DESC
|
|
LIMIT 20
|
|
""")
|
|
|
|
result = await self.session.execute(query, {"tenant_id": tenant_id})
|
|
return [dict(row._mapping) for row in result.fetchall()]
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get waste opportunities", error=str(e), tenant_id=str(tenant_id))
|
|
raise
|
|
|
|
async def get_reorder_recommendations(self, tenant_id: UUID) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get ingredients that need reordering based on stock levels and usage
|
|
"""
|
|
try:
|
|
from sqlalchemy import text
|
|
query = text("""
|
|
WITH usage_analysis AS (
|
|
SELECT
|
|
i.id,
|
|
i.name,
|
|
COALESCE(SUM(s.current_quantity), 0) as current_stock,
|
|
i.reorder_point,
|
|
i.low_stock_threshold,
|
|
COALESCE(SUM(sm.quantity) FILTER (WHERE sm.movement_date > NOW() - INTERVAL '7 days'), 0) / 7 as daily_usage,
|
|
i.preferred_supplier_id,
|
|
i.standard_order_quantity
|
|
FROM ingredients i
|
|
LEFT JOIN stock s ON s.ingredient_id = i.id AND s.is_available = true
|
|
LEFT JOIN stock_movements sm ON sm.ingredient_id = i.id
|
|
AND sm.movement_type = 'PRODUCTION_USE'
|
|
AND sm.movement_date > NOW() - INTERVAL '7 days'
|
|
WHERE i.tenant_id = :tenant_id
|
|
AND i.is_active = true
|
|
GROUP BY i.id, i.name, i.reorder_point, i.low_stock_threshold,
|
|
i.preferred_supplier_id, i.standard_order_quantity
|
|
)
|
|
SELECT *,
|
|
CASE
|
|
WHEN daily_usage > 0 THEN FLOOR(current_stock / NULLIF(daily_usage, 0))
|
|
ELSE 999
|
|
END as days_of_stock,
|
|
GREATEST(
|
|
standard_order_quantity,
|
|
CEIL(daily_usage * 14)
|
|
) as recommended_order_quantity
|
|
FROM usage_analysis
|
|
WHERE current_stock <= reorder_point
|
|
ORDER BY days_of_stock ASC, current_stock ASC
|
|
LIMIT 50
|
|
""")
|
|
|
|
result = await self.session.execute(query, {"tenant_id": tenant_id})
|
|
return [dict(row._mapping) for row in result.fetchall()]
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get reorder recommendations", error=str(e), tenant_id=str(tenant_id))
|
|
raise |