添加新版低吸中间服务器

admin

2025-06-04 287c506725b2d970f721f80169f83c2418cb0991

 middle_l1_data_server.py

@@ -2,22 +2,15 @@
import hashlib
import json
import logging
import os
import queue
import random
import socket
import socketserver
import threading
import time

import constant
import log
import socket_manager
from db import mysql_data
from db.redis_manager import RedisUtils, RedisManager
from log import logger_debug, logger_request_debug
from output import push_msg_manager
from utils import socket_util, kpl_api_util, hosting_api_util, kp_client_msg_manager, global_data_cache_util, tool
from utils.juejin_util import JueJinHttpApi
from log_module import log
from utils import socket_util

trade_data_request_queue = queue.Queue()

@@ -101,7 +94,14 @@
                    try:
                        if data_json["type"] == 'l1_data':
                            datas = data_json["data"]
                            L1DataManager.add_datas(datas)
                            L1DataManager().add_datas(datas)
                            break
                        elif data_json["type"] == 'get_l1_target_codes':
                            # 获取目标代码
                            codes = L1DataManager().get_target_codes()
                            sk.sendall(socket_util.load_header(json.dumps(
                                {"code": 0, "data": list(codes)}).encode(
                                encoding='utf-8')))
                            break
                    except Exception as e:
                        log.logger_tuoguan_request_debug.exception(e)
@@ -125,8 +125,28 @@
# L1数据管理
class L1DataManager:
    __l1_datas_dict = {}
    __target_codes = set()
    __instance = None

    def __new__(cls, *args, **kwargs):
        if not cls.__instance:
            cls.__instance = super(L1DataManager, cls).__new__(cls, *args, **kwargs)
            cls.__load_datas()
        return cls.__instance

    @classmethod
    def __load_datas(cls):
        try:
            with open(f"{constant.LOG_DIR}/l1_codes.txt", 'r', encoding='utf-8') as f:
                lines = f.readlines()
                for line in lines:
                    if line:
                        line = line.strip()
                        if line:
                            cls.__target_codes.add(line)
        except:
            pass

    def add_datas(cls, datas):
        for data in datas:
            """
@@ -134,9 +154,22 @@
            """
            cls.__l1_datas_dict[data[0]] = data

    @classmethod
    def get_current_l1_data(cls):
        return [cls.__l1_datas_dict[x] for x in cls.__l1_datas_dict]
    def get_current_l1_data(self):
        return [self.__l1_datas_dict[x] for x in self.__l1_datas_dict]

    def save_target_codes(self, codes):
        # 保存目标代码
        self.__target_codes = codes
        # 将代码保存到文件
        path = f"{constant.LOG_DIR}/l1_codes.txt"
        if not os.path.exists(constant.LOG_DIR):
            os.mkdir(constant.LOG_DIR)
        with open(path, 'w', encoding='utf-8') as f:
            for code in codes:
                f.write(f"{code}\n")

    def get_target_codes(self):
        return self.__target_codes


def run(port):
@@ -148,4 +181,4 @@


if __name__ == "__main__":
    print(builtins.type("") == str)
    print(L1DataManager().get_target_codes())