使用 httpx 直接访问 coinmarketcap 获取数据

archive/message_coinmarketcap.py

@@ -0,0 +1,77 @@
+# -*- coding: utf-8 -*-
+'''
+获取 coinmarketcap 数字货币实时价格
+'''
+import sys
+import os
+from datetime import datetime
+
+sys.path.append(os.path.join(os.path.abspath(__file__).split('auto')[0] + 'auto'))
+
+from base.base_load_config import load_config, get_base_path
+
+config_json = load_config()
+base_project = get_base_path()
+
+from utils.utils_check_base import *
+from utils.utils_send_gotify import *
+
+
+class CheckCoinmarketcap:
+    def __init__(self):
+        self.currency = '¥'
+        self.url_list = [
+            {'BTC': 'https://www.coinmarketcap.com/currencies/bitcoin/'},
+            {'ETH': 'https://www.coinmarketcap.com/currencies/ethereum/'},
+            {'DOGE': 'https://coinmarketcap.com/currencies/dogecoin/'},
+            {'ARB': 'https://coinmarketcap.com/currencies/arbitrum/'},
+            {'ATH': 'https://coinmarketcap.com/currencies/aethir/'},
+            {'SUI': 'https://coinmarketcap.com/currencies/sui/'},
+        ]
+
+        # self.selectors = ['#section-coin-overview > div.sc-65e7f566-0.czwNaM.flexStart.alignBaseline > span']
+        self.selectors = ['div#section-coin-overview > div:nth-of-type(2)']
+
+    def process_data(self, all_data):
+        result = {}
+        for data in all_data:
+            for key, value in data[0].items():
+                value_list = value.split('\xa0')
+                value_list = [item for item in value_list if item]
+                if key not in result:
+                    result[key] = value_list
+        return result
+
+    def send_data(self, all_data):
+        # 打印结果，只包含货币名称和价格
+        context = ''
+
+        for key, value in all_data.items():
+            context += f'{key}: {value[0]}     {value[1]}    {value[2]}\n'
+
+        if context:
+            context += '\n{}'.format(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
+            print(context)
+            # 推送到 message
+            GotifyNotifier('实时coin价格', context).send_message()
+        else:
+            print('no data!')
+
+    def main(self):
+        # 运行异步函数
+        all_data = None
+        crawler = CryptoCrawler(self.url_list, self.selectors, headless=True)
+        all_data = crawler.main()  # 返回的数据格式: [[{'BTC': '¥483,784.76\xa0\xa01.02%\xa0(1天)'}], ...
+
+        if not all_data:
+            print('获取数据失败')
+            return
+
+        # 清理一下数据
+        all_data = self.process_data(all_data)  # {'BTC': ['¥483,482.98', '0.95%', '(1天)'], ...
+
+        self.send_data(all_data)
+
+
+if __name__ == '__main__':
+    CheckCoinmarketcap().main()

message/message_coinmarketcap.py

@@ -1,77 +1,72 @@
 # -*- coding: utf-8 -*-
 '''
-获取 coinmarketcap 数字货币实时价格
+使用 httpx 获取 coinmarketcap 最新数字币数据
 '''
-import sys
-import os
-from datetime import datetime
 
-sys.path.append(os.path.join(os.path.abspath(__file__).split('auto')[0] + 'auto'))
+import httpx
+import re
 
-from base.base_load_config import load_config, get_base_path
 
-config_json = load_config()
-base_project = get_base_path()
+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)
 
-from utils.utils_check_base import *
-from utils.utils_send_gotify import *
+    if resp.status_code == 301:
+        print(resp.text)
+        exit(0)
+    elif resp.status_code != 200:
+        print(resp.status_code)
+        exit(0)
 
+    resp.encoding = 'utf-8'
+    page = resp.text
 
-class CheckCoinmarketcap:
-    def __init__(self):
-        self.currency = '¥'
-        self.url_list = [
-            {'BTC': 'https://www.coinmarketcap.com/currencies/bitcoin/'},
-            {'ETH': 'https://www.coinmarketcap.com/currencies/ethereum/'},
-            {'DOGE': 'https://coinmarketcap.com/currencies/dogecoin/'},
-            {'ARB': 'https://coinmarketcap.com/currencies/arbitrum/'},
-            {'ATH': 'https://coinmarketcap.com/currencies/aethir/'},
-            {'SUI': 'https://coinmarketcap.com/currencies/sui/'},
-        ]
+    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'
 
-        # self.selectors = ['#section-coin-overview > div.sc-65e7f566-0.czwNaM.flexStart.alignBaseline > span']
-        self.selectors = ['div#section-coin-overview > div:nth-of-type(2)']
+    text = re.sub(r'</strong>', '', text)
+    text = re.sub(r'<!-- -->', '', text)
 
-    def process_data(self, all_data):
-        result = {}
-        for data in all_data:
-            for key, value in data[0].items():
-                value_list = value.split('\xa0')
-                value_list = [item for item in value_list if item]
-                if key not in result:
-                    result[key] = value_list
-        return result
+    result = f'target: {target}\n'
+    prices = re.findall(r'\$(\d+\.\d+)', text)
+    if prices:
+        result += f'prices: ${prices[0]}\n'
 
-    def send_data(self, all_data):
-        # 打印结果，只包含货币名称和价格
-        context = ''
+    volumes = re.findall(r'\$(\d{1,3}(?:,\d{3})*(?:\.\d+)?)', text)
+    if volumes:
+        result += f'24-hour trading volume: ${volumes[1]}\n'
 
-        for key, value in all_data.items():
-            context += f'{key}: {value[0]}     {value[1]}    {value[2]}\n'
+    change = re.findall(r'(up|down) (\d+\.\d+)%', text)
+    if change:
+        c = ' '.join(change[0])
+        result += f'change: {c}%\n'
 
-        if context:
-            context += '\n{}'.format(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
-            print(context)
-            # 推送到 message
-            GotifyNotifier('实时coin价格', context).send_message()
-        else:
-            print('no data!')
+    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'
 
-    def main(self):
-        # 运行异步函数
-        all_data = None
-        crawler = CryptoCrawler(self.url_list, self.selectors, headless=True)
-        all_data = crawler.main()  # 返回的数据格式: [[{'BTC': '¥483,784.76\xa0\xa01.02%\xa0(1天)'}], ...
+    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]}'
 
-        if not all_data:
-            print('获取数据失败')
-            return
+    return result + '\n\n'
 
-        # 清理一下数据
-        all_data = self.process_data(all_data)  # {'BTC': ['¥483,482.98', '0.95%', '(1天)'], ...
 
-        self.send_data(all_data)
+if __name__ == '__main__':
+    url_list = [
+        # ['BTC', 'https://www.coinmarketcap.com/currencies/bitcoin/'],
+        # ['ETH', 'https://www.coinmarketcap.com/currencies/ethereum/'],
+        ['DOGE', 'https://coinmarketcap.com/currencies/dogecoin/'],
+        ['ARB', 'https://coinmarketcap.com/currencies/arbitrum/'],
+        ['ATH', 'https://coinmarketcap.com/currencies/aethir/'],
+        ['SUI', 'https://coinmarketcap.com/currencies/sui/'],
+    ]
 
+    text = ''
 
-if __name__ == '__main__':
-    CheckCoinmarketcap().main()
+    for data_list in url_list:
+        text += get_coinmarketcap_coin_price(data_list[1], data_list[0])
+
+    print(text)