scripts/seed.py - master - mugmod/AI_jogging_UrFU

scripts/seed.py
186 строк7 KB
mugmod
Fix: переместил проект в корень репозитория
06 июн 2026, 01:09
06 июн 2026, 01:09711580d
#!/usr/bin/env python3
"""
Seed script: создаёт 25 пробежек с прогрессом за последние 6 месяцев.
Использует /api/runs/analyze-json — ML не нужен.

Запуск:
  python3 scripts/seed.py
"""

import json
import random
import time
import urllib.request
import urllib.error
from datetime import datetime, timedelta, timezone

API = "http://localhost:3000"
EMAIL = "seed@test.com"

# ── Нормы для speed=10-12, height=165-175, weight=60-75 ──────────────────────
NORMS = {
    "head_angle":   {"optimal": 135.6, "sigma": 2.8},
    "elbow_angle":  {"optimal": 109.6, "sigma": 10.0},
    "arm_swing":    {"optimal": 146.3, "sigma": 6.0},
    "knee_angle":   {"optimal": 118.3, "sigma": 19.9},
    "hip_angle":    {"optimal": 24.0,  "sigma": 5.9},
    "ankle_angle":  {"optimal": 99.6,  "sigma": 8.7},
    "shank_angle":  {"optimal": 51.4,  "sigma": 17.6},
    "trunk_angle":  {"optimal": 175.2, "sigma": 1.5},
}


def post(path, body=None, token=None):
    data = json.dumps(body).encode() if body else None
    headers = {"Content-Type": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    req = urllib.request.Request(f"{API}{path}", data=data, headers=headers, method="POST")
    with urllib.request.urlopen(req, timeout=15) as r:
        return json.loads(r.read())


def get(path, token=None):
    headers = {}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    req = urllib.request.Request(f"{API}{path}", headers=headers)
    with urllib.request.urlopen(req, timeout=15) as r:
        return json.loads(r.read())


def get_token():
    """Логинимся через email (dev mode — код сразу в ответе)."""
    res = post("/api/auth/email/request", {"email": EMAIL})
    code = res["data"]["code"]
    res = post("/api/auth/email/verify", {"email": EMAIL, "code": code})
    return res["data"]["accessToken"]


def make_stat(optimal, sigma, quality, side_offset=0.0):
    """
    Генерирует {min, max, mean, std, count} для одного параметра.
    quality: 0.0 (плохо) → 1.0 (идеально)
    """
    # Чем хуже quality, тем дальше от оптимума
    deviation = (1.0 - quality) * 2.5 * sigma * random.uniform(-1, 1)
    mean = optimal + deviation + side_offset
    std = sigma * (0.3 + (1.0 - quality) * 0.7) * random.uniform(0.8, 1.2)
    count = random.randint(50, 80)
    lo = mean - 2.5 * std
    hi = mean + 2.5 * std
    return {"min": round(lo, 1), "max": round(hi, 1),
            "mean": round(mean, 1), "std": round(std, 1), "count": count}


def make_biomechanics(quality):
    """Строит biomechanicsRaw по нормам с заданным качеством."""
    n = NORMS
    # Небольшая асимметрия между сторонами
    r_off = random.uniform(-2, 2)

    return {
        "left_arm": {
            "arm_swing":   make_stat(n["arm_swing"]["optimal"],   n["arm_swing"]["sigma"],   quality),
            "elbow_angle": make_stat(n["elbow_angle"]["optimal"], n["elbow_angle"]["sigma"], quality),
        },
        "right_arm": {
            "arm_swing":   make_stat(n["arm_swing"]["optimal"],   n["arm_swing"]["sigma"],   quality, r_off),
            "elbow_angle": make_stat(n["elbow_angle"]["optimal"], n["elbow_angle"]["sigma"], quality, r_off),
        },
        "left_leg": {
            "knee_angle":  make_stat(n["knee_angle"]["optimal"],  n["knee_angle"]["sigma"],  quality),
            "ankle_angle": make_stat(n["ankle_angle"]["optimal"], n["ankle_angle"]["sigma"], quality),
            "hip_angle":   make_stat(n["hip_angle"]["optimal"],   n["hip_angle"]["sigma"],   quality),
            "shank_angle": make_stat(n["shank_angle"]["optimal"], n["shank_angle"]["sigma"], quality),
        },
        "right_leg": {
            "knee_angle":  make_stat(n["knee_angle"]["optimal"],  n["knee_angle"]["sigma"],  quality, r_off),
            "ankle_angle": make_stat(n["ankle_angle"]["optimal"], n["ankle_angle"]["sigma"], quality, r_off),
            "hip_angle":   make_stat(n["hip_angle"]["optimal"],   n["hip_angle"]["sigma"],   quality, r_off),
            "shank_angle": make_stat(n["shank_angle"]["optimal"], n["shank_angle"]["sigma"], quality, r_off),
        },
        "trunk": {
            "trunk_angle": make_stat(n["trunk_angle"]["optimal"], n["trunk_angle"]["sigma"], quality),
        },
        "head": {
            "head_angle":  make_stat(n["head_angle"]["optimal"],  n["head_angle"]["sigma"],  quality),
        },
    }


def main():
    random.seed(42)

    print("🔐 Авторизация...")
    token = get_token()
    print(f"   ✓ токен получен\n")

    # 25 пробежек за последние 6 месяцев — примерно раз в неделю
    now = datetime.now(timezone.utc)
    runs_meta = []

    print("🏃 Создаю пробежки...\n")

    for i in range(25):
        # Прогресс: quality растёт от ~0.35 до ~0.75 с небольшим шумом
        progress = i / 24
        quality = 0.35 + progress * 0.40 + random.uniform(-0.05, 0.05)
        quality = max(0.2, min(0.95, quality))

        # Скорость чуть растёт со временем
        speed = round(10.0 + progress * 2.0 + random.uniform(-0.5, 0.5), 1)

        body = {
            "weightKg": 70,
            "heightCm": 170,
            "speed": speed,
            "biomechanicsRaw": make_biomechanics(quality),
        }

        try:
            res = post("/api/runs/analyze-json", body, token)
            run_id = res["runId"]

            # Дата: равномерно за последние 6 месяцев
            days_ago = int((24 - i) * 7.3)
            run_date = now - timedelta(days=days_ago)

            runs_meta.append({"id": run_id, "date": run_date.isoformat(), "quality": round(quality, 2)})
            print(f"  [{i+1:2}/25] run {run_id[:8]}...  quality={quality:.2f}  speed={speed}")

            time.sleep(1.5)  # не спамим сервер

        except Exception as e:
            print(f"  [{i+1:2}/25] ❌ ошибка: {e}")

    print(f"\n✅ Создано {len(runs_meta)} пробежек")

    # Обновляем даты в MongoDB
    print("\n📅 Обновляю даты в MongoDB...")
    mongo_cmd = "db.runs.find({status:'Processed'}, {_id:1, uploadedAt:1}).sort({uploadedAt:-1}).limit(30).forEach(r => print(r._id + ' ' + r.uploadedAt))"
    print("   Чтобы раздать пробежкам даты за последние 6 месяцев, запусти:")
    print()

    # Генерируем JS-скрипт для MongoDB
    js_parts = []
    for meta in runs_meta:
        d = meta["date"].replace("+00:00", "Z")
        js_parts.append(
            f'db.runs.updateOne({{_id: ObjectId("{meta["id"]}")}}, '
            f'{{$set: {{uploadedAt: new Date("{d}"), analyzedAt: new Date("{d}")}}}});'
        )

    js_script = "\n".join(js_parts)
    script_path = "/home/keka/Desktop/dev/sberAI/AI_jogging/scripts/update_dates.js"
    with open(script_path, "w") as f:
        f.write(js_script)

    print(f"   sudo docker exec mongodb mongosh -u mongo_admin -p mongo_secret_2024 --authenticationDatabase admin aij --file /scripts/update_dates.js")
    print(f"\n   (скрипт сохранён в scripts/update_dates.js)")
    print()
    print("🎉 Готово! Открой http://localhost:8081 и посмотри пробежки.")


if __name__ == "__main__":
    main()