auto/message/message_vix.py

# -*- coding: utf-8 -*-
'''
检查 当前VIX恐慌指数
'''
import sys
import os
import re
from datetime import datetime

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

from message_check_base import *
from utils.utils_send_gotify import *
from utils.utils_send_serverchan import *


class CheckVIX:
    def __init__(self):
        self.url_list = [{'VIX恐慌指数': 'https://wallstreetcn.com/markets/codes/VIX.OTC'}]

        self.selectors = [
            '.price-lastpx'
        ]
        self.send_message = False
        self.check_value = 60

    def clean_string(self, input_string):
        """
        清除字符串中的特殊字符和HTML标签

        :param input_string: 要处理的字符串
        :return: 处理后的字符串
        """
        # 替换字符
        clean_list = ['\n', '\t', ' ']
        for char in clean_list:
            input_string = input_string.replace(char, '')

        # 使用正则表达式移除HTML标签
        cleaned_string = re.sub(r'<.*?>', '', input_string)

        return cleaned_string

    def main(self):
        all_data = None
        crawler = CryptoCrawler(self.url_list, self.selectors)
        all_data = crawler.main()  # 返回的数据格式: dict: [{关键词str: 数据any}]

        # 组成context信息
        context = ''

        # 如果有返回数据
        if all_data:
            all_data = all_data[0]
            for data in all_data:
                for key, value in data.items():
                    context += f"{key}: {value}%\n"

                    # 如果恐慌指数大于指定数值, 则发送消息, 小于则不发送
                    if float(value) >= self.check_value:
                        self.send_message = True
        else:
            print('no data!')

        if self.send_message and context:
            print(f'发送消息: {context}')
            context += '\n{}'.format(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
            # 推送到 message
            GotifyNotifier('vix恐慌指数', context).send_message()

            # 推送到 serverchan
            ServerChanNotifier('vix恐慌指数', context.replace('\n', '\n\n')).send_message()
        else:
            print(f"VIX恐慌指数小于{self.check_value}，不发送消息\n{context}")

        print('done!')


if __name__ == '__main__':
    CheckVIX().main()