from fastapi import FastAPI, Depends, HTTPException, APIRouter from fastapi.staticfiles import StaticFiles from fastapi.middleware.cors import CORSMiddleware from sqlalchemy.orm import Session import crud, models, schemas, security from database import SessionLocal, engine from typing import Optional, List from datetime import timedelta from security import create_access_token, ACCESS_TOKEN_EXPIRE_MINUTES, oauth2_scheme, SECRET_KEY, ALGORITHM from jose import JWTError, jwt models.Base.metadata.create_all(bind=engine) app = FastAPI() app.mount("/static", StaticFiles(directory="static"), name="static") # CORS Middleware origins = [ "http://localhost:5173", "http://12-7.0.0.1:5173", "http://localhost:5174", "http://127.0.0.1:5174", ] app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Dependency def get_db(): db = SessionLocal() try: yield db finally: db.close() async def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)): credentials_exception = HTTPException( status_code=401, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) subject = payload.get("sub") if subject is None: raise credentials_exception # Try to convert sub to int for user_id, if it fails, assume it's a phone number try: user_id = int(subject) user = db.query(models.User).filter(models.User.id == user_id).first() except ValueError: user = crud.get_user_by_phone_number(db, phone_number=subject) except JWTError: raise credentials_exception if user is None: raise credentials_exception return user async def get_current_user_optional(token: Optional[str] = Depends(oauth2_scheme), db: Session = Depends(get_db)): if token is None: return None try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) subject = payload.get("sub") if subject is None: return None try: user_id = int(subject) user = db.query(models.User).filter(models.User.id == user_id).first() except ValueError: user = crud.get_user_by_phone_number(db, phone_number=subject) return user except (JWTError, AttributeError): return None # --- Auth Router --- auth_router = APIRouter( prefix="/api/v1/auth", tags=["auth"], ) @auth_router.post("/send-verification-code") def send_verification_code(user: schemas.UserCreate): if not user.phone_number or len(user.phone_number) != 11: raise HTTPException(status_code=400, detail="Valid 11-digit phone number is required.") print(f"Simulating sending verification code to {user.phone_number}") return {"message": "Verification code sent successfully.", "code": "111111"} @auth_router.post("/login", response_model=schemas.Token) def login(form_data: schemas.UserCodeLogin, db: Session = Depends(get_db)): if form_data.code != "111111": raise HTTPException(status_code=400, detail="Invalid verification code.") user = crud.get_user_by_phone_number(db, phone_number=form_data.phone_number) is_new_user = False if not user: user = crud.create_user(db=db, user=schemas.UserCreate(phone_number=form_data.phone_number)) is_new_user = True access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( data={"sub": user.phone_number}, expires_delta=access_token_expires ) return {"access_token": access_token, "token_type": "bearer", "is_new_user": is_new_user} @auth_router.post("/login/password", response_model=schemas.Token) def login_with_password(form_data: schemas.UserPasswordLogin, db: Session = Depends(get_db)): user = crud.authenticate_user(db, phone_number=form_data.phone_number, password=form_data.password) if not user: raise HTTPException( status_code=401, detail="Incorrect phone number or password", headers={"WWW-Authenticate": "Bearer"}, ) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( data={"sub": user.phone_number}, expires_delta=access_token_expires ) return {"access_token": access_token, "token_type": "bearer", "is_new_user": False} @auth_router.put("/users/me/password", response_model=schemas.User) async def set_password_for_current_user( password_set: schemas.PasswordSet, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): return crud.set_user_password(db=db, user=current_user, password_set=password_set) @auth_router.post("/wechat/callback", response_model=schemas.Token) def wechat_login_callback(body: dict, db: Session = Depends(get_db)): code = body.get("code") if not code: raise HTTPException(status_code=400, detail="Code is required") user, is_new_user = security.authenticate_wechat_user(db, code=code) if not user: raise HTTPException( status_code=401, detail="Could not validate WeChat credentials", ) # For OAuth users, we might not have a phone number initially # The subject of the token should be something unique and stable. # Using user.id is a good choice. subject = str(user.id) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( data={"sub": subject}, expires_delta=access_token_expires ) return {"access_token": access_token, "token_type": "bearer", "is_new_user": is_new_user} app.include_router(auth_router) # --- Family & Health Profile Router --- family_router = APIRouter( prefix="/api/v1/family", tags=["family"], dependencies=[Depends(get_current_user)] ) @family_router.post("/members", response_model=schemas.FamilyMember) def create_family_member( member: schemas.FamilyMemberCreate, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): return crud.create_family_member(db=db, member=member, owner_id=current_user.id) @family_router.get("/members", response_model=List[schemas.FamilyMember]) def get_family_members( db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): return crud.get_family_members_by_user(db=db, user_id=current_user.id) @family_router.put("/members/{member_id}/health-profile", response_model=List[schemas.HealthProfile]) def update_member_health_profile( member_id: int, profiles: List[schemas.HealthProfileCreate], db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): db_member = crud.get_family_member(db, member_id=member_id, user_id=current_user.id) if db_member is None: raise HTTPException(status_code=404, detail="Family member not found") return crud.update_health_profile(db=db, member_id=member_id, profiles=profiles) app.include_router(family_router) # --- Food Router --- food_router = APIRouter( prefix="/api/v1/foods", tags=["foods"] ) @food_router.get("/{barcode}/details", response_model=schemas.FoodDetails) def read_food_details( barcode: str, db: Session = Depends(get_db), current_user: Optional[models.User] = Depends(get_current_user_optional) ): db_food = crud.get_food_by_barcode(db, barcode=barcode) if db_food is None: raise HTTPException(status_code=404, detail="Food not found") return db_food @food_router.post("/submit", response_model=schemas.SubmittedFood, dependencies=[Depends(get_current_user)]) def submit_new_food( food: schemas.SubmittedFoodCreate, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): return crud.create_submitted_food(db=db, food=food, user_id=current_user.id) app.include_router(food_router) # --- Content Center Router --- content_router = APIRouter( prefix="/api/v1/articles", tags=["articles"] ) @content_router.post("/", response_model=schemas.Article) def create_article(article: schemas.ArticleCreate, db: Session = Depends(get_db)): return crud.create_article(db=db, article=article) @content_router.get("/", response_model=List[schemas.Article]) def read_articles_by_category(category: str, skip: int = 0, limit: int = 10, db: Session = Depends(get_db)): return crud.get_articles_by_category(db=db, category=category, skip=skip, limit=limit) @content_router.get("/{article_id}", response_model=schemas.Article) def read_article(article_id: int, db: Session = Depends(get_db)): db_article = crud.get_article(db, article_id=article_id) if db_article is None: raise HTTPException(status_code=404, detail="Article not found") return db_article app.include_router(content_router) # --- Recipe Router --- recipe_router = APIRouter( prefix="/api/v1/recipes", tags=["recipes"] ) @recipe_router.post("/", response_model=schemas.Recipe, dependencies=[Depends(get_current_user)]) def create_recipe( recipe: schemas.RecipeCreate, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): return crud.create_recipe(db=db, recipe=recipe, author_id=current_user.id) @recipe_router.get("/{recipe_id}", response_model=schemas.Recipe) def get_recipe(recipe_id: int, db: Session = Depends(get_db)): db_recipe = crud.get_recipe_details(db, recipe_id=recipe_id) if db_recipe is None: raise HTTPException(status_code=404, detail="Recipe not found") return db_recipe @recipe_router.post("/suggest", response_model=List[schemas.Recipe], dependencies=[Depends(get_current_user)]) def suggest_recipes( db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): return db.query(models.Recipe).order_by(models.Recipe.id.desc()).limit(5).all() @recipe_router.post("/suggest-by-ingredients", response_model=List[schemas.Recipe]) def suggest_recipes_by_ingredients_api( ingredient_ids: List[int], db: Session = Depends(get_db) ): return crud.suggest_recipes_by_ingredients(db=db, ingredient_ids=ingredient_ids) app.include_router(recipe_router) # --- Community Router --- community_router = APIRouter( prefix="/api/v1", tags=["community"] ) @community_router.post("/posts", response_model=schemas.Post, dependencies=[Depends(get_current_user)]) def create_post( post: schemas.PostCreate, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): return crud.create_post(db=db, post=post, author_id=current_user.id) @community_router.get("/foods/{food_id}/posts", response_model=List[schemas.Post]) def get_posts_for_food(food_id: int, skip: int = 0, limit: int = 10, db: Session = Depends(get_db)): return crud.get_posts_by_food(db=db, food_id=food_id, skip=skip, limit=limit) @community_router.post("/posts/{post_id}/comments", response_model=schemas.Comment, dependencies=[Depends(get_current_user)]) def create_comment_for_post( post_id: int, comment: schemas.CommentBase, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): comment_create = schemas.CommentCreate(content=comment.content, post_id=post_id) return crud.create_comment(db=db, comment=comment_create, author_id=current_user.id) @community_router.get("/topics", response_model=List[schemas.Topic]) def get_topics(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): return crud.get_topics(db=db, skip=skip, limit=limit) @community_router.get("/topics/{topic_id}/posts", response_model=List[schemas.Post]) def get_posts_by_topic(topic_id: int, skip: int = 0, limit: int = 10, db: Session = Depends(get_db)): return crud.get_posts_by_topic(db=db, topic_id=topic_id, skip=skip, limit=limit) @community_router.get("/topics", response_model=List[schemas.Topic]) def get_topics(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): return crud.get_topics(db=db, skip=skip, limit=limit) @community_router.get("/topics/{topic_id}/posts", response_model=List[schemas.Post]) def get_posts_by_topic(topic_id: int, skip: int = 0, limit: int = 10, db: Session = Depends(get_db)): return crud.get_posts_by_topic(db=db, topic_id=topic_id, skip=skip, limit=limit) @community_router.post("/posts/{post_id}/like", dependencies=[Depends(get_current_user)]) def like_post( post_id: int, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): return crud.toggle_like(db=db, post_id=post_id, user_id=current_user.id) app.include_router(community_router) # --- E-commerce Router --- mall_router = APIRouter( prefix="/api/v1/mall", tags=["mall"] ) @mall_router.get("/products", response_model=List[schemas.Product]) def get_products(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)): return crud.get_products(db=db, skip=skip, limit=limit) @mall_router.get("/cart", response_model=List[schemas.CartItem], dependencies=[Depends(get_current_user)]) def get_cart(db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user)): return crud.get_cart_items(db=db, user_id=current_user.id) @mall_router.post("/cart/items", response_model=schemas.CartItem, dependencies=[Depends(get_current_user)]) def add_item_to_cart(item: schemas.CartItemCreate, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user)): return crud.add_to_cart(db=db, item=item, user_id=current_user.id) @mall_router.post("/orders", response_model=schemas.Order, dependencies=[Depends(get_current_user)]) def create_order(db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user)): return crud.create_order(db=db, user_id=current_user.id) app.include_router(mall_router) # --- Other Routers --- @app.get("/") def read_root(): return {"message": "Welcome to 食话食说 API"} @app.get("/api/v1/users/check-phone-existence/") def check_phone_existence(phone_number: str, db: Session = Depends(get_db)): user = crud.get_user_by_phone_number(db, phone_number=phone_number) return {"exists": user is not None} @app.post("/api/v1/food/", response_model=schemas.Food) def create_food_entry(food: schemas.FoodCreate, db: Session = Depends(get_db)): db_food = crud.get_food_by_barcode(db, barcode=food.barcode) if db_food: raise HTTPException(status_code=400, detail="Barcode already registered") return crud.create_food(db=db, food=food) @app.get("/api/v1/food/{barcode}", response_model=schemas.Food) def read_food_by_barcode(barcode: str, db: Session = Depends(get_db)): db_food = crud.get_food_by_barcode(db, barcode=barcode) if db_food is None: raise HTTPException(status_code=404, detail="Food not found") return db_food @app.post("/api/v1/users/me/preferences", status_code=204) async def update_user_preferences( preferences: schemas.OnboardingPreferences, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db) ): crud.create_user_preferences(db=db, user_id=current_user.id, preferences=preferences) return @app.post("/api/v1/search/history", response_model=schemas.SearchHistory) async def create_search_history_entry( search: schemas.SearchHistoryCreate, current_user: Optional[models.User] = Depends(get_current_user_optional), db: Session = Depends(get_db) ): user_id = current_user.id if current_user else None return crud.create_search_history(db=db, search=search, user_id=user_id) @app.get("/api/v1/search/history", response_model=List[schemas.SearchHistory]) async def get_user_search_history( current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db), skip: int = 0, limit: int = 10 ): return crud.get_search_history_by_user(db=db, user_id=current_user.id, skip=skip, limit=limit) @app.get("/api/v1/search/popular") async def get_popular_searches(db: Session = Depends(get_db), limit: int = 10): return crud.get_popular_searches(db=db, limit=limit)