You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
70 lines
2.0 KiB
70 lines
2.0 KiB
import threading
|
|
import time
|
|
import random
|
|
import string
|
|
from flask import jsonify
|
|
|
|
# 模拟队列
|
|
virus_queue = []
|
|
bugs_queue = []
|
|
|
|
virus_prefix = "py-virus"
|
|
bugs_prefix = "sec-bugs"
|
|
|
|
def generate_random_string(length=10):
|
|
letters = string.ascii_lowercase
|
|
return ''.join(random.choice(letters) for _ in range(length))
|
|
|
|
def initialize_virus_queue():
|
|
global virus_queue
|
|
for _ in range(5):
|
|
random_string = generate_random_string()
|
|
virus_queue.append(f'{virus_prefix}-{random_string}')
|
|
|
|
def initialize_bugs_queue():
|
|
global bugs_queue
|
|
for _ in range(5):
|
|
random_string = generate_random_string()
|
|
bugs_queue.append(f'{bugs_prefix}-{random_string}')
|
|
|
|
# 定时检查病毒和漏洞数量
|
|
def check_queue_length():
|
|
while True:
|
|
time.sleep(600) # 每10分钟检查一次
|
|
if len(virus_queue) < 5:
|
|
for _ in range(5 - len(virus_queue)):
|
|
random_string = generate_random_string()
|
|
virus_queue.append(f'{virus_prefix}-{random_string}')
|
|
|
|
if len(bugs_queue) < 5:
|
|
for _ in range(5 - len(bugs_queue)):
|
|
random_string = generate_random_string()
|
|
bugs_queue.append(f'{bugs_prefix}-{random_string}')
|
|
|
|
|
|
def get_virus_queue_data():
|
|
return jsonify({"virus_queue": virus_queue})
|
|
|
|
def delete_virus_queue_data(names_list):
|
|
global virus_queue
|
|
for name in names_list:
|
|
virus_queue = [data for data in virus_queue if data != name]
|
|
return jsonify({"message": "Data deleted successfully"})
|
|
|
|
def get_bugs_queue_data():
|
|
return jsonify({"bugs_queue": bugs_queue})
|
|
|
|
def delete_bugs_queue_data(names_list):
|
|
global bugs_queue
|
|
for name in names_list:
|
|
bugs_queue = [data for data in bugs_queue if data != name]
|
|
return jsonify({"message": "Data deleted successfully"})
|
|
|
|
# 初始化队列
|
|
initialize_virus_queue()
|
|
initialize_bugs_queue()
|
|
|
|
# 启动后台线程
|
|
thread = threading.Thread(target=check_queue_length)
|
|
thread.daemon = True
|
|
thread.start() |