""" app.py LeafDoctor AI FastAPI Application """ from pathlib import Path import tempfile from typing import Optional from fastapi import FastAPI, File, Form, HTTPException, UploadFile from fastapi.responses import JSONResponse from config import settings from database import SessionLocal, create_tables from service import DiagnosisService app = FastAPI( title=settings.APP_NAME, version=settings.APP_VERSION, ) @app.on_event("startup") def startup(): create_tables() settings.upload_path.mkdir( parents=True, exist_ok=True, ) settings.data_path.mkdir( parents=True, exist_ok=True, ) print("=" * 60) print("LeafDoctor AI Started") print("=" * 60) @app.get("/") def root(): return { "application": settings.APP_NAME, "version": settings.APP_VERSION, "status": "running", } @app.get("/health") def health(): db = SessionLocal() try: service = DiagnosisService(db) return service.health() finally: db.close() @app.get("/statistics") def statistics(): db = SessionLocal() try: service = DiagnosisService(db) return service.statistics() finally: db.close() @app.get("/history") def history(limit: int = 50): db = SessionLocal() try: service = DiagnosisService(db) histories = service.get_histories(limit) result = [] for item in histories: result.append( { "id": item.id, "user_id": getattr(item, "user_id", None), "user_ph": getattr(item, "user_ph", None), "image": item.image_path, "crop": item.crop_detected, "label": item.predicted_label, "class_id": item.class_id, "confidence": item.confidence, "created_at": item.created_at, } ) return result finally: db.close() @app.get("/history/{history_id}") def history_details(history_id: int): db = SessionLocal() try: service = DiagnosisService(db) history = service.get_history(history_id) if history is None: raise HTTPException( status_code=404, detail="History not found", ) return history finally: db.close() @app.delete("/history/{history_id}") def delete_history(history_id: int): db = SessionLocal() try: service = DiagnosisService(db) deleted = service.delete_history( history_id ) if not deleted: raise HTTPException( status_code=404, detail="History not found", ) return { "success": True } finally: db.close() @app.post("/diagnose") async def diagnose( image: UploadFile = File(...), user_id: Optional[str] = Form(None), user_ph: Optional[str] = Form(None), ): suffix = Path(image.filename).suffix.lower() if suffix not in settings.ALLOWED_IMAGE_EXTENSIONS: raise HTTPException( status_code=400, detail="Unsupported image type", ) with tempfile.NamedTemporaryFile( delete=False, suffix=suffix, ) as temp: temp.write(await image.read()) temp_path = Path(temp.name) db = SessionLocal() try: service = DiagnosisService(db) result = service.diagnose( image_path=temp_path, user_id=user_id, user_ph=user_ph, ) return JSONResponse(result) except Exception as ex: raise HTTPException( status_code=500, detail=str(ex), ) finally: db.close() try: temp_path.unlink() except Exception: pass @app.get("/history/user") def user_history( user_id: Optional[str] = None, user_ph: Optional[str] = None, limit: int = 50, ): if not user_id and not user_ph: raise HTTPException( status_code=400, detail="Either 'user_id' or 'user_ph' query parameter must be provided.", ) db = SessionLocal() try: service = DiagnosisService(db) histories = service.get_user_histories( user_id=user_id, user_ph=user_ph, limit=limit, ) result = [] for item in histories: result.append( { "id": item.id, "user_id": getattr(item, "user_id", None), "user_ph": getattr(item, "user_ph", None), "image": item.image_path, "crop": item.crop_detected, "label": item.predicted_label, "class_id": item.class_id, "confidence": item.confidence, "created_at": item.created_at, } ) return result finally: db.close() if __name__ == "__main__": import uvicorn uvicorn.run( "app:app", host="0.0.0.0", port=8000, reload=True, ) # """ # app.py # LeafDoctor AI # FastAPI Application # """ # from pathlib import Path # import tempfile # from fastapi import FastAPI, File, HTTPException, UploadFile # from fastapi.responses import JSONResponse # from config import settings # from database import SessionLocal, create_tables # from service import DiagnosisService # app = FastAPI( # title=settings.APP_NAME, # version=settings.APP_VERSION, # ) # @app.on_event("startup") # def startup(): # create_tables() # settings.upload_path.mkdir( # parents=True, # exist_ok=True, # ) # settings.data_path.mkdir( # parents=True, # exist_ok=True, # ) # print("=" * 60) # print("LeafDoctor AI Started") # print("=" * 60) # @app.get("/") # def root(): # return { # "application": settings.APP_NAME, # "version": settings.APP_VERSION, # "status": "running", # } # @app.get("/health") # def health(): # db = SessionLocal() # try: # service = DiagnosisService(db) # return service.health() # finally: # db.close() # @app.get("/statistics") # def statistics(): # db = SessionLocal() # try: # service = DiagnosisService(db) # return service.statistics() # finally: # db.close() # @app.get("/history") # def history(limit: int = 50): # db = SessionLocal() # try: # service = DiagnosisService(db) # histories = service.get_histories(limit) # result = [] # for item in histories: # result.append( # { # "id": item.id, # "image": item.image_path, # "crop": item.crop_detected, # "label": item.predicted_label, # "class_id": item.class_id, # "confidence": item.confidence, # "created_at": item.created_at, # } # ) # return result # finally: # db.close() # @app.get("/history/{history_id}") # def history_details(history_id: int): # db = SessionLocal() # try: # service = DiagnosisService(db) # history = service.get_history(history_id) # if history is None: # raise HTTPException( # status_code=404, # detail="History not found", # ) # return history # finally: # db.close() # @app.delete("/history/{history_id}") # def delete_history(history_id: int): # db = SessionLocal() # try: # service = DiagnosisService(db) # deleted = service.delete_history( # history_id # ) # if not deleted: # raise HTTPException( # status_code=404, # detail="History not found", # ) # return { # "success": True # } # finally: # db.close() # @app.post("/diagnose") # async def diagnose( # image: UploadFile = File(...), # ): # suffix = Path(image.filename).suffix.lower() # if suffix not in settings.ALLOWED_IMAGE_EXTENSIONS: # raise HTTPException( # status_code=400, # detail="Unsupported image type", # ) # with tempfile.NamedTemporaryFile( # delete=False, # suffix=suffix, # ) as temp: # temp.write(await image.read()) # temp_path = Path(temp.name) # db = SessionLocal() # try: # service = DiagnosisService(db) # result = service.diagnose( # temp_path # ) # return JSONResponse(result) # except Exception as ex: # raise HTTPException( # status_code=500, # detail=str(ex), # ) # finally: # db.close() # try: # temp_path.unlink() # except Exception: # pass # if __name__ == "__main__": # import uvicorn # uvicorn.run( # "app:app", # host="0.0.0.0", # port=8000, # reload=True, # )