feat:add price predict feature

Database/models.py

@@ -38,6 +38,7 @@ class Item(BaseEntity):
     item_category = Column(Enum(ItemCategory), nullable=False)
     base_consume_expectation = Column(Integer, nullable=False)
     base_price = Column(Integer, nullable=False)
+    base_count = Column(Integer, nullable=False)
     embedding = Column(JSON, nullable=True)
 
 

Router/item.py

@@ -9,6 +9,7 @@
 from fastapi import Query, Request
 from Data.item import Item, ItemAdd, UserItemAdd, UserItemConsume, ItemRead
 from Database.models import User
+from Service.item_service import ItemService
 from Service.purchase_service import PurchaseService
 from Service.useritem_service import UserItemService
 router = APIRouter(tags=["items"], prefix="/items")
@@ -52,6 +53,32 @@ def consume_item(request : Request, user_id: str, user_item_consume: UserItemCon
 @router.post("/addone")
 def add_item(request : Request, userItemAdd: UserItemAdd):
     UserItemService.add_userItem(userItemAdd)
-    purchaseHistory = PurchaseService.get_purchase_history(userItemAdd)
+    purchaseHistory = PurchaseService.purchase_history_db(userItemAdd)
     PurchaseService.purchase_history_save(purchaseHistory)
-    return JSONResponse(status_code=HTTP_200_OK, content={"message": "Item added successfully"})
+    return JSONResponse(status_code=HTTP_200_OK, content={"message": "Item added successfully"})
+
+
+@router.get("/{user_id}/price")
+def get_expected_price(request : Request, user_id: str, date : str = Query(...)):
+    try:
+        year, month = map(int, date.split("-"))
+
+        user_item_list = UserItemService.get_all_userItem_filtered_by_date(user_id, year, month)
+
+        price = 0
+        for useritem in user_item_list:
+            purchases = PurchaseService.get_purchase_histories_by_item_id(useritem.item_id)
+            if len(purchases) == 0:
+                item = ItemService.get_item(useritem.item_name)
+                price += item.base_price * item.base_count
+            else:
+                avg_count = round(sum([purchase.count for purchase in purchases]) / len(purchases))
+                avg_price = sum([purchase.price for purchase in purchases]) / len(purchases)
+                price += avg_count * avg_price
+
+        return JSONResponse(status_code=HTTP_200_OK, content={"price": price})
+
+    except ValueError:
+        return JSONResponse(status_code=HTTP_400_BAD_REQUEST, content={"message": "Invalid date format"})
+    except Exception as e:
+        return JSONResponse(status_code=HTTP_500_INTERNAL_SERVER_ERROR, content={"message": str(e)})

Service/purchase_service.py

@@ -18,12 +18,17 @@ def purchase_history_list_save(data_list: List[PurchaseHistory]):
             db.commit()
             return True
 
-    def get_purchase_history(data : UserItemAdd):
+    def purchase_history_db(data : UserItemAdd):
         with get_db() as db:
             item_id = db.query(UserItem).filter(UserItem.user_id == data.user_id, UserItem.item_name == data.item_name).first().item_id
             return PurchaseHistory(user_id=data.user_id, item_id=item_id, price=data.price, count=1, date=data.purchase_date)
 
     def purchase_history_save(data : PurchaseHistory):
         with get_db() as db:
             db.add(data)
-            db.commit()
+            db.commit()
+
+
+    def get_purchase_histories_by_item_id(item_id: str):
+        with get_db() as db:
+            return db.query(PurchaseHistory).filter(PurchaseHistory.item_id == item_id).all()

Service/useritem_service.py

@@ -3,6 +3,7 @@
 from typing import Optional, List, Annotated
 
 from sqlalchemy.orm import Session
+from sqlalchemy.sql.expression import extract
 
 from Data.item import UserItemAdd, ItemAdd
 from Database.database import get_db
@@ -29,6 +30,12 @@ def get_all_userItem(user_id: str):
             return db.query(UserItem).filter(UserItem.user_id == user_id).all()
 
 
+    def get_all_userItem_filtered_by_date(user_id, year: int, month: int):
+        with get_db() as db:
+            return db.query(UserItem).filter(UserItem.user_id == user_id, extract("year", UserItem.consume_date) == year,
+                    extract("month", UserItem.consume_date) == month).all()
+
+
     def to_userItem_dict(userItemList: List[UserItem]):
         itemlist = []
         with get_db() as db: