创建机器人¶

from lxmfy import LXMFBot

# 1. 实例化机器人
bot = LXMFBot(
    name="SimpleBot",
    command_prefix="!",
    storage_path="simple_data"
)

# 2. 定义命令
@bot.command(name="ping", description="Responds with pong")
def ping_command(ctx):
    # ctx 是包含消息信息的上下文对象
    # ctx.sender: 发送者的 LXMF 哈希
    # ctx.content: 完整消息内容
    # ctx.args: 命令之后的参数列表
    # ctx.reply(message): 发送回复的函数
    #   （也可以接受关键字参数，如 title="My Title", lxmf_fields=some_fields）
    ctx.reply("Pong!")

# 对于耗时任务，可以使用线程化命令：
# import time
# @bot.command(name="long_op", description="Performs a long operation in a separate thread", threaded=True)
# def long_op_command(ctx):
#     ctx.reply("Starting long operation...")
#     time.sleep(10) # 模拟耗时操作
#     ctx.reply("Long operation complete!")
# 重要：线程化命令不应直接与 RNS 或 lxmfy.transport.py 交互。

@bot.command(name="greet", description="Greets the user")
def greet_command(ctx):
    if ctx.args:
        name = " ".join(ctx.args)
        ctx.reply(f"Hello, {name}!")
    else:
        ctx.reply("Hello there! Tell me your name: !greet <your_name>")

# 3. 运行机器人
if __name__ == "__main__":
    print(f"Starting bot: {bot.config.name}")
    print(f"Bot LXMF Address: {bot.local.hash}")
    bot.run()

from lxmfy.templates import CogTestBot

if __name__ == "__main__":
    bot = CogTestBot() # 创建 CogTestBot 模板的实例
    # 可以选择性地覆盖默认名称：
    # bot.bot.name = "My Cog Test Bot"
    bot.run()

from lxmfy import LXMFBot

bot = LXMFBot(
    name="ConfiguredBot",
    announce=3600, # 每小时 announce 一次
    admins={"your_admin_hash_here"}, # 设置管理员用户
    command_prefix="$", # 使用 '$' 作为前缀
    storage_type="sqlite", # 使用 SQLite 数据库
    storage_path="data/my_bot_data.db", # 指定数据库文件路径
    rate_limit=10, # 允许每分钟 10 条消息
    cooldown=30, # 冷却 30 秒
    permissions_enabled=True # 启用基于角色的权限
)

if __name__ == "__main__":
    # 也可以在实例化之后修改配置
    # 注意：有些设置最好在初始化时设置
    bot.config.max_warnings = 5
    bot.spam_protection.config.max_warnings = 5 # 同步更新刷屏保护器

    bot.run()

from lxmfy import IconAppearance, pack_icon_appearance_field

# 在你的机器人类或初始化代码中
icon_data = IconAppearance(
    icon_name="robot_2",  # 从 Material Symbols 中选择
    fg_color=b'\x00\xFF\x00',  # 绿色
    bg_color=b'\x33\x33\x33'   # 深灰色
)
self.bot_icon_field = pack_icon_appearance_field(icon_data)

# 发送消息或回复时：
ctx.reply("Message from your bot!", lxmf_fields=self.bot_icon_field)
# 或
# bot.send(destination, "Another message", lxmf_fields=self.bot_icon_field)

from lxmfy import LXMFBot

bot = LXMFBot(name="FieldBot")

@bot.command(name="add", description="Add two numbers")
def add_command(ctx):
    if len(ctx.args) >= 2:
        try:
            result = float(ctx.args[0]) + float(ctx.args[1])
            ctx.reply(str(result))
        except ValueError:
            ctx.reply("Invalid numbers")
    else:
        ctx.reply("Usage: add <a> <b>")

import LXMF
from lxmfy import FIELD_COMMANDS

lxm = LXMF.LXMessage(
    destination,
    source,
    b"",  # 纯字段命令的内容可以为空
    desired_method=LXMF.LXMessage.DIRECT,
)
lxm.fields[FIELD_COMMANDS] = {
    "command": "add",
    "args": ["3", "5"],
    "request_id": "req-42",  # 可选，用于关联请求
}
router.handle_outbound(lxm)

bot = LXMFBot(
    name="TextOnlyBot",
    lxmf_commands_enabled=False,
)

from lxmfy import Command
from lxmfy.commands import Cog  # 如果要继承，导入 Cog
import time

class UtilityCog: # 或者 class UtilityCog(Cog):
    def __init__(self, bot):
        self.bot = bot
        self.start_time = time.time()

    @Command(name="uptime", description="Shows bot uptime")
    # 注意：Cog 中的方法通常接收 'self' 和 'ctx'
    def uptime_command(self, ctx):
        uptime_seconds = time.time() - self.start_time
        ctx.reply(f"Bot has been running for {uptime_seconds:.2f} seconds.")

    @Command(name="info", description="Shows bot info")
    def info_command(self, ctx):
        info = (
            f"Bot Name: {self.bot.config.name}\n"
            f"Owner(s): {', '.join(self.bot.config.admins) or 'None'}\n"
            f"Prefix: {self.bot.config.command_prefix}"
        )
        ctx.reply(info)

    @Command(name="threaded_cog_task", description="Performs a long task in a cog thread", threaded=True)
    def threaded_cog_task(self, ctx):
        ctx.reply("Starting a long cog task... this will run in a separate thread.")
        time.sleep(7) # 模拟耗时操作
        ctx.reply("Long cog task completed!")

# 加载该 Cog 必须有这个函数
def setup(bot):
    cog_instance = UtilityCog(bot)
    bot.add_cog(cog_instance) # 向机器人注册 Cog 实例

from lxmfy import LXMFBot

bot = LXMFBot(
    name="CogBot",
    cogs_enabled=True, # 确保启用了 Cog（默认开启）
    cogs_dir="cogs" # 指向该目录
)

if __name__ == "__main__":
    # 如果 cogs_enabled 为 True，
    # Cog 会在 LXMFBot 初始化期间自动加载。
    bot.run()

@bot.intent("help", examples=["how do I use this?", "show me commands", "help me please"])
def help_intent(msg):
    msg.reply("I can help! Try typing /help to see a list of commands.")

# 导出模型
model_data = bot.nlp.export_model()
# 把 model_data 保存到文件或数据库

# 之后重新导入
bot.nlp.import_model(model_data)

def handle_link(link):
    print(f"Link established with {RNS.hexrep(link.destination.hash)}")
    # 现在可以使用该 link 进行直接的 RNS 通信

bot.on_link(handle_link)

from lxmfy import LXMFBot

bot = LXMFBot(
    name="WelcomeBot",
    first_message_enabled=True  # 必须为 True（默认）
)

@bot.on_first_message()
def welcome_new_user(sender, message):
    content = message.content.decode("utf-8")
    bot.send(
        sender,
        f"Welcome to the bot! You said: {content}\n\n"
        "Type /help to see available commands."
    )
    return True  # 返回 True 以停止该消息的进一步处理

if __name__ == "__main__":
    bot.run()

from lxmfy import LXMFBot

bot = LXMFBot(name="EchoBot")

@bot.on_message()
def echo_non_commands(sender, message):
    content = message.content.decode("utf-8").strip()

    # 检查是否为命令，是则交给命令处理器
    if content.startswith(bot.config.command_prefix):
        command_name = content.split()[0][len(bot.config.command_prefix):]
        if command_name in bot.commands:
            return False  # 交给命令处理器处理

    # 不是命令，回显它
    bot.send(sender, f"You said: {content}")
    return False  # 返回 False 继续处理（虽然不会有命令匹配）

@bot.command(name="hello", description="Say hello")
def hello_command(ctx):
    ctx.reply("Hello! This is a command response.")

if __name__ == "__main__":
    bot.run()

from lxmfy import LXMFBot
from lxmfy.events import EventPriority # 可选，用于优先级

bot = LXMFBot(name="EventBot")

@bot.events.on("message_received")
def log_message(event):
    # Event 对象包含详细信息
    sender = event.data.get("sender")
    message_content = event.data.get("message").content.decode('utf-8', errors='ignore')
    print(f"Received message from {sender}: {message_content}")

    # 可以取消事件处理（例如停止消息处理）
    # if sender == "some_blocked_hash":
    #    event.cancel()

@bot.events.on("command_executed", priority=EventPriority.LOW)
def log_command(event):
    # 示例：event.data 可能包含 {'command_name': 'ping', 'sender': '...', ...}
    command_name = event.data.get('command_name', 'unknown')
    sender = event.data.get('sender', 'unknown')
    print(f"Command '{command_name}' executed by {sender}")

# 也可以定义自定义事件
@bot.command(name="special")
def special_command(ctx):
    ctx.reply("Doing something special!")
    # 派发一个自定义事件
    bot.events.dispatch(Event("special_action_taken", data={"user": ctx.sender}))

@bot.events.on("special_action_taken")
def handle_special(event):
    user = event.data.get("user")
    print(f"Special action was taken by user: {user}")


if __name__ == "__main__":
    bot.run()

# 保存数据
bot.storage.set("user_prefs:" + ctx.sender, {"theme": "dark"})

# 获取数据（带默认值）
prefs = bot.storage.get("user_prefs:" + ctx.sender, {})
theme = prefs.get("theme", "light")

# 检查数据是否存在
if bot.storage.exists("some_key"):
    print("Key exists!")

# 删除数据
bot.storage.delete("old_data_key")

# 按前缀扫描键（便于枚举用户数据）
user_keys = bot.storage.scan("user_prefs:")
for key in user_keys:
    user_data = bot.storage.get(key)
    print(f"Data for {key}: {user_data}")

bot = LXMFBot(
    name="SecureBot",
    signature_verification_enabled=True,  # 启用签名检查
    require_message_signatures=False      # 设为 True 可拒绝未签名消息
)

from lxmfy import LXMFBot

bot = LXMFBot(name="PropagationBot")

@bot.command(name="send", description="Send via propagation node")
def send_command(ctx):
    # 只设置一次特定的传播节点（配置层面）
    bot.set_propagation_node("<propagation_node_hash_here>")

    # 按配置好的投递策略发送
    bot.send(
        ctx.sender,
        "This message will use direct delivery with propagation fallback as configured"
    )

from lxmfy import LXMFBot

bot = LXMFBot(name="ReliableBot")

bot = LXMFBot(
    name="ReliableBot",
    direct_delivery_retries=5,  # 直接投递最多重试 5 次
    propagation_fallback_enabled=True
)

@bot.command(name="important", description="Send important message with retries")
def important_command(ctx):
    bot.send(ctx.sender, "This is an important message")

@bot.command(name="normal", description="Send with default retries")
def normal_command(ctx):
    # direct_delivery_retries 默认为 3
    bot.send(ctx.sender, "This message uses default retry settings")

from lxmfy import LXMFBot, RRCMessage

bot = LXMFBot(
    name="RoomBot",
    reticulum_config_dir="~/.reticulum",
    rrc_enabled=True,
    rrc_hubs=["your_rrc_hub_destination_hash"],
    rrc_rooms=["general"],
    rrc_nick="RoomBot",
    rrc_auto_reconnect=True,
    rrc_persist_sessions=True,
)

@bot.on_rrc
def on_rrc(event, client, payload):
    if event == "welcome":
        bot.logger.info("Welcomed by hub")
        return
    if event != "msg" or not isinstance(payload, RRCMessage):
        return
    if payload.mention and payload.room:
        client.send_message(
            payload.room,
            f"Heard you, {payload.nick}",
        )

bot.run()

bot.connect_rrc("hub_destination_hash", rooms=["general"])
bot.rrc.send_message("general", "hello room")
bot.rrc.send_action("general", "waves")
bot.disconnect_rrc()