import ipaddress
import json
import re
import sqlite3
import time
from pathlib import Path
from typing import Optional

# Use a workspace-relative absolute path so code always opens the same DB,
# even if the current working directory changes at runtime.
DB_PATH = str(Path(__file__).resolve().parent.parent / "database" / "ozpay.db")


def get_conn():
	Path(DB_PATH).parent.mkdir(parents=True, exist_ok=True)
	print(f"[db_api] opening DB: {DB_PATH}")
	conn = sqlite3.connect(DB_PATH)
	conn.row_factory = sqlite3.Row
	# Ensure the accounts table exists so callers can assume schema is present
	cur = conn.cursor()
	cur.execute(
		"""
		CREATE TABLE IF NOT EXISTS accounts (
			device TEXT PRIMARY KEY,
			ip TEXT,
			port INTEGER,
			number TEXT,
			password TEXT,
			name TEXT,
			balance REAL,
			income REAL,
			outcome REAL,
			cards TEXT,
			blocked INTEGER DEFAULT 0
		)
		"""
	)
	cur.execute("PRAGMA table_info(accounts)")
	cols = {row[1] for row in cur.fetchall()}
	if "blocked" not in cols:
		cur.execute("ALTER TABLE accounts ADD COLUMN blocked INTEGER DEFAULT 0")
	if "card_flags" not in cols:
		cur.execute("ALTER TABLE accounts ADD COLUMN card_flags TEXT")
	if "tg_id" not in cols:
		cur.execute("ALTER TABLE accounts ADD COLUMN tg_id INTEGER")
	if "leased_at" not in cols:
		cur.execute("ALTER TABLE accounts ADD COLUMN leased_at REAL")
	if "lease_price" not in cols:
		cur.execute("ALTER TABLE accounts ADD COLUMN lease_price REAL")
	if "blocked_at" not in cols:
		cur.execute("ALTER TABLE accounts ADD COLUMN blocked_at REAL")
	cur.execute(
		"""
		CREATE TABLE IF NOT EXISTS app_users (
			tg_id INTEGER PRIMARY KEY,
			is_admin INTEGER DEFAULT 0,
			username TEXT,
			first_name TEXT,
			seen_at TEXT
		)
		"""
	)
	cur.execute("PRAGMA table_info(app_users)")
	user_cols = {row[1] for row in cur.fetchall()}
	if "usd_balance" not in user_cols:
		cur.execute("ALTER TABLE app_users ADD COLUMN usd_balance REAL DEFAULT 0")
	if "sms_chat_id" not in user_cols:
		cur.execute("ALTER TABLE app_users ADD COLUMN sms_chat_id INTEGER")
	cur.execute(
		"""
		CREATE TABLE IF NOT EXISTS app_settings (
			key TEXT PRIMARY KEY,
			value TEXT
		)
		"""
	)
	cur.execute(
		"""
		CREATE TABLE IF NOT EXISTS panel_workers (
			tg_id INTEGER PRIMARY KEY,
			added_at REAL NOT NULL
		)
		"""
	)
	cur.execute(
		"""
		CREATE TABLE IF NOT EXISTS ip_bans (
			ip TEXT PRIMARY KEY,
			added_at REAL NOT NULL
		)
		"""
	)
	cur.execute(
		"""
		CREATE TABLE IF NOT EXISTS balance_ops (
			id INTEGER PRIMARY KEY AUTOINCREMENT,
			tg_id INTEGER NOT NULL,
			amount REAL NOT NULL,
			reason TEXT NOT NULL,
			device TEXT,
			admin_id INTEGER,
			created_at REAL NOT NULL
		)
		"""
	)
	conn.commit()
	return conn


def init_db():
	conn = get_conn()
	cur = conn.cursor()
	cur.execute(
		"""
		CREATE TABLE IF NOT EXISTS accounts (
			device TEXT PRIMARY KEY,
			ip TEXT,
			port INTEGER,
			number TEXT,
			password TEXT,
			name TEXT,
			balance REAL,
			income REAL,
			outcome REAL,
			cards TEXT
		)
		"""
	)
	conn.commit()
	conn.close()


def create_device(device: str, **fields):
	"""Create a device row. Pass any of the columns as keyword args."""
	init_db()
	conn = get_conn()
	cur = conn.cursor()
	cur.execute(
		"INSERT INTO accounts (device, ip, port, number, password, name, balance, income, outcome, cards) VALUES (?,?,?,?,?,?,?,?,?,?)",
		(
			device,
			fields.get('ip'),
			fields.get('port'),
			fields.get('number'),
			fields.get('password'),
			fields.get('name'),
			fields.get('balance'),
			fields.get('income'),
			fields.get('outcome'),
			fields.get('cards'),
		),
	)
	conn.commit()
	conn.close()


def find_device_by_ip_port(ip: str, port) -> Optional[dict]:
	if not ip or port in (None, ""):
		return None
	try:
		port = int(port)
	except (TypeError, ValueError):
		return None
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("SELECT * FROM accounts WHERE ip = ? AND port = ?", (ip, port))
	row = cur.fetchone()
	conn.close()
	if not row:
		return None
	return dict(row)


def delete_device(device: str) -> bool:
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("DELETE FROM accounts WHERE device = ?", (device,))
	conn.commit()
	changed = cur.rowcount
	conn.close()
	return changed > 0


def get_device(device: str):
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("SELECT * FROM accounts WHERE device = ?", (device,))
	row = cur.fetchone()
	conn.close()
	if not row:
		return None
	return dict(row)


def list_devices():
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("SELECT * FROM accounts")
	rows = cur.fetchall()
	conn.close()
	return [dict(r) for r in rows]


def list_devices_by_tg(tg_id: int):
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("SELECT * FROM accounts WHERE tg_id = ?", (int(tg_id),))
	rows = cur.fetchall()
	conn.close()
	return [dict(r) for r in rows]


def _is_blank(value) -> bool:
	return value is None or str(value).strip() == "" or str(value).strip() == "0"


def count_free_slots() -> int:
	conn = get_conn()
	cur = conn.cursor()
	cur.execute(
		"""
		SELECT COUNT(*) FROM accounts
		WHERE (number IS NULL OR TRIM(number) = '')
		  AND (tg_id IS NULL OR tg_id = 0)
		"""
	)
	n = cur.fetchone()[0]
	conn.close()
	return int(n)


def find_unlinked_owned(tg_id: int) -> Optional[dict]:
	conn = get_conn()
	cur = conn.cursor()
	cur.execute(
		"""
		SELECT * FROM accounts
		WHERE tg_id = ?
		  AND (number IS NULL OR TRIM(number) = '')
		ORDER BY device
		LIMIT 1
		""",
		(int(tg_id),),
	)
	row = cur.fetchone()
	conn.close()
	if not row:
		return None
	return dict(row)


def claim_free_device(tg_id: int) -> Optional[dict]:
	"""Занять свободный слот под пользователя. Атомарно, без гонки."""
	tg_id = int(tg_id)
	owned = find_unlinked_owned(tg_id)
	if owned:
		return owned
	conn = get_conn()
	device = None
	try:
		cur = conn.cursor()
		cur.execute("BEGIN IMMEDIATE")
		cur.execute(
			"""
			SELECT device FROM accounts
			WHERE (number IS NULL OR TRIM(number) = '')
			  AND (tg_id IS NULL OR tg_id = 0)
			ORDER BY device
			LIMIT 1
			"""
		)
		row = cur.fetchone()
		if not row:
			conn.rollback()
			return None
		device = row["device"]
		cur.execute("UPDATE accounts SET tg_id = ? WHERE device = ?", (tg_id, device))
		conn.commit()
	except Exception:
		conn.rollback()
		raise
	finally:
		conn.close()
	return get_device(device) if device else None


def release_device_owner(device: str) -> bool:
	"""Снять владельца, если ЛК ещё не привязан (отмена входа)."""
	row = get_device(device)
	if not row:
		return False
	if not _is_blank(row.get("number")):
		return False
	return update_device(device, {"tg_id": None})


def bind_device_owner(device: str, tg_id: int) -> bool:
	return update_device(device, {"tg_id": int(tg_id)})


def get_app_user(tg_id: int) -> Optional[dict]:
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("SELECT * FROM app_users WHERE tg_id = ?", (int(tg_id),))
	row = cur.fetchone()
	conn.close()
	if not row:
		return None
	return dict(row)


def count_admins() -> int:
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("SELECT COUNT(*) FROM app_users WHERE is_admin = 1")
	n = cur.fetchone()[0]
	conn.close()
	return int(n)


def upsert_app_user(
	tg_id: int,
	username: Optional[str] = None,
	first_name: Optional[str] = None,
	is_admin: Optional[bool] = None,
) -> dict:
	from datetime import datetime, timezone

	tg_id = int(tg_id)
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("SELECT * FROM app_users WHERE tg_id = ?", (tg_id,))
	existing = cur.fetchone()
	now = datetime.now(timezone.utc).isoformat()
	if existing:
		admin_val = existing["is_admin"]
		if is_admin is True:
			admin_val = 1
		cur.execute(
			"""
			UPDATE app_users
			SET username = COALESCE(?, username),
			    first_name = COALESCE(?, first_name),
			    is_admin = ?,
			    seen_at = ?
			WHERE tg_id = ?
			""",
			(username, first_name, admin_val, now, tg_id),
		)
	else:
		admin_val = 1 if is_admin else 0
		cur.execute(
			"""
			INSERT INTO app_users (tg_id, is_admin, username, first_name, seen_at)
			VALUES (?, ?, ?, ?, ?)
			""",
			(tg_id, admin_val, username, first_name, now),
		)
	conn.commit()
	conn.close()
	return get_app_user(tg_id)


def parse_sms_chat_id(value) -> Optional[int]:
	if value in (None, "", 0, "0"):
		return None
	try:
		chat_id = int(str(value).strip())
	except (TypeError, ValueError):
		return None
	if chat_id == 0:
		return None
	return chat_id


def get_sms_chat_id(tg_id: int) -> Optional[int]:
	row = get_app_user(int(tg_id))
	if not row:
		return None
	return parse_sms_chat_id(row.get("sms_chat_id"))


def set_sms_chat_id(tg_id: int, chat_id: Optional[int]) -> Optional[int]:
	tg_id = int(tg_id)
	upsert_app_user(tg_id)
	parsed = parse_sms_chat_id(chat_id)
	if parsed == tg_id:
		parsed = None
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("UPDATE app_users SET sms_chat_id = ? WHERE tg_id = ?", (parsed, tg_id))
	conn.commit()
	conn.close()
	return parsed


def resolve_sms_destination(owner_tg_id: int) -> tuple[int, bool]:
	owner = int(owner_tg_id)
	custom = get_sms_chat_id(owner)
	if custom is None:
		return owner, False
	return custom, True


SLOT_PRICE_KEY = "slot_price_usd"
DEFAULT_SLOT_PRICE = 20.0


def get_app_setting(key: str, default: Optional[str] = None) -> Optional[str]:
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("SELECT value FROM app_settings WHERE key = ?", (str(key),))
	row = cur.fetchone()
	conn.close()
	if not row or row[0] is None:
		return default
	return str(row[0])


def set_app_setting(key: str, value: str) -> None:
	conn = get_conn()
	cur = conn.cursor()
	cur.execute(
		"INSERT INTO app_settings (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value",
		(str(key), str(value)),
	)
	conn.commit()
	conn.close()


def get_slot_price() -> float:
	raw = get_app_setting(SLOT_PRICE_KEY)
	if raw in (None, ""):
		return DEFAULT_SLOT_PRICE
	try:
		price = float(str(raw).replace(",", ".").replace(" ", ""))
	except (TypeError, ValueError):
		return DEFAULT_SLOT_PRICE
	return max(0.0, round(price, 2))


def set_slot_price(price: float) -> float:
	value = round(float(price), 2)
	if value < 0:
		raise ValueError("Цена не может быть отрицательной")
	set_app_setting(SLOT_PRICE_KEY, str(value))
	return value


def get_usd_balance(tg_id: int) -> float:
	row = get_app_user(int(tg_id))
	if not row:
		return 0.0
	try:
		return round(float(row.get("usd_balance") or 0), 2)
	except (TypeError, ValueError):
		return 0.0


def adjust_usd_balance(
	tg_id: int,
	amount: float,
	*,
	reason: str,
	device: Optional[str] = None,
	admin_id: Optional[int] = None,
) -> float:
	tg_id = int(tg_id)
	amount = round(float(amount), 2)
	if amount == 0:
		return get_usd_balance(tg_id)
	upsert_app_user(tg_id)
	conn = get_conn()
	try:
		cur = conn.cursor()
		cur.execute("BEGIN IMMEDIATE")
		cur.execute("SELECT usd_balance FROM app_users WHERE tg_id = ?", (tg_id,))
		row = cur.fetchone()
		current = 0.0
		if row and row[0] not in (None, ""):
			try:
				current = float(row[0])
			except (TypeError, ValueError):
				current = 0.0
		new_balance = round(current + amount, 2)
		if new_balance < -0.0001:
			conn.rollback()
			raise ValueError("Недостаточно средств")
		cur.execute("UPDATE app_users SET usd_balance = ? WHERE tg_id = ?", (new_balance, tg_id))
		cur.execute(
			"""
			INSERT INTO balance_ops (tg_id, amount, reason, device, admin_id, created_at)
			VALUES (?, ?, ?, ?, ?, ?)
			""",
			(tg_id, amount, str(reason or "")[:40], device, admin_id, time.time()),
		)
		conn.commit()
	except Exception:
		conn.rollback()
		raise
	finally:
		conn.close()
	return new_balance


def list_app_users() -> list[dict]:
	conn = get_conn()
	cur = conn.cursor()
	cur.execute(
		"""
		SELECT tg_id, username, first_name, is_admin, usd_balance, seen_at
		FROM app_users
		ORDER BY seen_at DESC
		"""
	)
	rows = [dict(row) for row in cur.fetchall()]
	conn.close()
	return rows


def _mark_device_leased(device: str, price: float) -> None:
	update_device(device, {"leased_at": time.time(), "lease_price": round(float(price), 2)})


def refund_client_lease(device: str) -> None:
	row = get_device(device)
	if not row:
		return
	try:
		tg_id = int(row.get("tg_id") or 0)
	except (TypeError, ValueError):
		tg_id = 0
	try:
		price = float(row.get("lease_price") or 0)
	except (TypeError, ValueError):
		price = 0.0
	if tg_id and price > 0 and row.get("leased_at"):
		adjust_usd_balance(tg_id, price, reason="refund", device=device)
	update_device(device, {"leased_at": None, "lease_price": None})


def start_client_lease(tg_id: int) -> dict:
	tg_id = int(tg_id)
	price = get_slot_price()
	upsert_app_user(tg_id)
	conn = get_conn()
	device = None
	try:
		cur = conn.cursor()
		cur.execute("BEGIN IMMEDIATE")
		cur.execute(
			"""
			SELECT * FROM accounts
			WHERE tg_id = ?
			  AND (number IS NULL OR TRIM(number) = '')
			ORDER BY device
			LIMIT 1
			""",
			(tg_id,),
		)
		owned = cur.fetchone()
		if owned and owned["leased_at"]:
			conn.commit()
			return dict(owned)
		cur.execute("SELECT usd_balance FROM app_users WHERE tg_id = ?", (tg_id,))
		bal_row = cur.fetchone()
		balance = 0.0
		if bal_row and bal_row[0] not in (None, ""):
			try:
				balance = float(bal_row[0])
			except (TypeError, ValueError):
				balance = 0.0
		if owned:
			device = owned["device"]
		else:
			cur.execute(
				"""
				SELECT device FROM accounts
				WHERE (number IS NULL OR TRIM(number) = '')
				  AND (tg_id IS NULL OR tg_id = 0)
				ORDER BY device
				LIMIT 1
				"""
			)
			free = cur.fetchone()
			if not free:
				conn.rollback()
				raise ValueError("Нет свободных слотов")
			device = free["device"]
		if price > 0 and balance + 0.0001 < price:
			conn.rollback()
			raise ValueError(f"Недостаточно средств. Нужно {price:.2f} $, на балансе {balance:.2f} $")
		new_balance = round(balance - price, 2) if price > 0 else round(balance, 2)
		if price > 0:
			cur.execute("UPDATE app_users SET usd_balance = ? WHERE tg_id = ?", (new_balance, tg_id))
			cur.execute(
				"""
				INSERT INTO balance_ops (tg_id, amount, reason, device, admin_id, created_at)
				VALUES (?, ?, ?, ?, ?, ?)
				""",
				(tg_id, -price, "rent", device, None, time.time()),
			)
		cur.execute(
			"UPDATE accounts SET tg_id = ?, leased_at = ?, lease_price = ? WHERE device = ?",
			(tg_id, time.time(), price, device),
		)
		conn.commit()
	except Exception:
		conn.rollback()
		raise
	finally:
		conn.close()
	row = get_device(device)
	if not row:
		raise ValueError("Нет свободных слотов")
	return row


def ensure_client_lease(tg_id: int, device: str) -> dict:
	row = get_device(device)
	if not row:
		raise ValueError("Устройство не найдено")
	if row.get("leased_at"):
		return row
	price = get_slot_price()
	if price > 0:
		try:
			adjust_usd_balance(int(tg_id), -price, reason="rent", device=device)
		except ValueError as exc:
			raise ValueError(
				f"Недостаточно средств. Нужно {price:.2f} $, на балансе {get_usd_balance(tg_id):.2f} $"
			) from exc
	_mark_device_leased(device, price)
	return get_device(device) or row


def list_lock_logout_due(after_seconds: float) -> list[dict]:
	cutoff = time.time() - float(after_seconds)
	conn = get_conn()
	cur = conn.cursor()
	cur.execute(
		"""
		SELECT * FROM accounts
		WHERE blocked = 1
		  AND leased_at IS NOT NULL
		  AND blocked_at IS NOT NULL
		  AND blocked_at <= ?
		  AND number IS NOT NULL AND TRIM(number) != ''
		ORDER BY blocked_at
		""",
		(cutoff,),
	)
	rows = [dict(row) for row in cur.fetchall()]
	conn.close()
	return rows


def update_device(device: str, data: dict) -> bool:
	"""Generic update by device. `data` keys should be column names."""
	allowed = ['ip', 'port', 'number', 'password', 'name', 'balance', 'income', 'outcome', 'cards', 'blocked', 'card_flags', 'tg_id', 'leased_at', 'lease_price', 'blocked_at']
	set_parts = []
	params = []
	for k, v in data.items():
		if k in allowed:
			set_parts.append(f"{k} = ?")
			params.append(v)
	if not set_parts:
		return False
	params.append(device)
	conn = get_conn()
	cur = conn.cursor()
	cur.execute(f"UPDATE accounts SET {', '.join(set_parts)} WHERE device = ?", params)
	conn.commit()
	changed = cur.rowcount
	conn.close()
	return changed > 0


def rename_device(old: str, new: str) -> bool:
	old = (old or "").strip()
	new = (new or "").strip()
	if not old or not new or old == new:
		return False
	conn = get_conn()
	cur = conn.cursor()
	try:
		cur.execute("UPDATE accounts SET device = ? WHERE device = ?", (new, old))
		conn.commit()
		changed = cur.rowcount
	except sqlite3.IntegrityError:
		conn.close()
		return False
	conn.close()
	return changed > 0


# --- Simple get/update helpers for each column ---


def _get_field(device: str, field: str):
	conn = get_conn()
	cur = conn.cursor()
	cur.execute(f"SELECT {field} FROM accounts WHERE device = ?", (device,))
	row = cur.fetchone()
	conn.close()
	if not row:
		return None
	return row[0]


def _update_field(device: str, field: str, value) -> bool:
	conn = get_conn()
	cur = conn.cursor()
	cur.execute(f"UPDATE accounts SET {field} = ? WHERE device = ?", (value, device))
	conn.commit()
	changed = cur.rowcount
	conn.close()
	return changed > 0


# ip
def get_ip(device: str):
	return _get_field(device, 'ip')


def update_ip(device: str, ip: str) -> bool:
	return _update_field(device, 'ip', ip)


# port
def get_port(device: str):
	return _get_field(device, 'port')


def update_port(device: str, port: int) -> bool:
	return _update_field(device, 'port', port)


# number (phone) - user requested phone helpers
def get_number(device: str):
	return _get_field(device, 'number')


def update_number(device: str, number: str) -> bool:
	return _update_field(device, 'number', number)


def get_phone(device: str):
	return get_number(device)


def update_phone(device: str, phone: str) -> bool:
	return update_number(device, phone)


# name
def get_name(device: str):
	return _get_field(device, 'name')


def update_name(device: str, name: str) -> bool:
	return _update_field(device, 'name', name)


# balance
def get_balance(device: str):
	return _get_field(device, 'balance')


def update_balance(device: str, balance: float) -> bool:
	return _update_field(device, 'balance', balance)


# income
def get_income(device: str):
	return _get_field(device, 'income')


def update_income(device: str, income: float) -> bool:
	return _update_field(device, 'income', income)


# outcome
def get_outcome(device: str):
	return _get_field(device, 'outcome')


def update_outcome(device: str, outcome: float) -> bool:
	return _update_field(device, 'outcome', outcome)


# cards
def get_cards(device: str):
	return _get_field(device, 'cards')


def update_cards(device: str, cards: str) -> bool:
	return _update_field(device, 'cards', cards)


def get_card_flags(device: str):
	return _get_field(device, 'card_flags')


def update_card_flags(device: str, card_flags: str) -> bool:
	return _update_field(device, 'card_flags', card_flags)


# blocked (Ozon operations suspended)
def get_blocked(device: str):
	return _get_field(device, 'blocked')


def update_blocked(device: str, blocked: int) -> bool:
	blocked = int(bool(blocked))
	row = get_device(device)
	if not row:
		return False
	was = int(bool(row.get("blocked")))
	payload = {"blocked": blocked}
	if blocked and not was:
		payload["blocked_at"] = time.time()
	elif not blocked:
		payload["blocked_at"] = None
	return update_device(device, payload)


# password
def get_password(device: str):
	return _get_field(device, 'password')


def update_password(device: str, password: str) -> bool:
	return _update_field(device, 'password', password)


CARD_FLAG_KEYS = ("beeline", "yapay", "mts")
SYSTEM_BY_ARG = {"я": "yapay", "б": "beeline", "м": "mts"}


def _card_digits(value) -> str:
	return re.sub(r"\D", "", str(value or ""))


def _format_card_number(number: str) -> str:
	digits = _card_digits(number)
	if not digits:
		return number or ""
	return " ".join(digits[i:i + 4] for i in range(0, len(digits), 4))


def _format_expiry(raw: str) -> str:
	text = (raw or "").strip()
	if not text:
		return ""
	if re.fullmatch(r"(0[1-9]|1[0-2])/\d{2}(?:\d{2})?", text):
		return text[:5] if len(text) > 5 else text
	digits = _card_digits(text)
	if len(digits) == 4:
		month = int(digits[:2]) if digits[:2].isdigit() else 0
		if 1 <= month <= 12:
			return f"{digits[:2]}/{digits[2:]}"
	return ""


def _empty_flags() -> dict:
	return {key: False for key in CARD_FLAG_KEYS}


def _flags_from_value(value) -> dict:
	if not isinstance(value, dict):
		return _empty_flags()
	return {key: bool(value.get(key)) for key in CARD_FLAG_KEYS}


def parse_card_flags(raw) -> dict:
	if not raw:
		return {}
	if isinstance(raw, dict):
		data = raw
	else:
		try:
			data = json.loads(raw)
		except (TypeError, ValueError, json.JSONDecodeError):
			return {}
	if not isinstance(data, dict):
		return {}
	flags = {}
	for key, value in data.items():
		digits = _card_digits(key)
		if not digits:
			continue
		flags[digits] = _flags_from_value(value)
	return flags


def parse_cards(raw: Optional[str], flags_raw=None) -> list:
	"""cards в БД: number/expiry/cvv, карты через ':'. Старый формат: number/last4/cvv."""
	flags_map = parse_card_flags(flags_raw)
	if not raw:
		return []
	cards = []
	for chunk in str(raw).split(":"):
		chunk = chunk.strip()
		if not chunk:
			continue
		parts = chunk.split("/")
		number = parts[0] if parts else ""
		middle = parts[1] if len(parts) > 1 else ""
		cvv = parts[2] if len(parts) > 2 else ""
		number_digits = _card_digits(number)
		middle_digits = _card_digits(middle)
		if number_digits and middle_digits == number_digits[-4:]:
			expiry = ""
		else:
			expiry = _format_expiry(middle)
		flags = flags_map.get(number_digits) or _empty_flags()
		card = {
			"number": _format_card_number(number),
			"expiry": expiry,
			"cvv": cvv,
		}
		card.update(flags)
		cards.append(card)
	return cards


def _to_balance(value) -> float:
	if value is None or value == "":
		return 0.0
	if isinstance(value, (int, float)):
		return float(value)
	text = str(value).replace("\xa0", " ").replace(" ", "").replace(",", ".")
	text = re.sub(r"[^\d.]", "", text)
	try:
		return float(text) if text else 0.0
	except ValueError:
		return 0.0


def pick_card(*, smaller: bool = True, system: Optional[str] = None) -> Optional[dict]:
	"""Карта со всей панели: min/max баланс аккаунта, без галочки выбранной системы."""
	if system is not None and system not in CARD_FLAG_KEYS:
		return None
	candidates = []
	for row in list_devices():
		if row.get("blocked"):
			continue
		balance = _to_balance(row.get("balance"))
		device_id = row.get("device") or ""
		for card in parse_cards(row.get("cards"), row.get("card_flags")):
			number_digits = _card_digits(card.get("number"))
			if not number_digits:
				continue
			if system and card.get(system):
				continue
			candidates.append({
				"device": device_id,
				"name": (row.get("name") or "").strip(),
				"balance": balance,
				"number": card.get("number") or "",
				"number_digits": number_digits,
				"expiry": card.get("expiry") or "",
				"cvv": card.get("cvv") or "",
				"system": system,
			})
	if not candidates:
		return None
	candidates.sort(
		key=lambda item: (item["balance"], item["device"], item["number_digits"]),
		reverse=not smaller,
	)
	return candidates[0]


def set_card_flag(device: str, number: str, flag: str, value: bool = True) -> bool:
	if flag not in CARD_FLAG_KEYS:
		return False
	row = get_device(device)
	if not row:
		return False
	flags = parse_card_flags(row.get("card_flags"))
	digits = _card_digits(number)
	if not digits:
		return False
	current = flags.get(digits) or _empty_flags()
	current[flag] = bool(value)
	if any(current.get(key) for key in CARD_FLAG_KEYS):
		flags[digits] = current
	else:
		flags.pop(digits, None)
	return update_card_flags(device, json.dumps(flags, ensure_ascii=False))


_panel_workers_cache: Optional[set[int]] = None


def _load_panel_workers() -> set[int]:
	global _panel_workers_cache
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("SELECT tg_id FROM panel_workers")
	ids = set()
	for row in cur.fetchall():
		if not row or row[0] is None:
			continue
		try:
			ids.add(int(row[0]))
		except (TypeError, ValueError):
			continue
	conn.close()
	_panel_workers_cache = ids
	return ids


def list_panel_workers() -> list[int]:
	global _panel_workers_cache
	if _panel_workers_cache is None:
		_load_panel_workers()
	return sorted(_panel_workers_cache)


def is_panel_worker(tg_id) -> bool:
	try:
		uid = int(tg_id)
	except (TypeError, ValueError):
		return False
	global _panel_workers_cache
	if _panel_workers_cache is None:
		_load_panel_workers()
	return uid in _panel_workers_cache


def add_panel_worker(tg_id: int) -> bool:
	try:
		uid = int(tg_id)
	except (TypeError, ValueError):
		return False
	if uid <= 0:
		return False
	conn = get_conn()
	cur = conn.cursor()
	cur.execute(
		"INSERT OR IGNORE INTO panel_workers (tg_id, added_at) VALUES (?, ?)",
		(uid, time.time()),
	)
	conn.commit()
	conn.close()
	global _panel_workers_cache
	if _panel_workers_cache is None:
		_load_panel_workers()
	else:
		_panel_workers_cache.add(uid)
	return True


def remove_panel_worker(tg_id: int) -> bool:
	try:
		uid = int(tg_id)
	except (TypeError, ValueError):
		return False
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("DELETE FROM panel_workers WHERE tg_id = ?", (uid,))
	conn.commit()
	changed = cur.rowcount > 0
	conn.close()
	global _panel_workers_cache
	if _panel_workers_cache is None:
		_load_panel_workers()
	else:
		_panel_workers_cache.discard(uid)
	return changed


WATCH_SETTINGS_KEY = "userbot_watch"
_watch_cache: Optional[dict] = None


def _default_watch_settings() -> dict:
	return {
		"enabled": True,
		"keywords": [],
		"chats": [],
		"alert_chat_id": None,
	}


def _normalize_watch_keyword(value) -> str:
	return " ".join(str(value or "").split())


def _normalize_watch_chat(value) -> Optional[dict]:
	if not isinstance(value, dict):
		return None
	raw_id = value.get("id")
	if raw_id in (None, ""):
		return None
	try:
		chat_id = int(raw_id)
	except (TypeError, ValueError):
		return None
	title = " ".join(str(value.get("title") or chat_id).split())[:120]
	kind = str(value.get("kind") or "").strip()[:16]
	return {"id": chat_id, "title": title or str(chat_id), "kind": kind}


def _parse_watch_settings(raw) -> dict:
	data = _default_watch_settings()
	if not raw:
		return data
	if isinstance(raw, dict):
		payload = raw
	else:
		try:
			payload = json.loads(raw)
		except (TypeError, ValueError, json.JSONDecodeError):
			return data
	if not isinstance(payload, dict):
		return data
	if "enabled" in payload:
		data["enabled"] = bool(payload.get("enabled"))
	keywords = []
	seen = set()
	for item in payload.get("keywords") or []:
		word = _normalize_watch_keyword(item)
		key = word.casefold()
		if not word or key in seen:
			continue
		seen.add(key)
		keywords.append(word[:80])
		if len(keywords) >= 50:
			break
	data["keywords"] = keywords
	chats = []
	seen_ids = set()
	for item in payload.get("chats") or []:
		chat = _normalize_watch_chat(item)
		if not chat or chat["id"] in seen_ids:
			continue
		seen_ids.add(chat["id"])
		chats.append(chat)
		if len(chats) >= 100:
			break
	data["chats"] = chats
	alert = payload.get("alert_chat_id")
	if alert in (None, ""):
		data["alert_chat_id"] = None
	else:
		try:
			data["alert_chat_id"] = int(alert)
		except (TypeError, ValueError):
			data["alert_chat_id"] = None
	return data


def get_watch_settings() -> dict:
	global _watch_cache
	if _watch_cache is not None:
		return json.loads(json.dumps(_watch_cache))
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("SELECT value FROM app_settings WHERE key = ?", (WATCH_SETTINGS_KEY,))
	row = cur.fetchone()
	conn.close()
	data = _parse_watch_settings(row[0] if row else None)
	_watch_cache = data
	return json.loads(json.dumps(data))


def save_watch_settings(payload: dict) -> dict:
	global _watch_cache
	current = get_watch_settings()
	if not isinstance(payload, dict):
		payload = {}
	merged = {
		"enabled": payload["enabled"] if "enabled" in payload else current["enabled"],
		"keywords": payload["keywords"] if "keywords" in payload else current["keywords"],
		"chats": payload["chats"] if "chats" in payload else current["chats"],
		"alert_chat_id": payload["alert_chat_id"] if "alert_chat_id" in payload else current["alert_chat_id"],
	}
	data = _parse_watch_settings(merged)
	conn = get_conn()
	cur = conn.cursor()
	cur.execute(
		"INSERT INTO app_settings (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value",
		(WATCH_SETTINGS_KEY, json.dumps(data, ensure_ascii=False)),
	)
	conn.commit()
	conn.close()
	_watch_cache = data
	return json.loads(json.dumps(data))


_ip_bans_cache: Optional[list[str]] = None
_ip_ban_exact: Optional[set[str]] = None
_ip_ban_nets: Optional[list] = None
MAX_IP_BANS = 500


def normalize_ip_ban(value) -> Optional[str]:
	text = str(value or "").strip()
	if not text:
		return None
	if text.count(",") == 3 and "." not in text:
		text = text.replace(",", ".")
	try:
		if "/" in text:
			net = ipaddress.ip_network(text, strict=False)
			if net.version != 4:
				return None
			return str(net)
		addr = ipaddress.ip_address(text)
		if addr.version != 4:
			return None
		return str(addr)
	except ValueError:
		return None


def parse_ip_ban_inputs(raw) -> list[str]:
	text = str(raw or "")
	found: list[str] = []
	seen: set[str] = set()
	for part in re.split(r"[\s;]+", text):
		part = part.strip().strip(",")
		if not part:
			continue
		ip = normalize_ip_ban(part)
		if not ip or ip in seen:
			continue
		seen.add(ip)
		found.append(ip)
		if len(found) >= MAX_IP_BANS:
			break
	return found


def _set_ip_ban_cache(ips: list[str]) -> None:
	global _ip_bans_cache, _ip_ban_exact, _ip_ban_nets
	_ip_bans_cache = list(ips)
	exact: set[str] = set()
	nets = []
	for item in ips:
		if "/" in item:
			try:
				nets.append(ipaddress.ip_network(item, strict=False))
			except ValueError:
				continue
		else:
			exact.add(item)
	_ip_ban_exact = exact
	_ip_ban_nets = nets


def _load_ip_bans() -> list[str]:
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("SELECT ip FROM ip_bans ORDER BY ip")
	ips = []
	for row in cur.fetchall():
		if not row or not row[0]:
			continue
		ip = normalize_ip_ban(row[0])
		if ip:
			ips.append(ip)
	conn.close()
	_set_ip_ban_cache(ips)
	return ips


def list_ip_bans() -> list[str]:
	global _ip_bans_cache
	if _ip_bans_cache is None:
		_load_ip_bans()
	return list(_ip_bans_cache)


def is_ip_banned(ip: str) -> bool:
	candidate = str(ip or "").strip()
	if candidate.startswith("::ffff:"):
		candidate = candidate[7:]
	addr = None
	normalized = normalize_ip_ban(candidate)
	try:
		addr = ipaddress.ip_address(normalized or candidate)
	except ValueError:
		return False
	if addr.version != 4:
		return False
	global _ip_bans_cache
	if _ip_bans_cache is None:
		_load_ip_bans()
	if str(addr) in (_ip_ban_exact or ()):
		return True
	return any(addr in net for net in (_ip_ban_nets or ()))


def add_ip_ban(ip: str) -> Optional[str]:
	normalized = normalize_ip_ban(ip)
	if not normalized:
		return None
	current = list_ip_bans()
	if normalized not in current and len(current) >= MAX_IP_BANS:
		return None
	conn = get_conn()
	cur = conn.cursor()
	cur.execute(
		"INSERT OR IGNORE INTO ip_bans (ip, added_at) VALUES (?, ?)",
		(normalized, time.time()),
	)
	conn.commit()
	conn.close()
	global _ip_bans_cache
	if _ip_bans_cache is None:
		_load_ip_bans()
	elif normalized not in _ip_bans_cache:
		updated = sorted(_ip_bans_cache + [normalized])
		_set_ip_ban_cache(updated)
	return normalized


def remove_ip_ban(ip: str) -> bool:
	normalized = normalize_ip_ban(ip)
	if not normalized:
		return False
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("DELETE FROM ip_bans WHERE ip = ?", (normalized,))
	conn.commit()
	changed = cur.rowcount > 0
	conn.close()
	global _ip_bans_cache
	if _ip_bans_cache is None:
		_load_ip_bans()
	elif changed:
		_set_ip_ban_cache([item for item in _ip_bans_cache if item != normalized])
	return changed


def ip_covers_client(ban: str, client_ip: str) -> bool:
	target = normalize_ip_ban(client_ip)
	rule = normalize_ip_ban(ban)
	if not target or not rule:
		return False
	try:
		addr = ipaddress.ip_address(target)
		if "/" in rule:
			return addr in ipaddress.ip_network(rule, strict=False)
		return str(addr) == rule
	except ValueError:
		return False