Merge pull request #184 from UW-Macrostrat/stats-post

POST usage stats

Author: davenquinn

services/usage-stats/.gitignore

@@ -1 +1 @@
-.env
+.env

services/usage-stats/Dockerfile

@@ -0,0 +1,24 @@
+FROM python:3.11-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV PYTHONUNBUFFERED=1
+ENV POETRY_VERSION=1.8.2
+
+WORKDIR /app
+
+# Install system dependencies including PostgreSQL dev headers
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    libpq-dev gcc build-essential curl \
+    && curl -sSL https://install.python-poetry.org | python3 - --version $POETRY_VERSION \
+    && ln -s /root/.local/bin/poetry /usr/local/bin/poetry \
+    && apt-get remove -y curl \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY pyproject.toml poetry.lock ./
+
+RUN poetry config virtualenvs.create false \
+    && poetry install --no-root --no-interaction --no-ansi
+
+COPY . .
+
+CMD ["python", "worker.py"]

services/usage-stats/Makefile

@@ -0,0 +1,15 @@
+docker-build:
+	docker build -t usage-stats .
+
+docker-run:
+	docker run -p 8000:8000 usage-stats
+
+docker:
+	docker build -t usage-stats .
+	docker run -p 8000:8000 usage-stats
+
+worker:
+	python3 worker.py
+
+install:
+	pip install -r requirements.txt

services/usage-stats/README.md

@@ -0,0 +1,29 @@
+# Usage Stats
+
+This worker parses the Matomo database and post subsetted logs to the usage_stats schema for **Rockd** and **Macrostrat**
+
+## Requirements
+
+- Python 3.7+
+- Packages listed in `requirements.txt`
+
+## Local installation
+
+   ```bash
+   make install
+   ```
+
+## Running
+Running the worker reads data from the Matomo database and writes the parsed logs to the `usage_stats.macrostrat_stats` and `usage_stats.rockd_stats` tables in the Macrostrat database
+
+You can either run the app directly using
+
+   ```bash
+   make app
+   ```
+
+Or via docker using
+   
+   ```bash
+   make docker
+   ```

services/usage-stats/poetry.lock

@@ -0,0 +1,21 @@
+[tool.poetry]
+name = "usage-stats"
+version = "0.1.0"
+description = ""
+authors = ["davidsklar99 <[email protected]>"]
+readme = "README.md"
+packages = []  # ensures it's not treated as a package
+package-mode = false
+
+[tool.poetry.dependencies]
+python = ">=3.11,<4.0"
+requests = ">=2.32.4,<3.0.0"
+asyncpg = ">=0.30.0,<0.31.0"
+sqlalchemy = ">=2.0.41,<3.0.0"
+psycopg2 = ">=2.9.10,<3.0.0"
+asyncmy = ">=0.2.10,<0.3.0"
+dotenv = ">=0.9.9,<0.10.0"
+
+[build-system]
+requires = ["poetry-core>=2.0.0,<3.0.0"]
+build-backend = "poetry.core.masonry.api"

services/usage-stats/pyproject.toml

@@ -0,0 +1,5 @@
+# Matomo connection
+MARIADB_URL=<url>
+
+# Macrostrat connection
+DATABASE_URL=<url>

services/usage-stats/sample.env

@@ -0,0 +1,39 @@
+import asyncio
+import os
+
+from dotenv import load_dotenv
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import create_async_engine
+
+load_dotenv()
+
+DATABASE_URL = os.getenv("DATABASE_URL")
+
+engine = create_async_engine(DATABASE_URL, echo=True)
+
+
+async def insert(payload=None, table_name=None):
+    async with engine.connect() as conn:
+        if payload is None:
+            print("No payload provided")
+            return
+
+        if table_name is None:
+            print("No table name provided")
+            return
+
+        if table_name not in ["macrostrat", "rockd"]:
+            print("Invalid table name provided")
+            return
+
+        result = await conn.execute(
+            text(
+                f"""
+            INSERT INTO usage_stats.{table_name}_stats 
+                (lat, lng, date, ip, matomo_id) 
+            VALUES (:lat, :lng, :date, :ip, :matomo_id)
+        """
+            ),
+            (payload),
+        )
+        await conn.commit()

services/usage-stats/src/insert.py

@@ -0,0 +1,36 @@
+import asyncio
+import os
+
+from dotenv import load_dotenv
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import create_async_engine
+
+load_dotenv()
+
+DATABASE_URL = os.getenv("DATABASE_URL")
+
+engine = create_async_engine(DATABASE_URL, echo=True)
+
+
+async def get_last_id(table_name=None):
+    async with engine.connect() as conn:
+        if table_name is None:
+            print("No table name provided")
+            return
+
+        print("Fetching data")
+
+        result = await conn.execute(
+            text(f"SELECT MAX(matomo_id) FROM usage_stats.{table_name}_stats")
+        )
+        rows = result.fetchall()
+        id = rows[0][0]
+
+        # Note: If you want to regenerate the table, truncate it first, then
+        # this script will repopulate it automatically.
+
+        # Check if table is empty
+        if not id:
+            return 0
+        else:
+            return id

services/usage-stats/src/last_id.py

@@ -0,0 +1,72 @@
+import asyncio
+import os
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+from sqlalchemy.orm import sessionmaker
+from src.insert import insert
+from src.last_id import get_last_id
+
+BATCH_SIZE = 1000  # Adjust as needed
+
+# Database URL format: mysql+asyncmy://user:password@host:port/database
+DATABASE_URL = os.getenv(
+    "MARIADB_URL", "mysql+asyncmy://user:password@localhost:3306/database"
+)
+
+# Create async engine
+engine = create_async_engine(DATABASE_URL, echo=True)
+
+# Async session factory
+AsyncSessionLocal = sessionmaker(
+    bind=engine, expire_on_commit=False, class_=AsyncSession
+)
+
+
+async def get_data(last_id):
+    async with AsyncSessionLocal() as session:
+        query = text(
+            """
+            SELECT
+                location_latitude AS lat,
+                location_longitude AS lng,
+                visit_first_action_time AS date,
+                idvisitor AS ip,
+                idvisit as matomo_id
+            FROM matomo_log_visit
+            WHERE 
+                idvisit > :last_id
+                AND location_latitude IS NOT NULL
+                AND location_longitude IS NOT NULL
+                AND visit_first_action_time > '2025-07-02'
+            LIMIT :batch_size
+        """
+        )
+
+        result = await session.execute(
+            query, {"last_id": last_id, "batch_size": BATCH_SIZE}
+        )
+        rows = result.fetchall()
+
+        if not rows:
+            print("No more rows to process.")
+            return
+
+        payload = [
+            {
+                "lat": float(row.lat),
+                "lng": float(row.lng),
+                "date": row.date,
+                "ip": str(row.ip),
+                "matomo_id": row.matomo_id,
+            }
+            for row in rows
+        ]
+
+        await insert(payload, "macrostrat")
+
+
+async def get_macrostrat_data():
+    last_id = await get_last_id("macrostrat")
+    await get_data(last_id)
+    print("Data fetching completed.")