Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from typing import Optional, List
3import random
4import hashlib
5import secrets
6 
7app = FastAPI()
8 
9users = {}
10tokens = {}
11products = {}
12purchases = []
13co_purchase_matrix = {}
14user_purchase_history = {}
15next_user_id = 1
16next_product_id = 1
17next_purchase_id = 1
18 
19def get_current_user(authorization: str = Header(...)):
20    if not authorization.startswith("Bearer "):
21        raise HTTPException(status_code=401, detail="Invalid auth header")
22    token = authorization[7:]
23    if token not in tokens:
24        raise HTTPException(status_code=401, detail="Invalid token")
25    return tokens[token]
26 
27@app.post("/signup")
28def signup(username: str, password: str):
29    global next_user_id
30    for u in users.values():
31        if u["username"] == username:
32            raise HTTPException(status_code=400, detail="User exists")
33    user_id = next_user_id
34    next_user_id += 1
35    users[user_id] = {"id": user_id, "username": username, "password": password}
36    return {"user_id": user_id}
37 
38@app.post("/login")
39def login(username: str, password: str):
40    for u in users.values():
41        if u["username"] == username and u["password"] == password:
42            token = secrets.token_hex(16)
43            tokens[token] = u["id"]
44            return {"token": token}
45    raise HTTPException(status_code=401, detail="Invalid credentials")
46 
47@app.get("/products/{product_id}")
48def get_product(product_id: int):
49    if product_id not in products:
50        raise HTTPException(status_code=404, detail="Product not found")
51    return products[product_id]
52 
53@app.post("/products")
54def create_product(name: str, price: float):
55    global next_product_id
56    product_id = next_product_id
57    next_product_id += 1
58    products[product_id] = {"id": product_id, "name": name, "price": price}
59    return products[product_id]
60 
61@app.get("/products/{product_id}/related")
62def get_related_products(product_id: int):
63    if product_id not in products:
64        raise HTTPException(status_code=404, detail="Product not found")
65    related = co_purchase_matrix.get(product_id, [])
66    related_scores = [(pid, score) for pid, score in related if pid != product_id]
67    related_scores.sort(key=lambda x: -x[1])
68    top5 = [pid for pid, _ in related_scores[:5]]
69    result = []
70    for pid in top5:
71        if pid in products:
72            result.append(products[pid])
73    return result
74 
75@app.post("/purchase")
76def log_purchase(product_ids: List[int], user_id: int = Header(None)):
77    global next_purchase_id
78    purchase_id = next_purchase_id
79    next_purchase_id += 1
80    purchase = {"id": purchase_id, "user_id": user_id, "product_ids": product_ids}
81    purchases.append(purchase)
82    
83    for pid in product_ids:
84        if pid not in products:
85            continue
86        for pid2 in product_ids:
87            if pid == pid2:
88                continue
89            if pid not in co_purchase_matrix:
90                co_purchase_matrix[pid] = {}
91            if pid2 not in co_purchase_matrix[pid]:
92                co_purchase_matrix[pid][pid2] = 0
93            co_purchase_matrix[pid][pid2] += 1
94    
95    if user_id:
96        if user_id not in user_purchase_history:
97            user_purchase_history[user_id] = set()
98        for pid in product_ids:
99            user_purchase_history[user_id].add(pid)
100    
101    return {"purchase_id": purchase_id}
102 
103@app.get("/user/{user_id}/recommendations")
104def get_user_recommendations(user_id: int, authorization: str = Header(None)):
105    if user_id not in users:
106        raise HTTPException(status_code=404, detail="User not found")
107    
108    history = user_purchase_history.get(user_id, set())
109    candidate_scores = {}
110    
111    for pid in history:
112        related = co_purchase_matrix.get(pid, {})
113        for related_pid, score in related.items():
114            if related_pid not in history:
115                if related_pid not in candidate_scores:
116                    candidate_scores[related_pid] = 0
117                candidate_scores[related_pid] += score
118    
119    sorted_candidates = sorted(candidate_scores.items(), key=lambda x: -x[1])
120    top5 = [pid for pid, _ in sorted_candidates[:5]]
121    result = []
122    for pid in top5:
123        if pid in products:
124            result.append(products[pid])
125    return result

1fastapi
2uvicorn