auto/message/message_coinmarketcap.py

# -*- coding: utf-8 -*-
'''
使用 httpx 获取 coinmarketcap 最新数字币数据
'''
import time

import httpx
import re
from bs4 import BeautifulSoup

from utils.utils_send_gotify import *


def get_coinmarketcap_coin_price(url, target):
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36 Edg/107"
    }
    resp = httpx.get(url=url, headers=headers)

    if resp.status_code == 301:
        print(resp.text)
        return None
    elif resp.status_code != 200:
        print(resp.status_code)
        return None

    resp.encoding = 'utf-8'
    page = resp.text

    text = re.findall('<strong>(.*?)</p></div><div class="sc-65e7f566-0', page)[0] if re.findall(
        '<strong>(.*?)</p></div><div class="sc-65e7f566-0', page) else 'No Data'

    text = re.sub(r'</strong>', '', text)
    text = re.sub(r'<!-- -->', '', text)

    result = f'target: {target}\n'
    prices = re.findall(r'\$(\d+\.\d+)', text)
    if prices:
        result += f'prices: ${prices[0]}\n'

    volumes = re.findall(r'\$(\d{1,3}(?:,\d{3})*(?:\.\d+)?)', text)
    if volumes:
        result += f'24-hour trading volume: ${volumes[1]}\n'

    change = re.findall(r'(up|down) (\d+\.\d+)%', text)
    if change:
        c = ' '.join(change[0])
        result += f'change: {c}%\n'

    live_market_cap = re.findall(r'\$(\d{1,3}(?:,\d{3})*(?:\.\d+)?)', text)
    if live_market_cap:
        result += f'live market cap: ${live_market_cap[2]}\n'

    max_circulating_supply = re.findall(r'(\d{1,3}(?:,\d{3})*)', text)
    if max_circulating_supply:
        result += f'max circulating supply: {max_circulating_supply[-1]}'

    return result + '\n\n'


def get_vix_data():
    url = 'https://api-ddc-wscn.awtmt.com/market/real?'
    headers = {
        "Accept": "*/*",
        "Accept-Encoding": "gzip, deflate, br, zstd",
        "Accept-Language": "zh-CN,zh;q=0.9",
        "Connection": "keep-alive",
        "Host": "api-ddc-wscn.awtmt.com",
        "If-None-Match": "AMJ+W5ydwGIw9oqT3fOBeQ==",
        "Origin": "https://wallstreetcn.com",
        "Referer": "https://wallstreetcn.com/",
        "Sec-CH-UA": '"Chromium";v="130", "Brave";v="130", "Not?A_Brand";v="99"',
        "Sec-CH-UA-Mobile": "?0",
        "Sec-CH-UA-Platform": '"macOS"',
        "Sec-Fetch-Dest": "empty",
        "Sec-Fetch-Mode": "cors",
        "Sec-Fetch-Site": "cross-site",
        "Sec-GPC": "1",
        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36"
    }
    params = {
        "fields": "symbol,en_name,prod_name,last_px,px_change,px_change_rate,high_px,low_px,open_px,preclose_px,market_value,turnover_volume,turnover_ratio,turnover_value,dyn_pb_rate,amplitude,dyn_pe,trade_status,circulation_value,update_time,price_precision,week_52_high,week_52_low,static_pe,source",
        "prod_code": "VIX.OTC"
    }
    resp = httpx.get(url=url, headers=headers, params=params)

    if resp.status_code != 200:
        print(resp.status_code)
        return None

    data = resp.json()['data']['snapshot']['VIX.OTC']

    result = f'{data[1]}:{data[2]}'

    return result


if __name__ == '__main__':
    url_list = [
        ['BTC', 'https://coinmarketcap.com/currencies/bitcoin/'],
        ['ETH', 'https://coinmarketcap.com/currencies/ethereum/'],
        ['SOL', 'https://coinmarketcap.com/currencies/solana/'],
        ['SUI', 'https://coinmarketcap.com/currencies/sui/'],
        ['DOGE', 'https://coinmarketcap.com/currencies/dogecoin/'],
        ['ARB', 'https://coinmarketcap.com/currencies/arbitrum/'],
        ['ATH', 'https://coinmarketcap.com/currencies/aethir/']
    ]

    text = ''

    for data_list in url_list:
        res = get_coinmarketcap_coin_price(data_list[1], data_list[0])
        if res:
            text += res
        time.sleep(2)

    res = get_vix_data()
    if res:
        text += res

    print(text)

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