init

.gitattributes

@@ -0,0 +1 @@
+*.index filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -0,0 +1,11 @@
+__pycache__/
+env/
+venv/
+.venv/
+.idea/
+*.log
+*.egg-info/
+pip-wheel-EntityData/
+.env
+.DS_Store
+static/

Dockerfile

@@ -0,0 +1,17 @@
+FROM python:3.12.7
+
+RUN useradd -m -u 1000 user
+USER user
+ENV PATH="/home/user/.local/bin:$PATH"
+
+WORKDIR /app
+
+COPY --chown=user ./requirements.txt requirements.txt
+RUN pip install --no-cache-dir --upgrade -r requirements.txt
+
+USER root
+RUN apt-get update && apt-get install -y poppler-utils
+USER user
+
+COPY --chown=user . /app
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -0,0 +1,8 @@
+---
+title: Supplier Segmentation
+emoji: 💻
+colorFrom: yellow
+colorTo: blue
+sdk: docker
+pinned: false
+---

idreamers/__init__.py

@@ -0,0 +1,34 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from starlette.exceptions import HTTPException as StarletteHTTPException
+
+from idreamers.core.config import settings
+from idreamers.core.wrappers import PilotResponseWrapper, ErrorPilotResponse
+
+
+def create_app() -> FastAPI:
+    app = FastAPI()
+
+    from idreamers.api.calculation import calculation_router
+    app.include_router(calculation_router, tags=['calculation'])
+
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["*"],
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    @app.exception_handler(StarletteHTTPException)
+    async def http_exception_handler(_, exc):
+        return PilotResponseWrapper(
+            data=None,
+            successful=False,
+            error=ErrorPilotResponse(message=str(exc.detail))
+        ).response(exc.status_code)
+
+    @app.get("/")
+    async def read_root():
+        return {"calculation": "Hello world!"}
+
+    return app

idreamers/api/calculation/__init__.py

@@ -0,0 +1,7 @@
+from fastapi.routing import APIRouter
+
+calculation_router = APIRouter(
+    prefix="/api/calculation", tags=["calculation"]
+)
+
+from . import views

idreamers/api/calculation/openai_request.py

@@ -0,0 +1,27 @@
+from typing import Literal
+
+from idreamers.api.calculation.prompts import CalculationPrompts
+from idreamers.api.calculation.schemas import PilotRequest
+from idreamers.core.wrappers import openai_wrapper
+
+
+@openai_wrapper(is_json=True, return_='score')
+async def calculate_scores(data: PilotRequest, type_: Literal["business", "competitive", "supplier", "critical"]):
+    if type_ == "competitive":
+        prompt = CalculationPrompts.competitive
+    elif type_ == "supplier":
+        prompt = CalculationPrompts.supplier
+    elif type_ == "business":
+        prompt = CalculationPrompts.business
+    else:
+        prompt = CalculationPrompts.critical
+    messages = [
+        {
+            "role": "system",
+            "content": prompt
+            .replace("{supplier}", data.supplier)
+            .replace("{category}", data.category)
+            .replace("{buying_company}", data.buyingCompany)
+        }
+    ]
+    return messages

idreamers/api/calculation/prompts.py

@@ -0,0 +1,141 @@
+class CalculationPrompts:
+    competitive = """## Task
+You must answer on the question below and set a score. 
+
+Question:
+```
+How dependent is your business on this supplier for critical operations?
+```
+ 	 	
+Think about whether this supplier is essential for your day-to-day operations.	
+
+## Data
+
+**Supplier**:
+```
+{supplier}
+```
+
+**Category**:
+```
+{category}
+```
+
+**Buying Company**:
+```
+{buying_company}
+```
+
+## JSON Response format
+
+```json
+{
+  "score": integer
+}
+
+- [score]: Numerical score between 1 and 4. 1 = Many, 4 = Very few."""
+    supplier = """## Task
+You must answer on the question below and set a score. 
+
+Question:
+```
+How competitive is the market where this supplier operates?
+```
+ 	 	
+Evaluate whether this supplier faces competition or has significant market control.		
+
+## Data
+
+**Supplier**:
+```
+{supplier}
+```
+
+**Category**:
+```
+{category}
+```
+
+**Buying Company**:
+```
+{buying_company}
+```
+
+## JSON Response format
+
+```json
+{
+  "score": integer
+}
+
+- [score]: Numerical score between 1 and 4. 1 = Highly competitive, 4 = Monopoly/Oligopoly."""
+    business = """## Task
+You must answer on the question below and set a score. 
+
+Question:
+```
+How dependent is the supplier on your organization for their revenue?
+```
+ 	 	
+Reflect on how important your business is to this supplier’s overall revenue.		
+
+## Data
+
+**Supplier**:
+```
+{supplier}
+```
+
+**Category**:
+```
+{category}
+```
+
+**Buying Company**:
+```
+{buying_company}
+```
+
+## JSON Response format
+
+```json
+{
+  "score": integer
+}
+
+- [score]: Numerical score between 1 and 4. 1 = Not dependent, 4 = Highly dependent."""
+    critical = """## Task
+You must answer on the question below and set a score. 
+
+Question:
+```
+How critical is this supplier for delivering unique or differentiated products/services?
+```
+ 	 	
+Think about whether this supplier provides something that others cannot easily replicate.		
+
+## Data
+
+**Supplier**:
+```
+{supplier}
+```
+
+**Category**:
+```
+{category}
+```
+
+**Buying Company**:
+```
+{buying_company}
+```
+
+## JSON Response format
+
+```json
+{
+  "score": integer
+}
+
+- [score]: Numerical score between 1 and 4. Think about whether this supplier provides something that others cannot easily replicate."""

idreamers/api/calculation/schemas.py

@@ -0,0 +1,12 @@
+from pydantic import BaseModel
+
+
+class PilotRequest(BaseModel):
+    category: str
+    buyingCompany: str
+    supplier: str
+
+
+class PilotResponse(BaseModel):
+    x: float
+    y: float

idreamers/api/calculation/utils.py

@@ -0,0 +1,8 @@
+from idreamers.api.calculation.schemas import PilotResponse
+
+
+def calculate_pilot_axes(q1: int, q2: int, q3: int, q4: int) -> PilotResponse:
+    return PilotResponse(
+        x=round((q1 + q2) / 2, 2),
+        y=round((q3 + q4) / 2, 2),
+    )

idreamers/api/calculation/views.py

@@ -0,0 +1,21 @@
+import asyncio
+
+from idreamers.api.calculation import calculation_router
+from idreamers.api.calculation.openai_request import calculate_scores
+from idreamers.api.calculation.schemas import PilotRequest, PilotResponse
+from idreamers.api.calculation.utils import calculate_pilot_axes
+from idreamers.core.wrappers import PilotResponseWrapper
+
+
+@calculation_router.post('/pilot/calculate')
+async def calculate_pilot(
+        data: PilotRequest
+) -> PilotResponseWrapper[PilotResponse]:
+    depend_business, competitive, depend_supplier, critical_supplier = await asyncio.gather(
+        calculate_scores(data, 'business'),
+        calculate_scores(data, 'competitive'),
+        calculate_scores(data, 'supplier'),
+        calculate_scores(data, 'critical'),
+    )
+    response = calculate_pilot_axes(depend_business, competitive, depend_supplier, critical_supplier)
+    return PilotResponseWrapper(data=response)

idreamers/core/config.py

@@ -0,0 +1,36 @@
+import os
+import pathlib
+from functools import lru_cache
+
+from dotenv import load_dotenv
+from openai import AsyncClient
+
+load_dotenv()
+
+
+class BaseConfig:
+    BASE_DIR: pathlib.Path = pathlib.Path(__file__).parent.parent.parent
+    SECRET_KEY = os.getenv('SECRET')
+    OPENAI_CLIENT = AsyncClient(api_key=os.getenv('OPENAI_API_KEY'))
+
+
+class DevelopmentConfig(BaseConfig):
+    pass
+
+
+class ProductionConfig(BaseConfig):
+    pass
+
+
+@lru_cache()
+def get_settings() -> DevelopmentConfig | ProductionConfig:
+    config_cls_dict = {
+        'development': DevelopmentConfig,
+        'production': ProductionConfig,
+    }
+    config_name = os.getenv('FASTAPI_CONFIG', default='development')
+    config_cls = config_cls_dict[config_name]
+    return config_cls()
+
+
+settings = get_settings()

idreamers/core/wrappers.py

@@ -0,0 +1,57 @@
+import json
+from functools import wraps
+from typing import Generic, Optional, TypeVar
+
+import pydash
+from pydantic import BaseModel
+from starlette.responses import JSONResponse
+
+from idreamers.core.config import settings
+
+T = TypeVar('T')
+
+
+class ErrorPilotResponse(BaseModel):
+    message: str
+
+
+class PilotResponseWrapper(BaseModel, Generic[T]):
+    data: Optional[T] = None
+    successful: bool = True
+    error: Optional[ErrorPilotResponse] = None
+
+    def response(self, status_code: int):
+        return JSONResponse(
+            status_code=status_code,
+            content={
+                "data": self.data,
+                "successful": self.successful,
+                "error": self.error.model_dump(mode='json') if self.error else None
+            }
+        )
+
+
+def openai_wrapper(
+        temperature: int | float = 0, model: str = "gpt-4o-mini", is_json: bool = False, return_: str = None
+):
+    def decorator(func):
+        @wraps(func)
+        async def wrapper(*args, **kwargs) -> str:
+            messages = await func(*args, **kwargs)
+            completion = await settings.OPENAI_CLIENT.chat.completions.create(
+                messages=messages,
+                temperature=temperature,
+                n=1,
+                model=model,
+                response_format={"type": "json_object"} if is_json else {"type": "text"}
+            )
+            response = completion.choices[0].message.content
+            if is_json:
+                response = json.loads(response)
+                if return_:
+                    return pydash.get(response, return_)
+            return response
+
+        return wrapper
+
+    return decorator

main.py

@@ -0,0 +1,3 @@
+from idreamers import create_app
+
+app = create_app()

requirements.txt

@@ -0,0 +1,25 @@
+annotated-types==0.7.0
+anyio==4.8.0
+certifi==2024.12.14
+click==8.1.8
+distro==1.9.0
+fastapi==0.115.7
+h11==0.14.0
+httpcore==1.0.7
+httptools==0.6.4
+httpx==0.28.1
+idna==3.10
+jiter==0.8.2
+openai==1.60.1
+pydantic==2.10.6
+pydantic_core==2.27.2
+python-dotenv==1.0.1
+PyYAML==6.0.2
+sniffio==1.3.1
+starlette==0.45.3
+tqdm==4.67.1
+typing_extensions==4.12.2
+uvicorn==0.34.0
+uvloop==0.21.0
+watchfiles==1.0.4
+websockets==14.2