313 lines
12 KiB
Python
313 lines
12 KiB
Python
import json
|
|
import sqlite3
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
from app.config import DATABASE_PATH
|
|
|
|
ROBOTS_DEFAULT = [
|
|
{
|
|
"id": "RB-001", "number": 1, "name": "Альфа-7", "type": "Промышленный манипулятор",
|
|
"status": "online", "battery": 87, "speed": 1.2, "maxSpeed": 2.4, "baseTemp": 32,
|
|
"payload": 15, "temperature": 42, "uptime": "142ч", "firmware": "v3.2.1",
|
|
"mode": "auto", "speedSetting": 50, "sensitivity": 70, "emergency": False, "nightMode": False,
|
|
},
|
|
{
|
|
"id": "RB-002", "number": 2, "name": "Бета-3", "type": "Мобильный транспортёр",
|
|
"status": "idle", "battery": 64, "speed": 0.8, "maxSpeed": 1.6, "baseTemp": 30,
|
|
"payload": 50, "temperature": 38, "uptime": "89ч", "firmware": "v2.8.0",
|
|
"mode": "assist", "speedSetting": 35, "sensitivity": 55, "emergency": False, "nightMode": True,
|
|
},
|
|
{
|
|
"id": "RB-003", "number": 3, "name": "Гамма-X", "type": "Сервисный ассистент",
|
|
"status": "online", "battery": 95, "speed": 0.5, "maxSpeed": 1.0, "baseTemp": 28,
|
|
"payload": 5, "temperature": 35, "uptime": "210ч", "firmware": "v4.0.2",
|
|
"mode": "manual", "speedSetting": 60, "sensitivity": 80, "emergency": False, "nightMode": False,
|
|
},
|
|
{
|
|
"id": "RB-004", "number": 4, "name": "Дельта-9", "type": "Исследовательский дрон",
|
|
"status": "offline", "battery": 12, "speed": 0, "maxSpeed": 5.0, "baseTemp": 22,
|
|
"payload": 2, "temperature": 28, "uptime": "0ч", "firmware": "v1.9.5",
|
|
"mode": "standby", "speedSetting": 20, "sensitivity": 40, "emergency": True, "nightMode": False,
|
|
},
|
|
]
|
|
|
|
|
|
def utcnow() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
@contextmanager
|
|
def get_db():
|
|
Path(DATABASE_PATH).parent.mkdir(parents=True, exist_ok=True)
|
|
conn = sqlite3.connect(DATABASE_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("PRAGMA foreign_keys = ON")
|
|
try:
|
|
yield conn
|
|
conn.commit()
|
|
except Exception:
|
|
conn.rollback()
|
|
raise
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def init_db():
|
|
with get_db() as conn:
|
|
conn.executescript("""
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
email TEXT NOT NULL UNIQUE COLLATE NOCASE,
|
|
password_hash TEXT NOT NULL,
|
|
firstname TEXT DEFAULT '',
|
|
lastname TEXT DEFAULT '',
|
|
patronymic TEXT DEFAULT '',
|
|
age INTEGER,
|
|
phone TEXT DEFAULT '',
|
|
rating REAL NOT NULL DEFAULT 0,
|
|
points INTEGER NOT NULL DEFAULT 0,
|
|
selected_robot_id TEXT DEFAULT 'RB-001',
|
|
created_at TEXT NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS sessions (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER NOT NULL,
|
|
token_hash TEXT NOT NULL UNIQUE,
|
|
created_at TEXT NOT NULL,
|
|
expires_at TEXT NOT NULL,
|
|
is_active INTEGER NOT NULL DEFAULT 1,
|
|
user_agent TEXT DEFAULT '',
|
|
ip_address TEXT DEFAULT '',
|
|
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS robot_settings (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER NOT NULL,
|
|
robot_id TEXT NOT NULL,
|
|
settings_json TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
UNIQUE(user_id, robot_id),
|
|
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS external_data (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER,
|
|
source_app TEXT NOT NULL,
|
|
data_key TEXT NOT NULL,
|
|
data_value TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE SET NULL
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_sessions_token ON sessions(token_hash);
|
|
CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id);
|
|
CREATE INDEX IF NOT EXISTS idx_external_user ON external_data(user_id);
|
|
""")
|
|
|
|
|
|
def row_to_dict(row: sqlite3.Row) -> dict:
|
|
return dict(row) if row else None
|
|
|
|
|
|
def create_user(email: str, password_hash: str) -> dict:
|
|
now = utcnow().isoformat()
|
|
with get_db() as conn:
|
|
cur = conn.execute(
|
|
"INSERT INTO users (email, password_hash, created_at) VALUES (?, ?, ?)",
|
|
(email.lower().strip(), password_hash, now),
|
|
)
|
|
user_id = cur.lastrowid
|
|
for robot in ROBOTS_DEFAULT:
|
|
conn.execute(
|
|
"INSERT INTO robot_settings (user_id, robot_id, settings_json, updated_at) VALUES (?, ?, ?, ?)",
|
|
(user_id, robot["id"], json.dumps(robot, ensure_ascii=False), now),
|
|
)
|
|
row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
|
|
return row_to_dict(row)
|
|
|
|
|
|
def get_user_by_email(email: str) -> dict | None:
|
|
with get_db() as conn:
|
|
row = conn.execute(
|
|
"SELECT * FROM users WHERE email = ? COLLATE NOCASE",
|
|
(email.lower().strip(),),
|
|
).fetchone()
|
|
return row_to_dict(row)
|
|
|
|
|
|
def get_user_by_id(user_id: int) -> dict | None:
|
|
with get_db() as conn:
|
|
row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
|
|
return row_to_dict(row)
|
|
|
|
|
|
def update_user_profile(user_id: int, data: dict) -> dict:
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"""UPDATE users SET firstname=?, lastname=?, patronymic=?, age=?, phone=?
|
|
WHERE id=?""",
|
|
(
|
|
data.get("firstname", ""),
|
|
data.get("lastname", ""),
|
|
data.get("patronymic", ""),
|
|
data.get("age"),
|
|
data.get("phone", ""),
|
|
user_id,
|
|
),
|
|
)
|
|
row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
|
|
return row_to_dict(row)
|
|
|
|
|
|
def create_session(user_id: int, token_hash: str, user_agent: str, ip_address: str, ttl_hours: int) -> dict:
|
|
now = utcnow()
|
|
expires = now + timedelta(hours=ttl_hours)
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"""INSERT INTO sessions (user_id, token_hash, created_at, expires_at, user_agent, ip_address)
|
|
VALUES (?, ?, ?, ?, ?, ?)""",
|
|
(user_id, token_hash, now.isoformat(), expires.isoformat(), user_agent, ip_address),
|
|
)
|
|
row = conn.execute(
|
|
"SELECT * FROM sessions WHERE token_hash = ?", (token_hash,)
|
|
).fetchone()
|
|
return row_to_dict(row)
|
|
|
|
|
|
def get_session_by_token_hash(token_hash: str) -> dict | None:
|
|
with get_db() as conn:
|
|
row = conn.execute(
|
|
"SELECT * FROM sessions WHERE token_hash = ? AND is_active = 1",
|
|
(token_hash,),
|
|
).fetchone()
|
|
return row_to_dict(row)
|
|
|
|
|
|
def deactivate_session(token_hash: str) -> None:
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"UPDATE sessions SET is_active = 0 WHERE token_hash = ?",
|
|
(token_hash,),
|
|
)
|
|
|
|
|
|
def get_active_sessions(user_id: int | None = None) -> list[dict]:
|
|
now = utcnow().isoformat()
|
|
with get_db() as conn:
|
|
if user_id is not None:
|
|
rows = conn.execute(
|
|
"""SELECT s.*, u.email FROM sessions s
|
|
JOIN users u ON u.id = s.user_id
|
|
WHERE s.is_active = 1 AND s.expires_at > ? AND s.user_id = ?
|
|
ORDER BY s.created_at DESC""",
|
|
(now, user_id),
|
|
).fetchall()
|
|
else:
|
|
rows = conn.execute(
|
|
"""SELECT s.*, u.email FROM sessions s
|
|
JOIN users u ON u.id = s.user_id
|
|
WHERE s.is_active = 1 AND s.expires_at > ?
|
|
ORDER BY s.created_at DESC""",
|
|
(now,),
|
|
).fetchall()
|
|
return [row_to_dict(r) for r in rows]
|
|
|
|
|
|
def get_user_robots(user_id: int) -> list[dict]:
|
|
with get_db() as conn:
|
|
rows = conn.execute(
|
|
"SELECT robot_id, settings_json FROM robot_settings WHERE user_id = ? ORDER BY robot_id",
|
|
(user_id,),
|
|
).fetchall()
|
|
if not rows:
|
|
now = utcnow().isoformat()
|
|
for robot in ROBOTS_DEFAULT:
|
|
conn.execute(
|
|
"INSERT INTO robot_settings (user_id, robot_id, settings_json, updated_at) VALUES (?, ?, ?, ?)",
|
|
(user_id, robot["id"], json.dumps(robot, ensure_ascii=False), now),
|
|
)
|
|
rows = conn.execute(
|
|
"SELECT robot_id, settings_json FROM robot_settings WHERE user_id = ? ORDER BY robot_id",
|
|
(user_id,),
|
|
).fetchall()
|
|
return [json.loads(r["settings_json"]) for r in rows]
|
|
|
|
|
|
def update_robot_settings(user_id: int, robot_id: str, settings: dict) -> dict:
|
|
now = utcnow().isoformat()
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"""INSERT INTO robot_settings (user_id, robot_id, settings_json, updated_at)
|
|
VALUES (?, ?, ?, ?)
|
|
ON CONFLICT(user_id, robot_id) DO UPDATE SET
|
|
settings_json = excluded.settings_json, updated_at = excluded.updated_at""",
|
|
(user_id, robot_id, json.dumps(settings, ensure_ascii=False), now),
|
|
)
|
|
return settings
|
|
|
|
|
|
def set_selected_robot(user_id: int, robot_id: str) -> None:
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"UPDATE users SET selected_robot_id = ? WHERE id = ?",
|
|
(robot_id, user_id),
|
|
)
|
|
|
|
|
|
def add_external_data(user_id: int | None, source_app: str, data_key: str, data_value: dict) -> dict:
|
|
now = utcnow().isoformat()
|
|
with get_db() as conn:
|
|
cur = conn.execute(
|
|
"""INSERT INTO external_data (user_id, source_app, data_key, data_value, created_at)
|
|
VALUES (?, ?, ?, ?, ?)""",
|
|
(user_id, source_app, data_key, json.dumps(data_value, ensure_ascii=False), now),
|
|
)
|
|
if user_id and "points" in data_value:
|
|
points = int(data_value["points"])
|
|
conn.execute(
|
|
"UPDATE users SET points = points + ? WHERE id = ?",
|
|
(points, user_id),
|
|
)
|
|
if user_id and "rating" in data_value:
|
|
conn.execute(
|
|
"UPDATE users SET rating = ? WHERE id = ?",
|
|
(float(data_value["rating"]), user_id),
|
|
)
|
|
row = conn.execute(
|
|
"SELECT * FROM external_data WHERE id = ?", (cur.lastrowid,)
|
|
).fetchone()
|
|
return row_to_dict(row)
|
|
|
|
|
|
def get_external_data(user_id: int | None = None, limit: int = 100) -> list[dict]:
|
|
with get_db() as conn:
|
|
if user_id is not None:
|
|
rows = conn.execute(
|
|
"SELECT * FROM external_data WHERE user_id = ? ORDER BY created_at DESC LIMIT ?",
|
|
(user_id, limit),
|
|
).fetchall()
|
|
else:
|
|
rows = conn.execute(
|
|
"SELECT * FROM external_data ORDER BY created_at DESC LIMIT ?",
|
|
(limit,),
|
|
).fetchall()
|
|
result = []
|
|
for r in rows:
|
|
item = row_to_dict(r)
|
|
item["data_value"] = json.loads(item["data_value"])
|
|
result.append(item)
|
|
return result
|
|
|
|
|
|
def list_users_public() -> list[dict]:
|
|
with get_db() as conn:
|
|
rows = conn.execute(
|
|
"SELECT id, email, firstname, lastname, rating, points, created_at FROM users ORDER BY id"
|
|
).fetchall()
|
|
return [row_to_dict(r) for r in rows]
|