import json
import time
import configparser
from kafka import KafkaProducer
from threading import Timer

# 读取配置文件
def load_config(config_file):
    config = configparser.ConfigParser()
    config.read(config_file)
    return config

# 读取 JSON 文件内容
def load_json(json_file):
    with open(json_file, 'r') as f:
        return json.load(f)

# 发送消息到 Kafka
def send_message(producer, topic, message):
    producer.send(topic, value=json.dumps(message).encode('utf-8'))
    print(f"Sent message to {topic}: {message}")

# 定时发送消息
def send_periodically(producer, topic, message, interval, stop_flag):
    if stop_flag[0]:  # 如果 stop_flag 是 True，则停止发送
        send_message(producer, topic, message)
        # 设置定时器，按照给定的间隔继续发送
        Timer(interval, send_periodically, [producer, topic, message, interval, stop_flag]).start()
    else:
        print("Stopping periodic message sending.")

# 停止定时器
def stop_sending(stop_flag):
    stop_flag[0] = False
    print("Message sending stopped.")

def main():
    # 配置文件路径
    config_file = 'msg_config.ini'
    config = load_config(config_file)

    # Kafka 配置
    kafka_server = config['KAFKA']['server']
    topics = config['KAFKA']['topics'].split(',')  # topics 列表
    send_topics = config['SEND_MSG']['send_topics']
    run_send = config.getboolean('SEND_MSG', 'run_send')  # run_send 为布尔值
    send_interval = config.getint('SEND_MSG', 'send_interval')
    json_file = config['SEND_MSG']['json_file']

    # 读取消息内容
    message = load_json(json_file)

    # 创建 Kafka producer
    producer = KafkaProducer(bootstrap_servers=kafka_server)

    # 如果需要发送消息，则启动定时任务
    if run_send:
        print(f"Sending messages to {send_topics} every {send_interval} seconds...")
        
        # 使用列表来实现可修改的标志位
        stop_flag = [True]
        
        # 启动定时器
        send_periodically(producer, send_topics, message, send_interval, stop_flag)

        # 例如，可以在10秒后停止定时发送
        #time.sleep(10)
        #stop_sending(stop_flag)
    else:
        print("run_send is false. No messages will be sent.")

if __name__ == "__main__":
    main()