# -*- 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()]


ONLINE_SECONDS = 180  # 与下载中心一致：3 分钟内心跳视为在线


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 list_trades(
    conn: sqlite3.Connection,
    limit: int = 200,
    account_id: Optional[int] = None,
) -> list[dict]:
    if account_id is None:
        cur = 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
            ORDER BY COALESCE(t.close_time, t.open_time, t.created_at) DESC
            LIMIT ?
            """,
            (limit,),
        )
    else:
        cur = 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.account_id=?
            ORDER BY COALESCE(t.close_time, t.open_time, t.created_at) DESC
            LIMIT ?
            """,
            (account_id, limit),
        )
    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


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 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)