# -*- coding: utf-8 -*-
from __future__ import annotations

import json
import secrets
import sqlite3
import time
from contextlib import contextmanager
from typing import Any, Iterator, Optional

from app.config import Config


def _connect() -> sqlite3.Connection:
    Config.DATABASE_PATH.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(Config.DATABASE_PATH), check_same_thread=False)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    return conn


@contextmanager
def get_db() -> Iterator[sqlite3.Connection]:
    conn = _connect()
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def init_db() -> None:
    with get_db() as conn:
        conn.executescript(
            """
            CREATE TABLE IF NOT EXISTS accounts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                broker TEXT DEFAULT '',
                login_id TEXT DEFAULT '',
                account_type TEXT DEFAULT 'demo',
                note TEXT DEFAULT '',
                enabled INTEGER DEFAULT 1,
                created_at REAL NOT NULL
            );

            CREATE TABLE IF NOT EXISTS machines (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                device_fingerprint TEXT DEFAULT '',
                note TEXT DEFAULT '',
                enabled INTEGER DEFAULT 1,
                last_seen_at REAL,
                created_at REAL NOT NULL
            );

            CREATE TABLE IF NOT EXISTS activations (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                code TEXT NOT NULL UNIQUE,
                account_id INTEGER,
                machine_id INTEGER,
                token TEXT,
                status TEXT DEFAULT 'pending',
                bound_at REAL,
                created_at REAL NOT NULL,
                FOREIGN KEY(account_id) REFERENCES accounts(id),
                FOREIGN KEY(machine_id) REFERENCES machines(id)
            );

            CREATE TABLE IF NOT EXISTS trades (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                ticket TEXT,
                symbol TEXT NOT NULL,
                side TEXT NOT NULL,
                volume REAL,
                open_price REAL,
                close_price REAL,
                open_time REAL,
                close_time REAL,
                profit REAL,
                sl REAL,
                tp REAL,
                account_id INTEGER,
                machine_id INTEGER,
                entry_style TEXT DEFAULT '',
                review_note TEXT DEFAULT '',
                snapshot_json TEXT DEFAULT '{}',
                created_at REAL NOT NULL,
                FOREIGN KEY(account_id) REFERENCES accounts(id),
                FOREIGN KEY(machine_id) REFERENCES machines(id)
            );

            CREATE TABLE IF NOT EXISTS bars (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                symbol TEXT NOT NULL,
                timeframe TEXT NOT NULL,
                ts REAL NOT NULL,
                open REAL, high REAL, low REAL, close REAL, volume REAL,
                UNIQUE(symbol, timeframe, ts)
            );

            CREATE TABLE IF NOT EXISTS settings (
                key TEXT PRIMARY KEY,
                value TEXT NOT NULL
            );

            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                username TEXT NOT NULL UNIQUE,
                password_hash TEXT NOT NULL,
                role TEXT NOT NULL DEFAULT 'user',
                account_id INTEGER UNIQUE,
                enabled INTEGER DEFAULT 1,
                created_at REAL NOT NULL,
                FOREIGN KEY(account_id) REFERENCES accounts(id)
            );
            """
        )


def now() -> float:
    return time.time()


def row_to_dict(row: Optional[sqlite3.Row]) -> Optional[dict]:
    if row is None:
        return None
    return {k: row[k] for k in row.keys()}


def list_accounts(conn: sqlite3.Connection) -> list[dict]:
    cur = conn.execute("SELECT * FROM accounts ORDER BY id DESC")
    return [row_to_dict(r) for r in cur.fetchall()]


def list_machines(conn: sqlite3.Connection) -> list[dict]:
    cur = conn.execute("SELECT * FROM machines ORDER BY id DESC")
    return [row_to_dict(r) for r in cur.fetchall()]


# 客户端约 2 秒心跳一次；超过该秒数无心跳视为离线（秒级大屏）
ONLINE_SECONDS = 8


def list_machines_for_user(conn: sqlite3.Connection, user: dict) -> list[dict]:
    if user.get("role") == "admin":
        items = list_machines(conn)
    else:
        cur = conn.execute(
            """
            SELECT DISTINCT m.* FROM machines m
            JOIN activations act ON act.machine_id = m.id
            WHERE act.account_id=?
            ORDER BY m.id DESC
            """,
            (user.get("account_id"),),
        )
        items = [row_to_dict(r) for r in cur.fetchall()]
    return annotate_machines_online(items)


def annotate_machines_online(machines: list[dict], now_ts: Optional[float] = None) -> list[dict]:
    ts = now_ts if now_ts is not None else now()
    for m in machines:
        last = m.get("last_seen_at")
        try:
            online = bool(last and (ts - float(last)) <= ONLINE_SECONDS)
        except (TypeError, ValueError):
            online = False
        m["online"] = online
        m["online_label"] = "在线" if online else "离线"
    return machines


def create_activation(
    conn: sqlite3.Connection,
    account_id: int,
    machine_id: Optional[int],
    created_by_user_id: Optional[int] = None,
) -> str:
    if not conn.execute("SELECT id FROM accounts WHERE id=?", (account_id,)).fetchone():
        raise ValueError("账户不存在（须由管理员预先创建）")
    code = secrets.token_urlsafe(12)
    # created_by 存在 activations 可选列：兼容旧库无该列时忽略
    cols = {r["name"] for r in conn.execute("PRAGMA table_info(activations)").fetchall()}
    if "created_by" in cols:
        conn.execute(
            "INSERT INTO activations(code, account_id, machine_id, status, created_at, created_by) VALUES(?,?,?,?,?,?)",
            (code, account_id, machine_id, "pending", now(), created_by_user_id),
        )
    else:
        conn.execute(
            "INSERT INTO activations(code, account_id, machine_id, status, created_at) VALUES(?,?,?,?,?)",
            (code, account_id, machine_id, "pending", now()),
        )
    return code


def activate_device(conn: sqlite3.Connection, code: str, fingerprint: str, machine_name: str) -> dict[str, Any]:
    row = conn.execute("SELECT * FROM activations WHERE code=?", (code,)).fetchone()
    if row is None:
        raise ValueError("激活码无效")
    if row["status"] == "revoked":
        raise ValueError("激活码已作废")
    if not row["account_id"]:
        raise ValueError("激活码未绑定账户，请联系管理员重新发放")
    if not conn.execute("SELECT id FROM accounts WHERE id=? AND enabled=1", (row["account_id"],)).fetchone():
        raise ValueError("绑定账户不存在或已停用")

    token = secrets.token_urlsafe(32)
    machine_id = row["machine_id"]
    if machine_id:
        conn.execute(
            "UPDATE machines SET device_fingerprint=?, last_seen_at=?, name=COALESCE(NULLIF(name,''), ?) WHERE id=?",
            (fingerprint, now(), machine_name or f"PC-{fingerprint[:6]}", machine_id),
        )
    else:
        cur = conn.execute(
            "INSERT INTO machines(name, device_fingerprint, note, enabled, last_seen_at, created_at) VALUES(?,?,?,?,?,?)",
            (machine_name or f"PC-{fingerprint[:6]}", fingerprint, "激活绑定", 1, now(), now()),
        )
        machine_id = cur.lastrowid
        conn.execute("UPDATE activations SET machine_id=? WHERE id=?", (machine_id, row["id"]))

    conn.execute(
        "UPDATE activations SET token=?, status='active', bound_at=? WHERE id=?",
        (token, now(), row["id"]),
    )
    return {
        "token": token,
        "account_id": row["account_id"],
        "machine_id": machine_id,
        "public_base": Config.PUBLIC_BASE_URL,
    }


def auth_token(conn: sqlite3.Connection, token: str) -> Optional[dict]:
    if not token:
        return None
    row = conn.execute(
        "SELECT * FROM activations WHERE token=? AND status='active'",
        (token,),
    ).fetchone()
    return row_to_dict(row)


def resolve_web_user_for_activation(conn: sqlite3.Connection, act: dict) -> Optional[dict]:
    """设备 token → 网页登录用户：优先账户绑定用户，其次发放激活码的用户。"""
    account_id = act.get("account_id")
    if account_id:
        u = conn.execute(
            "SELECT * FROM users WHERE account_id=? AND enabled=1",
            (account_id,),
        ).fetchone()
        if u:
            return row_to_dict(u)
    created_by = act.get("created_by")
    if created_by:
        u = get_user(conn, int(created_by))
        if u and int(u.get("enabled") or 0) == 1:
            return u
    return None


def _trades_where(
    account_id: Optional[int] = None,
    machine_id: Optional[int] = None,
    open_only: bool = False,
) -> tuple[str, list[Any]]:
    where: list[str] = []
    params: list[Any] = []
    if account_id is not None:
        where.append("t.account_id=?")
        params.append(account_id)
    if machine_id is not None:
        where.append("t.machine_id=?")
        params.append(machine_id)
    if open_only:
        where.append("(t.close_time IS NULL OR t.close_time='')")
    clause = (" WHERE " + " AND ".join(where)) if where else ""
    return clause, params


def count_trades(
    conn: sqlite3.Connection,
    account_id: Optional[int] = None,
    machine_id: Optional[int] = None,
    open_only: bool = False,
) -> int:
    clause, params = _trades_where(account_id, machine_id, open_only=open_only)
    sql = f"SELECT COUNT(*) AS c FROM trades t{clause}"
    row = conn.execute(sql, tuple(params)).fetchone()
    return int(row["c"] if row else 0)


def list_trades(
    conn: sqlite3.Connection,
    limit: int = 200,
    account_id: Optional[int] = None,
    machine_id: Optional[int] = None,
    offset: int = 0,
    open_only: bool = False,
) -> list[dict]:
    clause, params = _trades_where(account_id, machine_id, open_only=open_only)
    if open_only:
        # 持仓列表：同一账户+单号只显示最新一条
        sql = f"""
            SELECT t.*, a.name AS account_name, m.name AS machine_name
            FROM trades t
            LEFT JOIN accounts a ON a.id = t.account_id
            LEFT JOIN machines m ON m.id = t.machine_id
            {clause}
              AND t.id = (
                SELECT MAX(t2.id) FROM trades t2
                WHERE t2.account_id = t.account_id
                  AND t2.ticket = t.ticket
                  AND (t2.close_time IS NULL OR t2.close_time='')
              )
            ORDER BY COALESCE(t.open_time, t.created_at) DESC
            LIMIT ? OFFSET ?
        """
    else:
        sql = f"""
            SELECT t.*, a.name AS account_name, m.name AS machine_name
            FROM trades t
            LEFT JOIN accounts a ON a.id = t.account_id
            LEFT JOIN machines m ON m.id = t.machine_id
            {clause}
            ORDER BY COALESCE(t.close_time, t.open_time, t.created_at) DESC
            LIMIT ? OFFSET ?
        """
    params.extend([limit, max(0, int(offset))])
    cur = conn.execute(sql, tuple(params))
    rows = []
    for r in cur.fetchall():
        d = row_to_dict(r)
        try:
            d["snapshot"] = json.loads(d.get("snapshot_json") or "{}")
        except json.JSONDecodeError:
            d["snapshot"] = {}
        rows.append(d)
    return rows


# 入场形式：顺势/反转各含三种结构手法；其它=乱作
ENTRY_PATTERN_TYPES = ("窄幅区间突破", "宽幅区间假突破", "波段高低点假突破")
ENTRY_STYLES = (
    [f"顺势 · {p}" for p in ENTRY_PATTERN_TYPES]
    + [f"反转 · {p}" for p in ENTRY_PATTERN_TYPES]
    + ["其他 · 乱作"]
)


def entry_category(style: str) -> str:
    s = (style or "").strip()
    if not s:
        return "未标注"
    if s.startswith("顺势"):
        return "顺势"
    if s.startswith("反转"):
        return "反转"
    if s.startswith("其他") or "乱作" in s:
        return "其他"
    return "旧分类"


def _hold_seconds(open_time: Any, close_time: Any, now_ts: Optional[float] = None) -> Optional[float]:
    if open_time in (None, ""):
        return None
    try:
        start = float(open_time)
        end = float(close_time) if close_time not in (None, "") else float(now_ts or now())
    except (TypeError, ValueError):
        return None
    return max(0.0, end - start)


def _fmt_hold_secs(secs: Optional[float]) -> str:
    if secs is None:
        return "—"
    s = int(secs)
    days, rem = divmod(s, 86400)
    hours, rem = divmod(rem, 3600)
    mins, sec = divmod(rem, 60)
    if days > 0:
        return f"{days}天{hours}小时{mins}分"
    if hours > 0:
        return f"{hours}小时{mins}分"
    if mins > 0:
        return f"{mins}分{sec}秒"
    return f"{sec}秒"


def _bucket_key(secs: Optional[float]) -> str:
    if secs is None:
        return "unknown"
    if secs < 300:
        return "lt_5m"
    if secs < 1800:
        return "m5_30"
    if secs < 7200:
        return "m30_2h"
    return "ge_2h"


def _empty_group(label: str) -> dict[str, Any]:
    return {
        "label": label,
        "count": 0,
        "wins": 0,
        "losses": 0,
        "breakeven": 0,
        "profit": 0.0,
        "hold_sum": 0.0,
        "hold_n": 0,
    }


def _finalize_group(g: dict[str, Any]) -> dict[str, Any]:
    n = g["count"]
    decided = g["wins"] + g["losses"]
    g["win_rate"] = (g["wins"] / decided * 100.0) if decided else 0.0
    g["avg_profit"] = (g["profit"] / n) if n else 0.0
    g["avg_hold_secs"] = (g["hold_sum"] / g["hold_n"]) if g["hold_n"] else None
    g["avg_hold"] = _fmt_hold_secs(g["avg_hold_secs"])
    return g


def _accumulate(g: dict[str, Any], profit: Optional[float], hold: Optional[float]) -> None:
    g["count"] += 1
    p = float(profit) if profit is not None else 0.0
    g["profit"] += p
    if profit is not None:
        if p > 1e-9:
            g["wins"] += 1
        elif p < -1e-9:
            g["losses"] += 1
        else:
            g["breakeven"] += 1
    if hold is not None:
        g["hold_sum"] += hold
        g["hold_n"] += 1


def trade_stats(
    conn: sqlite3.Connection,
    account_id: Optional[int] = None,
    machine_id: Optional[int] = None,
) -> dict[str, Any]:
    """程序统计看板：盈亏/胜率/持仓时长分段/按机器与方向。"""
    clause, params = _trades_where(account_id, machine_id)
    sql = f"""
        SELECT t.id, t.side, t.symbol, t.profit, t.open_time, t.close_time,
               t.volume, t.machine_id, t.entry_style, m.name AS machine_name
        FROM trades t
        LEFT JOIN machines m ON m.id = t.machine_id
        {clause}
        ORDER BY COALESCE(t.close_time, t.open_time, t.created_at) DESC
    """
    rows = [row_to_dict(r) for r in conn.execute(sql, tuple(params)).fetchall()]
    now_ts = now()

    bucket_labels = {
        "lt_5m": "< 5 分钟",
        "m5_30": "5–30 分钟",
        "m30_2h": "30 分钟–2 小时",
        "ge_2h": "≥ 2 小时",
        "unknown": "时长未知",
    }
    by_hold = {k: _empty_group(v) for k, v in bucket_labels.items()}
    by_machine: dict[str, dict[str, Any]] = {}
    by_side: dict[str, dict[str, Any]] = {}
    by_category: dict[str, dict[str, Any]] = {
        k: _empty_group(k) for k in ("顺势", "反转", "其他", "未标注", "旧分类")
    }
    by_style: dict[str, dict[str, Any]] = {s: _empty_group(s) for s in ENTRY_STYLES}

    total = len(rows)
    closed = 0
    open_pos = 0
    wins = losses = breakeven = 0
    profit_sum = 0.0
    gross_win = 0.0
    gross_loss = 0.0
    hold_sum = 0.0
    hold_n = 0
    annotated = 0

    for t in rows:
        profit = t.get("profit")
        hold = _hold_seconds(t.get("open_time"), t.get("close_time"), now_ts)
        closed_flag = t.get("close_time") not in (None, "")
        if closed_flag:
            closed += 1
        else:
            open_pos += 1

        if profit is not None:
            p = float(profit)
            profit_sum += p
            if p > 1e-9:
                wins += 1
                gross_win += p
            elif p < -1e-9:
                losses += 1
                gross_loss += abs(p)
            else:
                breakeven += 1

        if hold is not None:
            hold_sum += hold
            hold_n += 1

        style = (t.get("entry_style") or "").strip()
        if style:
            annotated += 1

        cat = entry_category(style)
        _accumulate(by_category[cat], profit, hold)
        if style in by_style:
            _accumulate(by_style[style], profit, hold)
        elif style:
            key = f"旧 · {style}"
            if key not in by_style:
                by_style[key] = _empty_group(key)
            _accumulate(by_style[key], profit, hold)

        bk = _bucket_key(hold)
        _accumulate(by_hold[bk], profit, hold)

        mid = t.get("machine_id")
        mkey = str(mid) if mid is not None else "none"
        if mkey not in by_machine:
            name = t.get("machine_name") or ("未绑定" if mid is None else f"#{mid}")
            by_machine[mkey] = _empty_group(name)
        _accumulate(by_machine[mkey], profit, hold)

        side = (t.get("side") or "—").lower()
        if side not in by_side:
            by_side[side] = _empty_group("买入" if side == "buy" else ("卖出" if side == "sell" else side))
        _accumulate(by_side[side], profit, hold)

    decided = wins + losses
    avg_hold = (hold_sum / hold_n) if hold_n else None
    style_order = list(ENTRY_STYLES) + [k for k in by_style if k not in ENTRY_STYLES]
    return {
        "total": total,
        "closed": closed,
        "open_pos": open_pos,
        "wins": wins,
        "losses": losses,
        "breakeven": breakeven,
        "win_rate": (wins / decided * 100.0) if decided else 0.0,
        "profit": profit_sum,
        "avg_profit": (profit_sum / total) if total else 0.0,
        "gross_win": gross_win,
        "gross_loss": gross_loss,
        "profit_factor": (gross_win / gross_loss) if gross_loss > 1e-9 else (None if gross_win <= 0 else 999.0),
        "avg_hold_secs": avg_hold,
        "avg_hold": _fmt_hold_secs(avg_hold),
        "annotated": annotated,
        "unannotated": total - annotated,
        "by_hold": [_finalize_group(by_hold[k]) for k in ("lt_5m", "m5_30", "m30_2h", "ge_2h", "unknown") if by_hold[k]["count"] or k != "unknown"],
        "by_machine": sorted(
            [_finalize_group(g) for g in by_machine.values()],
            key=lambda x: x["profit"],
            reverse=True,
        ),
        "by_side": [_finalize_group(g) for g in by_side.values()],
        "by_category": [
            _finalize_group(by_category[k])
            for k in ("顺势", "反转", "其他", "未标注", "旧分类")
            if by_category[k]["count"] or k in ("顺势", "反转", "其他", "未标注")
        ],
        "by_style": [
            _finalize_group(by_style[k])
            for k in style_order
            if k in by_style and (by_style[k]["count"] or k in ENTRY_STYLES)
        ],
    }


def get_trade(conn: sqlite3.Connection, trade_id: int) -> Optional[dict]:
    r = conn.execute(
        """
        SELECT t.*, a.name AS account_name, m.name AS machine_name
        FROM trades t
        LEFT JOIN accounts a ON a.id = t.account_id
        LEFT JOIN machines m ON m.id = t.machine_id
        WHERE t.id=?
        """,
        (trade_id,),
    ).fetchone()
    d = row_to_dict(r)
    if d:
        try:
            d["snapshot"] = json.loads(d.get("snapshot_json") or "{}")
        except json.JSONDecodeError:
            d["snapshot"] = {}
    return d


def list_users(conn: sqlite3.Connection) -> list[dict]:
    cur = conn.execute(
        """
        SELECT u.*, a.name AS account_name
        FROM users u
        LEFT JOIN accounts a ON a.id = u.account_id
        ORDER BY u.id ASC
        """
    )
    return [row_to_dict(r) for r in cur.fetchall()]


def get_user_by_username(conn: sqlite3.Connection, username: str) -> Optional[dict]:
    return row_to_dict(
        conn.execute("SELECT * FROM users WHERE username=? AND enabled=1", (username,)).fetchone()
    )


def get_user(conn: sqlite3.Connection, user_id: int) -> Optional[dict]:
    return row_to_dict(conn.execute("SELECT * FROM users WHERE id=?", (user_id,)).fetchone())


def upsert_bars(conn: sqlite3.Connection, symbol: str, timeframe: str, bars: list[dict]) -> int:
    n = 0
    for b in bars:
        conn.execute(
            """
            INSERT INTO bars(symbol, timeframe, ts, open, high, low, close, volume)
            VALUES(?,?,?,?,?,?,?,?)
            ON CONFLICT(symbol, timeframe, ts) DO UPDATE SET
              open=excluded.open, high=excluded.high, low=excluded.low,
              close=excluded.close, volume=excluded.volume
            """,
            (
                symbol,
                timeframe,
                float(b["ts"]),
                float(b["open"]),
                float(b["high"]),
                float(b["low"]),
                float(b["close"]),
                float(b.get("volume") or 0),
            ),
        )
        n += 1
    return n


def get_bars(conn: sqlite3.Connection, symbol: str, timeframe: str, start_ts: float, end_ts: float) -> list[dict]:
    cur = conn.execute(
        """
        SELECT ts, open, high, low, close, volume FROM bars
        WHERE symbol=? AND timeframe=? AND ts>=? AND ts<=?
        ORDER BY ts ASC
        """,
        (symbol, timeframe, start_ts, end_ts),
    )
    return [row_to_dict(r) for r in cur.fetchall()]


def get_setting(conn: sqlite3.Connection, key: str, default: str = "") -> str:
    row = conn.execute("SELECT value FROM settings WHERE key=?", (key,)).fetchone()
    return row["value"] if row else default


def set_setting(conn: sqlite3.Connection, key: str, value: str) -> None:
    conn.execute(
        "INSERT INTO settings(key, value) VALUES(?, ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value",
        (key, value),
    )


def seed_if_empty() -> None:
    """仅确保管理员用户存在；交易账户不自动创建，须管理员手工建。"""
    from werkzeug.security import generate_password_hash

    from app.config import Config

    with get_db() as conn:
        # 兼容旧库：补 activations.created_by
        cols = {r["name"] for r in conn.execute("PRAGMA table_info(activations)").fetchall()}
        if "created_by" not in cols:
            try:
                conn.execute("ALTER TABLE activations ADD COLUMN created_by INTEGER")
            except sqlite3.OperationalError:
                pass

        n_users = conn.execute("SELECT COUNT(*) AS c FROM users").fetchone()["c"]
        if n_users == 0:
            # 迁移旧 settings 密码，或使用 .env 默认
            old_hash = get_setting(conn, "admin_password_hash")
            pwd_hash = old_hash or generate_password_hash(Config.ADMIN_PASSWORD)
            username = get_setting(conn, "admin_user") or Config.ADMIN_USER
            conn.execute(
                """INSERT INTO users(username, password_hash, role, account_id, enabled, created_at)
                   VALUES(?,?,?,?,?,?)""",
                (username, pwd_hash, "admin", None, 1, now()),
            )


def verify_login(username: str, password: str) -> Optional[dict]:
    from werkzeug.security import check_password_hash

    with get_db() as conn:
        user = get_user_by_username(conn, username)
        if not user:
            return None
        if not check_password_hash(user["password_hash"], password):
            return None
        return user


def change_user_password(user_id: int, old_password: str, new_password: str) -> None:
    change_user_credentials(user_id, old_password, new_username=None, new_password=new_password)


def change_user_credentials(
    user_id: int,
    old_password: str,
    new_username: Optional[str] = None,
    new_password: Optional[str] = None,
) -> dict:
    """校验当前密码后，可改用户名和/或密码。返回更新后的用户信息。"""
    from werkzeug.security import check_password_hash, generate_password_hash

    new_username = (new_username or "").strip() or None
    new_password = new_password if new_password else None
    if new_password is not None and len(new_password) < 6:
        raise ValueError("新密码至少 6 位")
    if new_username is not None and len(new_username) < 2:
        raise ValueError("用户名至少 2 位")

    with get_db() as conn:
        user = get_user(conn, user_id)
        if not user:
            raise ValueError("用户不存在")
        if not check_password_hash(user["password_hash"], old_password):
            raise ValueError("当前密码不正确")

        username_changed = bool(new_username and new_username != user["username"])
        password_changed = new_password is not None
        if not username_changed and not password_changed:
            raise ValueError("请修改用户名或填写新密码")

        if username_changed:
            taken = conn.execute(
                "SELECT id FROM users WHERE username=? AND id!=?",
                (new_username, user_id),
            ).fetchone()
            if taken:
                raise ValueError("用户名已被占用")
            conn.execute("UPDATE users SET username=? WHERE id=?", (new_username, user_id))

        if password_changed:
            conn.execute(
                "UPDATE users SET password_hash=? WHERE id=?",
                (generate_password_hash(new_password), user_id),
            )

        # 同步旧 settings 键（兼容）
        updated = get_user(conn, user_id)
        if updated and updated.get("role") == "admin":
            set_setting(conn, "admin_user", updated["username"])
            if password_changed:
                set_setting(conn, "admin_password_hash", updated["password_hash"])
        return updated or user


def delete_account(conn: sqlite3.Connection, account_id: int) -> None:
    """删除交易账户；若仍被用户绑定则拒绝。"""
    bound = conn.execute(
        "SELECT id, username FROM users WHERE account_id=?", (account_id,)
    ).fetchone()
    if bound:
        raise ValueError(f"账户已绑定用户 {bound['username']}，请先调整用户绑定")
    pending = conn.execute(
        "SELECT COUNT(*) AS c FROM activations WHERE account_id=? AND status='pending'",
        (account_id,),
    ).fetchone()["c"]
    if pending:
        raise ValueError("该账户仍有未使用的激活码，请先作废或用完后再删")
    if not conn.execute("SELECT id FROM accounts WHERE id=?", (account_id,)).fetchone():
        raise ValueError("账户不存在")
    conn.execute("DELETE FROM accounts WHERE id=?", (account_id,))


def delete_machine(conn: sqlite3.Connection, machine_id: int) -> None:
    """删除机器。在线中的不允许删；相关激活/成交仅解除 machine 引用。"""
    m = conn.execute("SELECT * FROM machines WHERE id=?", (machine_id,)).fetchone()
    if not m:
        raise ValueError("机器不存在")
    last = m["last_seen_at"]
    online = False
    try:
        online = bool(last and (now() - float(last)) <= ONLINE_SECONDS)
    except (TypeError, ValueError):
        online = False
    if online:
        raise ValueError("机器当前在线，请先关闭客户端后再删")
    conn.execute("UPDATE activations SET machine_id=NULL WHERE machine_id=?", (machine_id,))
    conn.execute("UPDATE trades SET machine_id=NULL WHERE machine_id=?", (machine_id,))
    conn.execute("DELETE FROM machines WHERE id=?", (machine_id,))


def create_user(
    conn: sqlite3.Connection,
    username: str,
    password: str,
    role: str,
    account_id: Optional[int],
) -> int:
    from werkzeug.security import generate_password_hash

    username = username.strip()
    if not username:
        raise ValueError("用户名不能为空")
    if len(password) < 6:
        raise ValueError("密码至少 6 位")
    if role not in ("admin", "user"):
        raise ValueError("角色无效")
    if role == "user":
        if not account_id:
            raise ValueError("普通用户必须绑定一个交易账户")
        if not conn.execute("SELECT id FROM accounts WHERE id=?", (account_id,)).fetchone():
            raise ValueError("交易账户不存在")
        taken = conn.execute(
            "SELECT id FROM users WHERE account_id=? AND role='user'",
            (account_id,),
        ).fetchone()
        if taken:
            raise ValueError("该交易账户已绑定其他用户（一人一账户）")
    else:
        account_id = None

    try:
        cur = conn.execute(
            """INSERT INTO users(username, password_hash, role, account_id, enabled, created_at)
               VALUES(?,?,?,?,?,?)""",
            (username, generate_password_hash(password), role, account_id, 1, now()),
        )
    except sqlite3.IntegrityError as exc:
        raise ValueError("用户名已存在或账户已被占用") from exc
    return int(cur.lastrowid)


def user_can_access_account(user: dict, account_id: Optional[int]) -> bool:
    if not user:
        return False
    if user.get("role") == "admin":
        return True
    return account_id is not None and int(user.get("account_id") or 0) == int(account_id)


# 兼容旧调用名
def verify_admin(username: str, password: str) -> bool:
    u = verify_login(username, password)
    return bool(u and u.get("role") == "admin")


def change_admin_password(old_password: str, new_password: str) -> None:
    """兼容：修改名为 admin 的管理员密码。"""
    with get_db() as conn:
        u = get_user_by_username(conn, "admin") or conn.execute(
            "SELECT * FROM users WHERE role='admin' ORDER BY id LIMIT 1"
        ).fetchone()
        if not u:
            raise ValueError("未找到管理员用户")
        uid = u["id"] if isinstance(u, sqlite3.Row) else u["id"]
    change_user_password(uid, old_password, new_password)