Merge pull request #184 from J-A-A-M/analytics

Технічне: Додано аналітику

.github/workflows/deploy_server.yml

@@ -73,7 +73,7 @@ jobs:
                 git fetch --all
                 git checkout ${{ github.ref_name }}
                 git pull
-                bash redeploy_websocket_server.sh -m ${{ secrets.MEMCACHED_HOST }}
+                bash redeploy_websocket_server.sh -m ${{ secrets.MEMCACHED_HOST }} -s ${{ secrets.API_SECRET }} -i ${{ secrets.MEASUREMENT_ID }}
           - name: Redeploy Update Server (bin list)
             if: ${{ inputs.redeploy_update_server || inputs.redeploy_all }}
             uses: appleboy/ssh-action@v1.0.3

.github/workflows/deploy_websockets_dev.yml

@@ -0,0 +1,30 @@
+name: 🧪 Deploy Dev Websockets
+on:
+  workflow_dispatch:
+    inputs:
+      random_mode:
+        type: boolean
+        default: false
+        description: Random Alerst Mode
+      fetch_interval:
+        type: number
+        default: 1
+        description: Memcached Fetch Interval (seconds)
+
+run-name: Deploy Dev Websockets from '${{ github.ref_name }}' branch
+jobs:
+    deploy_websocket_dev:
+        runs-on: ubuntu-latest
+        steps:
+          - name: Redeploy Dev WebSockets
+            uses: appleboy/ssh-action@v1.0.3
+            with:
+              host: ${{ secrets.SERVER_IP }}
+              username: ${{ secrets.SERVER_SSH_USER }}
+              password: ${{ secrets.SERVER_SSH_PASSWORD }}
+              script: |
+                cd /root/ukraine_alarm_map/deploy/
+                git fetch --all
+                git checkout ${{ github.ref_name }}
+                git pull
+                bash redeploy_websocket_server_dev.sh -m ${{ secrets.MEMCACHED_HOST }} -s ${{ secrets.API_SECRET }} -i ${{ secrets.MEASUREMENT_ID }} -r ${{ inputs.random_mode && 'True' || 'False' }} -f ${{ inputs.fetch_interval }}

deploy/redeploy_websocket_server.sh

@@ -5,6 +5,7 @@ MEMCACHED_HOST=""
 WEBSOCKET_PORT=38440
 DEBUG_LEVEL="INFO"
 PING_INTERVAL=60
+ENVIRONMENT="PROD"
 
 # Check for arguments
 while [[ $# -gt 0 ]]; do
@@ -17,6 +18,14 @@ while [[ $# -gt 0 ]]; do
             MEMCACHED_HOST="$2"
             shift 2
             ;;
+        -s|--api-secret)
+            API_SECRET="$2"
+            shift 2
+            ;;
+        -i|--measurement-id)
+            MEASUREMENT_ID="$2"
+            shift 2
+            ;;
         -d|--debug-level)
             DEBUG_LEVEL="$2"
             shift 2
@@ -60,7 +69,7 @@ docker rm map_websocket_server || true
 
 # Deploying the new container
 echo "Deploying new container..."
-docker run --name map_websocket_server --restart unless-stopped -d  -p "$WEBSOCKET_PORT":"$WEBSOCKET_PORT" --env WEBSOCKET_PORT="$WEBSOCKET_PORT" --env DEBUG_LEVEL="$DEBUG_LEVEL" --env PING_INTERVAL="$PING_INTERVAL" --env MEMCACHED_HOST="$MEMCACHED_HOST" map_websocket_server
+docker run --name map_websocket_server --restart unless-stopped -d  -p "$WEBSOCKET_PORT":"$WEBSOCKET_PORT" --env WEBSOCKET_PORT="$WEBSOCKET_PORT" --env API_SECRET="$API_SECRET" --env MEASUREMENT_ID="$MEASUREMENT_ID" --env DEBUG_LEVEL="$DEBUG_LEVEL" --env PING_INTERVAL="$PING_INTERVAL" --env MEMCACHED_HOST="$MEMCACHED_HOST" --env ENVIRONMENT="$ENVIRONMENT" map_websocket_server
 
 echo "Container deployed successfully!"
 

deploy/redeploy_websocket_server_dev.sh

@@ -0,0 +1,88 @@
+#!/bin/bash
+
+# Default values
+MEMCACHED_HOST=""
+WEBSOCKET_PORT=38447
+DEBUG_LEVEL="INFO"
+PING_INTERVAL=60
+ENVIRONMENT="DEV"
+RANDOM_MODE="False"
+MEMCACHE_FETCH_INTERVAL=1
+
+# Check for arguments
+while [[ $# -gt 0 ]]; do
+    case "$1" in
+        -a|--tcp-port)
+            WEBSOCKET_PORT="$2"
+            shift 2
+            ;;
+        -m|--memcached-host)
+            MEMCACHED_HOST="$2"
+            shift 2
+            ;;
+        -s|--api-secret)
+            API_SECRET="$2"
+            shift 2
+            ;;
+        -i|--measurement-id)
+            MEASUREMENT_ID="$2"
+            shift 2
+            ;;
+        -d|--debug-level)
+            DEBUG_LEVEL="$2"
+            shift 2
+            ;;
+        -p|--ping-interval)
+            PING_INTERVAL="$2"
+            shift 2
+            ;;
+        -r|--random-mode)
+            RANDOM_MODE="$2"
+            shift 2
+            ;;
+        -f|--fetch-interval)
+            MEMCACHE_FETCH_INTERVAL="$2"
+            shift 2
+            ;;
+        *)
+            echo "Unknown argument: $1"
+            exit 1
+            ;;
+    esac
+done
+
+echo "WEBSOCKET SERVER DEV"
+
+echo "MEMCACHED_HOST: $MEMCACHED_HOST"
+echo "WEBSOCKET_PORT: $WEBSOCKET_PORT"
+echo "DEBUG_LEVEL: $DEBUG_LEVEL"
+echo "PING_INTERVAL: $PING_INTERVAL"
+echo "ENVIRONMENT: $ENVIRONMENT"
+echo "RANDOM_MODE: $RANDOM_MODE"
+echo "MEMCACHE_FETCH_INTERVAL: $MEMCACHE_FETCH_INTERVAL"
+
+
+# Updating the Git repo
+echo "Updating Git repo..."
+#cd /path/to/your/git/repo
+git pull
+
+# Moving to the deployment directory
+echo "Moving to deployment directory..."
+cd websocket_server
+
+# Building Docker image
+echo "Building Docker image..."
+docker build -t map_websocket_server_dev -f Dockerfile .
+
+# Stopping and removing the old container (if exists)
+echo "Stopping and removing old container..."
+docker stop map_websocket_server_dev || true
+docker rm map_websocket_server_dev || true
+
+# Deploying the new container
+echo "Deploying new container..."
+docker run --name map_websocket_server_dev --restart unless-stopped -d  -p "$WEBSOCKET_PORT":"$WEBSOCKET_PORT" --env WEBSOCKET_PORT="$WEBSOCKET_PORT" --env API_SECRET="$API_SECRET" --env MEASUREMENT_ID="$MEASUREMENT_ID" --env DEBUG_LEVEL="$DEBUG_LEVEL" --env PING_INTERVAL="$PING_INTERVAL" --env MEMCACHED_HOST="$MEMCACHED_HOST" --env ENVIRONMENT="$ENVIRONMENT" --env RANDOM_MODE="$RANDOM_MODE" --env MEMCACHE_FETCH_INTERVAL="$MEMCACHE_FETCH_INTERVAL" map_websocket_server_dev
+
+echo "Container deployed successfully!"
+

deploy/web_server/web_server.py

@@ -416,11 +416,14 @@ async def stats(request):
 
         websocket_clients = await mc.get(b'websocket_clients')
         websocket_clients_data = json.loads(websocket_clients.decode('utf-8')) if websocket_clients else {}
+        websocket_clients_dev = await mc.get(b'websocket_clients_dev')
+        websocket_clients_dev_data = json.loads(websocket_clients_dev.decode('utf-8')) if websocket_clients_dev else {}
 
         tcp_clients = await dataparcer(tcp_clients_data, 'tcp')
         websocket_clients = await dataparcer(websocket_clients_data, 'websockets')
+        websocket_clients_dev = await dataparcer(websocket_clients_dev_data, 'websockets_dev')
 
-        map_clients_data = tcp_clients + websocket_clients
+        map_clients_data = tcp_clients + websocket_clients + websocket_clients_dev
 
         return JSONResponse ({
 

deploy/websocket_server/requirements.txt

@@ -1,3 +1,4 @@
 aiomcache==0.8.1
 websockets==12.0
-geoip2==4.8.0
+geoip2==4.8.0
+ga4mp==2.0.4

deploy/websocket_server/websocket_server.py

@@ -7,13 +7,18 @@
 from aiomcache import Client
 from geoip2 import database, errors
 from functools import partial
-from datetime import datetime, timezone
+from datetime import datetime, timezone, timedelta
+from ga4mp import GtagMP
+from ga4mp.store import DictStore
 
 debug_level = os.environ.get('DEBUG_LEVEL') or 'INFO'
 websocket_port = os.environ.get('WEBSOCKET_PORT') or 38440
 ping_interval = int(os.environ.get('PING_INTERVAL', 60))
 memcache_fetch_interval = int(os.environ.get('MEMCACHE_FETCH_INTERVAL', 1))
-random_mode = os.environ.get('RANDOM_MODE') or False
+random_mode = os.environ.get('RANDOM_MODE', 'False').lower() in ('true', '1', 't')
+api_secret = os.environ.get('API_SECRET') or ''
+measurement_id = os.environ.get('MEASUREMENT_ID') or ''
+environment = os.environ.get('ENVIRONMENT') or 'PROD'
 
 logging.basicConfig(level=debug_level,
                     format='%(asctime)s %(levelname)s : %(message)s')
@@ -35,6 +40,7 @@ def __init__(self):
         self.bins = '[]'
         self.test_bins = '[]'
         self.clients = {}
+        self.trackers = {}
         self.blocked_ips = []
 
 
@@ -142,9 +148,11 @@ async def echo(websocket, path):
         response = geo.city(client_ip)
         city = response.city.name or 'not-in-db'
         region = response.subdivisions.most_specific.name or 'not-in-db'
+        country = response.country.iso_code or 'not-in-db'
     except errors.AddressNotFoundError:
         city = 'not-found'
         region = 'not-found'
+        country = 'not-found'
 
     # if response.country.iso_code != 'UA' and response.continent.code != 'EU':
     #     shared_data.blocked_ips.append(client_ip)
@@ -159,9 +167,12 @@ async def echo(websocket, path):
         'firmware': 'unknown',
         'chip_id': 'unknown',
         'city': city,
-        'region': region
+        'region': region,
+        'country': country,
     }
 
+    tracker = shared_data.trackers[f'{client_ip}_{client_port}'] = GtagMP(api_secret=api_secret, measurement_id=measurement_id, client_id='temp_id')
+
     match path:
         case "/data_v1":
             data_task = asyncio.create_task(alerts_data(websocket, client, shared_data, AlertVersion.v1))
@@ -196,18 +207,52 @@ def split_message(message):
                         logger.info(f"{client_ip}:{client_id} <<< district {payload} ")
                     case 'firmware':
                         client['firmware'] = data
+                        parts = data.split('_', 1)
+                        tracker.store.set_user_property('firmware_v', parts[0])
+                        tracker.store.set_user_property('identifier', parts[1])
                         logger.warning(f"{client_ip}:{client_id} >>> firmware saved")
+                    case 'user_info':
+                        json_data = json.loads(data)
+                        for key, value in json_data.items():
+                            tracker.store.set_user_property(key, value)
                     case 'chip_id':
                         client['chip_id'] = data
+                        tracker.client_id = data
+                        tracker.store.set_session_parameter('session_id', f'{data}_{datetime.now().timestamp()}')
+                        tracker.store.set_user_property('user_id', data)
+                        tracker.store.set_user_property('chip_id', data)
+                        tracker.store.set_user_property('country', country)
+                        tracker.store.set_user_property('region', region)
+                        tracker.store.set_user_property('city', city)
+                        tracker.store.set_user_property('ip', client_ip)
+                        online_event = tracker.create_new_event('status')
+                        online_event.set_event_param('online', 'true')
+                        tracker.send(events=[online_event], date=datetime.now())
                         logger.info(f"{client_ip}:{client_id} >>> chip_id saved")
+                    case 'pong':
+                        ping_event = tracker.create_new_event('ping')
+                        ping_event.set_event_param('state', 'alive')
+                        tracker.send(events=[ping_event], date=datetime.now())
+                        logger.info(f"{client_ip}:{client_id} >>> ping analytics sent")
+                    case 'settings':
+                        json_data = json.loads(data)
+                        settings_event = tracker.create_new_event('settings')
+                        for key, value in json_data.items():
+                            settings_event.set_event_param(key, value)
+                        tracker.send(events=[settings_event], date=datetime.now())
+                        logger.info(f"{client_ip}:{client_id} >>> settings analytics sent")
                     case _:
                         logger.info(f"{client_ip}:{client_id} !!! unknown data request")
     except websockets.exceptions.ConnectionClosedError as e:
         logger.error(f"Connection closed with error - {e}")
     except Exception as e:
         pass
     finally:
+        offline_event = tracker.create_new_event('status')
+        offline_event.set_event_param('online', 'false')
+        tracker.send(events=[offline_event], date=datetime.now())
         data_task.cancel()
+        del shared_data.trackers[f'{client_ip}_{client_port}']
         del shared_data.clients[f'{client_ip}_{client_port}']
         try:
             await data_task
@@ -297,7 +342,8 @@ async def print_clients(shared_data, mc):
             logger.info(f"Clients:")
             for client, data in shared_data.clients.items():
                 logger.info(client)
-            await mc.set(b"websocket_clients", json.dumps(shared_data.clients).encode('utf-8'))
+            websoket_key = b"websocket_clients" if environment == 'PROD' else b"websocket_clients_dev"
+            await mc.set(websoket_key, json.dumps(shared_data.clients).encode('utf-8'))
         except Exception as e:
             logger.error(f"Error in update_shared_data: {e}")
 
@@ -312,11 +358,14 @@ async def get_data_from_memcached(mc):
     weather_full_cached = await mc.get(b"weather")
 
     if random_mode:
-        values = [0] * 25
-        position = random.randint(0, 25)
-        values.insert(position, 1)
-        alerts_cached_data_v1 = json.dumps(values[:26])
-        alerts_cached_data_v2 = json.dumps(values[:26])
+        values_v1 = []
+        values_v2 = []
+        for i in range(26):
+            values_v1.append(random.randint(0, 3))
+            diff = random.randint(0, 600)
+            values_v2.append([random.randint(0, 1), (datetime.now() - timedelta(seconds=diff)).strftime('%Y-%m-%dT%H:%M:%SZ')])
+        alerts_cached_data_v1 = json.dumps(values_v1[:26])
+        alerts_cached_data_v2 = json.dumps(values_v2[:26])
     else:
         if alerts_cached_v1:
             alerts_cached_data_v1 = alerts_cached_v1.decode('utf-8')