cron.add_task("periodic_task", lambda: print("🔁 周期性任务执行"), interval_seconds=5)

import schedule
import time

def job():
    print("Scheduled task executed")

schedule.every(5).seconds.do(job)

while True:
    schedule.run_pending()
    time.sleep(1)

from apscheduler.schedulers.blocking import BlockingScheduler

def job():
    print("APScheduled task executed")

sched = BlockingScheduler()
sched.add_job(job, 'interval', seconds=5)
sched.start()

import time
import datetime
import threading
from typing import Dict, Optional, Callable, Tuple

class SimpleCronJob:
    def __init__(self):
        self.tasks: Dict[str, Tuple[datetime.datetime, datetime.timedelta, bool, bool, Callable]] = {}
        self.lock = threading.Lock()
        self.stop_flag = False

    def add_task(self, task_id: str, func: Callable, delay_seconds: int = 0, interval_seconds: int = 0, once: bool = False):
        """
        添加一个定时任务。

        :param task_id: 任务唯一标识符
        :param func: 要执行的函数
        :param delay_seconds: 首次执行延迟时间（秒）
        :param interval_seconds: 周期执行间隔时间（秒）
        :param once: 是否为一次性任务
        """
        with self.lock:
            self.tasks[task_id] = (
                None,  # last_time
                datetime.timedelta(seconds=delay_seconds),
                False,  # first_executed
                datetime.timedelta(seconds=interval_seconds),
                once,
                func
            )
        print(f"任务 {task_id} 已注册")

    def start(self):
        print("开始执行定时任务服务...")
        while not self.stop_flag:
            now = datetime.datetime.now()

            with self.lock:
                for task_id, (last_time, delay_time, first_executed, interval_time, once, func) in list(self.tasks.items()):
                    if delay_time.total_seconds() > 0 and not first_executed:
                        if last_time is None:
                            self.tasks[task_id] = (now, delay_time, first_executed, interval_time, once, func)
                            continue
                        elif (now - last_time) < delay_time:
                            continue
                        else:
                            self.tasks[task_id] = (now, delay_time, True, interval_time, once, func)

                    elif last_time is None:
                        self.tasks[task_id] = (now, delay_time, first_executed, interval_time, once, func)
                        continue
                    elif (now - last_time) < interval_time:
                        continue

                    # 执行任务
                    try:
                        print(f"执行任务 {task_id}")
                        func()
                    except Exception as e:
                        print(f"任务 {task_id} 执行出错: {e}")

                    # 更新最后执行时间
                    self.tasks[task_id] = (now, delay_time, True, interval_time, once, func)

                    # 如果是一次性任务，执行后移除
                    if once:
                        del self.tasks[task_id]
                        print(f"任务 {task_id} 已完成并移除")

            time.sleep(1)  # 模拟 CRON_TIME_INTERVAL

    def stop(self):
        self.stop_flag = True
        print("定时任务服务已停止")


# 示例任务函数
def example_task():
    print("✅ 示例任务正在执行...")

if __name__ == "__main__":
    cron = SimpleCronJob()

    # 1. 周期性任务：每5秒执行一次
    cron.add_task("periodic_task", lambda: print("🔁 周期性任务执行"), interval_seconds=5)

    # 2. 延迟性任务：首次延迟3秒后开始，之后每6秒执行一次
    cron.add_task("delayed_periodic_task",
                  lambda: print("🕒 延迟+周期任务执行"),
                  delay_seconds=3,
                  interval_seconds=6)

    # 3. 一次性任务：立即执行一次（无延迟）
    cron.add_task("one_time_task",
                  lambda: print("📎 一次性任务执行"),
                  once=True)

    # 4. 一次性延迟任务：延迟5秒后执行一次
    cron.add_task("one_time_delayed_task",
                  lambda: print("⏳ 一次性延迟任务执行"),
                  delay_seconds=5,
                  once=True)

    try:
        cron.start()
    except KeyboardInterrupt:
        cron.stop()

import redis
import threading
import time

class RedisPubSub:
    def __init__(self, host='localhost', port=6379, db=0, channel='default'):
        self.channel = channel  # 频道
        self.redis = redis.Redis(host=host, port=port, db=db, decode_responses=True)  # redis客户端
        self.pubsub = self.redis.pubsub() # redis发布订阅对象
        self._running = False  # 是否开启订阅
	
    # 向self.channel频道发送消息message
    def publish(self, message: str):
        """发布消息"""
        self.redis.publish(self.channel, message)
	
    # 订阅self.channel消息
    def subscribe(self, callback):
        """
        订阅并启动监听线程。
        参数 callback: 接收一个函数，在收到消息时被调用。
        """
        def listen():
            # 订阅self.channel频道
            self.pubsub.subscribe(self.channel)
            print(f"Subscribed to {self.channel}")
            
            # 监听
            for message in self.pubsub.listen():
                if not self._running:
                    break
                if message['type'] == 'message':
                    callback(message['data'])
		
        # 开启订阅
        self._running = True
        
        # 创建监听线程
        self.listen_thread = threading.Thread(target=listen, daemon=True)
        # 启动监听线程
        self.listen_thread.start()

    def stop(self):
        """停止订阅"""
        self._running = False
        self.pubsub.unsubscribe()
        print(f"Unsubscribed from {self.channel}")

from redis_pubsub import RedisPubSub
import time

pub = RedisPubSub(channel='chat')
for i in range(5):
    pub.publish(f"Message {i}")
    print(f"Sent: Message {i}")
    time.sleep(1)

from redis_pubsub import RedisPubSub
import time

def handle_message(msg):
    print(f"Received: {msg}")

sub = RedisPubSub(channel='chat')
sub.subscribe(callback=handle_message)

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    sub.stop()

import torch
import torch.nn as nn
import torch.nn.functional as F
import math


class PositionalEncoding(nn.Module):
    def __init__(self, d_model, max_len=5000):
        super(PositionalEncoding, self).__init__()
        pe = torch.zeros(max_len, d_model)
        position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
        div_term = torch.exp(torch.arange(0, d_model, 2).float() * -(math.log(10000.0) / d_model))
        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)
        pe = pe.unsqueeze(0)
        self.register_buffer('pe', pe)

    def forward(self, x):
        return x + self.pe[:, :x.size(1)].detach()


class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, num_heads):
        super(MultiHeadAttention, self).__init__()
        self.d_model = d_model
        self.num_heads = num_heads
        assert d_model % self.num_heads == 0  # 确保 d_model 能被 num_heads 整除

        self.depth = d_model // self.num_heads

        self.wq = nn.Linear(d_model, d_model)
        self.wk = nn.Linear(d_model, d_model)
        self.wv = nn.Linear(d_model, d_model)

        self.dense = nn.Linear(d_model, d_model)

    def split_heads(self, x, batch_size):
        x = x.view(batch_size, -1, self.num_heads, self.depth)
        return x.permute(0, 2, 1, 3)

    def attention(self, query, key, value, mask=None, dropout=None):
        matmul_qk = torch.matmul(query, key.transpose(-2, -1))  # QK^T
        dk = query.size(-1)
        scaled_attention_logits = matmul_qk / math.sqrt(dk)

        if mask is not None:
            scaled_attention_logits += (mask * -1e9)  # 避免pad部分被注意到

        attention_weights = torch.softmax(scaled_attention_logits, dim=-1)
        if dropout is not None:
            attention_weights = dropout(attention_weights)

        output = torch.matmul(attention_weights, value)
        return output, attention_weights

    def forward(self, query, key, value, mask=None, dropout=None):
        batch_size = query.size(0)

        query = self.split_heads(self.wq(query), batch_size)
        key = self.split_heads(self.wk(key), batch_size)
        value = self.split_heads(self.wv(value), batch_size)

        output, attention_weights = self.attention(query, key, value, mask, dropout)
        output = output.permute(0, 2, 1, 3).contiguous().view(batch_size, -1, self.d_model)

        return self.dense(output)


class FeedForwardNetwork(nn.Module):
    def __init__(self, d_model, d_ff=2048, dropout=0.1):
        super(FeedForwardNetwork, self).__init__()
        self.linear1 = nn.Linear(d_model, d_ff)
        self.linear2 = nn.Linear(d_ff, d_model)
        self.dropout = nn.Dropout(dropout)

    def forward(self, x):
        x = F.relu(self.linear1(x))
        x = self.dropout(x)
        return self.linear2(x)


class EncoderLayer(nn.Module):
    def __init__(self, d_model, num_heads, d_ff=2048, dropout=0.1):
        super(EncoderLayer, self).__init__()
        self.attention = MultiHeadAttention(d_model, num_heads)
        self.ffn = FeedForwardNetwork(d_model, d_ff, dropout)
        self.layernorm1 = nn.LayerNorm(d_model)
        self.layernorm2 = nn.LayerNorm(d_model)
        self.dropout = nn.Dropout(dropout)

    def forward(self, x, mask=None):
        # 自注意力层
        attn_output = self.attention(x, x, x, mask, self.dropout)
        x = self.layernorm1(x + attn_output)  # 残差连接 + LayerNorm

        # 前馈网络层
        ffn_output = self.ffn(x)
        x = self.layernorm2(x + ffn_output)  # 残差连接 + LayerNorm

        return x


class DecoderLayer(nn.Module):
    def __init__(self, d_model, num_heads, d_ff=2048, dropout=0.1):
        super(DecoderLayer, self).__init__()
        self.attention1 = MultiHeadAttention(d_model, num_heads)
        self.attention2 = MultiHeadAttention(d_model, num_heads)
        self.ffn = FeedForwardNetwork(d_model, d_ff, dropout)
        self.layernorm1 = nn.LayerNorm(d_model)
        self.layernorm2 = nn.LayerNorm(d_model)
        self.layernorm3 = nn.LayerNorm(d_model)
        self.dropout = nn.Dropout(dropout)

    def forward(self, x, enc_output, look_ahead_mask=None, padding_mask=None):
        # 解码器中的自注意力层
        attn1_output = self.attention1(x, x, x, look_ahead_mask, self.dropout)
        x = self.layernorm1(attn1_output + x)

        # 编码器-解码器注意力层
        attn2_output = self.attention2(x, enc_output, enc_output, padding_mask, self.dropout)
        x = self.layernorm2(attn2_output + x)

        # 前馈网络层
        ffn_output = self.ffn(x)
        x = self.layernorm3(ffn_output + x)

        return x


class TransformerEncoder(nn.Module):
    def __init__(self, vocab_size, d_model, num_heads, num_layers, d_ff=2048, dropout=0.1):
        super(TransformerEncoder, self).__init__()
        self.embedding = nn.Embedding(vocab_size, d_model)
        self.positional_encoding = PositionalEncoding(d_model)
        self.layers = nn.ModuleList([
            EncoderLayer(d_model, num_heads, d_ff, dropout) for _ in range(num_layers)
        ])
        self.d_model = d_model

    def forward(self, x, mask=None):
        x = self.embedding(x) * math.sqrt(self.d_model)  # 嵌入 + 缩放
        x = self.positional_encoding(x)

        for layer in self.layers:
            x = layer(x, mask)

        return x


class TransformerDecoder(nn.Module):
    def __init__(self, vocab_size, d_model, num_heads, num_layers, d_ff=2048, dropout=0.1):
        super(TransformerDecoder, self).__init__()
        self.embedding = nn.Embedding(vocab_size, d_model)
        self.positional_encoding = PositionalEncoding(d_model)
        self.layers = nn.ModuleList([
            DecoderLayer(d_model, num_heads, d_ff, dropout) for _ in range(num_layers)
        ])
        self.d_model = d_model

    def forward(self, x, enc_output, look_ahead_mask=None, padding_mask=None):
        x = self.embedding(x) * math.sqrt(self.d_model)
        x = self.positional_encoding(x)

        for layer in self.layers:
            x = layer(x, enc_output, look_ahead_mask, padding_mask)

        return x


class Transformer(nn.Module):
    def __init__(self, vocab_size, d_model, num_heads, num_layers, d_ff=2048, dropout=0.1):
        super(Transformer, self).__init__()

        self.encoder = TransformerEncoder(vocab_size, d_model, num_heads, num_layers, d_ff, dropout)
        self.decoder = TransformerDecoder(vocab_size, d_model, num_heads, num_layers, d_ff, dropout)
        self.output_layer = nn.Linear(d_model, vocab_size)

    def forward(self, src, tgt, src_mask=None, tgt_mask=None):
        # 编码器部分
        enc_output = self.encoder(src, src_mask)

        # 解码器部分
        dec_output = self.decoder(tgt, enc_output, tgt_mask, src_mask)

        # 输出层
        return self.output_layer(dec_output)


vocab_size = 10000  # 词汇表大小
d_model = 512  # 特征维度
num_heads = 8  # 注意力头数
num_layers = 6  # 编码器和解码器层数
dropout = 0.1  # Dropout 比例

# 初始化 Transformer 模型
transformer = Transformer(vocab_size, d_model, num_heads, num_layers, dropout=dropout)

# 输入张量（batch_size, sequence_length）
src = torch.randint(0, vocab_size, (32, 100))  # 假设 source 语言输入 batch_size=32，序列长度=100
tgt = torch.randint(0, vocab_size, (32, 100))  # 假设 target 语言输入 batch_size=32，序列长度=100

# 创建遮罩（假设没有 padding）
src_mask = None
tgt_mask = None

# 前向传播
output = transformer(src, tgt, src_mask, tgt_mask)

print("Output shape:", output.shape)  # 输出的形状 (batch_size, tgt_sequence_length, vocab_size)