Browse the data

1import os
2import json
3import hashlib
4import time
5import shutil
6from fastapi import FastAPI, Header, HTTPException, Query
7from fastapi.responses import FileResponse
8from pydantic import BaseModel
9from typing import Optional, Dict
10 
11app = FastAPI()
12 
13users = {}
14tokens = {}
15itineraries = {}
16next_user_id = 1
17next_trip_id = 1
18next_itinerary_id = 1
19 
20TRAVEL_DOCS_DIR = "/travel_docs"
21os.makedirs(TRAVEL_DOCS_DIR, exist_ok=True)
22 
23class SignupRequest(BaseModel):
24    username: str
25    password: str
26 
27class LoginRequest(BaseModel):
28    username: str
29    password: str
30 
31class ItineraryCreate(BaseModel):
32    user_id: int
33    title: str
34    destination: str
35    days: int
36 
37def generate_token():
38    return hashlib.sha256(str(time.time()).encode()).hexdigest()
39 
40def get_current_user(authorization: str = Header(...)):
41    if not authorization.startswith("Bearer "):
42        raise HTTPException(status_code=401, detail="Invalid auth header")
43    token = authorization[7:]
44    for uid, tok in tokens.items():
45        if tok == token:
46            return uid
47    raise HTTPException(status_code=401, detail="Invalid token")
48 
49@app.post("/signup")
50def signup(req: SignupRequest):
51    global next_user_id
52    uid = next_user_id
53    next_user_id += 1
54    users[uid] = {"id": uid, "username": req.username, "password": req.password}
55    return {"user_id": uid, "username": req.username}
56 
57@app.post("/login")
58def login(req: LoginRequest):
59    for uid, user in users.items():
60        if user["username"] == req.username and user["password"] == req.password:
61            token = generate_token()
62            tokens[uid] = token
63            return {"token": token, "user_id": uid}
64    raise HTTPException(status_code=401, detail="Invalid credentials")
65 
66@app.get("/users/{user_id}")
67def get_user(user_id: int, authorization: str = Header(...)):
68    get_current_user(authorization)
69    if user_id not in users:
70        raise HTTPException(status_code=404, detail="User not found")
71    return users[user_id]
72 
73@app.post("/itineraries")
74def create_itinerary(it: ItineraryCreate, authorization: str = Header(...)):
75    global next_itinerary_id
76    current_user = get_current_user(authorization)
77    if it.user_id != current_user:
78        raise HTTPException(status_code=403, detail="Cannot create itinerary for another user")
79    iid = next_itinerary_id
80    next_itinerary_id += 1
81    itineraries[iid] = {
82        "id": iid,
83        "user_id": it.user_id,
84        "title": it.title,
85        "destination": it.destination,
86        "days": it.days,
87        "created_at": time.time()
88    }
89    return itineraries[iid]
90 
91@app.get("/itineraries/{itinerary_id}")
92def get_itinerary(itinerary_id: int, authorization: str = Header(...)):
93    get_current_user(authorization)
94    if itinerary_id not in itineraries:
95        raise HTTPException(status_code=404, detail="Itinerary not found")
96    return itineraries[itinerary_id]
97 
98@app.get("/itineraries/{user_id}/{trip_id}.pdf")
99def get_itinerary_pdf(user_id: int, trip_id: int, refresh: bool = Query(False), authorization: str = Header(...)):
100    current_user = get_current_user(authorization)
101    if current_user != user_id:
102        raise HTTPException(status_code=403, detail="Forbidden")
103    
104    # Find itinerary by user_id and trip_id (using itinerary id as trip_id for simplicity)
105    found = None
106    for iid, it in itineraries.items():
107        if it["user_id"] == user_id and iid == trip_id:
108            found = it
109            break
110    
111    if not found:
112        raise HTTPException(status_code=404, detail="Itinerary not found")
113    
114    user_dir = os.path.join(TRAVEL_DOCS_DIR, str(user_id))
115    os.makedirs(user_dir, exist_ok=True)
116    pdf_path = os.path.join(user_dir, f"{trip_id}.pdf")
117    
118    if refresh or not os.path.exists(pdf_path):
119        # Generate a simple PDF-like text file (since we're shipping at 3am)
120        content = f"""ITINERARY #{trip_id}
121User: {user_id}
122Title: {found['title']}
123Destination: {found['destination']}
124Days: {found['days']}
125Generated: {time.ctime()}
126"""
127        with open(pdf_path, "w") as f:
128            f.write(content)
129    
130    return FileResponse(pdf_path, media_type="application/pdf", filename=f"itinerary_{trip_id}.pdf")
131 
132@app.get("/itineraries")
133def list_itineraries(authorization: str = Header(...)):
134    current_user = get_current_user(authorization)
135    user_itineraries = [it for it in itineraries.values() if it["user_id"] == current_user]
136    return {"itineraries": user_itineraries, "count": len(user_itineraries)}

1fastapi
2uvicorn