auto/message/message_coin_detail.py

# -*- coding: utf-8 -*-
'''
使用 httpx 获取 coinmarketcap 最新数字币数据
'''
import time
import os
import sys
import httpx
import re
from playwright.sync_api import sync_playwright

sys.path.append(os.path.join(os.path.abspath(__file__).split('auto')[0] + 'auto'))

from utils.utils_send_gotify import *


def get_chainalert():
    coin_follow_list = ['BTC', 'ETH', 'SOL', 'GRASS', 'SUI', 'DOGE', 'ARB', 'ATH', 'MOVE']
    # coin_follow_list = ['DOGE']

    text = ''

    data = {}
    with sync_playwright() as playwright:
        browser = playwright.chromium.launch(headless=True)
        context = browser.new_context()
        page = context.new_page()

        page.goto("https://chainalert.me/data/marketprice")

        for coin in coin_follow_list:
            page.fill("#marketprice_search_input", coin)

            page.click("#marketprice_search_btn")

            page.wait_for_timeout(1500)

            elements = page.query_selector_all("#listdata_header > li:nth-child(2) > div")
            text_content = [element.text_content() for element in elements]

            temp_data = [s.replace('↑', '').replace('↓', '').split() for s in text_content]
            print(temp_data[0])

            # 去重
            if temp_data[0][1] not in data:
                data[temp_data[0][1]] = temp_data[0]

        # 将去重后的数据拼接到 text 中
        for coin, value in data.items():
            text += f'Name: {coin}\n'
            text += f'Ranking: {value[0]}\n'
            text += f'Price: {value[2]}\n'
            text += f'24H Transaction Volume: {value[3]}\n'
            text += f'24H Price Change: {value[4]}\n'
            text += f'Market Capitalization: {value[5]}\n'
            text += f'Diluted Market Value: {value[6]}\n'
            text += '\n\n'

        time.sleep(1)

        page.goto("https://chainalert.me/data/")
        page.wait_for_timeout(1500)
        elements = page.query_selector_all(
            '#greedy-galay-div > div.greedy_index-svg-value-div > div.greedy-svg-value > strong')
        text_content = [element.text_content() for element in elements]
        print(f'VIX: {text_content[0] if text_content else "No Data"}')
        if text_content:
            text += f'VIX: {text_content[0]}'
        else:
            text += 'VIX: No Data'

        context.close()
        browser.close()

        return text


def get_gas_value():
    url = "https://a5.maiziqianbao.net/api/v1/chains/EVM/1/gas_price"

    headers = {
        "Host": "a5.maiziqianbao.net",
        "Connection": "keep-alive",
        "x-req-token": "MDbO4FsaSUPdjCdvTUs2zY4V3rnvvYatvYyjz7SfY+aCJ8r+RFm06X2dGR8eEDK7Gc5g1TLEQySEhGerRXbDT/NS+e5QAWRU68yD8m4y/aKK+TBkIv90VwvxmvYId2BVoDPDHQCGG4o3EqRWkS93eV0twYQ7w7qvNUj2e3tpDcUZYuplPyLozgYVTegFPnDk",
        "Accept": "*/*",
        "x-app-type": "iOS-5",
        "x-app-ver": "1.0.1",
        "x-app-udid": "419815AD-3015-4B5A-92CA-3BCBED24ACEC",
        "x-app-locale": "en",
        "Accept-Language": "zh-Hans-CN;q=1.0, en-CN;q=0.9",
        "Accept-Encoding": "br;q=1.0, gzip;q=0.9, deflate;q=0.8",
        "User-Agent": "MathGas/1.0.1 (MathWallet.MathGas; build:3; macOS 13.5.0) Alamofire/5.4.4"
    }

    response = httpx.get(url, headers=headers)
    if response.status_code != 200:
        print("Error:", response.status_code)
        return None

    if not response.json():
        print("Not Find GAS Data. Error: No response")
        return None

    remove_last_n_chars = lambda n, n_chars=9: int(str(n)[:-n_chars]) if len(str(n)) > n_chars else n

    result = '\nGAS:\n'

    try:
        data = response.json()['data']

        fastest = remove_last_n_chars(data['fastest']['price'])
        fast = remove_last_n_chars(data['fast']['price'])
        standard = remove_last_n_chars(data['standard']['price'])
        low = remove_last_n_chars(data['low']['price'])
        base = remove_last_n_chars(data['base']['price'])
        result += f'fastest: {fastest}\nfast: {fast}\nstandard: {standard}\nlow: {low}\nbase: {base}'
        return result
    except Exception as e:
        print(e)
        return None


def main():
    retry = 5
    text = ''
    for i in range(retry):
        try:
            text = get_chainalert()

            gas = get_gas_value()
            if gas:
                text += gas

            break
        except Exception as e:
            print(f'retry {i + 1}')
            print(e)
            time.sleep(5)

    if text:
        GotifyNotifier('Real-time coin price\n', text, 'AgfOJESqDKftBTQ').send_message()
    else:
        print('No Data')


if __name__ == '__main__':
    main()