Sync from GitHub with Git LFS

agents/peer_sync.py

@@ -7,18 +7,43 @@ import json
 import uuid
 import ipaddress
 from tools.storage import Storage
+from datetime import datetime
 
 storage = Storage()
-my_id = storage.get_config_value("agent_id", "")
+my_id = storage.get_config_value("agent_id", str(uuid.uuid4()))
+agent_name = storage.get_config_value("agent_name", "HMP-Agent")
+
+# ======================
+# Формируем TCP/UDP порты для прослушивания
+# ======================
+def get_listening_ports():
+    tcp_ports = set()
+    udp_ports = set()
+
+    for key in ["global_addresses", "local_addresses"]:
+        addresses = json.loads(storage.get_config_value(key, "[]"))
+        for a in addresses:
+            proto, hostport = a.split("://")
+            host, port = hostport.split(":")
+            port = int(port)
+            if proto == "tcp":
+                tcp_ports.add(port)
+            elif proto in ["udp", "utp"]:
+                udp_ports.add(port)
+            elif proto == "any":
+                tcp_ports.add(port)
+                udp_ports.add(port)
+
+    return sorted(tcp_ports), sorted(udp_ports)
+
+tcp_ports, udp_ports = get_listening_ports()
 
 # ======================
 # LAN Discovery
 # ======================
-def lan_discovery(udp_port: int):
-    """
-    Периодический поиск локальных агентов в сети.
-    """
+def lan_discovery():
     DISCOVERY_INTERVAL = 300  # каждые 5 минут
+    udp_port_set = set(udp_ports)
 
     while True:
         local_ip = get_local_ip()
@@ -26,23 +51,22 @@ def lan_discovery(udp_port: int):
             time.sleep(DISCOVERY_INTERVAL)
             continue
 
-        net = ipaddress.ip_network(local_ip + '/24', strict=False)  # /24 по умолчанию
+        net = ipaddress.ip_network(local_ip + '/24', strict=False)
         for ip in net.hosts():
             if str(ip) == local_ip:
-                continue  # пропускаем себя
-            try:
-                sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-                sock.settimeout(0.5)
-                msg = json.dumps({"ping": "HMP"}).encode("utf-8")
-                sock.sendto(msg, (str(ip), udp_port))
-                sock.close()
-            except:
                 continue
-
+            for port in udp_port_set:
+                try:
+                    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+                    sock.settimeout(0.5)
+                    msg = json.dumps({"id": my_id, "name": agent_name}).encode("utf-8")
+                    sock.sendto(msg, (str(ip), port))
+                    sock.close()
+                except:
+                    continue
         time.sleep(DISCOVERY_INTERVAL)
 
 def get_local_ip():
-    """Возвращает локальный IP хоста."""
     try:
         s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
         s.connect(("8.8.8.8", 80))
@@ -53,46 +77,64 @@ def get_local_ip():
         return None
 
 # ======================
-# UDP Discovery
+# UDP Discovery Listener
 # ======================
-def udp_discovery_listener(udp_port: int):
-    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-    sock.bind(("", udp_port))
+def udp_discovery_listener():
+    sockets = []
+    for port in udp_ports:
+        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        sock.bind(("", port))
+        sockets.append(sock)
+
     while True:
-        data, addr = sock.recvfrom(1024)
-        try:
-            msg = json.loads(data.decode("utf-8"))
-            peer_id = msg.get("id")
-            name = msg.get("name", "unknown")
-            addresses = msg.get("addresses", [f"{addr[0]}:{msg.get('tcp_port', udp_port)}"])
-            normalized_addresses = []
-            for a in addresses:
-                norm = storage.normalize_address(a)
-                if norm is None:
-                    continue
-                proto, hostport = norm.split("://")
-                if proto in ["udp", "any"]:
-                    normalized_addresses.append(norm)
-            if normalized_addresses:
-                storage.add_or_update_peer(peer_id, name, normalized_addresses, "discovery", "online")
-        except Exception as e:
-            print("[PeerSync] Ошибка при обработке UDP пакета:", e)
-
-def udp_discovery_sender(agent_id: str, agent_name: str, udp_port: int, tcp_port: int):
+        for sock in sockets:
+            try:
+                data, addr = sock.recvfrom(1024)
+                msg = json.loads(data.decode("utf-8"))
+                peer_id = msg.get("id")
+                if peer_id == my_id:
+                    continue  # не добавляем себя
+
+                name = msg.get("name", "unknown")
+                addresses = msg.get("addresses", [f"{addr[0]}:{sock.getsockname()[1]}"])
+                normalized_addresses = []
+                for a in addresses:
+                    norm = storage.normalize_address(a)
+                    if norm is None:
+                        continue
+                    proto, _ = norm.split("://")
+                    if proto in ["udp", "any"]:
+                        normalized_addresses.append(norm)
+
+                if normalized_addresses:
+                    storage.add_or_update_peer(peer_id, name, normalized_addresses, "discovery", "online")
+            except:
+                continue
+
+# ======================
+# UDP Discovery Sender
+# ======================
+def udp_discovery_sender():
     sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
     sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 2)
-    msg = {"id": agent_id, "name": agent_name, "tcp_port": tcp_port, "addresses": [f"127.0.0.1:{tcp_port}"]}
+
+    global_addresses = json.loads(storage.get_config_value("global_addresses", "[]"))
+    msg = {"id": my_id, "name": agent_name, "addresses": global_addresses}
+
     last_broadcast = 0
     DISCOVERY_INTERVAL = 60
     BROADCAST_INTERVAL = 600
+
     while True:
         now = time.time()
         if int(now) % DISCOVERY_INTERVAL == 0:
-            sock.sendto(json.dumps(msg).encode("utf-8"), ("239.255.0.1", udp_port))
+            for port in udp_ports:
+                sock.sendto(json.dumps(msg).encode("utf-8"), ("239.255.0.1", port))
         if now - last_broadcast > BROADCAST_INTERVAL:
             sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
-            sock.sendto(json.dumps(msg).encode("utf-8"), ("255.255.255.255", udp_port))
+            for port in udp_ports:
+                sock.sendto(json.dumps(msg).encode("utf-8"), ("255.255.255.255", port))
             last_broadcast = now
         time.sleep(1)
 
@@ -105,9 +147,8 @@ def peer_exchange():
         peers = storage.get_online_peers(limit=50)
         for peer in peers:
             peer_id, addresses = peer["id"], peer["addresses"]
-
             if peer_id == my_id:
-                continue  # пропускаем собственный агент
+                continue  # пропускаем себя
 
             try:
                 addr_list = json.loads(addresses)
@@ -125,12 +166,11 @@ def peer_exchange():
                         s.connect((host, int(port)))
                         s.sendall(b"PEER_EXCHANGE_REQUEST")
                         s.close()
-                        print(f"[PeerSync] Успешно подключились к {peer_id} ({norm})")
                         break
-                    except Exception as e:
-                        print(f"[PeerSync] Не удалось подключиться к {peer_id} ({norm}): {e}")
-            except Exception as e:
-                print(f"[PeerSync] Ошибка обработки адресов {peer_id} ({addresses}): {e}")
+                    except:
+                        continue
+            except:
+                continue
         time.sleep(PEER_EXCHANGE_INTERVAL)
 
 # ======================
@@ -138,17 +178,11 @@ def peer_exchange():
 # ======================
 def start_sync():
     print("[PeerSync] Запуск фоновой синхронизации")
-    udp_port = int(storage.get_config_value("udp_port", 4000))
-    tcp_port = int(storage.get_config_value("tcp_port", 5000))
-    agent_id = storage.get_config_value("agent_id", str(uuid.uuid4()))
-    agent_name = storage.get_config_value("agent_name", "HMP-Agent")
-
     storage.load_bootstrap()
-
-    threading.Thread(target=udp_discovery_listener, args=(udp_port,), daemon=True).start()
-    threading.Thread(target=udp_discovery_sender, args=(agent_id, agent_name, udp_port, tcp_port), daemon=True).start()
+    threading.Thread(target=udp_discovery_listener, daemon=True).start()
+    threading.Thread(target=udp_discovery_sender, daemon=True).start()
     threading.Thread(target=peer_exchange, daemon=True).start()
-    threading.Thread(target=lan_discovery, args=(udp_port,), daemon=True).start()  # новый поток
+    threading.Thread(target=lan_discovery, daemon=True).start()
 
     while True:
         time.sleep(60)

agents/tools/storage.py

@@ -881,41 +881,59 @@ class Storage:
         return None
 
     # Работа с пирам (agent_peers)
-    def add_or_update_peer(self, peer_id, name, addresses, source="discovery", status="unknown"):
+    def add_or_update_peer(self, peer_id, name, addresses, source="discovery", status="unknown", pubkey=None, capabilities=None):
         c = self.conn.cursor()
 
         # ищем существующий peer по любому совпадающему адресу
-        c.execute("SELECT id, addresses FROM agent_peers")
+        c.execute("SELECT id, addresses, pubkey, capabilities FROM agent_peers")
         rows = c.fetchall()
         existing_id = None
+        existing_addresses = []
+        existing_pubkey = None
+        existing_capabilities = {}
+
         for row in rows:
-            db_id, db_addresses_json = row
+            db_id, db_addresses_json, db_pubkey, db_capabilities_json = row
             try:
                 db_addresses = json.loads(db_addresses_json)
             except Exception:
                 db_addresses = []
+
             if any(addr in db_addresses for addr in addresses):
                 existing_id = db_id
+                existing_addresses = db_addresses
+                existing_pubkey = db_pubkey
+                try:
+                    existing_capabilities = json.loads(db_capabilities_json) if db_capabilities_json else {}
+                except:
+                    existing_capabilities = {}
                 break
 
-        if existing_id:
-            peer_id = existing_id  # используем существующий ID
+        combined_addresses = list(set(existing_addresses) | set(addresses))
+        final_peer_id = existing_id or peer_id
+        final_pubkey = pubkey or existing_pubkey
+        final_capabilities = capabilities or existing_capabilities
 
         c.execute("""
-            INSERT INTO agent_peers (id, name, addresses, source, status, last_seen)
-            VALUES (?, ?, ?, ?, ?, ?)
+            INSERT INTO agent_peers (id, name, addresses, source, status, last_seen, pubkey, capabilities)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
             ON CONFLICT(id) DO UPDATE SET
+                name=excluded.name,
                 addresses=excluded.addresses,
                 source=excluded.source,
                 status=excluded.status,
-                last_seen=excluded.last_seen
+                last_seen=excluded.last_seen,
+                pubkey=excluded.pubkey,
+                capabilities=excluded.capabilities
         """, (
-            peer_id,
+            final_peer_id,
             name,
-            json.dumps(addresses),
+            json.dumps(combined_addresses),
             source,
             status,
-            datetime.now(UTC).isoformat()
+            datetime.now(UTC).isoformat(),
+            final_pubkey,
+            json.dumps(final_capabilities)
         ))
         self.conn.commit()