scripts/get_mech_info.py

from dataclasses import dataclass
from string import Template
from typing import Any
from datetime import datetime, timedelta, UTC
from utils import SUBGRAPH_API_KEY
import requests

OLD_MECH_SUBGRAPH_URL = (
    "https://api.thegraph.com/subgraphs/name/stakewise/ethereum-gnosis"
)
# MECH_SUBGRAPH_URL = "https://api.studio.thegraph.com/query/57238/mech/0.0.2"
NETWORK_SUBGRAPH_URL = Template(
    """https://gateway-arbitrum.network.thegraph.com/api/${subgraph_api_key}/subgraphs/id/FxV6YUix58SpYmLBwc9gEHkwjfkqwe1X5FJQjn8nKPyA"""
)

SUBGRAPH_HEADERS = {
    "Accept": "application/json, multipart/mixed",
    "Content-Type": "application/json",
}
QUERY_BATCH_SIZE = 1000
DATETIME_60_DAYS_AGO = datetime.now(UTC) - timedelta(days=60)
BLOCK_NUMBER = Template(
    """
    { 
        blocks(
            first: 1,
            orderBy: timestamp,
            orderDirection: asc,
            where: {
                timestamp_gte: "${timestamp_from}",
                timestamp_lte: "${timestamp_to}"
            }
        ){
            id,
            number,
        }
    }
    """
)


def fetch_block_number(timestamp_from: int, timestamp_to: int) -> dict:
    """Get a block number by its timestamp margins."""

    query = BLOCK_NUMBER.substitute(
        timestamp_from=timestamp_from, timestamp_to=timestamp_to
    )
    # print(f"Sending query for the subgraph = {query}")
    network_subgraph_url = NETWORK_SUBGRAPH_URL.substitute(
        subgraph_api_key=SUBGRAPH_API_KEY
    )
    response = requests.post(
        network_subgraph_url,
        headers=SUBGRAPH_HEADERS,
        json={"query": query},
        timeout=300,
    )

    result_json = response.json()
    print(f"Response of the query={result_json}")
    blocks = result_json.get("data", {}).get("blocks", "")
    if len(blocks) == 0:
        raise ValueError(f"The query {query} did not return any results")
    return blocks[0]


def get_mech_info_2024() -> dict[str, Any]:
    """Query the subgraph to get the 2024 information from mech."""

    date = "2024-01-01"
    datetime_jan_2024 = datetime.strptime(date, "%Y-%m-%d")
    timestamp_jan_2024 = int(datetime_jan_2024.timestamp())
    margin = timedelta(seconds=5)
    timestamp_jan_2024_plus_margin = int((datetime_jan_2024 + margin).timestamp())

    jan_block_number = fetch_block_number(
        timestamp_jan_2024, timestamp_jan_2024_plus_margin
    )
    # expecting only one block
    jan_block_number = jan_block_number.get("number", "")
    if jan_block_number.isdigit():
        jan_block_number = int(jan_block_number)

    if jan_block_number == "":
        raise ValueError(
            "Could not find a valid block number for the first of January 2024"
        )
    MECH_TO_INFO = {
        # this block number is when the creator had its first tx ever, and after this mech's creation
        "0xff82123dfb52ab75c417195c5fdb87630145ae81": (
            "old_mech_abi.json",
            jan_block_number,
        ),
        # this block number is when this mech was created
        "0x77af31de935740567cf4ff1986d04b2c964a786a": (
            "new_mech_abi.json",
            jan_block_number,
        ),
    }
    return MECH_TO_INFO


def get_mech_info_last_60_days() -> dict[str, Any]:
    """Query the subgraph to get the last 60 days of information from mech."""

    timestamp_60_days_ago = int((DATETIME_60_DAYS_AGO).timestamp())
    margin = timedelta(seconds=5)
    timestamp_60_days_ago_plus_margin = int((DATETIME_60_DAYS_AGO + margin).timestamp())

    last_month_block_number = fetch_block_number(
        timestamp_60_days_ago, timestamp_60_days_ago_plus_margin
    )
    # expecting only one block
    last_month_block_number = last_month_block_number.get("number", "")
    if last_month_block_number.isdigit():
        last_month_block_number = int(last_month_block_number)

    if last_month_block_number == "":
        raise ValueError("Could not find a valid block number for last month data")

    MECH_TO_INFO = {
        # this block number is when the creator had its first tx ever, and after this mech's creation
        "0xff82123dfb52ab75c417195c5fdb87630145ae81": (
            "old_mech_abi.json",
            last_month_block_number,
        ),
        # this block number is when this mech was created
        "0x77af31de935740567cf4ff1986d04b2c964a786a": (
            "new_mech_abi.json",
            last_month_block_number,
        ),
    }
    return MECH_TO_INFO


if __name__ == "__main__":
    result = get_mech_info_last_60_days()
    print(result)