| from fastapi import FastAPI, UploadFile, File, Form, Depends, HTTPException |
| from fastapi.staticfiles import StaticFiles |
| from fastapi.responses import FileResponse |
| import os |
| import auth, storage, ai_search |
|
|
| app = FastAPI(title="云学习平台") |
|
|
| |
| BASE_DIR = os.path.dirname(os.path.abspath(__file__)) |
| STATIC_DIR = os.path.join(BASE_DIR, "static") |
| os.makedirs(STATIC_DIR, exist_ok=True) |
|
|
| INDEX_PATH = os.path.join(STATIC_DIR, "index.html") |
| if not os.path.exists(INDEX_PATH): |
| with open(INDEX_PATH, "w", encoding="utf-8") as f: |
| f.write("<h1>云学习平台已启动</h1><p>请访问 <a href='/docs'>/docs</a></p>") |
|
|
| app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") |
| def startup(): |
| auth.init_db() |
| storage.ensure_bucket() |
|
|
| |
| @app.post("/auth/register") |
| def register(username: str = Form(...), password: str = Form(...)): |
| conn = auth.sqlite3.connect(auth.DB_PATH) |
| try: |
| conn.execute( |
| "INSERT INTO users (username, password_hash, created_at) VALUES (?,?,?)", |
| (username, auth.hash_password(password), auth.datetime.now().isoformat()) |
| ) |
| conn.commit() |
| except auth.sqlite3.IntegrityError: |
| raise HTTPException(400, "用户名已存在") |
| finally: |
| conn.close() |
| token = auth.create_access_token({"sub": username}) |
| return {"access_token": token, "token_type": "bearer"} |
|
|
| @app.post("/auth/login") |
| def login(username: str = Form(...), password: str = Form(...)): |
| conn = auth.sqlite3.connect(auth.DB_PATH) |
| row = conn.execute("SELECT password_hash FROM users WHERE username=?", (username,)).fetchone() |
| conn.close() |
| if not row or not auth.verify_password(password, row[0]): |
| raise HTTPException(401, "用户名或密码错误") |
| token = auth.create_access_token({"sub": username}) |
| return {"access_token": token, "token_type": "bearer"} |
|
|
| |
| @app.post("/files/upload") |
| async def upload(file: UploadFile = File(...), user: str = Depends(auth.get_current_user)): |
| content = await file.read() |
| path = storage.upload_file(content, file.filename, user, file.content_type or "application/octet-stream") |
| return {"path": path, "filename": file.filename} |
|
|
| @app.get("/files/list") |
| def list_files(user: str = Depends(auth.get_current_user)): |
| return {"files": storage.list_user_files(user)} |
|
|
| |
| @app.post("/ai/search-by-image") |
| async def search_by_image(file: UploadFile = File(...), user: str = Depends(auth.get_current_user)): |
| image_bytes = await file.read() |
| description = await ai_search.analyze_image(image_bytes) |
| results = await ai_search.search_in_library(description, user, storage) |
| return {"ai_description": description, "results": results} |
|
|
| @app.get("/") |
| def root(): |
| return FileResponse("static/index.html") |
|
|
| if __name__ == "__main__": |
| import uvicorn |
| uvicorn.run(app, host="0.0.0.0", port=7860) |
