highperfocused/nostrdvm
1
0
Fork 0
mirror of https://github.com/believethehype/nostrdvm.git synced 2025-11-19 22:36:25 +01:00
Code Issues Packages Projects Releases Wiki Activity

add chatbot example and option to bot

Browse Source
This commit is contained in:
Believethehype
2024-09-21 12:54:49 +02:00
parent 802d63b003
commit 9686644afb
5 changed files with 307 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files

357
nostr_dvm/bot.py
View File

@@ -41,6 +41,9 @@ class Bot:
self.dvm_config.NIP89 = nip89config self.dvm_config.NIP89 = nip89config
self.admin_config = admin_config self.admin_config = admin_config
self.keys = Keys.parse(dvm_config.PRIVATE_KEY) self.keys = Keys.parse(dvm_config.PRIVATE_KEY)
self.CHATBOT = False
wait_for_send = True wait_for_send = True
skip_disconnected_relays = True skip_disconnected_relays = True
@@ -57,6 +60,11 @@ class Bot:
print("Nostr BOT public key: " + str(pk.to_bech32()) + " Hex: " + str(pk.to_hex()) + " Name: " + self.NAME) # + print("Nostr BOT public key: " + str(pk.to_bech32()) + " Hex: " + str(pk.to_hex()) + " Name: " + self.NAME) # +
# " Supported DVM tasks: " + # " Supported DVM tasks: " +
# ', '.join(p.NAME + ":" + p.TASK for p in self.dvm_config.SUPPORTED_DVMS) + "\n") # ', '.join(p.NAME + ":" + p.TASK for p in self.dvm_config.SUPPORTED_DVMS) + "\n")
if dvm_config.CHATBOT is not None:
if dvm_config.CHATBOT is True:
self.CHATBOT = True
self.DVM_KEY = dvm_config.DVM_KEY
for relay in self.dvm_config.RELAY_LIST: for relay in self.dvm_config.RELAY_LIST:
await self.client.add_relay(relay) await self.client.add_relay(relay)
@@ -69,6 +77,8 @@ class Bot:
for dvm in self.dvm_config.SUPPORTED_DVMS: for dvm in self.dvm_config.SUPPORTED_DVMS:
if dvm.KIND not in kinds: if dvm.KIND not in kinds:
kinds.append(Kind(dvm.KIND.as_u16() + 1000)) kinds.append(Kind(dvm.KIND.as_u16() + 1000))
if self.CHATBOT:
kinds.append(Kind(6050))
dvm_filter = (Filter().kinds(kinds).since(Timestamp.now())) dvm_filter = (Filter().kinds(kinds).since(Timestamp.now()))
await self.client.subscribe([zap_filter, dm_filter, nip17_filter, dvm_filter], None) await self.client.subscribe([zap_filter, dm_filter, nip17_filter, dvm_filter], None)
@@ -106,6 +116,7 @@ class Bot:
return return
async def handle_dm(nostr_event, giftwrap): async def handle_dm(nostr_event, giftwrap):
sender = nostr_event.author().to_hex() sender = nostr_event.author().to_hex()
if sender == self.keys.public_key().to_hex(): if sender == self.keys.public_key().to_hex():
return return
@@ -145,149 +156,180 @@ class Bot:
print("[" + self.NAME + "]" + sealed + "Message from " + user.name + ": " + decrypted_text) print("[" + self.NAME + "]" + sealed + "Message from " + user.name + ": " + decrypted_text)
# if user selects an index from the overview list...
if decrypted_text != "" and decrypted_text[0].isdigit():
split = decrypted_text.split(' ') if not self.CHATBOT:
index = int(split[0]) - 1 # if user selects an index from the overview list...
# if user sends index info, e.g. 1 info, we fetch the nip89 information and reply with it. if decrypted_text != "" and decrypted_text[0].isdigit():
if len(split) > 1 and split[1].lower() == "info":
await answer_nip89(nostr_event, index, giftwrap, sender)
# otherwise we probably have to do some work, so build an event from input and send it to the DVM
else:
task = self.dvm_config.SUPPORTED_DVMS[index].TASK
print("[" + self.NAME + "] Request from " + str(user.name) + " (" + str(user.nip05) +
", Balance: " + str(user.balance) + " Sats) Task: " + str(task))
if user.isblacklisted:
# If users are blacklisted for some reason, tell them.
answer_blacklisted(nostr_event, giftwrap, sender)
split = decrypted_text.split(' ')
index = int(split[0]) - 1
# if user sends index info, e.g. 1 info, we fetch the nip89 information and reply with it.
if len(split) > 1 and split[1].lower() == "info":
await answer_nip89(nostr_event, index, giftwrap, sender)
# otherwise we probably have to do some work, so build an event from input and send it to the DVM
else: else:
# Parse inputs to params task = self.dvm_config.SUPPORTED_DVMS[index].TASK
tags = build_params(decrypted_text, sender, index) print("[" + self.NAME + "] Request from " + str(user.name) + " (" + str(user.nip05) +
p_tag = Tag.parse(['p', self.dvm_config.SUPPORTED_DVMS[index].PUBLIC_KEY]) ", Balance: " + str(user.balance) + " Sats) Task: " + str(task))
if user.isblacklisted:
# If users are blacklisted for some reason, tell them.
await answer_blacklisted(nostr_event, giftwrap, sender)
if self.dvm_config.SUPPORTED_DVMS[index].SUPPORTS_ENCRYPTION:
tags_str = []
for tag in tags:
tags_str.append(tag.as_vec())
params_as_str = json.dumps(tags_str)
print(params_as_str)
# and encrypt them
encrypted_params = nip04_encrypt(self.keys.secret_key(),
PublicKey.from_hex(
self.dvm_config.SUPPORTED_DVMS[
index].PUBLIC_KEY),
params_as_str)
# add encrypted and p tag on the outside
encrypted_tag = Tag.parse(['encrypted'])
# add the encrypted params to the content
nip90request = (EventBuilder(self.dvm_config.SUPPORTED_DVMS[index].KIND,
encrypted_params, [p_tag, encrypted_tag]).
to_event(self.keys))
else: else:
tags.append(p_tag) # Parse inputs to params
tags = build_params(decrypted_text, sender, index)
p_tag = Tag.parse(['p', self.dvm_config.SUPPORTED_DVMS[index].PUBLIC_KEY])
nip90request = (EventBuilder(self.dvm_config.SUPPORTED_DVMS[index].KIND, if self.dvm_config.SUPPORTED_DVMS[index].SUPPORTS_ENCRYPTION:
"", tags). tags_str = []
to_event(self.keys)) for tag in tags:
tags_str.append(tag.as_vec())
params_as_str = json.dumps(tags_str)
print(params_as_str)
# and encrypt them
encrypted_params = nip04_encrypt(self.keys.secret_key(),
PublicKey.from_hex(
self.dvm_config.SUPPORTED_DVMS[
index].PUBLIC_KEY),
params_as_str)
# add encrypted and p tag on the outside
encrypted_tag = Tag.parse(['encrypted'])
# add the encrypted params to the content
nip90request = (EventBuilder(self.dvm_config.SUPPORTED_DVMS[index].KIND,
encrypted_params, [p_tag, encrypted_tag]).
to_event(self.keys))
else:
tags.append(p_tag)
# remember in the job_list that we have made an event, if anybody asks for payment, nip90request = (EventBuilder(self.dvm_config.SUPPORTED_DVMS[index].KIND,
# we know we actually sent the request "", tags).
entry = {"npub": user.npub, "event_id": nip90request.id().to_hex(), to_event(self.keys))
"dvm_key": self.dvm_config.SUPPORTED_DVMS[index].PUBLIC_KEY, "is_paid": False,
"giftwrap": giftwrap}
self.job_list.append(entry)
# send the event to the DVM # remember in the job_list that we have made an event, if anybody asks for payment,
await send_event(nip90request, client=self.client, dvm_config=self.dvm_config) # we know we actually sent the request
# print(nip90request.as_json()) entry = {"npub": user.npub, "event_id": nip90request.id().to_hex(),
"dvm_key": self.dvm_config.SUPPORTED_DVMS[index].PUBLIC_KEY, "is_paid": False,
"giftwrap": giftwrap}
self.job_list.append(entry)
# send the event to the DVM
await send_event(nip90request, client=self.client, dvm_config=self.dvm_config)
# print(nip90request.as_json())
elif decrypted_text.lower().startswith("invoice"): elif decrypted_text.lower().startswith("invoice"):
requests_rq = False requests_rq = False
amount_str = decrypted_text.lower().split(" ")[1] amount_str = decrypted_text.lower().split(" ")[1]
if amount_str == "qr": if amount_str == "qr":
amount_str = decrypted_text.lower().split(" ")[2] amount_str = decrypted_text.lower().split(" ")[2]
requests_rq = True requests_rq = True
try: try:
amount = int(amount_str) amount = int(amount_str)
except: except:
amount = 100 amount = 100
invoice, hash = create_bolt11_ln_bits(amount, self.dvm_config) invoice, hash = create_bolt11_ln_bits(amount, self.dvm_config)
expires = nostr_event.created_at().as_secs() + (60 * 60 * 24) expires = nostr_event.created_at().as_secs() + (60 * 60 * 24)
qr_code = "https://qrcode.tec-it.com/API/QRCode?data=" + invoice + "&backcolor=%23ffffff&size=small&quietzone=1&errorcorrection=H" qr_code = "https://qrcode.tec-it.com/API/QRCode?data=" + invoice + "&backcolor=%23ffffff&size=small&quietzone=1&errorcorrection=H"
self.invoice_list.append( self.invoice_list.append(
InvoiceToWatch(sender=sender, bolt11=invoice, payment_hash=hash, is_paid=False, InvoiceToWatch(sender=sender, bolt11=invoice, payment_hash=hash, is_paid=False,
expires=expires, amount=amount)) expires=expires, amount=amount))
if requests_rq:
message = invoice + "\n" + qr_code
else:
message = invoice
if giftwrap:
await self.client.send_private_msg(PublicKey.parse(sender), message, None)
else:
await asyncio.sleep(2.0)
evt = EventBuilder.encrypted_direct_msg(self.keys, PublicKey.parse(sender),
message, None).to_event(self.keys)
await send_event(evt, client=self.client, dvm_config=dvm_config)
elif decrypted_text.lower().startswith("balance"):
await asyncio.sleep(2.0)
message = "Your current balance is " + str(user.balance) + (" Sats. Zap me to add to your "
"balance. I will use your "
"balance interact with the DVMs "
"for you.\nI support both "
"public and private Zaps, "
"as well as "
"Zapplepay.\nOr write \"invoice "
"100\" to receive an invoice of "
"100 sats (or any other amount) "
"to top up your balance")
if giftwrap:
await self.client.send_private_msg(PublicKey.parse(sender), message, None)
else:
evt = EventBuilder.encrypted_direct_msg(self.keys, PublicKey.parse(sender),
message, None).to_event(self.keys)
await send_event(evt, client=self.client, dvm_config=dvm_config)
elif decrypted_text.startswith("cashuA"):
print("Received Cashu token:" + decrypted_text)
cashu_redeemed, cashu_message, total_amount, fees = await redeem_cashu(decrypted_text,
self.dvm_config,
self.client)
print(cashu_message)
if cashu_message == "success":
await update_user_balance(self.dvm_config.DB, sender, total_amount, client=self.client,
config=self.dvm_config)
else:
await asyncio.sleep(2.0)
message = "Error: " + cashu_message + ". Token has not been redeemed."
if requests_rq:
message = invoice + "\n" + qr_code
else:
message = invoice
if giftwrap: if giftwrap:
await self.client.send_private_msg(PublicKey.parse(sender), message, None) await self.client.send_private_msg(PublicKey.parse(sender), message, None)
else: else:
evt = EventBuilder.encrypted_direct_msg(self.keys, PublicKey.from_hex(sender), message, #await self.client.send_direct_msg(PublicKey.parse(sender), message, None)
None).to_event(self.keys) await self.client.send_private_msg(PublicKey.parse(sender), message, None)
await send_event(evt, client=self.client, dvm_config=self.dvm_config)
elif decrypted_text.lower().startswith("what's the second best"):
await asyncio.sleep(2.0)
message = "No, there is no second best.\n\nhttps://cdn.nostr.build/p/mYLv.mp4"
if giftwrap:
await self.client.send_private_msg(PublicKey.parse(sender), message, None)
else:
evt = await EventBuilder.encrypted_direct_msg(self.keys, PublicKey.parse(sender),
message,
nostr_event.id()).to_event(self.keys)
await send_event(evt, client=self.client, dvm_config=self.dvm_config)
elif decrypted_text.lower().startswith("balance"):
await asyncio.sleep(2.0)
message = "Your current balance is " + str(user.balance) + (" Sats. Zap me to add to your "
"balance. I will use your "
"balance interact with the DVMs "
"for you.\nI support both "
"public and private Zaps, "
"as well as "
"Zapplepay.\nOr write \"invoice "
"100\" to receive an invoice of "
"100 sats (or any other amount) "
"to top up your balance")
if giftwrap:
await self.client.send_private_msg(PublicKey.parse(sender), message, None)
else:
#await self.client.send_direct_msg(PublicKey.parse(sender), message, None)
await self.client.send_private_msg(PublicKey.parse(sender), message, None)
elif decrypted_text.startswith("cashuA"):
print("Received Cashu token:" + decrypted_text)
cashu_redeemed, cashu_message, total_amount, fees = await redeem_cashu(decrypted_text,
self.dvm_config,
self.client)
print(cashu_message)
if cashu_message == "success":
await update_user_balance(self.dvm_config.DB, sender, total_amount, client=self.client,
config=self.dvm_config)
else:
await asyncio.sleep(2.0)
message = "Error: " + cashu_message + ". Token has not been redeemed."
if giftwrap:
await self.client.send_private_msg(PublicKey.parse(sender), message, None)
else:
#await self.client.send_direct_msg(PublicKey.parse(sender), message, None)
await self.client.send_private_msg(PublicKey.parse(sender), message, None)
elif decrypted_text.lower().startswith("what's the second best"):
await asyncio.sleep(2.0)
message = "No, there is no second best.\n\nhttps://cdn.nostr.build/p/mYLv.mp4"
if giftwrap:
await self.client.send_private_msg(PublicKey.parse(sender), message, None)
else:
#await self.client.send_direct_msg(PublicKey.parse(sender), message, nostr_event.id())
await self.client.send_private_msg(PublicKey.parse(sender), message, None)
else:
# Build an overview of known DVMs and send it to the user
await answer_overview(nostr_event, giftwrap, sender)
else: else:
# Build an overview of known DVMs and send it to the user kind = 5050
await answer_overview(nostr_event, giftwrap, sender) tags = []
print(decrypted_text)
input_tag = Tag.parse(["i", decrypted_text, "text"])
tags.append(input_tag)
alt_tag = Tag.parse(["alt", "text-generation"])
tags.append(alt_tag)
relaylist = ["relays"]
for relay in self.dvm_config.RELAY_LIST:
relaylist.append(relay)
relays_tag = Tag.parse(relaylist)
tags.append(relays_tag)
output_tag = Tag.parse(["output", "text/plain"])
tags.append(output_tag)
p_tag = Tag.parse(['p', self.DVM_KEY])
tags.append(p_tag)
nip90request = (EventBuilder(Kind(kind),
"", tags).
to_event(self.keys))
entry = {"npub": user.npub, "event_id": nip90request.id().to_hex(),
"dvm_key": self.DVM_KEY, "is_paid": False,
"giftwrap": giftwrap}
self.job_list.append(entry)
# send the event to the DVM
await send_event(nip90request, client=self.client, dvm_config=self.dvm_config)
except Exception as e: except Exception as e:
print("Error in bot " + str(e)) print("Error in bot " + str(e))
@@ -346,12 +388,9 @@ class Bot:
if entry["giftwrap"]: if entry["giftwrap"]:
await self.client.send_private_msg(PublicKey.parse(entry["npub"]), content, None) await self.client.send_private_msg(PublicKey.parse(entry["npub"]), content, None)
else: else:
reply_event = EventBuilder.encrypted_direct_msg(self.keys, #await self.client.send_direct_msg(PublicKey.from_hex(entry['npub']), content, None)
PublicKey.from_hex(entry['npub']), await self.client.send_private_msg(PublicKey.parse(entry['npub']),
content, content, None)
None).to_event(self.keys)
await send_event(reply_event, client=self.client, dvm_config=dvm_config)
print(status + ": " + content) print(status + ": " + content)
print( print(
"[" + self.NAME + "] Received reaction from " + nostr_event.author().to_hex() + " message to orignal sender " + user.name) "[" + self.NAME + "] Received reaction from " + nostr_event.author().to_hex() + " message to orignal sender " + user.name)
@@ -381,26 +420,20 @@ class Bot:
await self.client.send_private_msg(PublicKey.parse(entry["npub"]), message, await self.client.send_private_msg(PublicKey.parse(entry["npub"]), message,
None) None)
else: else:
evt = EventBuilder.encrypted_direct_msg(self.keys, #await self.client.send_direct_msg(PublicKey.parse(PublicKey.parse(entry["npub"])), message, None)
PublicKey.parse(entry["npub"]), await self.client.send_private_msg(PublicKey.parse(entry["npub"]), message, None)
message,
None).to_event(self.keys)
await send_event(evt, client=self.client, dvm_config=dvm_config)
print( print(
"[" + self.NAME + "] Replying " + user.name + " with \"scheduled\" confirmation") "[" + self.NAME + "] Replying " + user.name + " with \"scheduled\" confirmation")
else: else:
print("Bot payment-required") print("Bot payment-required")
await asyncio.sleep(2.0) await asyncio.sleep(2.0)
evt = EventBuilder.encrypted_direct_msg(self.keys, message = "Current balance: " + str( user.balance) + " Sats. Balance of " + str(amount) + " Sats required. Please zap me with at least " + str(int(amount - user.balance))+ " Sats, then try again.",
PublicKey.parse(entry["npub"]), #await self.client.send_direct_msg(PublicKey.parse(PublicKey.parse(entry["npub"])),
"Current balance: " + str( # message, None)
user.balance) + " Sats. Balance of " + str( await self.client.send_private_msg(PublicKey.parse(entry["npub"]),
amount) + " Sats required. Please zap me with at least " + message, None)
str(int(amount - user.balance))
+ " Sats, then try again.",
None).to_event(self.keys)
await send_event(evt, client=self.client, dvm_config=dvm_config)
return return
if len(tag.as_vec()) > 2: if len(tag.as_vec()) > 2:
@@ -432,6 +465,7 @@ class Bot:
async def handle_nip90_response_event(nostr_event: Event): async def handle_nip90_response_event(nostr_event: Event):
try: try:
ptag = "" ptag = ""
etag = "" etag = ""
is_encrypted = False is_encrypted = False
@@ -473,11 +507,8 @@ class Bot:
if entry["giftwrap"]: if entry["giftwrap"]:
await self.client.send_private_msg(PublicKey.parse(user.npub), content, None) await self.client.send_private_msg(PublicKey.parse(user.npub), content, None)
else: else:
reply_event = EventBuilder.encrypted_direct_msg(self.keys, #await self.client.send_direct_msg(PublicKey.parse(user.npub), content, None)
PublicKey.parse(user.npub), await self.client.send_private_msg(PublicKey.parse(user.npub), content, None)
content,
None).to_event(self.keys)
await send_event(reply_event, client=self.client, dvm_config=dvm_config)
except Exception as e: except Exception as e:
print(e) print(e)
@@ -547,23 +578,20 @@ class Bot:
text = message + "\nSelect an Index and provide an input (e.g. \"2 A purple ostrich\")\nType \"index info\" to learn more about each DVM. (e.g. \"2 info\")\n\n Type \"balance\" to see your current balance" text = message + "\nSelect an Index and provide an input (e.g. \"2 A purple ostrich\")\nType \"index info\" to learn more about each DVM. (e.g. \"2 info\")\n\n Type \"balance\" to see your current balance"
if giftwrap: if giftwrap:
await self.client.send_private_msg(PublicKey.parse(sender), text, None) await self.client.send_private_msg(PublicKey.parse(sender), text, nostr_event.id())
else: else:
evt = EventBuilder.encrypted_direct_msg(self.keys, PublicKey.parse(sender), #await self.client.send_direct_msg(PublicKey.parse(sender), text, nostr_event.id())
text, await self.client.send_private_msg(PublicKey.parse(sender), text, nostr_event.id())
nostr_event.id()).to_event(self.keys)
await send_event(evt, client=self.client, dvm_config=dvm_config) async def answer_blacklisted(nostr_event, giftwrap, sender):
def answer_blacklisted(nostr_event, giftwrap, sender):
message = "Your are currently blocked from this service." message = "Your are currently blocked from this service."
if giftwrap: if giftwrap:
self.client.send_sealed_msg(PublicKey.parse(sender), message, None) await self.client.send_private_msg(PublicKey.parse(sender), message, None)
else: else:
# For some reason an admin might blacklist npubs, e.g. for abusing the service #await self.client.send_direct_msg(nostr_event.author(), message, None)
evt = EventBuilder.encrypted_direct_msg(self.keys, nostr_event.author(), await self.client.send_private_msg(PublicKey.parse(sender), message, None)
message, None).to_event(self.keys)
send_event(evt, client=self.client, dvm_config=dvm_config)
async def answer_nip89(nostr_event, index, giftwrap, sender): async def answer_nip89(nostr_event, index, giftwrap, sender):
info = await print_dvm_info(self.client, index) info = await print_dvm_info(self.client, index)
@@ -574,9 +602,8 @@ class Bot:
if giftwrap: if giftwrap:
await self.client.send_private_msg(PublicKey.parse(sender), info, None) await self.client.send_private_msg(PublicKey.parse(sender), info, None)
else: else:
evt = EventBuilder.encrypted_direct_msg(self.keys, nostr_event.author(), #await self.client.send_direct_msg(nostr_event.author(), info, None)
info, None).to_event(self.keys) await self.client.send_private_msg(PublicKey.parse(sender), info, None)
await send_event(evt, client=self.client, dvm_config=dvm_config)
def build_params(decrypted_text, author, index): def build_params(decrypted_text, author, index):
tags = [] tags = []

14
nostr_dvm/subscription.py
View File

@@ -350,13 +350,12 @@ class Subscription:
await send_status_success(nostr_event, "noogle.lol") await send_status_success(nostr_event, "noogle.lol")
keys = Keys.parse(dvm_config.PRIVATE_KEY)
message = ("Subscribed to DVM " + tier + ". Renewing on: " + str( message = ("Subscribed to DVM " + tier + ". Renewing on: " + str(
Timestamp.from_secs(end).to_human_datetime().replace("Z", " ").replace("T", Timestamp.from_secs(end).to_human_datetime().replace("Z", " ").replace("T",
" ") + " GMT")) " ") + " GMT"))
evt = EventBuilder.encrypted_direct_msg(keys, PublicKey.parse(subscriber), message,
None).to_event(keys) self.client.send_private_msg(PublicKey.parse(subscriber), message, None)
await send_event(evt, client=self.client, dvm_config=dvm_config)
@@ -415,10 +414,9 @@ class Subscription:
"Renewed Subscription to DVM " + subscription.tier + ". Next renewal: " + str( "Renewed Subscription to DVM " + subscription.tier + ". Next renewal: " + str(
Timestamp.from_secs(end).to_human_datetime().replace("Z", " ").replace("T", Timestamp.from_secs(end).to_human_datetime().replace("Z", " ").replace("T",
" "))) " ")))
evt = EventBuilder.encrypted_direct_msg(keys, PublicKey.parse(subscription.subscriber), #await self.client.send_direct_msg(PublicKey.parse(subscription.subscriber), message, None)
message, await self.client.send_private_msg(PublicKey.parse(subscription.subscriber), message, None)
None).to_event(keys)
await send_event(evt, client=self.client, dvm_config=dvm_config)
async def check_subscriptions(): async def check_subscriptions():
try: try:

5
nostr_dvm/utils/database_utils.py
View File

@@ -196,9 +196,8 @@ async def update_user_balance(db, npub, additional_sats, client, config, giftwra
if giftwrap: if giftwrap:
await client.send_private_msg(PublicKey.parse(npub), message, None) await client.send_private_msg(PublicKey.parse(npub), message, None)
else: else:
evt = EventBuilder.encrypted_direct_msg(keys, PublicKey.parse(npub), message, #await client.send_direct_msg(PublicKey.parse(npub), message, None)
None).to_event(keys) await client.send_private_msg(PublicKey.parse(npub), message, None)
await send_event(evt, client=client, dvm_config=config)
def update_user_subscription(npub, subscribed_until, client, dvm_config): def update_user_subscription(npub, subscribed_until, client, dvm_config):

103
tests/chat_bot.py Normal file
View File

@@ -0,0 +1,103 @@
import asyncio
import json
import os
import threading
from pathlib import Path
import dotenv
from duck_chat import ModelType
from nostr_sdk import Keys, Kind
from nostr_dvm.bot import Bot
from nostr_dvm.tasks import textextraction_pdf, convert_media, discovery_inactive_follows, translation_google
from nostr_dvm.tasks.generic_dvm import GenericDVM
from nostr_dvm.utils.admin_utils import AdminConfig
from nostr_dvm.utils.backend_utils import keep_alive
from nostr_dvm.utils.definitions import EventDefinitions
from nostr_dvm.utils.dvmconfig import DVMConfig, build_default_config
from nostr_dvm.utils.external_dvm_utils import build_external_dvm
from nostr_dvm.utils.nip89_utils import NIP89Config, check_and_set_d_tag
from nostr_dvm.utils.nostr_utils import check_and_set_private_key
from nostr_dvm.utils.output_utils import PostProcessFunctionType
from nostr_dvm.utils.zap_utils import check_and_set_ln_bits_keys
def playground(announce = False):
identifier = "bot_test"
bot_config = build_default_config(identifier)
bot_config.CHATBOT = True
bot_config.DVM_KEY = "aa8ab5b774d47e7b29a985dd739cfdcccf93451678bf7977ba1b2e094ecd8b30"
admin_config = AdminConfig()
admin_config.REBROADCAST_NIP65_RELAY_LIST = True
admin_config.UPDATE_PROFILE = True
x = threading.Thread(target=Bot, args=([bot_config, admin_config]))
x.start()
kind = 5050
admin_config = AdminConfig()
admin_config.REBROADCAST_NIP89 = announce
admin_config.REBROADCAST_NIP65_RELAY_LIST = announce
admin_config.UPDATE_PROFILE = announce
name = "DuckChat"
identifier = "duckduckchat" # Chose a unique identifier in order to get a lnaddress
dvm_config = build_default_config(identifier)
dvm_config.KIND = Kind(kind) # Manually set the Kind Number (see data-vending-machines.org)
dvm_config.SEND_FEEDBACK_EVENTS = False
# Add NIP89
nip89info = {
"name": name,
"image": "https://image.nostr.build/28da676a19841dcfa7dcf7124be6816842d14b84f6046462d2a3f1268fe58d03.png",
"about": "I'm briding DuckDuckAI'",
"encryptionSupported": True,
"cashuAccepted": True,
"nip90Params": {
}
}
nip89config = NIP89Config()
nip89config.KIND = kind
nip89config.DTAG = check_and_set_d_tag(identifier, name, dvm_config.PRIVATE_KEY, nip89info["image"])
nip89config.CONTENT = json.dumps(nip89info)
options = {
"input": "",
}
dvm = GenericDVM(name=name, dvm_config=dvm_config, nip89config=nip89config,
admin_config=admin_config, options=options)
async def process(request_form):
# pip install -U https://github.com/mrgick/duckduckgo-chat-ai/archive/master.zip
from duck_chat import DuckChat
options = dvm.set_options(request_form)
async with DuckChat(model=ModelType.GPT4o) as chat:
query = options["input"]
result = await chat.ask_question(query)
print(result)
return result
dvm.process = process # overwrite the process function with the above one
dvm.run(True)
if __name__ == '__main__':
env_path = Path('.env')
if not env_path.is_file():
with open('.env', 'w') as f:
print("Writing new .env file")
f.write('')
if env_path.is_file():
print(f'loading environment from {env_path.resolve()}')
dotenv.load_dotenv(env_path, verbose=True, override=True)
else:
raise FileNotFoundError(f'.env file not found at {env_path} ')
playground(False)

5
tests/simplebot.py
View File

@@ -39,7 +39,10 @@ async def test():
try: try:
msg = nip04_decrypt(sk, event.author(), event.content()) msg = nip04_decrypt(sk, event.author(), event.content())
print(f"Received new msg: {msg}") print(f"Received new msg: {msg}")
await client.send_direct_msg(event.author(), f"Echo: {msg}", event.id()) #await client.send_direct_msg(event.author(), f"Echo: {msg}", event.id())
await client.send_private_msg(event.author(), f"Echo: {msg}", event.id())
except Exception as e: except Exception as e:
print(f"Error during content NIP04 decryption: {e}") print(f"Error during content NIP04 decryption: {e}")
elif event.kind().as_enum() == KindEnum.GIFT_WRAP(): elif event.kind().as_enum() == KindEnum.GIFT_WRAP():