fix(automations): operate check new feed on cron

dineshpinto · dineshpinto · commit 26c759bd47a8 · 2025-07-22T17:53:42.000+04:00
diff --git a/.github/workflows/automatic-feed-request.yml b/.github/workflows/automatic-feed-request.yml
@@ -1,18 +1,20 @@
 name: Automatic Feed Request
 
 on:
+  schedule:
+    # Once per day at 03:00 UTC
+    - cron: "0 3 * * *"
+  workflow_dispatch:
   push:
-    branches:
-      - main
-
-permissions:
-  contents: read
-  issues: write
+    branches: [main]
 
 jobs:
   check_and_report:
     name: Run automatic feed request
     runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      issues: write
     defaults:
       run:
         working-directory: ./automations
@@ -25,19 +27,12 @@ jobs:
           enable-cache: true
       - name: Install dependencies
         run: uv sync
-      - name: Generate issues file
+      - name: Check for new feeds
         run: uv run check_new_feed.py
-      - name: Check if issues file is empty
-        id: check_file
-        run: |
-          if [ -s "automations/issues.md" ]; then
-            echo "file_is_empty=false" >> $GITHUB_OUTPUT
-          else
-            echo "file_is_empty=true" >> $GITHUB_OUTPUT
-          fi
+        id: feed
       - name: Create issue
         uses: JasonEtco/create-an-issue@v2
-        if: steps.check_file.outputs.file_is_empty == 'false'
+        if: steps.feed.outcome == 'success' && steps.feed.conclusion != 'neutral'
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
         with:
diff --git a/automations/check_new_feed.py b/automations/check_new_feed.py
@@ -2,25 +2,28 @@
 import logging
 import sys
 from pathlib import Path
-from typing import Any
+from typing import Any, Final
 
 import requests
-from pycoingecko import CoinGeckoAPI
+from pycoingecko import CoinGeckoAPI  # pyright: ignore[reportMissingTypeStubs]
+from tenacity import retry, stop_after_attempt, wait_exponential_jitter
 from web3 import Web3
 from web3.contract import Contract
 
 # Configuration
-RPC_URL = "https://flare-api.flare.network/ext/C/rpc"
-REGISTRY_ADDRESS = "0xaD67FE66660Fb8dFE9d6b1b4240d8650e30F6019"
-EXPLORER_API_URL = "https://flare-explorer.flare.network/api"
-ISSUES_FILE = Path("issues.md")
-MAX_MARKET_CAP_RANK = 100
-HEADER_TEMPLATE = """---
+RPC_URL: Final[str] = "https://flare-api.flare.network/ext/C/rpc"
+REGISTRY_ADDRESS: Final[str] = "0xaD67FE66660Fb8dFE9d6b1b4240d8650e30F6019"
+EXPLORER_API_URL: Final[str] = "https://flare-explorer.flare.network/api"
+ISSUES_FILE: Final[Path] = Path("issues.md")
+MAX_MARKET_CAP_RANK: Final[int] = 100
+HEADER_TEMPLATE: Final[str] = """---
 title: "[auto_req]: Potential New Feeds"
 assignees: dineshpinto
 labels: "enhancement"
 ---
 """
+
+# Logging
 logging.basicConfig(
     encoding="utf-8",
     level=logging.INFO,
@@ -29,101 +32,113 @@
 logger = logging.getLogger(__name__)
 
 
-def get_contract_abi(contract_address: str) -> dict[str, Any]:
-    """Fetch the ABI for a contract from the Chain Explorer API."""
-    params = {"module": "contract", "action": "getabi", "address": contract_address}
-    headers = {"accept": "application/json"}
+class ExplorerError(RuntimeError):
+    """Raised when the chain explorer cannot provide a contract ABI."""
+
 
+@retry(stop=stop_after_attempt(3), wait=wait_exponential_jitter(initial=1, max=8))
+def _get(
+    session: requests.Session, **kwargs: dict[str, str] | str
+) -> requests.Response:
+    """`requests.get` with automatic retry and logging."""
+    logger.debug("GET %s", kwargs.get("url") or kwargs.get("url"))
+    resp = session.get(**kwargs, timeout=10)  # pyright: ignore[reportArgumentType]
+    resp.raise_for_status()
+    return resp
+
+
+def get_contract_abi(
+    contract_address: str, session: requests.Session
+) -> list[dict[str, Any]]:
+    """Return ABI for `contract_address` from the explorer API."""
+    params = {"module": "contract", "action": "getabi", "address": contract_address}
     try:
-        response = requests.get(
-            EXPLORER_API_URL, params=params, headers=headers, timeout=10
+        r = _get(
+            session,
+            url=EXPLORER_API_URL,
+            params=params,
+            headers={"accept": "application/json"},
         )
-        response.raise_for_status()
-        result = response.json().get("result")
-        return json.loads(result)
-    except (requests.RequestException, ValueError, json.JSONDecodeError):
-        logger.exception("Error fetching ABI for contract")
-        sys.exit(1)
+        return json.loads(r.json()["result"])
+    except (requests.RequestException, json.JSONDecodeError, KeyError) as exc:
+        msg = f"Could not fetch ABI for {contract_address}"
+        raise ExplorerError(msg) from exc
 
 
-def format_coin_info(coin_data: dict[str, Any]) -> str:
-    """Format coin data dictionary to a readable string."""
-    coin_info = {
-        "name": coin_data.get("name", "N/A"),
-        "symbol": coin_data.get("symbol", "N/A"),
-        "coingecko_id": coin_data.get("id", "N/A"),
-        "price_change_percentage_24h": coin_data.get("data", {})
-        .get("price_change_percentage_24h", {})
-        .get("usd", "N/A"),
-        "total_volume": coin_data.get("data", {}).get("total_volume", "N/A"),
-        "coingecko_link": f"https://www.coingecko.com/en/coins/{coin_data.get('id', '')}",
-        "description": coin_data.get("data", {}).get("content", {}).get("description")
-        if coin_data.get("data", {}).get("content", {})
-        else "",
-    }
+def decode_feed_name(feed: bytes) -> str:
+    """Convert feed bytes (returned by contract) to a plain symbol."""
+    return feed[1:].decode().rstrip("\x00").split("/")[0]
 
-    return "\n".join(f"{key}: {value}" for key, value in coin_info.items())
 
+def get_current_feeds(contract: Contract) -> set[str]:
+    """Return a **set** of existing feed symbols."""
+    feeds: list[bytes] = contract.functions.fetchAllCurrentFeeds().call()[0]
+    return {decode_feed_name(f) for f in feeds}
 
-def get_current_feeds(contract: Contract) -> list[str]:
-    """Fetch the current block latency feeds from the contract."""
-    try:
-        feeds = contract.functions.fetchAllCurrentFeeds().call()
-        return [
-            feed[1:].decode("utf-8").rstrip("\x00").split("/")[0] for feed in feeds[0]
-        ]
-    except Exception:
-        logger.exception("Error fetching current feeds")
-        sys.exit(1)
 
+def prettify_coin(coin: dict[str, Any]) -> str:
+    """Convert `coin` dict into a readable block of Markdown."""
+    d = coin.get("data", {})
+    change_24h = d.get("price_change_percentage_24h", {}).get("usd", "N/A")
+    total_vol = d.get("total_volume", "N/A")
 
-def write_issues_file(path: Path, header: str, coins: list[dict[str, Any]]) -> None:
-    """Write coin data to the issues.md file."""
-    with path.open("w", encoding="utf-8") as file:
-        file.write(header)
-        file.write("Coins matching criteria:\n\n")
-        for coin in coins:
-            file.write(f"## {coin['name']}\n")
-            file.write(format_coin_info(coin))
-            file.write("\n\n")
-    logger.info("New feeds written to %s", path)
+    return (
+        f"name: {coin.get('name', 'N/A')}\n"
+        f"symbol: {coin.get('symbol', 'N/A')}\n"
+        f"coingecko_id: {coin.get('id', 'N/A')}\n"
+        f"price_change_percentage_24h: {change_24h}\n"
+        f"total_volume: {total_vol}\n"
+        f"coingecko_link: https://www.coingecko.com/en/coins/{coin.get('id', '')}\n"
+    )
 
 
-if __name__ == "__main__":
+def write_issue(coins: list[dict[str, Any]]) -> None:
+    """Write the Markdown issue file."""
+    body = ["Coins matching criteria:\n"]
+    body += [f"## {c['name']}\n{prettify_coin(c)}\n" for c in coins]
+    ISSUES_FILE.write_text(HEADER_TEMPLATE + "\n".join(body), encoding="utf-8")
+    logger.info("Wrote %d coin(s) to %s", len(coins), ISSUES_FILE)
+
+
+def main() -> None:
+    cg = CoinGeckoAPI()
+    session = requests.Session()
+
     w3 = Web3(Web3.HTTPProvider(RPC_URL))
-    logger.info("Connected to RPC `%s`", RPC_URL)
+    if not w3.is_connected():
+        msg = f"Cannot reach RPC at {RPC_URL}"
+        raise ConnectionError(msg)
+    logger.info("Connected to RPC %s", RPC_URL)
 
-    # Get contract registry
     registry = w3.eth.contract(
         address=Web3.to_checksum_address(REGISTRY_ADDRESS),
-        abi=get_contract_abi(REGISTRY_ADDRESS),
+        abi=get_contract_abi(REGISTRY_ADDRESS, session),
     )
-
-    # Set up contract
-    fast_updater_address = registry.functions.getContractAddressByName(
-        "FastUpdater"
-    ).call()
+    updater_addr = registry.functions.getContractAddressByName("FastUpdater").call()
     fast_updater = w3.eth.contract(
-        address=Web3.to_checksum_address(fast_updater_address),
-        abi=get_contract_abi(fast_updater_address),
+        address=Web3.to_checksum_address(updater_addr),
+        abi=get_contract_abi(updater_addr, session),
     )
-    logger.info("Connected to FastUpdater contract `%s`", fast_updater_address)
+    logger.info("FastUpdater contract %s", fast_updater.address)
 
-    # Query block latency feeds
     current_feeds = get_current_feeds(fast_updater)
 
-    # Fetch trending coins
-    cg = CoinGeckoAPI()
-    trending = cg.get_search_trending()
-
-    # Filter coins based on criteria
-    selected_coins = [
-        coin["item"]
-        for coin in trending["coins"]
-        if coin["item"].get("market_cap_rank", float("inf")) < MAX_MARKET_CAP_RANK
-        and coin["item"].get("symbol") not in current_feeds
+    trending = cg.get_search_trending()["coins"]  # pyright: ignore[reportUnknownMemberType]
+    logger.info("Found potential coins %s", [f"{c['item']['name']}" for c in trending])
+    candidates = [
+        c["item"]
+        for c in trending
+        if c["item"].get("market_cap_rank", float("inf")) < MAX_MARKET_CAP_RANK
+        and c["item"]["symbol"] not in current_feeds
     ]
+    if not candidates:
+        sys.exit(78) # Skips GitHub workflow 'if' steps
+    write_issue(candidates)
 
-    # Write results to issues file
-    if selected_coins:
-        write_issues_file(ISSUES_FILE, HEADER_TEMPLATE, selected_coins)
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception:
+        logger.exception("Fatal error")
+        sys.exit(1)
diff --git a/automations/feed_table_generator.py b/automations/feed_table_generator.py
@@ -156,7 +156,7 @@ def generate_feed_data(
 
 
 if __name__ == "__main__":
-    logging.info("Running Feed Table automation...")
+    logger.info("Running Feed Table automation...")
 
     w3 = Web3(Web3.HTTPProvider(RPC_URL))
     logger.debug("Connected to RPC `%s`", RPC_URL)
@@ -199,7 +199,7 @@ def generate_feed_data(
     anchor_data = generate_feed_data(feed_names, anchor_risk, coins_list)
     write_data_to_file(ANCHOR_FEEDS_PATH, anchor_data)
 
-    logging.info(
+    logger.info(
         "Feed Table automation: Data successfully saved to %s and %s",
         BLOCK_LATENCY_FEEDS_PATH,
         ANCHOR_FEEDS_PATH,
diff --git a/automations/pyproject.toml b/automations/pyproject.toml
@@ -6,6 +6,7 @@ readme = "README.md"
 requires-python = ">=3.12"
 dependencies = [
     "pycoingecko>=3.1.0",
+    "tenacity>=9.1.2",
     "tqdm>=4.67.1",
     "web3>=7.2.0",
 ]
@@ -16,6 +17,13 @@ dev-dependencies = [
     "ruff>=0.6.4",
 ]
 
+[tool.ruff]
+target-version = "py312"
+
 [tool.ruff.lint]
 select = ["ALL"]
 ignore = ["S101", "D", "ISC001", "COM812", "T201", "E501"]
+
+[tool.pyright]
+pythonVersion = "3.12"
+typeCheckingMode = "strict"
diff --git a/automations/solidity_reference_table_generator.py b/automations/solidity_reference_table_generator.py
@@ -5,6 +5,7 @@
 from web3 import Web3
 
 logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
 
 
 NETWORK_RPCS = {
@@ -57,13 +58,13 @@ def get_solidity_reference(
                     {"name": name, "address": address}
                 )
         except Exception:
-            logging.exception("Error fetching data from %s", network_name)
+            logger.exception("Error fetching data from %s", network_name)
 
     return solidity_reference
 
 
 if __name__ == "__main__":
-    logging.info("Running Solidity Reference automation...")
+    logger.info("Running Solidity Reference automation...")
 
     solidity_reference = get_solidity_reference(
         NETWORK_RPCS, REGISTRY_ADDRESS, REGISTRY_ABI
@@ -73,6 +74,6 @@ def get_solidity_reference(
     with output_file.open("w") as f:
         json.dump(solidity_reference, f, indent=4)
 
-    logging.info(
+    logger.info(
         "Solidity Reference automation: Data successfully saved to %s", output_file
     )
diff --git a/automations/uv.lock b/automations/uv.lock
