File size: 2,590 Bytes
6bfebc3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | import os, json, sqlite3
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer
from huggingface_hub import HfApi
SECRET_KEY = os.environ.get("JWT_SECRET", "dev-secret-change-me")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7 # 7天
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
security = HTTPBearer()
DB_PATH = "/tmp/users.db"
BUCKET_ID = os.environ.get("HF_BUCKET_ID", "your-username/learning-platform-data")
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
password_hash TEXT NOT NULL,
email TEXT,
created_at TEXT
)
""")
conn.commit()
conn.close()
# 尝试从 Buckets 恢复用户数据库
_restore_db_from_bucket()
def _restore_db_from_bucket():
try:
from huggingface_hub import hf_hub_download
path = hf_hub_download(
repo_id=BUCKET_ID.replace("/", "_"), # Buckets 访问方式
filename="users.db",
repo_type="bucket"
)
import shutil
shutil.copy(path, DB_PATH)
except Exception:
pass # 首次运行没有备份
def _backup_db_to_bucket():
try:
from huggingface_hub import upload_file
upload_file(
path_or_fileobj=DB_PATH,
path_in_repo="users.db",
repo_id=BUCKET_ID.replace("/", "_"),
repo_type="bucket"
)
except Exception:
pass
def hash_password(password: str) -> str:
return pwd_context.hash(password)
def verify_password(plain: str, hashed: str) -> bool:
return pwd_context.verify(plain, hashed)
def create_access_token(data: dict):
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def get_current_user(credentials=Depends(security)):
token = credentials.credentials
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise HTTPException(status_code=401, detail="Invalid token")
return username
except JWTError:
raise HTTPException(status_code=401, detail="Invalid token") |