shivace007/enrollmentAPI

1from pydantic import BaseModel #to have data checks in place #Library ensures that the data inputs are in place2from fastapi import FastAPI    #Python library to create APIs3from typing import Optional, List4from enum import Enum5import csv6import os7from pathlib import Path8from fastapi.middleware.cors import CORSMiddleware9 10class Role(str, Enum):11    SOFTWARE_ENGINEER = "Software Engineer"12    ML_ENGINEER = "ML Engineer"13    DATA_SCIENTIST = "Data Scientist"14    TECH_PM = "Technical Product Manager"15    OTHER = "Other"16 17class CustomerSegment(str, Enum):18    TIER_1 = "Tier 1: Core Fit"19    TIER_2 = "Tier 2: Potential Fit"20    TIER_3 = "Tier 3: Low Fit"21 22class Customers(BaseModel):23    id: int24    name: str25    age: int26    email: Optional[str] = None27    role: Role28    years_experience: int29    tools_familiarity: List[str]30    use_case: str31    technical_background_score: int = 032    work_experience_score: int = 033    use_case_clarity_score: int = 034    tool_familiarity_score: int = 035    engagement_willingness_score: int = 036    total_score: int = 037    segment: Optional[CustomerSegment] = None38 39# Initialize FastAPI app40app = FastAPI()41 42# Add CORS middleware43app.add_middleware(44    CORSMiddleware,45    allow_origins=["*"],  # Allows all origins46    allow_credentials=True,47    allow_methods=["*"],  # Allows all methods48    allow_headers=["*"],  # Allows all headers49)50 51# Get the base directory for file storage52BASE_DIR = Path(__file__).resolve().parent53CSV_FILE = BASE_DIR / "customers.csv"54 55# Ensure CSV file exists with headers56def init_csv():57    try:58        if not os.path.exists(CSV_FILE):59            with open(CSV_FILE, 'w', newline='') as f:60                writer = csv.writer(f)61                writer.writerow([62                    'id', 'name', 'age', 'email', 'role', 'years_experience',63                    'tools_familiarity', 'use_case', 'technical_background_score',64                    'work_experience_score', 'use_case_clarity_score',65                    'tool_familiarity_score', 'engagement_willingness_score',66                    'total_score', 'segment'67                ])68    except Exception as e:69        print(f"Error initializing CSV: {str(e)}")70        # Create a temporary in-memory storage if file operations fail71        return []72 73# Read all customers from CSV74def read_customers():75    try:76        if not os.path.exists(CSV_FILE):77            return []78            79        customers = []80        with open(CSV_FILE, 'r', newline='') as f:81            reader = csv.DictReader(f)82            for row in reader:83                try:84                    # Convert string representation of list back to list85                    row['tools_familiarity'] = eval(row['tools_familiarity'])86                    # Convert numeric fields87                    for field in ['id', 'age', 'years_experience', 'technical_background_score',88                                'work_experience_score', 'use_case_clarity_score',89                                'tool_familiarity_score', 'engagement_willingness_score',90                                'total_score']:91                        row[field] = int(row[field])92                    customers.append(row)93                except Exception as e:94                    print(f"Error processing row: {str(e)}")95                    continue96        return customers97    except Exception as e:98        print(f"Error reading CSV: {str(e)}")99        return []100 101# Write customers to CSV102def write_customers(customers):103    try:104        with open(CSV_FILE, 'w', newline='') as f:105            writer = csv.DictWriter(f, fieldnames=[106                'id', 'name', 'age', 'email', 'role', 'years_experience',107                'tools_familiarity', 'use_case', 'technical_background_score',108                'work_experience_score', 'use_case_clarity_score',109                'tool_familiarity_score', 'engagement_willingness_score',110                'total_score', 'segment'111            ])112            writer.writeheader()113            for customer in customers:114                try:115                    # Create a copy to avoid modifying the original116                    customer_copy = customer.copy()117                    # Convert list to string for CSV storage118                    customer_copy['tools_familiarity'] = str(customer_copy['tools_familiarity'])119                    writer.writerow(customer_copy)120                except Exception as e:121                    print(f"Error writing customer: {str(e)}")122                    continue123    except Exception as e:124        print(f"Error writing to CSV: {str(e)}")125 126def calculate_scores(customer: Customers) -> Customers:127    # Technical Background Score (0-3)128    if customer.role in [Role.SOFTWARE_ENGINEER, Role.ML_ENGINEER, Role.DATA_SCIENTIST]:129        customer.technical_background_score = 3130    elif customer.role == Role.TECH_PM:131        customer.technical_background_score = 1132    else:133        customer.technical_background_score = 0134 135    # Work Experience Score (0-2)136    if customer.years_experience >= 5:137        customer.work_experience_score = 2138    elif customer.years_experience >= 2:139        customer.work_experience_score = 1140    else:141        customer.work_experience_score = 0142 143    # Use Case Clarity Score (0-2)144    if len(customer.use_case.split()) > 50:  # Detailed use case145        customer.use_case_clarity_score = 2146    elif len(customer.use_case.split()) > 20:  # Basic use case147        customer.use_case_clarity_score = 1148    else:149        customer.use_case_clarity_score = 0150 151    # Tool Familiarity Score (0-2)152    advanced_tools = ["LangChain", "vector DB", "Python", "OpenAI API"]153    basic_tools = ["ChatGPT", "API"]154    155    advanced_count = sum(1 for tool in customer.tools_familiarity if tool in advanced_tools)156    basic_count = sum(1 for tool in customer.tools_familiarity if tool in basic_tools)157    158    if advanced_count >= 2:159        customer.tool_familiarity_score = 2160    elif advanced_count >= 1 or basic_count >= 2:161        customer.tool_familiarity_score = 1162    else:163        customer.tool_familiarity_score = 0164 165    # Calculate total score166    customer.total_score = (167        customer.technical_background_score +168        customer.work_experience_score +169        customer.use_case_clarity_score +170        customer.tool_familiarity_score +171        customer.engagement_willingness_score172    )173 174    # Assign segment based on total score175    if customer.total_score >= 8:176        customer.segment = CustomerSegment.TIER_1177    elif customer.total_score >= 5:178        customer.segment = CustomerSegment.TIER_2179    else:180        customer.segment = CustomerSegment.TIER_3181 182    return customer183 184# Initialize CSV file on startup185init_csv()186 187@app.get("/")188def read_root():189    return {"message": "Customer Management API is running"}190 191@app.get("/customers", response_model=list[Customers])192def get_customers():193    return read_customers()194 195@app.post("/new_customers", response_model=Customers)196def add_customers(customer: Customers):197    try:198        customer = calculate_scores(customer)199        customers = read_customers()200        customers.append(customer.dict())201        write_customers(customers)202        return customer203    except Exception as e:204        print(f"Error adding customer: {str(e)}")205        return {"error": str(e)}206 207@app.put("/update_customers/{cu_id}", response_model=Customers)208def upd_customers(cu_id: int, upd_customer: Customers):209    try:210        customers = read_customers()211        for i, customer in enumerate(customers):212            if customer['id'] == cu_id:213                upd_customer = calculate_scores(upd_customer)214                customers[i] = upd_customer.dict()215                write_customers(customers)216                return upd_customer217        return {"error": "Customer not found"}218    except Exception as e:219        print(f"Error updating customer: {str(e)}")220        return {"error": str(e)}221 222@app.delete("/delete_customers/{cu_id}", response_model=Customers)223def del_customers(cu_id: int):224    try:225        customers = read_customers()226        for i, customer in enumerate(customers):227            if customer['id'] == cu_id:228                deleted_customer = customers.pop(i)229                write_customers(customers)230                return deleted_customer231        return {"error": "Customer not found"}232    except Exception as e:233        print(f"Error deleting customer: {str(e)}")234        return {"error": str(e)}235 236@app.get("/customers/segment/{segment}", response_model=list[Customers])237def get_customers_by_segment(segment: CustomerSegment):238    try:239        customers = read_customers()240        return [customer for customer in customers if customer['segment'] == segment]241    except Exception as e:242        print(f"Error getting customers by segment: {str(e)}")243        return []244 245@app.get("/customers/{cu_id}/score", response_model=Customers)246def get_customer_score(cu_id: int):247    try:248        customers = read_customers()249        for customer in customers:250            if customer['id'] == cu_id:251                return customer252        return {"error": "Customer not found"}253    except Exception as e:254        print(f"Error getting customer score: {str(e)}")255        return {"error": str(e)}256 257 258