#!/usr/bin/env python3
"""
ПЦН-приёмник (ru3): Surgard MLR2-DG / Contact ID (:5111) + SIA DC-09 (:5112).
Декодирует account + код события, публикует в mosquitto (raw MQTT 3.1.1, stdlib-only)
топик guard/<account>/event JSON.

ПЕРВАЯ рабочая версия — минимальный, но живой приёмник.
TODO: полная валидация SIA DC-09 CRC/шифрование; edge-cases Contact ID checksum.
"""
import asyncio
import json
import os
import re
import socket
import struct
import tempfile
import threading
import time
import logging

# ---- Конфиг ----
MQTT_HOST = "10.50.0.48"   # mosquitto vm-mts1 по туннелю awg-mesh
MQTT_PORT = 1883
MQTT_KEEPALIVE = 60
SURGARD_PORT = 5111
SIA_PORT = 5112
CLIENT_ID = "pcn-ru3"

# ---- Дисковый спул (буферизация при обрыве связи с брокером) ----
SPOOL_DIR = "/var/spool/pcn"
SPOOL_FILE = os.path.join(SPOOL_DIR, "queue.jsonl")
SPOOL_MAX_BYTES = 100 * 1024 * 1024   # 100 МБ — лимит спула
SPOOL_TRIM_BYTES = 80 * 1024 * 1024   # при ротации ужимаем до 80 МБ (дропаем старейшие)
DRAIN_INTERVAL = 5                    # секунд между попытками до-досыла
DRAIN_BATCH = 200                     # сколько сообщений максимум за один проход

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
log = logging.getLogger("pcn")

# ---- Соответствие SIA <-> Surgard Contact ID (из src/data/security.ts) ----
SIA_TO_CID = {
    "BA": "E130", "FA": "E110", "PA": "E120", "HA": "E121", "TA": "E137",
    "OP": "E401", "CL": "R401", "AT": "E301", "AR": "R301", "BT": "E302",
    "YC": "E354", "NL": "E381",
}
CID_TO_SIA = {v: k for k, v in SIA_TO_CID.items()}


# ============================================================
#  Минимальный raw-MQTT publisher (3.1.1), без внешних зависимостей
# ============================================================
def _mqtt_remaining_length(n: int) -> bytes:
    out = bytearray()
    while True:
        b = n % 128
        n //= 128
        if n > 0:
            b |= 0x80
        out.append(b)
        if n == 0:
            break
    return bytes(out)


def _mqtt_str(s: str) -> bytes:
    b = s.encode("utf-8")
    return struct.pack("!H", len(b)) + b


class MqttPublisher:
    """Синхронный публикатор в отдельном потоке-неблокирующий он не нужен:
    вызываем из executor, чтобы не блокировать asyncio-loop."""

    def __init__(self, host, port, client_id, keepalive=60):
        self.host = host
        self.port = port
        self.client_id = client_id
        self.keepalive = keepalive
        self.sock = None
        # publish() и дренаж исполняются в разных потоках executor'а — сериализуем сокет.
        self._lock = threading.Lock()

    def _connect(self):
        s = socket.create_connection((self.host, self.port), timeout=8)
        # CONNECT packet
        payload = _mqtt_str(self.client_id)
        var_header = _mqtt_str("MQTT") + bytes([0x04, 0x02]) + struct.pack("!H", self.keepalive)
        body = var_header + payload
        pkt = bytes([0x10]) + _mqtt_remaining_length(len(body)) + body
        s.sendall(pkt)
        # CONNACK (4 bytes)
        resp = s.recv(4)
        if len(resp) < 4 or resp[0] != 0x20 or resp[3] != 0x00:
            s.close()
            raise ConnectionError(f"MQTT CONNACK failed: {resp!r}")
        self.sock = s

    def publish(self, topic, payload, qos=0):
        with self._lock:
            if self.sock is None:
                self._connect()
            try:
                self._publish(topic, payload)
            except (OSError, ConnectionError) as e:
                # reconnect once
                log.warning("MQTT publish err (%s), reconnecting", e)
                try:
                    if self.sock:
                        self.sock.close()
                except OSError:
                    pass
                self.sock = None
                self._connect()
                self._publish(topic, payload)

    def _publish(self, topic, payload):
        body = _mqtt_str(topic) + payload.encode("utf-8")
        pkt = bytes([0x30]) + _mqtt_remaining_length(len(body)) + body
        self.sock.sendall(pkt)


_pub = MqttPublisher(MQTT_HOST, MQTT_PORT, CLIENT_ID, MQTT_KEEPALIVE)


# ============================================================
#  Дисковый спул: append+fsync при обрыве, дренаж при восстановлении
# ============================================================
def _spool_ensure():
    try:
        os.makedirs(SPOOL_DIR, exist_ok=True)
    except OSError as e:
        log.error("spool mkdir err: %s", e)


def _spool_append(topic, payload):
    """Атомарно (append+fsync) дописать одно сообщение в очередь.
    Строка формата: {"topic":..,"payload":..}\n"""
    _spool_ensure()
    line = json.dumps({"topic": topic, "payload": payload}, ensure_ascii=False) + "\n"
    try:
        _spool_rotate_if_needed()
        with open(SPOOL_FILE, "a", encoding="utf-8") as f:
            f.write(line)
            f.flush()
            os.fsync(f.fileno())
        log.warning("SPOOLED %s (queue on disk)", topic)
    except OSError as e:
        log.error("spool append err: %s (СООБЩЕНИЕ ПОТЕРЯНО)", e)


def _spool_rotate_if_needed():
    """Ограничение размера: если спул > SPOOL_MAX, дропаем старейшие строки
    до SPOOL_TRIM (сохраняем хвост = самые свежие события)."""
    try:
        if not os.path.exists(SPOOL_FILE):
            return
        size = os.path.getsize(SPOOL_FILE)
        if size <= SPOOL_MAX_BYTES:
            return
        log.warning("spool rotate: %d байт > лимит, ужимаю до %d", size, SPOOL_TRIM_BYTES)
        with open(SPOOL_FILE, "rb") as f:
            f.seek(max(0, size - SPOOL_TRIM_BYTES))
            f.readline()          # выравниваемся на границу строки
            tail = f.read()
        tmp = SPOOL_FILE + ".tmp"
        with open(tmp, "wb") as f:
            f.write(tail)
            f.flush()
            os.fsync(f.fileno())
        os.replace(tmp, SPOOL_FILE)
    except OSError as e:
        log.error("spool rotate err: %s", e)


def _drain_once():
    """Синхронно (в executor): попытаться до-дослать спул.
    Читает весь файл, публикует по одному; при первой ошибке — сохраняет
    непосланный остаток обратно (атомарно) и выходит. Возвращает кол-во посланных."""
    if not os.path.exists(SPOOL_FILE) or os.path.getsize(SPOOL_FILE) == 0:
        return 0
    sent = 0
    remainder = None
    try:
        with open(SPOOL_FILE, "r", encoding="utf-8") as f:
            lines = f.readlines()
    except OSError as e:
        log.error("drain read err: %s", e)
        return 0
    for i, line in enumerate(lines[:DRAIN_BATCH]):
        line = line.strip()
        if not line:
            sent += 1
            continue
        try:
            rec = json.loads(line)
            _pub.publish(rec["topic"], rec["payload"])
            sent += 1
        except (OSError, ConnectionError) as e:
            log.warning("drain: брокер снова недоступен (%s), стоп на %d", e, i)
            remainder = lines[i:]
            break
        except Exception as e:
            log.error("drain: битая строка спула, дроп: %s", e)
            sent += 1
    if remainder is None:
        remainder = lines[DRAIN_BATCH:]
    # Перезаписываем спул оставшимся хвостом атомарно
    try:
        if remainder:
            fd, tmp = tempfile.mkstemp(dir=SPOOL_DIR, prefix="q.", suffix=".tmp")
            with os.fdopen(fd, "w", encoding="utf-8") as f:
                f.writelines(remainder)
                f.flush()
                os.fsync(f.fileno())
            os.replace(tmp, SPOOL_FILE)
        else:
            os.remove(SPOOL_FILE)
    except OSError as e:
        log.error("drain rewrite err: %s", e)
    if sent:
        log.info("DRAINED %d из спула, осталось %d", sent, len(remainder))
    return sent


async def drain_loop():
    """Фоновая задача: раз в DRAIN_INTERVAL секунд пытается до-дослать спул."""
    loop = asyncio.get_event_loop()
    while True:
        await asyncio.sleep(DRAIN_INTERVAL)
        try:
            if os.path.exists(SPOOL_FILE) and os.path.getsize(SPOOL_FILE) > 0:
                await loop.run_in_executor(None, _drain_once)
        except Exception as e:
            log.error("drain_loop err: %s", e)


async def publish_event(account, proto, cid, sia, zone, raw):
    doc = {
        "account": account,
        "proto": proto,
        "cid": cid,
        "sia": sia,
        "zone": zone,
        "raw": raw,
        "ts": int(time.time()),
    }
    topic = f"guard/{account}/event"
    payload = json.dumps(doc, ensure_ascii=False)
    loop = asyncio.get_event_loop()
    try:
        await loop.run_in_executor(None, _pub.publish, topic, payload)
        log.info("PUB %s %s", topic, payload)
    except Exception as e:
        # Брокер/туннель недоступны — НЕ теряем: спуливаем на диск.
        log.error("MQTT publish failed: %s -> спул", e)
        await loop.run_in_executor(None, _spool_append, topic, payload)


# ============================================================
#  Разбор Contact ID кода -> (cid, sia)
# ============================================================
def cid_to_pair(qualifier, event3):
    """qualifier '1'=new/E , '3'=restore/R ; event3 = 3 цифры (130,110,...)"""
    prefix = "R" if qualifier == "3" else "E"
    cid = f"{prefix}{event3}"
    sia = CID_TO_SIA.get(cid, "")
    return cid, sia


# ============================================================
#  Surgard MLR2-DG / Contact ID приёмник (:5111)
# ============================================================
# Форматы, которые пытаемся распознать:
#  1) Sur-Gard 5-2-1: "5BB 18 <account> 18 <CID> <qual> <zone>"
#     напр. "5011 18 1042 18 130 01 002"
#  2) ADM-CID Contact ID: "<ACCT> 18 <MT> <Q> <XYZ> <GG> <CCC>"
#     где MT=18(DTMF)/98, Q=1/3, XYZ=код, GG=group/partition, CCC=zone/user
SURGARD_ACK = b"\x06"  # приёмник подтверждает событие символом ACK


def parse_surgard(line: str):
    line = line.strip()
    if not line:
        return None
    toks = line.split()
    # Попытка: найти шаблон Contact ID где-то в строке.
    # Ищем последовательность: <acct> 18 <q><xyz> ... либо явные поля.
    # Нормализуем: убираем ведущий "5xx" код линии Sur-Gard если есть.
    # Универсальный разбор по токенам:
    #   ищем токен "18" (Contact ID message type), после него qual+code.
    account = None
    cid = None
    sia = None
    zone = None

    # Вариант A: пробельно-разделённые поля (тестовый формат)
    #   "5011 18 1042 18 130 01 002"
    #   idx: 0=lineprefix 1=18 2=account 3=18 4=code 5=group 6=zone
    if len(toks) >= 7 and toks[1] == "18" and toks[3] == "18":
        account = toks[2]
        code = toks[4]              # 130
        # qualifier может идти как отдельная 1/3 или быть слит; тут по группе.
        # В классике group '01'/'02' — event; берём E по умолчанию.
        qual = "1"
        cid, sia = cid_to_pair(qual, code.zfill(3)[:3])
        zone = toks[6]
        return account, cid, sia, zone

    # Вариант B: слитный Contact ID "ACCT 18 Q XYZ GG CCC"
    #   напр. "1042 18 1 130 01 002"
    if len(toks) >= 6 and toks[1] in ("18", "98"):
        account = toks[0]
        qual = toks[2]
        code = toks[3]
        cid, sia = cid_to_pair(qual, code.zfill(3)[:3])
        zone = toks[-1]
        return account, cid, sia, zone

    # Вариант C: полностью слитная ADM-CID строка "10421813001002"
    #   ACCT(4) 18 Q(1) XYZ(3) GG(2) CCC(3)
    m = re.match(r"^(\d{3,4})18([13])(\d{3})(\d{2})(\d{3})$", line.replace(" ", ""))
    if m:
        account = m.group(1)
        qual = m.group(2)
        code = m.group(3)
        zone = m.group(5)
        cid, sia = cid_to_pair(qual, code)
        return account, cid, sia, zone

    # Не распознали — вернём хотя бы raw с пустыми полями (account=UNKNOWN)
    return None


async def handle_surgard(reader, writer):
    peer = writer.get_extra_info("peername")
    log.info("Surgard conn from %s", peer)
    try:
        while True:
            data = await asyncio.wait_for(reader.readline(), timeout=120)
            if not data:
                break
            raw = data.decode("latin-1", errors="replace").rstrip("\r\n")
            if not raw:
                continue
            log.info("Surgard raw: %r", raw)
            parsed = parse_surgard(raw)
            if parsed:
                account, cid, sia, zone = parsed
                await publish_event(account, "surgard", cid, sia, zone, raw)
            else:
                log.warning("Surgard: unparsed %r -> publishing raw under UNKNOWN", raw)
                await publish_event("UNKNOWN", "surgard", "", "", None, raw)
            # ACK приёмника
            writer.write(SURGARD_ACK)
            await writer.drain()
    except asyncio.TimeoutError:
        log.info("Surgard %s idle timeout", peer)
    except Exception as e:
        log.error("Surgard handler err: %s", e)
    finally:
        try:
            writer.close()
        except Exception:
            pass
        log.info("Surgard conn closed %s", peer)


# ============================================================
#  SIA DC-09 приёмник (:5112)
# ============================================================
# Кадр DC-09 (упрощённо):
#  \x0A <CRC4hex> <LEN4hex> "<id>" <seq> R<rcvr> L<line> #<account> [<data>] <ts> \x0D
#  id: "SIA-DCS" | "ADM-CID" | "NULL" | ...
#  data для ADM-CID: |< contact-id ...>  ; для SIA-DCS: |Nri1/BA01]
#  Ответ ACK: \x0A <CRC> <LEN> "ACK" <seq> R.. L.. #acct [] <ts> \x0D
LF = 0x0A
CR = 0x0D


def _dc09_crc(msg: bytes) -> int:
    """CRC-16 (polynomial 0x8005, reflected) как в DC-09 (для ACK)."""
    crc = 0
    for b in msg:
        crc ^= b
        for _ in range(8):
            if crc & 1:
                crc = (crc >> 1) ^ 0xA001
            else:
                crc >>= 1
    return crc & 0xFFFF


def parse_sia(frame: str):
    """frame — содержимое между LF и CR (без обрамляющих).
    Возвращает (account, cid, sia, zone, seq, id_token)."""
    account = None
    cid = None
    sia = None
    zone = None
    seq = "0000"
    id_token = ""

    # id в кавычках
    m_id = re.search(r'"([A-Z\-]+)"', frame)
    if m_id:
        id_token = m_id.group(1)
    # seq — 4 цифры сразу после закрывающей кавычки id
    m_seq = re.search(r'"[A-Z\-]+"(\d{4})', frame)
    if m_seq:
        seq = m_seq.group(1)
    # account: #ACCT
    m_acct = re.search(r"#([0-9A-Fa-f]+)", frame)
    if m_acct:
        account = m_acct.group(1)

    # payload между [ ... ] (у SIA-DCS) или после последнего |
    # SIA-DCS data: |Nri1/BA01]  -> code=BA zone=01
    m_sia = re.search(r"/([A-Z]{2})(\d{1,3})", frame)
    if m_sia:
        sia = m_sia.group(1)
        zone = m_sia.group(2)
        cid = SIA_TO_CID.get(sia, "")
    else:
        # ADM-CID внутри DC-09: |1 XYZ GG CCC| или похожее
        m_cid = re.search(r"\|(\d)?\s?(\d{3})\s?(\d{2})?\s?(\d{2,3})?", frame)
        if m_cid:
            qual = m_cid.group(1) or "1"
            code = m_cid.group(2)
            zone = m_cid.group(4)
            prefix = "R" if qual == "3" else "E"
            cid = f"{prefix}{code}"
            sia = CID_TO_SIA.get(cid, "")

    return account, cid, sia, zone, seq, id_token


def build_sia_ack(seq: str, account: str) -> bytes:
    """Строим минимальный DC-09 ACK-кадр."""
    ts = time.strftime("_%H:%M:%S,%m-%d-%Y", time.gmtime())
    acct = account or "0"
    body = f'"ACK"{seq}R0L0#{acct}[]{ts}'
    b = body.encode("ascii", errors="replace")
    crc = _dc09_crc(b)
    length = len(b)
    frame = f'\n{crc:04X}{length:04X}{body}\r'
    return frame.encode("ascii", errors="replace")


async def handle_sia(reader, writer):
    peer = writer.get_extra_info("peername")
    log.info("SIA conn from %s", peer)
    try:
        while True:
            # читаем до CR (0x0D) — конец кадра DC-09
            data = await asyncio.wait_for(reader.readuntil(b"\r"), timeout=120)
            if not data:
                break
            raw = data.decode("latin-1", errors="replace")
            log.info("SIA raw: %r", raw)
            # вырезаем содержимое между LF и CR
            inner = raw
            if inner and inner[0] == "\n":
                inner = inner[1:]
            inner = inner.rstrip("\r")
            account, cid, sia, zone, seq, id_token = parse_sia(inner)
            if account is None:
                account = "UNKNOWN"
            await publish_event(account, "sia", cid, sia, zone, raw.strip())
            # ACK-кадр
            try:
                ack = build_sia_ack(seq, account if account != "UNKNOWN" else "0")
                writer.write(ack)
                await writer.drain()
            except Exception as e:
                log.warning("SIA ACK build/send err: %s", e)
    except asyncio.IncompleteReadError:
        log.info("SIA %s incomplete/closed", peer)
    except asyncio.TimeoutError:
        log.info("SIA %s idle timeout", peer)
    except Exception as e:
        log.error("SIA handler err: %s", e)
    finally:
        try:
            writer.close()
        except Exception:
            pass
        log.info("SIA conn closed %s", peer)


# ============================================================
async def main():
    srv_sur = await asyncio.start_server(handle_surgard, "0.0.0.0", SURGARD_PORT)
    srv_sia = await asyncio.start_server(handle_sia, "0.0.0.0", SIA_PORT)
    _spool_ensure()
    drain = asyncio.ensure_future(drain_loop())
    log.info("PCN receiver up: Surgard :%d, SIA DC-09 :%d -> mqtt %s:%d (spool %s)",
             SURGARD_PORT, SIA_PORT, MQTT_HOST, MQTT_PORT, SPOOL_FILE)
    async with srv_sur, srv_sia:
        await asyncio.gather(srv_sur.serve_forever(), srv_sia.serve_forever(), drain)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass