#!/usr/bin/env python3
"""
MQTT брокер - исправленная версия
"""

import os
import json
import time
import threading
import socket
import struct
import sys
from datetime import datetime
from typing import Dict, Set
from collections import defaultdict


class ClientSession:
    def __init__(self, client_ip: str, sessions_dir: str, max_points: int = 1000):
        print(f"    [SESSION_INIT] НАЧАЛО", file=sys.stderr, flush=True)
        print(f"      client_ip={client_ip}", file=sys.stderr, flush=True)
        print(f"      sessions_dir={sessions_dir}", file=sys.stderr, flush=True)
        print(f"      max_points={max_points}", file=sys.stderr, flush=True)
        
        self.client_ip = client_ip
        self.session_id = f"{client_ip}_{int(time.time())}"
        self.ip_dir_name = client_ip.replace('.', '_')
        self.start_time = datetime.now()
        self.last_activity = time.time()
        self.session_time_str = self.start_time.strftime("%Y-%m-%d_%H-%M-%S")  # Заменяем : на -
        self.max_points = max_points
        self.point_count = 0
        
        self.ip_dir = os.path.join(sessions_dir, self.ip_dir_name)
        print(f"    [SESSION_INIT] ip_dir={self.ip_dir}", file=sys.stderr, flush=True)
        
        try:
            os.makedirs(self.ip_dir, exist_ok=True)
        except Exception as e:
            raise
        
        self.st_data = None
    
    def update_activity(self):
        self.last_activity = time.time()
    
    def is_active(self, timeout: int) -> bool:
        return (time.time() - self.last_activity) <= timeout
    
    def get_file_path(self, file_type: str) -> str:
        return os.path.join(self.ip_dir, f"{self.session_time_str}_{file_type}.json")
    
    def append_json(self, file_type: str, data: dict) -> bool:
        filename = self.get_file_path(file_type)
        file_exists = os.path.exists(filename) and os.path.getsize(filename) > 0
        
        try:
            if not file_exists:
                with open(filename, 'w', encoding='utf-8') as f:
                    f.write('[\n')
                    f.write(json.dumps(data, ensure_ascii=False, indent=2))
                    f.write('\n]')
            else:
                with open(filename, 'r+', encoding='utf-8') as f:
                    f.seek(0, 2)
                    current_pos = f.tell()
                    
                    if current_pos >= 1:
                        f.seek(current_pos - 1)
                        last_byte = f.read(1)
                        
                        if last_byte == ']':
                            f.seek(current_pos - 1)
                            f.truncate()
                            f.write(',\n')
                            f.write(json.dumps(data, ensure_ascii=False, indent=2))
                            f.write('\n]')
                        else:
                            f.seek(0, 2)
                            f.write(',\n')
                            f.write(json.dumps(data, ensure_ascii=False, indent=2))
                            f.write('\n]')
                    else:
                        f.seek(0, 2)
                        f.write('[\n')
                        f.write(json.dumps(data, ensure_ascii=False, indent=2))
                        f.write('\n]')
            
            if file_type == 'log':
                self.point_count += 1
            
            return True
        except Exception:
            return False
    
    def is_full(self) -> bool:
        return self.point_count >= self.max_points


class MQTTBroker:
    def __init__(self, host: str = "0.0.0.0", port: int = 1883,
                 sessions_dir: str = "sessions", session_timeout: int = 180,
                 max_points_per_session: int = 1000):
        self.host = host
        self.port = port
        self.sessions_dir = sessions_dir
        self.session_timeout = session_timeout
        self.max_points_per_session = max_points_per_session
        
        self.running = False
        
        self.clients: Dict[socket.socket, dict] = {}
        self.subscriptions: Dict[str, Set[socket.socket]] = defaultdict(set)
        self.client_topics: Dict[socket.socket, Set[str]] = defaultdict(set)
        self.pending_qos2: Dict[tuple, dict] = {}
        
        self.sessions: Dict[str, ClientSession] = {}
        self.sessions_lock = threading.Lock()
        
        try:
            os.makedirs(sessions_dir, exist_ok=True)
        except Exception as e:
            self.sessions_dir = os.getcwd()
        
    
    def start(self):
        self.running = True
        server_thread = threading.Thread(target=self._run_server, daemon=True)
        server_thread.start()
        print(f"MQTT брокер запущен на {self.host}:{self.port}")
    
    def _run_server(self):
        server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server_socket.bind((self.host, self.port))
        server_socket.listen(10)
        server_socket.settimeout(1.0)
        
        print(f"Слушаем {self.host}:{self.port}")
        
        while self.running:
            try:
                client_socket, client_addr = server_socket.accept()
                client_ip = client_addr[0]
                print(f"Клиент подключился: {client_ip}")
                
                client_thread = threading.Thread(
                    target=self._handle_client,
                    args=(client_socket, client_ip),
                    daemon=True
                )
                client_thread.start()
                
            except socket.timeout:
                continue
            except Exception:
                pass
        
        server_socket.close()
    
    def _handle_client(self, client_socket: socket.socket, client_ip: str):
        client_socket.settimeout(5.0)
        
        try:
            while self.running:
                try:
                    header = client_socket.recv(1)
                    if not header:
                        break
                    
                    packet_type = (header[0] >> 4) & 0x0F
                    flags = header[0] & 0x0F
                    
                    length = self._read_remaining_length(client_socket)
                    if length is None:
                        break
                    
                    if length > 0:
                        payload = client_socket.recv(length)
                    else:
                        payload = b''
                    
                    if packet_type == 1:
                        self._handle_connect(payload, client_socket, client_ip)
                    elif packet_type == 3:
                        self._handle_publish(payload, client_socket, client_ip, flags)
                    elif packet_type == 6:
                        self._handle_pubrel(payload, client_socket)
                    elif packet_type == 8:
                        self._handle_subscribe(payload, client_socket)
                    elif packet_type == 10:
                        self._handle_unsubscribe(payload, client_socket)
                    elif packet_type == 12:
                        client_socket.send(bytes([0xD0, 0x00]))
                    elif packet_type == 14:
                        break
                        
                except socket.timeout:
                    continue
                except ConnectionResetError:
                    break
                except Exception:
                    break
                    
        except Exception:
            pass
        finally:
            if client_socket in self.clients:
                del self.clients[client_socket]
            
            if client_socket in self.client_topics:
                for topic in self.client_topics[client_socket]:
                    if topic in self.subscriptions:
                        self.subscriptions[topic].discard(client_socket)
                del self.client_topics[client_socket]
            
            client_socket.close()
    
    def _read_remaining_length(self, sock: socket.socket) -> int:
        length = 0
        multiplier = 1
        
        while True:
            byte = sock.recv(1)
            if not byte:
                return None
            length += (byte[0] & 0x7F) * multiplier
            multiplier *= 128
            if not (byte[0] & 0x80):
                break
        
        return length
    
    def _encode_remaining_length(self, length: int) -> bytes:
        encoded = bytearray()
        while True:
            digit = length % 128
            length //= 128
            if length > 0:
                digit |= 0x80
            encoded.append(digit)
            if length == 0:
                break
        return bytes(encoded)
    
    def _handle_connect(self, data: bytes, sock: socket.socket, client_ip: str):
        try:
            if len(data) < 10:
                return
            
            pos = 0
            proto_len = struct.unpack('>H', data[pos:pos+2])[0]
            pos += 2
            pos += proto_len
            pos += 1
            pos += 1
            pos += 2
            
            if pos + 2 <= len(data):
                client_id_len = struct.unpack('>H', data[pos:pos+2])[0]
                pos += 2
                if client_id_len > 0 and pos + client_id_len <= len(data):
                    client_id = data[pos:pos+client_id_len].decode('utf-8', errors='ignore')
                else:
                    client_id = f"client_{client_ip}"
            else:
                client_id = f"client_{client_ip}"
            
            self.clients[sock] = {'client_id': client_id, 'client_ip': client_ip}
            
            connack = bytes([0x20, 0x02, 0x00, 0x00])
            sock.send(connack)
            
        except Exception:
            pass
    
    def _handle_publish(self, data: bytes, sock: socket.socket, client_ip: str, flags: int):
    
        qos = (flags >> 1) & 0x03
        
        if len(data) < 2:
            return
        
        pos = 0
        topic_len = struct.unpack('>H', data[pos:pos+2])[0]
        pos += 2
        
        if len(data) < pos + topic_len:
            return
        
        topic = data[pos:pos+topic_len].decode('utf-8', errors='ignore')
        pos += topic_len
        
        packet_id = None
        if qos > 0:
            if pos + 2 <= len(data):
                packet_id = struct.unpack('>H', data[pos:pos+2])[0]
                pos += 2
        
        payload = data[pos:].decode('utf-8', errors='ignore')
        
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        
        if qos == 2 and packet_id is not None:
            pubrec = bytes([0x50, 0x02]) + struct.pack('>H', packet_id)
            sock.send(pubrec)
            self.pending_qos2[(sock, packet_id)] = {
                'topic': topic,
                'payload': payload,
                'timestamp': timestamp,
                'client_ip': client_ip,
                'data': data
            }
            return
        
        topic_bytes = topic.encode()
        topic_len_bytes = struct.pack('>H', len(topic_bytes))
        payload_bytes = payload.encode()
        remaining = topic_len_bytes + topic_bytes + payload_bytes
        remaining_length = self._encode_remaining_length(len(remaining))
        new_packet = bytes([0x30]) + remaining_length + remaining
        
        for sub_topic, subscribers in list(self.subscriptions.items()):
            if self._topic_matches(sub_topic, topic):
                for subscriber in list(subscribers):
                    if subscriber != sock:
                        try:
                            subscriber.send(new_packet)
                        except:
                            pass
        
        if self._is_smv_topic(topic):
            self._log_smv_message(client_ip, topic, payload, timestamp)
        
        if qos == 1 and packet_id is not None:
            puback = bytes([0x40, 0x02]) + struct.pack('>H', packet_id)
            sock.send(puback)
    
    def _handle_pubrel(self, data: bytes, sock: socket.socket):
      
        if len(data) < 2:
            return
        
        packet_id = struct.unpack('>H', data[0:2])[0]
        
        pubcomp = bytes([0x70, 0x02]) + struct.pack('>H', packet_id)
        sock.send(pubcomp)
        
        key = (sock, packet_id)
        if key in self.pending_qos2:
            msg = self.pending_qos2[key]
            
            topic_bytes = msg['topic'].encode()
            topic_len_bytes = struct.pack('>H', len(topic_bytes))
            payload_bytes = msg['payload'].encode()
            remaining = topic_len_bytes + topic_bytes + payload_bytes
            remaining_length = self._encode_remaining_length(len(remaining))
            new_packet = bytes([0x30]) + remaining_length + remaining
            
            for sub_topic, subscribers in list(self.subscriptions.items()):
                if self._topic_matches(sub_topic, msg['topic']):
                    for subscriber in list(subscribers):
                        if subscriber != sock:
                            try:
                                subscriber.send(new_packet)
                            except:
                                pass
            
            if self._is_smv_topic(msg['topic']):
                self._log_smv_message(
                    msg['client_ip'], 
                    msg['topic'], 
                    msg['payload'], 
                    msg['timestamp']
                )
            
            del self.pending_qos2[key]
    
    def _topic_matches(self, subscription: str, topic: str) -> bool:
        if subscription == topic:
            return True
        if subscription == "#":
            return True
        if subscription.endswith("/#"):
            prefix = subscription[:-2]
            if topic == prefix or topic.startswith(prefix + "/"):
                return True
        
        sub_parts = subscription.split('/')
        topic_parts = topic.split('/')
        
        if len(sub_parts) > len(topic_parts):
            return False
        
        for i, sub_part in enumerate(sub_parts):
            if sub_part == '+':
                continue
            if sub_part == '#':
                return True
            if i >= len(topic_parts):
                return False
            if sub_part != topic_parts[i]:
                return False
        
        return True
    
    def _is_smv_topic(self, topic: str) -> bool:
        parts = topic.split('/')
        return len(parts) == 4 and parts[0] == "SMV" and parts[3] == "3"
    
    def _get_session_for_ip(self, client_ip: str) -> ClientSession:
        
        # Убираем блокировку для отладки
        if client_ip in self.sessions:
            session = self.sessions[client_ip]
            if session.is_full():
                print(f"    [SESSION_GET] Сессия заполнена, создаём новую", file=sys.stderr, flush=True)
                new_session = ClientSession(client_ip, self.sessions_dir, self.max_points_per_session)
                self.sessions[client_ip] = new_session
                return new_session
            return session
        else:
            try:
                session = ClientSession(client_ip, self.sessions_dir, self.max_points_per_session)
                self.sessions[client_ip] = session
                print(f"    [SESSION_GET] Создана сессия: {session.session_time_str}", file=sys.stderr, flush=True)
                return session
            except Exception as e:
                print(f"    [SESSION_GET] ОШИБКА: {e}", file=sys.stderr, flush=True)
                import traceback
                traceback.print_exc(file=sys.stderr)
                raise
    
    def _log_smv_message(self, client_ip: str, topic: str, payload: str, timestamp: str):
        
        try:
            session = self._get_session_for_ip(client_ip)
            if session is None:
                print(f"    [LOG_SMV] ОШИБКА: session is None!", file=sys.stderr, flush=True)
                return
            session.update_activity()
            
            parts = topic.split('/')
            subtopic = parts[2] if len(parts) >= 3 else "unknown"
            
            if subtopic == "st":
                st_data = {"type": "session", "time": timestamp, "data": payload}
                session.append_json("st", st_data)
                session.st_data = payload
                
            elif subtopic == "log":
                log_data = self._csv_to_json(payload, timestamp)
                session.append_json("log", log_data)
                
            elif subtopic == "msg":
                msg_data = {"type": "message", "time": timestamp, "message": payload}
                session.append_json("msg", msg_data)
        except Exception as e:
            print(f"    [LOG_SMV] ОШИБКА: {e}", file=sys.stderr, flush=True)
            import traceback
            traceback.print_exc(file=sys.stderr)
    
    def _csv_to_json(self, payload: str, timestamp: str) -> dict:
        result = {"Date": timestamp}
        
        first_comma = payload.find(',')
        if first_comma < 0:
            return result
        
        remaining = payload[first_comma + 1:]
        values = remaining.split(',')
        
        fields = ['Tp', 'Tc', 'Tw', 'Tb', 'Pa', 'N', 'Tt', 'V', 'Dw', 'P', 'Ab', 'Ap', 'Pb']
        
        for i, val in enumerate(values):
            if i < len(fields):
                try:
                    val = val.strip()
                    if val:
                        result[fields[i]] = float(val)
                except:
                    result[fields[i]] = val
        
        return result
    
    def _handle_subscribe(self, data: bytes, sock: socket.socket):
        try:
            if len(data) < 2:
                return
            
            packet_id = struct.unpack('>H', data[0:2])[0]
            pos = 2
            return_codes = []
            
            while pos < len(data):
                if pos + 2 > len(data):
                    break
                topic_len = struct.unpack('>H', data[pos:pos+2])[0]
                pos += 2
                
                if pos + topic_len > len(data):
                    break
                topic = data[pos:pos+topic_len].decode('utf-8', errors='ignore')
                pos += topic_len
                
                if pos >= len(data):
                    break
                qos = data[pos] & 0x03
                pos += 1
                
                self.subscriptions[topic].add(sock)
                self.client_topics[sock].add(topic)
                return_codes.append(qos)
                print(f"  Подписка: {topic}")
            
            suback = bytearray()
            suback.append(0x90)
            suback.extend(self._encode_remaining_length(2 + len(return_codes)))
            suback.extend(struct.pack('>H', packet_id))
            suback.extend(bytes(return_codes))
            sock.send(bytes(suback))
            
        except Exception:
            pass
    
    def _handle_unsubscribe(self, data: bytes, sock: socket.socket):
        try:
            if len(data) < 2:
                return
            
            packet_id = struct.unpack('>H', data[0:2])[0]
            pos = 2
            
            while pos < len(data):
                if pos + 2 > len(data):
                    break
                topic_len = struct.unpack('>H', data[pos:pos+2])[0]
                pos += 2
                
                if pos + topic_len > len(data):
                    break
                topic = data[pos:pos+topic_len].decode('utf-8', errors='ignore')
                pos += topic_len
                
                if topic in self.subscriptions:
                    self.subscriptions[topic].discard(sock)
                if sock in self.client_topics and topic in self.client_topics[sock]:
                    self.client_topics[sock].discard(topic)
            
            unsuback = bytearray()
            unsuback.append(0xB0)
            unsuback.extend(self._encode_remaining_length(2))
            unsuback.extend(struct.pack('>H', packet_id))
            sock.send(bytes(unsuback))
            
        except Exception:
            pass
    
    def stop(self):
        self.running = False
        for sock in list(self.clients.keys()):
            try:
                sock.close()
            except:
                pass
        print("MQTT брокер остановлен")