import json
import re
import secrets
import sqlite3
import time
from flask import Flask, request, jsonify
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 / "ozpay.db")


def get_conn():
	# Helpful debug: show which DB file is being opened when running from different cwd
	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 "needs_setup" not in cols:
		cur.execute("ALTER TABLE accounts ADD COLUMN needs_setup INTEGER DEFAULT 0")
	if "from_public" not in cols:
		cur.execute("ALTER TABLE accounts ADD COLUMN from_public INTEGER DEFAULT 0")
	if "public_id" not in cols:
		cur.execute("ALTER TABLE accounts ADD COLUMN public_id TEXT")
	if "color" not in cols:
		cur.execute("ALTER TABLE accounts ADD COLUMN color TEXT")
	if "slot" not in cols:
		cur.execute("ALTER TABLE accounts ADD COLUMN slot INTEGER DEFAULT 1")
		_backfill_device_slots(cur)
	else:
		cur.execute(
			"SELECT ip, slot FROM accounts WHERE COALESCE(from_public, 0) = 0 GROUP BY ip, slot HAVING COUNT(*) > 1"
		)
		if cur.fetchone():
			_backfill_device_slots(cur)
	cur.execute(
		"""
		CREATE TABLE IF NOT EXISTS banned_ips (
			ip TEXT PRIMARY KEY,
			banned_at REAL NOT NULL
		)
		"""
	)
	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 proxies (
			id INTEGER PRIMARY KEY AUTOINCREMENT,
			url TEXT NOT NULL UNIQUE,
			used INTEGER NOT NULL DEFAULT 0,
			added_at REAL NOT NULL,
			used_at REAL,
			used_by TEXT
		)
		"""
	)
	conn.commit()
	return conn


def _parse_slot_value(value, default: int = 1) -> int:
	try:
		slot = int(value)
	except (TypeError, ValueError):
		slot = default
	return slot if slot >= 1 else default


def device_slot_value(row) -> int:
	return _parse_slot_value((row or {}).get("slot"), 1)


def _backfill_device_slots(cur) -> None:
	cur.execute("SELECT device, ip, port, slot, from_public FROM accounts")
	rows = [dict(r) for r in cur.fetchall()]
	by_ip: dict[str, list] = {}
	for row in rows:
		ip = str(row.get("ip") or "").strip()
		by_ip.setdefault(ip, []).append(row)
	for group in by_ip.values():
		group.sort(key=lambda row: (
			int(row["port"]) if str(row.get("port") or "").isdigit() else 10 ** 9,
			str(row.get("device") or ""),
		))
		counts: dict[int, int] = {}
		for row in group:
			if row.get("from_public"):
				continue
			slot = _parse_slot_value(row.get("slot"), 0)
			if slot >= 1:
				counts[slot] = counts.get(slot, 0) + 1
		used: set[int] = set()
		assigned: dict[str, int] = {}
		for row in group:
			device = row.get("device")
			slot = _parse_slot_value(row.get("slot"), 0)
			if not device:
				continue
			if row.get("from_public"):
				if slot < 1:
					slot = 1
				used.add(slot)
				assigned[device] = slot
				continue
			if slot >= 1 and counts.get(slot) == 1 and slot not in used:
				used.add(slot)
				assigned[device] = slot
		n = 1
		for row in group:
			device = row.get("device")
			if not device or device in assigned:
				continue
			while n in used:
				n += 1
			used.add(n)
			assigned[device] = n
			n += 1
		for device, slot in assigned.items():
			cur.execute("UPDATE accounts SET slot = ? WHERE device = ?", (slot, device))


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, slot) 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'),
			_parse_slot_value(fields.get('slot'), 1),
		),
	)
	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 find_device_by_public_id(public_id: str) -> Optional[dict]:
	public_id = str(public_id or "").strip()
	if not public_id:
		return None
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("SELECT * FROM accounts WHERE public_id = ?", (public_id,))
	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 _device_sort_key(row):
	device = str(row.get("device") or "")
	return [int(part) if part.isdigit() else part.lower() for part in re.split(r"(\d+)", device)]


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


def list_devices_by_ip(ip: str, exclude: str = "") -> list:
	wanted = str(ip or "").strip()
	skip = str(exclude or "").strip()
	items = []
	for row in list_devices():
		if skip and str(row.get("device") or "") == skip:
			continue
		if str(row.get("ip") or "").strip() != wanted:
			continue
		items.append(row)
	return items


def next_device_slot(ip: str, exclude: str = "", min_slot: int = 1) -> int:
	used = {device_slot_value(row) for row in list_devices_by_ip(ip, exclude=exclude)}
	try:
		n = int(min_slot)
	except (TypeError, ValueError):
		n = 1
	if n < 1:
		n = 1
	while n in used:
		n += 1
	return n


def used_device_ports(ip: str, exclude: str = "") -> set[int]:
	used: set[int] = set()
	for row in list_devices_by_ip(ip, exclude=exclude):
		try:
			port = int(row.get("port"))
		except (TypeError, ValueError):
			continue
		if 1 <= port <= 65535:
			used.add(port)
	return used


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', 'needs_setup', 'slot', 'from_public', 'public_id', 'color']
	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)


CARD_FLAG_SETTINGS_KEY = "card_flag_defs"
_FLAG_ID_RE = re.compile(r"^[a-z][a-z0-9_]{0,23}$")
_HEX_COLOR_RE = re.compile(r"^#[0-9a-f]{6}$")
_DEFAULT_CARD_FLAGS = (
	{"id": "beeline", "name": "Билайн", "color": "#facc15", "alias": "б"},
	{"id": "yapay", "name": "Япей", "color": "#ef4444", "alias": "я"},
	{"id": "mts", "name": "МТС", "color": "#a855f7", "alias": "м"},
)
_flag_defs_cache: Optional[list] = None


def default_card_flag_defs() -> list:
	return [dict(item) for item in _DEFAULT_CARD_FLAGS]


def _normalize_flag_alias(value) -> str:
	text = str(value or "").strip()
	if not text:
		return ""
	chars = list(text)
	if len(chars) != 1 or chars[0].isspace() or not chars[0].isalnum():
		raise ValueError("Буква флага — один символ")
	return chars[0].casefold()


def _new_flag_id(used: set) -> str:
	while True:
		candidate = "f" + secrets.token_hex(4)
		if candidate not in used:
			return candidate


def _parse_card_flag_defs(raw) -> list:
	if isinstance(raw, list):
		payload = raw
	else:
		try:
			payload = json.loads(raw) if raw else None
		except (TypeError, ValueError, json.JSONDecodeError):
			payload = None
	if not isinstance(payload, list):
		return default_card_flag_defs()
	flags = []
	used_ids = set()
	used_names = set()
	used_aliases = set()
	for item in payload:
		if not isinstance(item, dict):
			continue
		name = " ".join(str(item.get("name") or "").split())[:24]
		color = str(item.get("color") or "").strip().lower()
		if not name or not _HEX_COLOR_RE.fullmatch(color):
			continue
		name_key = name.casefold()
		if name_key in used_names:
			continue
		try:
			alias = _normalize_flag_alias(item.get("alias"))
		except ValueError:
			alias = ""
		if alias and alias in used_aliases:
			alias = ""
		raw_id = str(item.get("id") or "").strip().lower()
		flag_id = raw_id if _FLAG_ID_RE.fullmatch(raw_id) and raw_id not in used_ids else _new_flag_id(used_ids)
		used_ids.add(flag_id)
		used_names.add(name_key)
		if alias:
			used_aliases.add(alias)
		flags.append({"id": flag_id, "name": name, "color": color, "alias": alias})
		if len(flags) >= 24:
			break
	return flags


def get_card_flag_defs() -> list:
	global _flag_defs_cache
	if _flag_defs_cache is not None:
		return json.loads(json.dumps(_flag_defs_cache))
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("SELECT value FROM app_settings WHERE key = ?", (CARD_FLAG_SETTINGS_KEY,))
	row = cur.fetchone()
	conn.close()
	data = _parse_card_flag_defs(row[0] if row else None)
	_flag_defs_cache = data
	return json.loads(json.dumps(data))


def card_flag_keys() -> tuple:
	return tuple(item["id"] for item in get_card_flag_defs())


def system_by_alias() -> dict:
	mapping = {}
	for item in get_card_flag_defs():
		alias = item.get("alias") or ""
		if alias:
			mapping[alias] = item["id"]
	return mapping


def save_card_flag_defs(items) -> list:
	global _flag_defs_cache
	if not isinstance(items, list):
		raise ValueError("Некорректный список флагов")
	if len(items) > 24:
		raise ValueError("Слишком много флагов")
	flags = []
	used_ids = set()
	used_names = set()
	used_aliases = set()
	for item in items:
		if not isinstance(item, dict):
			raise ValueError("Некорректный флаг")
		name = " ".join(str(item.get("name") or "").split())
		if not name:
			raise ValueError("Укажите название флага")
		if len(name) > 24:
			raise ValueError("Название флага слишком длинное")
		name_key = name.casefold()
		if name_key in used_names:
			raise ValueError("Названия флагов должны отличаться")
		color = str(item.get("color") or "").strip().lower()
		if not _HEX_COLOR_RE.fullmatch(color):
			raise ValueError("Укажите цвет флага")
		alias = _normalize_flag_alias(item.get("alias"))
		if alias and alias in used_aliases:
			raise ValueError("Буквы флагов должны отличаться")
		raw_id = str(item.get("id") or "").strip().lower()
		flag_id = raw_id if _FLAG_ID_RE.fullmatch(raw_id) and raw_id not in used_ids else _new_flag_id(used_ids)
		used_ids.add(flag_id)
		used_names.add(name_key)
		if alias:
			used_aliases.add(alias)
		flags.append({"id": flag_id, "name": name, "color": color, "alias": alias})
	conn = get_conn()
	cur = conn.cursor()
	cur.execute(
		"INSERT INTO app_settings (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value",
		(CARD_FLAG_SETTINGS_KEY, json.dumps(flags, ensure_ascii=False)),
	)
	conn.commit()
	conn.close()
	_flag_defs_cache = flags
	return json.loads(json.dumps(flags))


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)
	keys = card_flag_keys()
	if any(current.get(key) for key in keys):
		flags[digits] = {key: bool(current.get(key)) for key in keys}
	else:
		flags.pop(digits, None)
	return update_card_flags(device, json.dumps(flags, ensure_ascii=False))


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


def update_blocked(device: str, blocked: int) -> bool:
	return _update_field(device, 'blocked', int(bool(blocked)))


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


def list_banned_ips() -> list[str]:
	return [row["ip"] for row in list_banned_ip_rows()]


def list_banned_ip_rows() -> list[dict]:
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("SELECT ip, banned_at FROM banned_ips ORDER BY banned_at DESC, ip ASC")
	rows = []
	for row in cur.fetchall():
		if not row or not row[0]:
			continue
		try:
			banned_at = float(row[1] or 0)
		except (TypeError, ValueError):
			banned_at = 0.0
		rows.append({"ip": str(row[0]), "banned_at": banned_at})
	conn.close()
	return rows


def save_banned_ip(ip: str) -> None:
	ip = (ip or "").strip()
	if not ip:
		return
	conn = get_conn()
	cur = conn.cursor()
	cur.execute(
		"INSERT OR IGNORE INTO banned_ips (ip, banned_at) VALUES (?, ?)",
		(ip, time.time()),
	)
	conn.commit()
	conn.close()


def remove_banned_ip(ip: str) -> bool:
	ip = (ip or "").strip()
	if not ip:
		return False
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("DELETE FROM banned_ips WHERE ip = ?", (ip,))
	conn.commit()
	changed = cur.rowcount
	conn.close()
	return changed > 0


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


def _proxy_row(row) -> dict:
	return {
		"id": int(row["id"]),
		"url": str(row["url"] or ""),
		"used": bool(row["used"]),
		"added_at": float(row["added_at"] or 0),
		"used_at": float(row["used_at"]) if row["used_at"] not in (None, "") else None,
		"used_by": str(row["used_by"]) if row["used_by"] not in (None, "") else None,
	}


def add_proxies(urls) -> dict:
	"""Добавляет прокси-ссылки в пул. Дубликаты пропускаются. Возвращает счётчики."""
	if isinstance(urls, str):
		urls = urls.splitlines()
	if not isinstance(urls, (list, tuple)):
		return {"added": 0, "skipped": 0}
	seen = set()
	cleaned = []
	for item in urls:
		url = str(item or "").strip()
		if not url or url in seen:
			continue
		seen.add(url)
		cleaned.append(url)
	if not cleaned:
		return {"added": 0, "skipped": 0}
	conn = get_conn()
	cur = conn.cursor()
	now = time.time()
	added = 0
	for url in cleaned:
		cur.execute(
			"INSERT OR IGNORE INTO proxies (url, used, added_at) VALUES (?, 0, ?)",
			(url, now),
		)
		if cur.rowcount:
			added += 1
	conn.commit()
	conn.close()
	return {"added": added, "skipped": len(cleaned) - added}


def list_proxies(used: Optional[bool] = None) -> list[dict]:
	conn = get_conn()
	cur = conn.cursor()
	if used is None:
		cur.execute("SELECT * FROM proxies ORDER BY id ASC")
	else:
		cur.execute(
			"SELECT * FROM proxies WHERE used = ? ORDER BY used_at DESC, id ASC" if used
			else "SELECT * FROM proxies WHERE used = ? ORDER BY id ASC",
			(1 if used else 0,),
		)
	rows = cur.fetchall()
	conn.close()
	return [_proxy_row(r) for r in rows]


def proxy_counts() -> dict:
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("SELECT COALESCE(used, 0) AS used, COUNT(*) AS n FROM proxies GROUP BY COALESCE(used, 0)")
	available = 0
	used = 0
	for row in cur.fetchall():
		if int(row["used"]):
			used = int(row["n"])
		else:
			available = int(row["n"])
	conn.close()
	return {"available": available, "used": used, "total": available + used}


def take_next_proxy(device: str = "") -> Optional[dict]:
	"""Берёт самую старую свободную прокси, помечает использованной и возвращает её."""
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("SELECT * FROM proxies WHERE used = 0 ORDER BY id ASC LIMIT 1")
	row = cur.fetchone()
	if not row:
		conn.close()
		return None
	proxy = _proxy_row(row)
	cur.execute(
		"UPDATE proxies SET used = 1, used_at = ?, used_by = ? WHERE id = ? AND used = 0",
		(time.time(), str(device or "") or None, proxy["id"]),
	)
	if not cur.rowcount:
		conn.close()
		return take_next_proxy(device)
	conn.commit()
	conn.close()
	proxy["used"] = True
	proxy["used_by"] = str(device or "") or None
	return proxy


def delete_proxy(proxy_id) -> bool:
	try:
		pid = int(proxy_id)
	except (TypeError, ValueError):
		return False
	conn = get_conn()
	cur = conn.cursor()
	cur.execute("DELETE FROM proxies WHERE id = ?", (pid,))
	conn.commit()
	changed = cur.rowcount
	conn.close()
	return changed > 0


def restore_proxy(proxy_id) -> bool:
	"""Возвращает использованную прокси обратно в пул доступных."""
	try:
		pid = int(proxy_id)
	except (TypeError, ValueError):
		return False
	conn = get_conn()
	cur = conn.cursor()
	cur.execute(
		"UPDATE proxies SET used = 0, used_at = NULL, used_by = NULL WHERE id = ?",
		(pid,),
	)
	conn.commit()
	changed = cur.rowcount
	conn.close()
	return changed > 0