281 lines
9.0 KiB
Python
281 lines
9.0 KiB
Python
from datetime import datetime, timezone
|
|
|
|
from fastapi import Cookie, Depends, FastAPI, HTTPException, Request, Response
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from app import database as db
|
|
from app.config import EXTERNAL_API_KEY, SESSION_COOKIE, SESSION_TTL_HOURS, START_VIA_SCRIPT
|
|
from app.database import init_db
|
|
from app.schemas import (
|
|
ExternalDataWrite,
|
|
LoginRequest,
|
|
ProfileUpdate,
|
|
RegisterRequest,
|
|
RobotSettingsUpdate,
|
|
RobotsBulkUpdate,
|
|
)
|
|
from app.security import (
|
|
generate_session_token,
|
|
hash_password,
|
|
hash_token,
|
|
verify_password,
|
|
)
|
|
|
|
app = FastAPI(title="ROBO-OPS", version="2.0.0")
|
|
|
|
STATIC_DIR = __import__("pathlib").Path(__file__).resolve().parent.parent / "static"
|
|
|
|
|
|
@app.on_event("startup")
|
|
def on_startup():
|
|
if not START_VIA_SCRIPT:
|
|
raise RuntimeError(
|
|
"Приложение запускается только через ./start.sh\n"
|
|
"См. README.md — раздел «Запуск приложения»."
|
|
)
|
|
init_db()
|
|
|
|
|
|
def user_public(user: dict) -> dict:
|
|
return {
|
|
"id": user["id"],
|
|
"email": user["email"],
|
|
"firstname": user["firstname"],
|
|
"lastname": user["lastname"],
|
|
"patronymic": user["patronymic"],
|
|
"age": user["age"],
|
|
"phone": user["phone"],
|
|
"rating": user["rating"],
|
|
"points": user["points"],
|
|
"selected_robot_id": user["selected_robot_id"],
|
|
}
|
|
|
|
|
|
def session_public(session: dict) -> dict:
|
|
return {
|
|
"id": session["id"],
|
|
"user_id": session["user_id"],
|
|
"email": session.get("email", ""),
|
|
"created_at": session["created_at"],
|
|
"expires_at": session["expires_at"],
|
|
"user_agent": session["user_agent"],
|
|
"ip_address": session["ip_address"],
|
|
}
|
|
|
|
|
|
async def get_current_user(
|
|
request: Request,
|
|
session_token: str | None = Cookie(default=None, alias=SESSION_COOKIE),
|
|
) -> dict:
|
|
if not session_token:
|
|
raise HTTPException(status_code=401, detail="Требуется авторизация")
|
|
token_hash = hash_token(session_token)
|
|
session = db.get_session_by_token_hash(token_hash)
|
|
if not session:
|
|
raise HTTPException(status_code=401, detail="Сессия недействительна")
|
|
expires = datetime.fromisoformat(session["expires_at"])
|
|
if expires.tzinfo is None:
|
|
expires = expires.replace(tzinfo=timezone.utc)
|
|
if expires < datetime.now(timezone.utc):
|
|
db.deactivate_session(token_hash)
|
|
raise HTTPException(status_code=401, detail="Сессия истекла")
|
|
user = db.get_user_by_id(session["user_id"])
|
|
if not user:
|
|
raise HTTPException(status_code=401, detail="Пользователь не найден")
|
|
return user
|
|
|
|
|
|
def verify_external_api_key(request: Request) -> None:
|
|
api_key = request.headers.get("X-API-Key")
|
|
if not api_key or api_key != EXTERNAL_API_KEY:
|
|
raise HTTPException(status_code=403, detail="Неверный API-ключ")
|
|
|
|
|
|
# --- Auth ---
|
|
|
|
@app.post("/api/auth/register")
|
|
def register(data: RegisterRequest, response: Response, request: Request):
|
|
if db.get_user_by_email(data.email):
|
|
raise HTTPException(status_code=400, detail="Email уже зарегистрирован")
|
|
user = db.create_user(data.email, hash_password(data.password))
|
|
token = generate_session_token()
|
|
db.create_session(
|
|
user["id"],
|
|
hash_token(token),
|
|
request.headers.get("user-agent", ""),
|
|
request.client.host if request.client else "",
|
|
SESSION_TTL_HOURS,
|
|
)
|
|
response.set_cookie(
|
|
key=SESSION_COOKIE,
|
|
value=token,
|
|
httponly=True,
|
|
samesite="lax",
|
|
max_age=SESSION_TTL_HOURS * 3600,
|
|
)
|
|
return {"user": user_public(user), "message": "Регистрация успешна"}
|
|
|
|
|
|
@app.post("/api/auth/login")
|
|
def login(data: LoginRequest, response: Response, request: Request):
|
|
user = db.get_user_by_email(data.email)
|
|
if not user or not verify_password(data.password, user["password_hash"]):
|
|
raise HTTPException(status_code=401, detail="Неверный email или пароль")
|
|
token = generate_session_token()
|
|
db.create_session(
|
|
user["id"],
|
|
hash_token(token),
|
|
request.headers.get("user-agent", ""),
|
|
request.client.host if request.client else "",
|
|
SESSION_TTL_HOURS,
|
|
)
|
|
response.set_cookie(
|
|
key=SESSION_COOKIE,
|
|
value=token,
|
|
httponly=True,
|
|
samesite="lax",
|
|
max_age=SESSION_TTL_HOURS * 3600,
|
|
)
|
|
return {"user": user_public(user), "message": "Вход выполнен"}
|
|
|
|
|
|
@app.post("/api/auth/logout")
|
|
def logout(
|
|
response: Response,
|
|
session_token: str | None = Cookie(default=None, alias=SESSION_COOKIE),
|
|
):
|
|
if session_token:
|
|
db.deactivate_session(hash_token(session_token))
|
|
response.delete_cookie(SESSION_COOKIE)
|
|
return {"message": "Выход выполнен"}
|
|
|
|
|
|
@app.get("/api/auth/me")
|
|
def me(user: dict = Depends(get_current_user)):
|
|
return user_public(user)
|
|
|
|
|
|
# --- Profile ---
|
|
|
|
@app.get("/api/profile")
|
|
def get_profile(user: dict = Depends(get_current_user)):
|
|
return user_public(user)
|
|
|
|
|
|
@app.put("/api/profile")
|
|
def update_profile(data: ProfileUpdate, user: dict = Depends(get_current_user)):
|
|
updated = db.update_user_profile(user["id"], data.model_dump())
|
|
return user_public(updated)
|
|
|
|
|
|
# --- Sessions ---
|
|
|
|
@app.get("/api/sessions")
|
|
def my_sessions(user: dict = Depends(get_current_user)):
|
|
sessions = db.get_active_sessions(user["id"])
|
|
return {"sessions": [session_public(s) for s in sessions]}
|
|
|
|
|
|
# --- Robots ---
|
|
|
|
@app.get("/api/robots")
|
|
def get_robots(user: dict = Depends(get_current_user)):
|
|
robots = db.get_user_robots(user["id"])
|
|
return {
|
|
"robots": robots,
|
|
"selected_robot_id": user["selected_robot_id"],
|
|
}
|
|
|
|
|
|
@app.put("/api/robots")
|
|
def save_robots(data: RobotsBulkUpdate, user: dict = Depends(get_current_user)):
|
|
for robot in data.robots:
|
|
robot_id = robot.get("id")
|
|
if robot_id:
|
|
db.update_robot_settings(user["id"], robot_id, robot)
|
|
if data.selected_robot_id:
|
|
db.set_selected_robot(user["id"], data.selected_robot_id)
|
|
user = db.get_user_by_id(user["id"])
|
|
return {
|
|
"robots": db.get_user_robots(user["id"]),
|
|
"selected_robot_id": user["selected_robot_id"],
|
|
}
|
|
|
|
|
|
@app.put("/api/robots/{robot_id}")
|
|
def save_robot(robot_id: str, data: RobotSettingsUpdate, user: dict = Depends(get_current_user)):
|
|
settings = {**data.settings, "id": robot_id}
|
|
db.update_robot_settings(user["id"], robot_id, settings)
|
|
return settings
|
|
|
|
|
|
# --- External API ---
|
|
|
|
@app.get("/api/external/users")
|
|
def external_list_users(_: None = Depends(verify_external_api_key)):
|
|
return {"users": db.list_users_public()}
|
|
|
|
|
|
@app.get("/api/external/users/{user_id}")
|
|
def external_get_user(user_id: int, _: None = Depends(verify_external_api_key)):
|
|
user = db.get_user_by_id(user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="Пользователь не найден")
|
|
return user_public(user)
|
|
|
|
|
|
@app.get("/api/external/users/by-email/{email}")
|
|
def external_get_user_by_email(email: str, _: None = Depends(verify_external_api_key)):
|
|
user = db.get_user_by_email(email)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="Пользователь не найден")
|
|
return user_public(user)
|
|
|
|
|
|
@app.get("/api/external/sessions")
|
|
def external_sessions(
|
|
user_id: int | None = None,
|
|
_: None = Depends(verify_external_api_key),
|
|
):
|
|
sessions = db.get_active_sessions(user_id)
|
|
return {"sessions": [session_public(s) for s in sessions]}
|
|
|
|
|
|
@app.post("/api/external/data")
|
|
def external_write_data(data: ExternalDataWrite, _: None = Depends(verify_external_api_key)):
|
|
if data.user_id is not None and not db.get_user_by_id(data.user_id):
|
|
raise HTTPException(status_code=404, detail="Пользователь не найден")
|
|
record = db.add_external_data(data.user_id, data.source_app, data.data_key, data.data_value)
|
|
record["data_value"] = __import__("json").loads(record["data_value"])
|
|
return {"record": record, "message": "Данные записаны"}
|
|
|
|
|
|
@app.get("/api/external/data")
|
|
def external_get_data(
|
|
user_id: int | None = None,
|
|
limit: int = 100,
|
|
_: None = Depends(verify_external_api_key),
|
|
):
|
|
return {"data": db.get_external_data(user_id, min(limit, 500))}
|
|
|
|
|
|
# --- Static pages ---
|
|
|
|
@app.get("/")
|
|
def index():
|
|
return FileResponse(STATIC_DIR / "index.html")
|
|
|
|
|
|
@app.get("/login")
|
|
def login_page():
|
|
return FileResponse(STATIC_DIR / "login.html")
|
|
|
|
|
|
@app.get("/register")
|
|
def register_page():
|
|
return FileResponse(STATIC_DIR / "register.html")
|
|
|
|
|
|
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
|