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

future: rework functions to async to be compatible with newest nostrsdk

Browse Source
This commit is contained in:
Believethehype 2024-06-06 02:25:34 +02:00
parent f75ca73284
commit a286f4946e
43 changed files with 562 additions and 599 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

81
nostr_dvm/bot.py
View File

@ -1,3 +1,4 @@
import asyncio
import json
import os
import signal
@ -23,15 +24,7 @@ class Bot:
job_list: list
# This is a simple list just to keep track which events we created and manage, so we don't pay for other requests
def __init__(self, dvm_config, admin_config=None):
self.NAME = "Bot"
dvm_config.DB = "db/" + self.NAME + ".db"
self.dvm_config = dvm_config
nip89config = NIP89Config()
nip89config.NAME = self.NAME
self.dvm_config.NIP89 = nip89config
self.admin_config = admin_config
self.keys = Keys.parse(dvm_config.PRIVATE_KEY)
async def init_bot(self, dvm_config, admin_config=None):
wait_for_send = True
skip_disconnected_relays = True
opts = (Options().wait_for_send(wait_for_send).send_timeout(timedelta(seconds=self.dvm_config.RELAY_TIMEOUT))
@ -48,8 +41,8 @@ class Bot:
', '.join(p.NAME + ":" + p.TASK for p in self.dvm_config.SUPPORTED_DVMS) + "\n")
for relay in self.dvm_config.RELAY_LIST:
self.client.add_relay(relay)
self.client.connect()
await self.client.add_relay(relay)
await self.client.connect()
zap_filter = Filter().pubkey(pk).kinds([EventDefinitions.KIND_ZAP]).since(Timestamp.now())
dm_filter = Filter().pubkey(pk).kinds([EventDefinitions.KIND_DM]).since(Timestamp.now())
@ -64,40 +57,54 @@ class Bot:
self.client.subscribe([zap_filter, dm_filter, nip59_filter, dvm_filter], None)
create_sql_table(self.dvm_config.DB)
admin_make_database_updates(adminconfig=self.admin_config, dvmconfig=self.dvm_config, client=self.client)
await admin_make_database_updates(adminconfig=self.admin_config, dvmconfig=self.dvm_config, client=self.client)
def __init__(self, dvm_config, admin_config=None):
self.NAME = "Bot"
dvm_config.DB = "db/" + self.NAME + ".db"
self.dvm_config = dvm_config
nip89config = NIP89Config()
nip89config.NAME = self.NAME
self.dvm_config.NIP89 = nip89config
self.admin_config = admin_config
self.keys = Keys.parse(dvm_config.PRIVATE_KEY)
asyncio.run(self.init_bot(dvm_config, admin_config))
asyncio.run(self.run_bot(dvm_config))
# add_sql_table_column(dvm_config.DB)
async def run_bot(self, dvm_config):
class NotificationHandler(HandleNotification):
client = self.client
dvm_config = self.dvm_config
keys = self.keys
def handle(self, relay_url, subscription_id, nostr_event):
async def handle(self, relay_url, subscription_id, nostr_event):
if (EventDefinitions.KIND_NIP90_EXTRACT_TEXT.as_u64() + 1000 <= nostr_event.kind().as_u64()
<= EventDefinitions.KIND_NIP90_GENERIC.as_u64() + 1000):
handle_nip90_response_event(nostr_event)
await handle_nip90_response_event(nostr_event)
elif nostr_event.kind() == EventDefinitions.KIND_FEEDBACK:
handle_nip90_feedback(nostr_event)
await handle_nip90_feedback(nostr_event)
elif nostr_event.kind() == EventDefinitions.KIND_ZAP:
handle_zap(nostr_event)
await handle_zap(nostr_event)
elif nostr_event.kind() == EventDefinitions.KIND_DM:
try:
handle_dm(nostr_event, False)
await handle_dm(nostr_event, False)
except Exception as e:
print(f"Error during content NIP04 decryption: {e}")
elif nostr_event.kind().match_enum(KindEnum.GIFT_WRAP()):
try:
handle_dm(nostr_event, True)
await handle_dm(nostr_event, True)
except Exception as e:
print(f"Error during content NIP59 decryption: {e}")
def handle_msg(self, relay_url, msg):
async def handle_msg(self, relay_url, msg):
return
def handle_dm(nostr_event, giftwrap):
async def handle_dm(nostr_event, giftwrap):
sender = nostr_event.author().to_hex()
if sender == self.keys.public_key().to_hex():
return
@ -131,7 +138,7 @@ class Bot:
print(f"Error during content NIP04 decryption: {e}")
if decrypted_text != "":
user = get_or_add_user(db=self.dvm_config.DB, npub=sender, client=self.client,
user = await get_or_add_user(db=self.dvm_config.DB, npub=sender, client=self.client,
config=self.dvm_config)
print("[" + self.NAME + "]" + sealed + "Message from " + user.name + ": " + decrypted_text)
@ -217,10 +224,10 @@ class Bot:
else:
evt = EventBuilder.encrypted_direct_msg(self.keys, PublicKey.parse(sender),
message,None).to_event(self.keys)
send_event(evt, client=self.client, dvm_config=dvm_config)
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 = redeem_cashu(decrypted_text,
cashu_redeemed, cashu_message, total_amount, fees = await redeem_cashu(decrypted_text,
self.dvm_config,
self.client)
print(cashu_message)
@ -255,7 +262,7 @@ class Bot:
except Exception as e:
print("Error in bot " + str(e))
def handle_nip90_feedback(nostr_event):
async def handle_nip90_feedback(nostr_event):
# print(nostr_event.as_json())
try:
is_encrypted = False
@ -303,7 +310,7 @@ class Bot:
if status == "success" or status == "error" or status == "processing" or status == "partial" and content != "":
entry = next((x for x in self.job_list if x['event_id'] == etag), None)
if entry is not None and entry['dvm_key'] == nostr_event.author().to_hex():
user = get_or_add_user(db=self.dvm_config.DB, npub=entry['npub'],
user = await get_or_add_user(db=self.dvm_config.DB, npub=entry['npub'],
client=self.client, config=self.dvm_config)
time.sleep(2.0)
if entry["giftwrap"]:
@ -328,7 +335,7 @@ class Bot:
if entry is not None and entry['is_paid'] is False and entry[
'dvm_key'] == nostr_event.author().to_hex():
# if we get a bolt11, we pay and move on
user = get_or_add_user(db=self.dvm_config.DB, npub=entry["npub"],
user = await get_or_add_user(db=self.dvm_config.DB, npub=entry["npub"],
client=self.client, config=self.dvm_config)
if user.balance >= amount:
balance = max(user.balance - amount, 0)
@ -345,7 +352,7 @@ class Bot:
PublicKey.parse(entry["npub"]),
message,
None).to_event(self.keys)
send_event(evt, client=self.client, dvm_config=dvm_config)
await send_event(evt, client=self.client, dvm_config=dvm_config)
print(
"[" + self.NAME + "] Replying " + user.name + " with \"scheduled\" confirmation")
@ -360,14 +367,14 @@ class Bot:
str(int(amount - user.balance))
+ " Sats, then try again.",
None).to_event(self.keys)
send_event(evt, client=self.client, dvm_config=dvm_config)
await send_event(evt, client=self.client, dvm_config=dvm_config)
return
if len(tag.as_vec()) > 2:
bolt11 = tag.as_vec()[2]
# else we create a zap
else:
user = get_or_add_user(db=self.dvm_config.DB, npub=nostr_event.author().to_hex(),
user = await get_or_add_user(db=self.dvm_config.DB, npub=nostr_event.author().to_hex(),
client=self.client, config=self.dvm_config)
print("Paying: " + user.name)
bolt11 = zaprequest(user.lud16, amount, "Zap", nostr_event, self.keys,
@ -389,7 +396,7 @@ class Bot:
except Exception as e:
print(e)
def handle_nip90_response_event(nostr_event: Event):
async def handle_nip90_response_event(nostr_event: Event):
try:
ptag = ""
etag = ""
@ -406,7 +413,7 @@ class Bot:
if entry is not None and entry[
'dvm_key'] == nostr_event.author().to_hex():
print(entry)
user = get_or_add_user(db=self.dvm_config.DB, npub=entry['npub'],
user = await get_or_add_user(db=self.dvm_config.DB, npub=entry['npub'],
client=self.client, config=self.dvm_config)
self.job_list.remove(entry)
@ -436,12 +443,12 @@ class Bot:
PublicKey.parse(user.npub),
content,
None).to_event(self.keys)
send_event(reply_event, client=self.client, dvm_config=dvm_config)
await send_event(reply_event, client=self.client, dvm_config=dvm_config)
except Exception as e:
print(e)
def handle_zap(zap_event):
async def handle_zap(zap_event):
print("[" + self.NAME + "] Zap received")
try:
invoice_amount, zapped_event, sender, message, anon = parse_zap_event_tags(zap_event,
@ -455,7 +462,7 @@ class Bot:
if tag.as_vec()[0] == "e":
etag = tag.as_vec()[1]
user = get_or_add_user(self.dvm_config.DB, sender, client=self.client, config=self.dvm_config)
user = await get_or_add_user(self.dvm_config.DB, sender, client=self.client, config=self.dvm_config)
entry = next((x for x in self.job_list if x['event_id'] == etag), None)
print(entry)
@ -465,7 +472,7 @@ class Bot:
print(sender)
if entry is not None and entry['is_paid'] is True and entry['dvm_key'] == sender:
# if we get a bolt11, we pay and move on
user = get_or_add_user(db=self.dvm_config.DB, npub=entry["npub"],
user = await get_or_add_user(db=self.dvm_config.DB, npub=entry["npub"],
client=self.client, config=self.dvm_config)
sender = user.npub
@ -692,7 +699,7 @@ class Bot:
return None
self.client.handle_notifications(NotificationHandler())
await self.client.handle_notifications(NotificationHandler())
try:
while True:

158
nostr_dvm/dvm.py
View File

@ -1,3 +1,4 @@
import asyncio
import json
import os
import subprocess
@ -34,15 +35,13 @@ class DVM:
job_list: list
jobs_on_hold_list: list
def __init__(self, dvm_config, admin_config=None):
self.dvm_config = dvm_config
self.admin_config = admin_config
self.keys = Keys.parse(dvm_config.PRIVATE_KEY)
async def init_dvm(self, dvm_config, admin_config=None):
wait_for_send = False
skip_disconnected_relays = True
relaylimits = RelayLimits.disable()
opts = (Options().wait_for_send(wait_for_send).send_timeout(timedelta(seconds=self.dvm_config.RELAY_TIMEOUT))
.skip_disconnected_relays(skip_disconnected_relays).relay_limits(relaylimits))
opts = (
Options().wait_for_send(wait_for_send).send_timeout(timedelta(seconds=self.dvm_config.RELAY_TIMEOUT))
.skip_disconnected_relays(skip_disconnected_relays).relay_limits(relaylimits))
signer = NostrSigner.keys(self.keys)
self.client = Client.with_opts(signer, opts)
@ -55,8 +54,8 @@ class DVM:
', '.join(p.NAME + ":" + p.TASK for p in self.dvm_config.SUPPORTED_DVMS) + bcolors.ENDC)
for relay in self.dvm_config.RELAY_LIST:
self.client.add_relay(relay)
self.client.connect()
await self.client.add_relay(relay)
await self.client.connect()
zap_filter = Filter().pubkey(pk).kinds([EventDefinitions.KIND_ZAP]).since(Timestamp.now())
kinds = [EventDefinitions.KIND_NIP90_GENERIC]
@ -64,28 +63,35 @@ class DVM:
if dvm.KIND not in kinds:
kinds.append(dvm.KIND)
dvm_filter = (Filter().kinds(kinds).since(Timestamp.now()))
self.client.subscribe([dvm_filter, zap_filter], None)
create_sql_table(self.dvm_config.DB)
admin_make_database_updates(adminconfig=self.admin_config, dvmconfig=self.dvm_config, client=self.client)
await admin_make_database_updates(adminconfig=self.admin_config, dvmconfig=self.dvm_config, client=self.client)
await self.client.subscribe([dvm_filter, zap_filter], None)
def __init__(self, dvm_config, admin_config=None):
self.dvm_config = dvm_config
self.admin_config = admin_config
self.keys = Keys.parse(dvm_config.PRIVATE_KEY)
asyncio.run(self.init_dvm(dvm_config, admin_config))
asyncio.run(self.run_dvm(dvm_config))
async def run_dvm(self, dvm_config):
class NotificationHandler(HandleNotification):
client = self.client
dvm_config = self.dvm_config
keys = self.keys
def handle(self, relay_url, subscription_id, nostr_event: Event):
async def handle(self, relay_url, subscription_id, nostr_event: Event):
if EventDefinitions.KIND_NIP90_EXTRACT_TEXT.as_u64() <= nostr_event.kind().as_u64() <= EventDefinitions.KIND_NIP90_GENERIC.as_u64():
handle_nip90_job_event(nostr_event)
await handle_nip90_job_event(nostr_event)
elif nostr_event.kind().as_u64() == EventDefinitions.KIND_ZAP.as_u64():
handle_zap(nostr_event)
await handle_zap(nostr_event)
def handle_msg(self, relay_url, msg):
async def handle_msg(self, relay_url, msg):
return
def handle_nip90_job_event(nip90_event):
async def handle_nip90_job_event(nip90_event):
# decrypted encrypted events
nip90_event = check_and_decrypt_tags(nip90_event, self.dvm_config)
# if event is encrypted, but we can't decrypt it (e.g. because its directed to someone else), return
@ -113,11 +119,11 @@ class DVM:
# if task is supported, continue, else do nothing.
if task_supported:
# fetch or add user contacting the DVM from/to local database
user = get_or_add_user(self.dvm_config.DB, nip90_event.author().to_hex(), client=self.client,
user = await get_or_add_user(self.dvm_config.DB, nip90_event.author().to_hex(), client=self.client,
config=self.dvm_config, skip_meta=False)
# if user is blacklisted for some reason, send an error reaction and return
if user.isblacklisted:
send_job_status_reaction(nip90_event, "error", client=self.client, dvm_config=self.dvm_config)
await send_job_status_reaction(nip90_event, "error", client=self.client, dvm_config=self.dvm_config)
print("[" + self.dvm_config.NIP89.NAME + "] Request by blacklisted user, skipped")
return
@ -130,7 +136,7 @@ class DVM:
# If this is a subscription DVM and the Task is directed to us, check for active subscription
if dvm_config.NIP88 is not None and p_tag_str == self.dvm_config.PUBLIC_KEY:
send_job_status_reaction(nip90_event, "subscription-required", True, amount, self.client,
await send_job_status_reaction(nip90_event, "subscription-required", True, amount, self.client,
"Checking Subscription Status, please wait..", self.dvm_config)
# if we stored in the database that the user has an active subscription, we don't need to check it
print("User Subscription: " + str(user.subscribed) + " Current time: " + str(
@ -139,14 +145,14 @@ class DVM:
if int(user.subscribed) > int(Timestamp.now().as_secs()):
print("User subscribed until: " + str(Timestamp.from_secs(user.subscribed).to_human_datetime()))
user_has_active_subscription = True
send_job_status_reaction(nip90_event, "subscription-required", True, amount,
await send_job_status_reaction(nip90_event, "subscription-required", True, amount,
self.client, "User subscripton active until " +
Timestamp.from_secs(int(user.subscribed)).to_human_datetime().replace(
"Z", " ").replace("T", " ") + " GMT", self.dvm_config)
# otherwise we check for an active subscription by checking recipie events
else:
print("[" + self.dvm_config.NIP89.NAME + "] Checking Subscription status")
send_job_status_reaction(nip90_event, "subscription-required", True, amount, self.client,
await send_job_status_reaction(nip90_event, "subscription-required", True, amount, self.client,
"I Don't have information about subscription status, checking on the Nostr. This might take a few seconds",
self.dvm_config)
@ -155,7 +161,7 @@ class DVM:
self.dvm_config.PUBLIC_KEY)
if subscription_status["isActive"]:
send_job_status_reaction(nip90_event, "subscription-required", True, amount, self.client,
await send_job_status_reaction(nip90_event, "subscription-required", True, amount, self.client,
"User subscripton active until " + Timestamp.from_secs(int(
subscription_status[
"validUntil"])).to_human_datetime().replace("Z",
@ -170,7 +176,7 @@ class DVM:
self.client, self.dvm_config)
else:
print("No active subscription found")
send_job_status_reaction(nip90_event, "subscription-required", True, amount, self.client,
await send_job_status_reaction(nip90_event, "subscription-required", True, amount, self.client,
"No active subscription found. Manage your subscription at: " + self.dvm_config.SUBSCRIPTION_MANAGEMENT,
self.dvm_config)
@ -181,11 +187,11 @@ class DVM:
cashu_redeemed = False
if cashu != "":
print(cashu)
cashu_redeemed, cashu_message, redeem_amount, fees = redeem_cashu(cashu, self.dvm_config,
cashu_redeemed, cashu_message, redeem_amount, fees = await redeem_cashu(cashu, self.dvm_config,
self.client, int(amount))
print(cashu_message)
if cashu_message != "success":
send_job_status_reaction(nip90_event, "error", False, amount, self.client, cashu_message,
await send_job_status_reaction(nip90_event, "error", False, amount, self.client, cashu_message,
self.dvm_config)
return
# if user is whitelisted or task is free, just do the job
@ -197,14 +203,14 @@ class DVM:
". Starting processing..")
if dvm_config.SEND_FEEDBACK_EVENTS:
send_job_status_reaction(nip90_event, "processing", True, 0,
await send_job_status_reaction(nip90_event, "processing", True, 0,
content=self.dvm_config.CUSTOM_PROCESSING_MESSAGE,
client=self.client, dvm_config=self.dvm_config, user=user)
# when we reimburse users on error make sure to not send anything if it was free
if user.iswhitelisted or task_is_free:
amount = 0
do_work(nip90_event, amount)
await do_work(nip90_event, amount)
# if task is directed to us via p tag and user has balance or is subscribed, do the job and update balance
elif (p_tag_str == self.dvm_config.PUBLIC_KEY and (
user.balance >= int(
@ -225,11 +231,11 @@ class DVM:
print("[" + self.dvm_config.NIP89.NAME + "] User has active subscription for task: " + task +
". Starting processing.. Balance remains at: " + str(user.balance))
send_job_status_reaction(nip90_event, "processing", True, 0,
await send_job_status_reaction(nip90_event, "processing", True, 0,
content=self.dvm_config.CUSTOM_PROCESSING_MESSAGE,
client=self.client, dvm_config=self.dvm_config)
do_work(nip90_event, amount)
await do_work(nip90_event, amount)
# else send a payment required event to user
elif p_tag_str == "" or p_tag_str == self.dvm_config.PUBLIC_KEY:
@ -238,7 +244,7 @@ class DVM:
print(
"[" + self.dvm_config.NIP89.NAME + "] Hinting user for Subscription: " +
nip90_event.id().to_hex())
send_job_status_reaction(nip90_event, "subscription-required",
await send_job_status_reaction(nip90_event, "subscription-required",
False, 0, client=self.client,
dvm_config=self.dvm_config)
else:
@ -253,7 +259,7 @@ class DVM:
if bid > 0:
bid_offer = int(bid / 1000)
if bid_offer >= int(amount):
send_job_status_reaction(nip90_event, "payment-required", False,
await send_job_status_reaction(nip90_event, "payment-required", False,
int(amount), # bid_offer
client=self.client, dvm_config=self.dvm_config)
@ -261,7 +267,7 @@ class DVM:
print(
"[" + self.dvm_config.NIP89.NAME + "] Requesting payment for Event: " +
nip90_event.id().to_hex())
send_job_status_reaction(nip90_event, "payment-required",
await send_job_status_reaction(nip90_event, "payment-required",
False, int(amount), client=self.client, dvm_config=self.dvm_config)
@ -272,13 +278,13 @@ class DVM:
# else:
# print("[" + self.dvm_config.NIP89.NAME + "] Task " + task + " not supported on this DVM, skipping..")
def handle_zap(zap_event):
async def handle_zap(zap_event):
try:
invoice_amount, zapped_event, sender, message, anon = parse_zap_event_tags(zap_event,
self.keys,
self.dvm_config.NIP89.NAME,
self.client, self.dvm_config)
user = get_or_add_user(db=self.dvm_config.DB, npub=sender, client=self.client, config=self.dvm_config)
user = await get_or_add_user(db=self.dvm_config.DB, npub=sender, client=self.client, config=self.dvm_config)
if zapped_event is not None:
if zapped_event.kind() == EventDefinitions.KIND_FEEDBACK:
@ -305,7 +311,7 @@ class DVM:
# if a reaction by us got zapped
print(status)
if job_event.kind() == EventDefinitions.KIND_NIP88_SUBSCRIBE_EVENT:
send_job_status_reaction(job_event, "subscription-success", client=self.client,
await send_job_status_reaction(job_event, "subscription-success", client=self.client,
dvm_config=self.dvm_config, user=user)
@ -318,7 +324,7 @@ class DVM:
user.name))
if amount <= invoice_amount:
print("[" + self.dvm_config.NIP89.NAME + "] Payment-request fulfilled...")
send_job_status_reaction(job_event, "processing", client=self.client,
await send_job_status_reaction(job_event, "processing", client=self.client,
content=self.dvm_config.CUSTOM_PROCESSING_MESSAGE,
dvm_config=self.dvm_config, user=user)
indices = [i for i, x in enumerate(self.job_list) if
@ -329,18 +335,18 @@ class DVM:
if index > -1:
if self.job_list[index].is_processed:
self.job_list[index].is_paid = True
check_and_return_event(self.job_list[index].result, job_event)
await check_and_return_event(self.job_list[index].result, job_event)
elif not (self.job_list[index]).is_processed:
# If payment-required appears before processing
self.job_list.pop(index)
print("Starting work...")
do_work(job_event, invoice_amount)
await do_work(job_event, invoice_amount)
else:
print("Job not in List, but starting work...")
do_work(job_event, invoice_amount)
await do_work(job_event, invoice_amount)
else:
send_job_status_reaction(job_event, "payment-rejected",
await send_job_status_reaction(job_event, "payment-rejected",
False, invoice_amount, client=self.client,
dvm_config=self.dvm_config)
print("[" + self.dvm_config.NIP89.NAME + "] Invoice was not paid sufficiently")
@ -366,7 +372,7 @@ class DVM:
except Exception as e:
print("[" + self.dvm_config.NIP89.NAME + "] Error during content decryption: " + str(e))
def check_event_has_not_unfinished_job_input(nevent, append, client, dvmconfig):
async def check_event_has_not_unfinished_job_input(nevent, append, client, dvmconfig):
task_supported, task = check_task_is_supported(nevent, client, config=dvmconfig)
if not task_supported:
return False
@ -387,14 +393,14 @@ class DVM:
if append:
job_ = RequiredJobToWatch(event=nevent, timestamp=Timestamp.now().as_secs())
self.jobs_on_hold_list.append(job_)
send_job_status_reaction(nevent, "chain-scheduled", True, 0,
await send_job_status_reaction(nevent, "chain-scheduled", True, 0,
client=client, dvm_config=dvmconfig)
return False
else:
return True
def check_and_return_event(data, original_event: Event):
async def check_and_return_event(data, original_event: Event):
amount = 0
for x in self.job_list:
if x.event == original_event:
@ -403,12 +409,12 @@ class DVM:
x.result = data
x.is_processed = True
if self.dvm_config.SHOW_RESULT_BEFORE_PAYMENT and not is_paid:
send_nostr_reply_event(data, original_event.as_json())
send_job_status_reaction(original_event, "success", amount,
await send_nostr_reply_event(data, original_event.as_json())
await send_job_status_reaction(original_event, "success", amount,
dvm_config=self.dvm_config,
) # or payment-required, or both?
elif not self.dvm_config.SHOW_RESULT_BEFORE_PAYMENT and not is_paid:
send_job_status_reaction(original_event, "success", amount,
await send_job_status_reaction(original_event, "success", amount,
dvm_config=self.dvm_config,
) # or payment-required, or both?
@ -416,7 +422,7 @@ class DVM:
self.job_list.remove(x)
elif not self.dvm_config.SHOW_RESULT_BEFORE_PAYMENT and is_paid:
self.job_list.remove(x)
send_nostr_reply_event(data, original_event.as_json())
await send_nostr_reply_event(data, original_event.as_json())
break
task = get_task(original_event, self.client, self.dvm_config)
@ -424,17 +430,17 @@ class DVM:
if task == dvm.TASK:
try:
post_processed = dvm.post_process(data, original_event)
send_nostr_reply_event(post_processed, original_event.as_json())
await send_nostr_reply_event(post_processed, original_event.as_json())
except Exception as e:
print(e)
# Zapping back by error in post-processing is a risk for the DVM because work has been done,
# but maybe something with parsing/uploading failed. Try to avoid errors here as good as possible
send_job_status_reaction(original_event, "error",
await send_job_status_reaction(original_event, "error",
content="Error in Post-processing: " + str(e),
dvm_config=self.dvm_config,
)
if amount > 0 and self.dvm_config.LNBITS_ADMIN_KEY != "":
user = get_or_add_user(self.dvm_config.DB, original_event.author().to_hex(),
user = await get_or_add_user(self.dvm_config.DB, original_event.author().to_hex(),
client=self.client, config=self.dvm_config)
print(user.lud16 + " " + str(amount))
bolt11 = zaprequest(user.lud16, amount, "Couldn't finish job, returning sats",
@ -448,7 +454,7 @@ class DVM:
except Exception as e:
print(e)
def send_nostr_reply_event(content, original_event_as_str):
async def send_nostr_reply_event(content, original_event_as_str):
original_event = Event.from_json(original_event_as_str)
request_tag = Tag.parse(["request", original_event_as_str])
e_tag = Tag.parse(["e", original_event.id().to_hex()])
@ -466,7 +472,6 @@ class DVM:
if relay_tag is not None:
reply_tags.append(relay_tag)
encrypted = False
for tag in original_event.tags():
if tag.as_vec()[0] == "encrypted":
@ -489,12 +494,12 @@ class DVM:
self.keys)
# send_event(reply_event, client=self.client, dvm_config=self.dvm_config)
send_event_outbox(reply_event, client=self.client, dvm_config=self.dvm_config)
await send_event_outbox(reply_event, client=self.client, dvm_config=self.dvm_config)
print(bcolors.GREEN + "[" + self.dvm_config.NIP89.NAME + "] " + str(
original_event.kind().as_u64() + 1000) + " Job Response event sent: " + reply_event.as_json() + bcolors.ENDC)
def send_job_status_reaction(original_event, status, is_paid=True, amount=0, client=None,
async def send_job_status_reaction(original_event, status, is_paid=True, amount=0, client=None,
content=None,
dvm_config=None, user=None):
@ -600,13 +605,13 @@ class DVM:
keys = Keys.parse(dvm_config.PRIVATE_KEY)
reaction_event = EventBuilder(EventDefinitions.KIND_FEEDBACK, str(content), reply_tags).to_event(keys)
# send_event(reaction_event, client=self.client, dvm_config=self.dvm_config)
send_event_outbox(reaction_event, client=self.client, dvm_config=self.dvm_config)
await send_event_outbox(reaction_event, client=self.client, dvm_config=self.dvm_config)
print(bcolors.YELLOW + "[" + self.dvm_config.NIP89.NAME + "]" + " Sent Kind " + str(
EventDefinitions.KIND_FEEDBACK.as_u64()) + " Reaction: " + status + " " + reaction_event.as_json() + bcolors.ENDC)
return reaction_event.as_json()
def do_work(job_event, amount):
async def do_work(job_event, amount):
if ((
EventDefinitions.KIND_NIP90_EXTRACT_TEXT.as_u64() <= job_event.kind().as_u64() <= EventDefinitions.KIND_NIP90_GENERIC.as_u64())
or job_event.kind().as_u64() == EventDefinitions.KIND_DM.as_u64()):
@ -643,25 +648,25 @@ class DVM:
else: # Some components might have issues with running code in otuside venv.
# We install locally in these cases for now
result = dvm.process(request_form)
result = await dvm.process(request_form)
try:
post_processed = dvm.post_process(result, job_event)
send_nostr_reply_event(post_processed, job_event.as_json())
await send_nostr_reply_event(post_processed, job_event.as_json())
except Exception as e:
print(bcolors.RED + "[" + self.dvm_config.NIP89.NAME + "] Error: " + str(
e) + bcolors.ENDC)
send_job_status_reaction(job_event, "error", content=str(e),
await send_job_status_reaction(job_event, "error", content=str(e),
dvm_config=self.dvm_config)
except Exception as e:
print(
bcolors.RED + "[" + self.dvm_config.NIP89.NAME + "] Error: " + str(e) + bcolors.ENDC)
# we could send the exception here to the user, but maybe that's not a good idea after all.
send_job_status_reaction(job_event, "error", content=result,
await send_job_status_reaction(job_event, "error", content=result,
dvm_config=self.dvm_config)
# Zapping back the user on error
if amount > 0 and self.dvm_config.LNBITS_ADMIN_KEY != "":
user = get_or_add_user(self.dvm_config.DB, job_event.author().to_hex(),
user = await get_or_add_user(self.dvm_config.DB, job_event.author().to_hex(),
client=self.client, config=self.dvm_config)
print(user.lud16 + " " + str(amount))
bolt11 = zaprequest(user.lud16, amount, "Couldn't finish job, returning sats", job_event,
@ -677,11 +682,11 @@ class DVM:
return
self.client.handle_notifications(NotificationHandler())
await self.client.handle_notifications(NotificationHandler())
while True:
for dvm in self.dvm_config.SUPPORTED_DVMS:
scheduled_result = dvm.schedule(self.dvm_config)
scheduled_result = await dvm.schedule(self.dvm_config)
for job in self.job_list:
if job.bolt11 != "" and job.payment_hash != "" and not job.payment_hash is None and not job.is_paid:
@ -692,12 +697,12 @@ class DVM:
amount = parse_amount_from_bolt11_invoice(job.bolt11)
job.is_paid = True
send_job_status_reaction(job.event, "processing", True, 0,
await send_job_status_reaction(job.event, "processing", True, 0,
content=self.dvm_config.CUSTOM_PROCESSING_MESSAGE,
client=self.client,
dvm_config=self.dvm_config)
print("[" + self.dvm_config.NIP89.NAME + "] doing work from joblist")
do_work(job.event, amount)
await do_work(job.event, amount)
elif ispaid is None: # invoice expired
self.job_list.remove(job)
@ -705,9 +710,9 @@ class DVM:
self.job_list.remove(job)
for job in self.jobs_on_hold_list:
if check_event_has_not_unfinished_job_input(job.event, False, client=self.client,
if await check_event_has_not_unfinished_job_input(job.event, False, client=self.client,
dvmconfig=self.dvm_config):
handle_nip90_job_event(nip90_event=job.event)
await handle_nip90_job_event(nip90_event=job.event)
try:
self.jobs_on_hold_list.remove(job)
except:
@ -715,22 +720,5 @@ class DVM:
if Timestamp.now().as_secs() > job.timestamp + 60 * 20: # remove jobs to look for after 20 minutes..
self.jobs_on_hold_list.remove(job)
advanced_log = False
if advanced_log:
for url, relay in self.client.relays().items():
stats = relay.stats()
print(f"Relay: {url}")
print(f"Connected: {relay.is_connected()}")
print(f"Status: {relay.status()}")
print("Stats:")
print(f" Attempts: {stats.attempts()}")
print(f" Success: {stats.success()}")
print(f" Bytes sent: {stats.bytes_sent()}")
print(f" Bytes received: {stats.bytes_received()}")
print(f" Connected at: {stats.connected_at().to_human_datetime()}")
if stats.latency() is not None:
print(f" Latency: {stats.latency().total_seconds() * 1000} ms")
print("###########################################")
time.sleep(1.0)

20
nostr_dvm/interfaces/dvmtaskinterface.py
View File

@ -1,3 +1,4 @@
import asyncio
import json
import os
import subprocess
@ -34,11 +35,14 @@ class DVMTaskInterface:
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None,
options=None, task=None):
if options is None:
self.options = {}
else:
self.options = options
self.init(name, dvm_config, admin_config, nip88config, nip89config, task)
self.options = options
self.install_dependencies(dvm_config)
def init(self, name, dvm_config, admin_config=None, nip88config=None, nip89config=None, task=None):
def init(self, name, dvm_config, admin_config=None, nip88config=None, nip89config=None, task=None, options=None):
self.NAME = name
self.PRIVATE_KEY = dvm_config.PRIVATE_KEY
if dvm_config.PUBLIC_KEY == "" or dvm_config.PUBLIC_KEY is None:
@ -66,6 +70,14 @@ class DVMTaskInterface:
self.dvm_config = dvm_config
self.admin_config = admin_config
asyncio.run(self.init_dvm(name, dvm_config, nip89config, nip88config,
admin_config, options))
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
pass
def install_dependencies(self, dvm_config):
if dvm_config.SCRIPT != "":
if self.dvm_config.USE_OWN_VENV:
@ -94,7 +106,7 @@ class DVMTaskInterface:
nostr_dvm_thread = Thread(target=self.DVM, args=[self.dvm_config, self.admin_config])
nostr_dvm_thread.start()
def schedule(self, dvm_config):
async def schedule(self, dvm_config):
"""schedule something, e.g. define some time to update or to post, does nothing by default"""
pass
@ -115,7 +127,7 @@ class DVMTaskInterface:
"""Parse input into a request form that will be given to the process method"""
pass
def process(self, request_form):
async def process(self, request_form):
"Process the data and return the result"
pass

112
nostr_dvm/subscription.py
View File

@ -1,3 +1,4 @@
import asyncio
import json
import math
import os
@ -24,16 +25,7 @@ from nostr_dvm.utils.zap_utils import create_bolt11_lud16, zaprequest
class Subscription:
job_list: list
# This is a simple list just to keep track which events we created and manage, so we don't pay for other requests
def __init__(self, dvm_config, admin_config=None):
self.NAME = "Subscription Handler"
dvm_config.DB = "db/" + "subscriptions" + ".db"
self.dvm_config = dvm_config
nip89config = NIP89Config()
nip89config.NAME = self.NAME
self.dvm_config.NIP89 = nip89config
self.admin_config = admin_config
self.keys = Keys.parse(dvm_config.PRIVATE_KEY)
async def init_subscription(self, dvm_config, admin_config=None):
wait_for_send = False
skip_disconnected_relays = True
opts = (Options().wait_for_send(wait_for_send).send_timeout(timedelta(seconds=self.dvm_config.RELAY_TIMEOUT))
@ -49,18 +41,19 @@ class Subscription:
pk.to_hex()) + "\n")
for relay in self.dvm_config.RELAY_LIST:
self.client.add_relay(relay)
self.client.connect()
await self.client.add_relay(relay)
await self.client.connect()
zap_filter = Filter().pubkey(pk).kinds([EventDefinitions.KIND_ZAP]).since(Timestamp.now())
cancel_subscription_filter = Filter().kinds([EventDefinitions.KIND_NIP88_STOP_SUBSCRIPTION_EVENT]).since(
Timestamp.now())
authors = []
if admin_config is not None and len(admin_config.USERNPUBS) > 0:
#we might want to limit which services can connect to the subscription handler
# we might want to limit which services can connect to the subscription handler
for key in admin_config.USERNPUBS:
authors.append(PublicKey.parse(key))
dvm_filter = Filter().authors(authors).pubkey(pk).kinds([EventDefinitions.KIND_NIP90_DVM_SUBSCRIPTION]).since(
dvm_filter = Filter().authors(authors).pubkey(pk).kinds(
[EventDefinitions.KIND_NIP90_DVM_SUBSCRIPTION]).since(
Timestamp.now())
else:
# or we don't
@ -68,25 +61,41 @@ class Subscription:
[EventDefinitions.KIND_NIP90_DVM_SUBSCRIPTION]).since(
Timestamp.now())
self.client.subscribe([zap_filter, dvm_filter, cancel_subscription_filter], None)
await self.client.subscribe([zap_filter, dvm_filter, cancel_subscription_filter], None)
create_subscription_sql_table(dvm_config.DB)
# This is a simple list just to keep track which events we created and manage, so we don't pay for other requests
def __init__(self, dvm_config, admin_config=None):
self.NAME = "Subscription Handler"
dvm_config.DB = "db/" + "subscriptions" + ".db"
self.dvm_config = dvm_config
nip89config = NIP89Config()
nip89config.NAME = self.NAME
self.dvm_config.NIP89 = nip89config
self.admin_config = admin_config
self.keys = Keys.parse(dvm_config.PRIVATE_KEY)
asyncio.run(self.init_subscription(dvm_config, admin_config))
asyncio.run(self.run_subscription(dvm_config))
async def run_subscription(self, dvm_config):
class NotificationHandler(HandleNotification):
client = self.client
dvm_config = self.dvm_config
keys = self.keys
def handle(self, relay_url, subscription_id, nostr_event: Event):
async def handle(self, relay_url, subscription_id, nostr_event: Event):
if nostr_event.kind() == EventDefinitions.KIND_NIP90_DVM_SUBSCRIPTION:
handle_nwc_request(nostr_event)
await handle_nwc_request(nostr_event)
elif nostr_event.kind() == EventDefinitions.KIND_NIP88_STOP_SUBSCRIPTION_EVENT:
handle_cancel(nostr_event)
await handle_cancel(nostr_event)
def handle_msg(self, relay_url, msg):
async def handle_msg(self, relay_url, msg):
return
def handle_cancel(nostr_event):
async def handle_cancel(nostr_event):
print(nostr_event.as_json())
sender = nostr_event.author().to_hex()
kind7001eventid = ""
@ -127,7 +136,7 @@ class Subscription:
end = start + 60 * 60 * 24 * 356
return end
def send_status_success(original_event, domain):
async def send_status_success(original_event, domain):
e_tag = Tag.parse(["e", original_event.id().to_hex()])
p_tag = Tag.parse(["p", original_event.author().to_hex()])
@ -151,11 +160,11 @@ class Subscription:
keys = Keys.parse(dvm_config.PRIVATE_KEY)
reaction_event = EventBuilder(EventDefinitions.KIND_FEEDBACK, str(content), reply_tags).to_event(keys)
send_event(reaction_event, client=self.client, dvm_config=self.dvm_config)
await send_event(reaction_event, client=self.client, dvm_config=self.dvm_config)
print("[" + self.dvm_config.NIP89.NAME + "]" + ": Sent Kind " + str(
EventDefinitions.KIND_FEEDBACK.as_u64()) + " Reaction: " + "success" + " " + reaction_event.as_json())
def pay_zap_split(nwc, overall_amount, zaps, tier, unit="msats"):
async def pay_zap_split(nwc, overall_amount, zaps, tier, unit="msats"):
overallsplit = 0
for zap in zaps:
@ -166,10 +175,10 @@ class Subscription:
zapped_amount = 0
for zap in zaps:
if zap[1] == "":
#If the client did decide to not add itself to the zap split, or if a slot is left we add the subscription service in the empty space
# If the client did decide to not add itself to the zap split, or if a slot is left we add the subscription service in the empty space
zap[1] = Keys.parse(self.dvm_config.PRIVATE_KEY).public_key().to_hex()
name, nip05, lud16 = fetch_user_metadata(zap[1], self.client)
name, nip05, lud16 = await fetch_user_metadata(zap[1], self.client)
splitted_amount = math.floor(
(int(zap[3]) / overallsplit) * int(overall_amount) / 1000)
# invoice = create_bolt11_lud16(lud16, splitted_amount)
@ -187,7 +196,7 @@ class Subscription:
print(str(zapped_amount) + "/" + str(overall_amount))
if zapped_amount < overall_amount * 0.8: # TODO how do we handle failed zaps for some addresses? we are ok with 80% for now
#if zapped_amount < int(zaps[0][3]):
# if zapped_amount < int(zaps[0][3]):
success = False
else:
@ -196,7 +205,7 @@ class Subscription:
return success
def make_subscription_zap_recipe(event7001, recipient, subscriber, start, end, tier_dtag):
async def make_subscription_zap_recipe(event7001, recipient, subscriber, start, end, tier_dtag):
message = "payed by subscription service"
pTag = Tag.parse(["p", recipient])
PTag = Tag.parse(["P", subscriber])
@ -214,13 +223,14 @@ class Subscription:
signer = NostrSigner.keys(self.keys)
client = Client(signer)
for relay in dvmconfig.RELAY_LIST:
client.add_relay(relay)
client.connect()
recipeid = client.send_event(event)
await client.add_relay(relay)
await client.connect()
recipeid = await client.send_event(event)
await client.disconnect()
recipe = recipeid.to_hex()
return recipe
def handle_nwc_request(nostr_event):
async def handle_nwc_request(nostr_event):
print(nostr_event.as_json())
sender = nostr_event.author().to_hex()
if sender == self.keys.public_key().to_hex():
@ -228,6 +238,8 @@ class Subscription:
try:
decrypted_text = nip04_decrypt(self.keys.secret_key(), nostr_event.author(), nostr_event.content())
subscriber = ""
nwc = ""
try:
jsonevent = json.loads(decrypted_text)
for entry in jsonevent:
@ -238,7 +250,7 @@ class Subscription:
subscriptionfilter = Filter().kind(EventDefinitions.KIND_NIP88_SUBSCRIBE_EVENT).author(
PublicKey.parse(subscriber)).limit(1)
evts = self.client.get_events_of([subscriptionfilter], timedelta(seconds=3))
evts = await self.client.get_events_of([subscriptionfilter], timedelta(seconds=3))
print(evts)
if len(evts) > 0:
event7001id = evts[0].id().to_hex()
@ -250,6 +262,7 @@ class Subscription:
zaps = []
tier = "DVM"
overall_amount = 0
subscription_event_id = ""
for tag in evts[0].tags():
if tag.as_vec()[0] == "amount":
overall_amount = int(tag.as_vec()[1])
@ -315,7 +328,7 @@ class Subscription:
print("updated subscription entry before payment")
# we attempt to pay the subscription
success = pay_zap_split(nwc, overall_amount, zaps, tier, unit)
success = await pay_zap_split(nwc, overall_amount, zaps, tier, unit)
else:
start = Timestamp.now().as_secs()
@ -326,8 +339,8 @@ class Subscription:
if success:
# we create a payment recipe
recipe = make_subscription_zap_recipe(event7001id, recipient, subscriber, start, end,
tier_dtag)
recipe = await make_subscription_zap_recipe(event7001id, recipient, subscriber, start, end,
tier_dtag)
print("RECIPE " + recipe)
isactivesubscription = True
@ -338,14 +351,15 @@ class Subscription:
Timestamp.now().as_secs(), tier)
print("updated subscription entry after payment")
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(
Timestamp.from_secs(end).to_human_datetime().replace("Z", " ").replace("T", " ") + " GMT"))
Timestamp.from_secs(end).to_human_datetime().replace("Z", " ").replace("T",
" ") + " GMT"))
evt = EventBuilder.encrypted_direct_msg(keys, PublicKey.parse(subscriber), message,
None).to_event(keys)
send_event(evt, client=self.client, dvm_config=dvm_config)
await send_event(evt, client=self.client, dvm_config=dvm_config)
@ -370,15 +384,15 @@ class Subscription:
delete_from_subscription_sql_table(dvm_config.DB, subscription.id)
print("Delete expired subscription")
def handle_subscription_renewal(subscription):
async def handle_subscription_renewal(subscription):
zaps = json.loads(subscription.zaps)
success = pay_zap_split(subscription.nwc, subscription.amount, zaps, subscription.tier,
subscription.unit)
success = await pay_zap_split(subscription.nwc, subscription.amount, zaps, subscription.tier,
subscription.unit)
if success:
end = infer_subscription_end_time(Timestamp.now().as_secs(), subscription.cadence)
recipe = make_subscription_zap_recipe(subscription.id, subscription.recipent,
subscription.subscriber, subscription.begin,
end, subscription.tier_dtag)
recipe = await make_subscription_zap_recipe(subscription.id, subscription.recipent,
subscription.subscriber, subscription.begin,
end, subscription.tier_dtag)
else:
end = Timestamp.now().as_secs()
recipe = subscription.recipe
@ -403,9 +417,9 @@ class Subscription:
evt = EventBuilder.encrypted_direct_msg(keys, PublicKey.parse(subscription.subscriber),
message,
None).to_event(keys)
send_event(evt, client=self.client, dvm_config=dvm_config)
await send_event(evt, client=self.client, dvm_config=dvm_config)
def check_subscriptions():
async def check_subscriptions():
try:
subscriptions = get_all_subscriptions_from_sql_table(dvm_config.DB)
@ -430,7 +444,7 @@ class Subscription:
False,
Timestamp.now().as_secs(), subscription.tier)
else:
handle_subscription_renewal(subscription)
await handle_subscription_renewal(subscription)
else:
handle_expired_subscription(subscription)
@ -440,12 +454,12 @@ class Subscription:
except Exception as e:
print(e)
self.client.handle_notifications(NotificationHandler())
await self.client.handle_notifications(NotificationHandler())
try:
while True:
time.sleep(60.0)
check_subscriptions()
await check_subscriptions()
except KeyboardInterrupt:
print('Stay weird!')
os.kill(os.getpid(), signal.SIGTERM)

46
nostr_dvm/tasks/advanced_search.py
View File

@ -25,11 +25,9 @@ class AdvancedSearch(DVMTaskInterface):
FIX_COST: float = 0
dvm_config: DVMConfig
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -54,7 +52,7 @@ class AdvancedSearch(DVMTaskInterface):
search = ""
max_results = 100
relay = "wss://relay.nostr.band"
for tag in event.tags():
if tag.as_vec()[0] == 'i':
input_type = tag.as_vec()[2]
@ -63,7 +61,7 @@ class AdvancedSearch(DVMTaskInterface):
elif tag.as_vec()[0] == 'param':
param = tag.as_vec()[1]
if param == "user": # check for param type
#user = tag.as_vec()[2]
# user = tag.as_vec()[2]
users.append(Tag.parse(["p", tag.as_vec()[2]]))
elif param == "users": # check for param type
users = json.loads(tag.as_vec()[2])
@ -85,7 +83,7 @@ class AdvancedSearch(DVMTaskInterface):
request_form['options'] = json.dumps(options)
return request_form
def process(self, request_form):
async def process(self, request_form):
from nostr_sdk import Filter
options = self.set_options(request_form)
@ -96,24 +94,24 @@ class AdvancedSearch(DVMTaskInterface):
cli = Client.with_opts(signer, opts)
ropts = RelayOptions().ping(False)
cli.add_relay_with_opts(options["relay"], ropts)
await cli.add_relay_with_opts(options["relay"], ropts)
cli.connect()
await cli.connect()
#earch_since_seconds = int(options["since"]) * 24 * 60 * 60
#dif = Timestamp.now().as_secs() - search_since_seconds
#search_since = Timestamp.from_secs(dif)
# earch_since_seconds = int(options["since"]) * 24 * 60 * 60
# dif = Timestamp.now().as_secs() - search_since_seconds
# search_since = Timestamp.from_secs(dif)
search_since = Timestamp.from_secs(int(options["since"]))
#search_until_seconds = int(options["until"]) * 24 * 60 * 60
#dif = Timestamp.now().as_secs() - search_until_seconds
#search_until = Timestamp.from_secs(dif)
# search_until_seconds = int(options["until"]) * 24 * 60 * 60
# dif = Timestamp.now().as_secs() - search_until_seconds
# search_until = Timestamp.from_secs(dif)
search_until = Timestamp.from_secs(int(options["until"]))
userkeys = []
for user in options["users"]:
tag = Tag.parse(user)
user = tag.as_vec()[1]
#user = user[1]
# user = user[1]
user = str(user).lstrip("@")
if str(user).startswith('npub'):
userkey = PublicKey.from_bech32(user)
@ -125,25 +123,25 @@ class AdvancedSearch(DVMTaskInterface):
userkeys.append(userkey)
if not options["users"]:
notes_filter = Filter().kind(Kind(1)).search(options["search"]).since(search_since).until(search_until).limit(options["max_results"])
notes_filter = Filter().kind(Kind(1)).search(options["search"]).since(search_since).until(
search_until).limit(options["max_results"])
elif options["search"] == "":
notes_filter = Filter().kind(Kind(1)).authors(userkeys).since(search_since).until(
search_until).limit(options["max_results"])
notes_filter = Filter().kind(Kind(1)).authors(userkeys).since(search_since).until(
search_until).limit(options["max_results"])
else:
notes_filter = Filter().kind(Kind(1)).authors(userkeys).search(options["search"]).since(
search_since).until(search_until).limit(options["max_results"])
notes_filter = Filter().kind(Kind(1)).authors(userkeys).search(options["search"]).since(
search_since).until(search_until).limit(options["max_results"])
events = cli.get_events_of([notes_filter], timedelta(seconds=5))
events = await cli.get_events_of([notes_filter], timedelta(seconds=5))
result_list = []
if len(events) > 0:
for event in events:
e_tag = Tag.parse(["e", event.id().to_hex()])
#print(e_tag.as_vec())
result_list.append(e_tag.as_vec())
await cli.shutdown()
return json.dumps(result_list)
def post_process(self, result, event):

22
nostr_dvm/tasks/advanced_search_wine.py
View File

@ -27,11 +27,9 @@ class AdvancedSearchWine(DVMTaskInterface):
FIX_COST: float = 0
dvm_config: DVMConfig
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -56,7 +54,6 @@ class AdvancedSearchWine(DVMTaskInterface):
search = ""
max_results = 100
for tag in event.tags():
if tag.as_vec()[0] == 'i':
input_type = tag.as_vec()[2]
@ -87,7 +84,7 @@ class AdvancedSearchWine(DVMTaskInterface):
request_form['options'] = json.dumps(options)
return request_form
def process(self, request_form):
async def process(self, request_form):
from nostr_sdk import Filter
options = self.set_options(request_form)
userkeys = []
@ -106,13 +103,17 @@ class AdvancedSearchWine(DVMTaskInterface):
print("Sending job to nostr.wine API")
if options["users"]:
url = ('https://api.nostr.wine/search?query=' + options["search"] + '&kind=1' + '&pubkey=' + options["users"][0][1] + "&limit=100" + "&sort=time" + "&until=" + str(options["until"]) + "&since=" + str(options["since"]))
url = ('https://api.nostr.wine/search?query=' + options["search"] + '&kind=1' + '&pubkey=' +
options["users"][0][1] + "&limit=100" + "&sort=time" + "&until=" + str(
options["until"]) + "&since=" + str(options["since"]))
else:
url = ('https://api.nostr.wine/search?query=' + options["search"] + '&kind=1' + "&limit=100" + "&sort=time" + "&until=" + str(options["until"]) + "&since=" + str(options["since"]))
url = ('https://api.nostr.wine/search?query=' + options[
"search"] + '&kind=1' + "&limit=100" + "&sort=time" + "&until=" + str(
options["until"]) + "&since=" + str(options["since"]))
headers = {'Content-type': 'application/x-www-form-urlencoded'}
response = requests.get(url, headers=headers)
#print(response.text)
# print(response.text)
result_list = []
try:
ob = json.loads(response.text)
@ -126,7 +127,6 @@ class AdvancedSearchWine(DVMTaskInterface):
except Exception as e:
print(e)
return json.dumps(result_list)
def post_process(self, result, event):
@ -183,7 +183,7 @@ def build_example(name, identifier, admin_config):
nip89config.CONTENT = json.dumps(nip89info)
return AdvancedSearchWine(name=name, dvm_config=dvm_config, nip89config=nip89config,
admin_config=admin_config)
admin_config=admin_config)
if __name__ == '__main__':

8
nostr_dvm/tasks/audiogeneration_suno_ai.py
View File

@ -30,11 +30,9 @@ class AudioGenerationSonoAI(DVMTaskInterface):
TASK: str = "prompt-to-music"
FIX_COST: float = 120
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
self.base_url = 'http://localhost:3000'
def is_input_supported(self, tags, client=None, dvm_config=None):
@ -100,7 +98,7 @@ class AudioGenerationSonoAI(DVMTaskInterface):
response = requests.post(url, json=payload)
return response.json()
def process(self, request_form):
async def process(self, request_form):
try:
options = self.set_options(request_form)
has_quota = False

70
nostr_dvm/tasks/content_discovery_currently_popular.py
View File

@ -1,3 +1,4 @@
import asyncio
import json
import os
from datetime import timedelta
@ -27,6 +28,7 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface):
TASK: str = "discover-content"
FIX_COST: float = 0
dvm_config: DVMConfig
request_form = None
last_schedule: int
db_since = 3600
db_name = "db/nostr_recent_notes.db"
@ -34,12 +36,10 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface):
personalized = False
result = ""
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
self.request_form = {"jobID": "generic"}
opts = {
"max_results": 200,
@ -53,16 +53,14 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface):
if self.options.get("db_since"):
self.db_since = int(self.options.get("db_since"))
use_logger = False
if use_logger:
init_logger(LogLevel.DEBUG)
if self.dvm_config.UPDATE_DATABASE:
self.sync_db()
await self.sync_db()
if not self.personalized:
self.result = self.calculate_result(self.request_form)
self.result = await self.calculate_result(self.request_form)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -96,7 +94,7 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface):
request_form['options'] = json.dumps(options)
return request_form
def process(self, request_form):
async def process(self, request_form):
# if the dvm supports individual results, recalculate it every time for the request
if self.personalized:
return self.calculate_result(request_form)
@ -104,7 +102,7 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface):
else:
return self.result
def calculate_result(self, request_form):
async def calculate_result(self, request_form):
from nostr_sdk import Filter
from types import SimpleNamespace
ns = SimpleNamespace()
@ -116,9 +114,9 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface):
keys = Keys.parse(sk.to_hex())
signer = NostrSigner.keys(keys)
database = NostrDatabase.sqlite(self.db_name)
database = await NostrDatabase.sqlite(self.db_name)
cli = ClientBuilder().database(database).signer(signer).opts(opts).build()
cli.connect()
await cli.connect()
# Negentropy reconciliation
# Query events from database
@ -126,7 +124,7 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface):
since = Timestamp.from_secs(timestamp_hour_ago)
filter1 = Filter().kind(definitions.EventDefinitions.KIND_NOTE).since(since)
events = cli.database().query([filter1])
events = await cli.database().query([filter1])
print("[" + self.dvm_config.NIP89.NAME + "] Considering " + str(len(events)) + " Events")
ns.finallist = {}
for event in events:
@ -134,11 +132,11 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface):
filt = Filter().kinds([definitions.EventDefinitions.KIND_ZAP, definitions.EventDefinitions.KIND_REPOST,
definitions.EventDefinitions.KIND_REACTION,
definitions.EventDefinitions.KIND_NOTE]).event(event.id()).since(since)
reactions = cli.database().query([filt])
reactions = await cli.database().query([filt])
if len(reactions) >= self.min_reactions:
ns.finallist[event.id().to_hex()] = len(reactions)
if len(ns.finallist) == 0:
cli.shutdown()
await cli.shutdown()
return self.result
result_list = []
@ -147,7 +145,7 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface):
# print(EventId.parse(entry[0]).to_bech32() + "/" + EventId.parse(entry[0]).to_hex() + ": " + str(entry[1]))
e_tag = Tag.parse(["e", entry[0]])
result_list.append(e_tag.as_vec())
cli.shutdown()
await cli.shutdown()
print("[" + self.dvm_config.NIP89.NAME + "] Filtered " + str(
len(result_list)) + " fitting events.")
return json.dumps(result_list)
@ -163,29 +161,29 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface):
# if not text/plain, don't post-process
return result
def schedule(self, dvm_config):
async def schedule(self, dvm_config):
if dvm_config.SCHEDULE_UPDATES_SECONDS == 0:
return 0
else:
if Timestamp.now().as_secs() >= self.last_schedule + dvm_config.SCHEDULE_UPDATES_SECONDS:
if self.dvm_config.UPDATE_DATABASE:
self.sync_db()
await self.sync_db()
self.last_schedule = Timestamp.now().as_secs()
self.result = self.calculate_result(self.request_form)
self.result = await self.calculate_result(self.request_form)
return 1
def sync_db(self):
async def sync_db(self):
opts = (Options().wait_for_send(False).send_timeout(timedelta(seconds=self.dvm_config.RELAY_LONG_TIMEOUT)))
sk = SecretKey.from_hex(self.dvm_config.PRIVATE_KEY)
keys = Keys.parse(sk.to_hex())
signer = NostrSigner.keys(keys)
database = NostrDatabase.sqlite(self.db_name)
database = await NostrDatabase.sqlite(self.db_name)
cli = ClientBuilder().signer(signer).database(database).opts(opts).build()
for relay in self.dvm_config.RECONCILE_DB_RELAY_LIST:
cli.add_relay(relay)
await cli.add_relay(relay)
cli.connect()
await cli.connect()
timestamp_since = Timestamp.now().as_secs() - self.db_since
since = Timestamp.from_secs(timestamp_since)
@ -197,10 +195,10 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface):
print("[" + self.dvm_config.NIP89.NAME + "] Syncing notes of the last " + str(
self.db_since) + " seconds.. this might take a while..")
dbopts = NegentropyOptions().direction(NegentropyDirection.DOWN)
cli.reconcile(filter1, dbopts)
cli.database().delete(Filter().until(Timestamp.from_secs(
await cli.reconcile(filter1, dbopts)
await cli.database().delete(Filter().until(Timestamp.from_secs(
Timestamp.now().as_secs() - self.db_since))) # Clear old events so db doesn't get too full.
cli.shutdown()
await cli.shutdown()
print(
"[" + self.dvm_config.NIP89.NAME + "] Done Syncing Notes of the last " + str(self.db_since) + " seconds..")
@ -208,7 +206,8 @@ class DicoverContentCurrentlyPopular(DVMTaskInterface):
# We build an example here that we can call by either calling this file directly from the main directory,
# or by adding it to our playground. You can call the example and adjust it to your needs or redefine it in the
# playground or elsewhere
def build_example(name, identifier, admin_config, options, cost=0, update_rate=180, processing_msg=None, update_db=True):
def build_example(name, identifier, admin_config, options, cost=0, update_rate=180, processing_msg=None,
update_db=True):
dvm_config = build_default_config(identifier)
dvm_config.USE_OWN_VENV = False
dvm_config.SHOWLOG = True
@ -247,14 +246,15 @@ def build_example(name, identifier, admin_config, options, cost=0, update_rate=
nip89config.DTAG = check_and_set_d_tag(identifier, name, dvm_config.PRIVATE_KEY, nip89info["image"])
nip89config.CONTENT = json.dumps(nip89info)
#admin_config.UPDATE_PROFILE = False
#admin_config.REBROADCAST_NIP89 = False
# admin_config.UPDATE_PROFILE = False
# admin_config.REBROADCAST_NIP89 = False
return DicoverContentCurrentlyPopular(name=name, dvm_config=dvm_config, nip89config=nip89config,
admin_config=admin_config, options=options)
def build_example_subscription(name, identifier, admin_config, options, update_rate=180, processing_msg=None, update_db=True):
def build_example_subscription(name, identifier, admin_config, options, update_rate=180, processing_msg=None,
update_db=True):
dvm_config = build_default_config(identifier)
dvm_config.USE_OWN_VENV = False
dvm_config.SHOWLOG = True
@ -304,9 +304,9 @@ def build_example_subscription(name, identifier, admin_config, options, update_r
nip88config.PERK2DESC = "Support NostrDVM & NostrSDK development"
nip88config.PAYMENT_VERIFIER_PUBKEY = "5b5c045ecdf66fb540bdf2049fe0ef7f1a566fa427a4fe50d400a011b65a3a7e"
#admin_config.UPDATE_PROFILE = False
#admin_config.REBROADCAST_NIP89 = False
#admin_config.REBROADCAST_NIP88 = False
# admin_config.UPDATE_PROFILE = False
# admin_config.REBROADCAST_NIP89 = False
# admin_config.REBROADCAST_NIP88 = False
# admin_config.FETCH_NIP88 = True
# admin_config.EVENTID = ""
@ -318,4 +318,4 @@ def build_example_subscription(name, identifier, admin_config, options, update_r
if __name__ == '__main__':
process_venv(DicoverContentCurrentlyPopular)
process_venv(DicoverContentCurrentlyPopular)

53
nostr_dvm/tasks/content_discovery_currently_popular_by_top_zaps.py
View File

@ -1,3 +1,4 @@
import asyncio
import json
import os
from datetime import timedelta
@ -28,6 +29,7 @@ class DicoverContentCurrentlyPopularZaps(DVMTaskInterface):
TASK: str = "discover-content"
FIX_COST: float = 0
dvm_config: DVMConfig
request_form = None
last_schedule: int
db_since = 3600
db_name = "db/nostr_recent_notes.db"
@ -36,11 +38,8 @@ class DicoverContentCurrentlyPopularZaps(DVMTaskInterface):
result = ""
logger = False
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
self.request_form = {"jobID": "generic"}
opts = {
@ -60,12 +59,10 @@ class DicoverContentCurrentlyPopularZaps(DVMTaskInterface):
if self.logger:
init_logger(LogLevel.DEBUG)
if self.dvm_config.UPDATE_DATABASE:
self.sync_db()
await self.sync_db()
if not self.personalized:
self.result = self.calculate_result(self.request_form)
self.result = await self.calculate_result(self.request_form)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -98,7 +95,7 @@ class DicoverContentCurrentlyPopularZaps(DVMTaskInterface):
request_form['options'] = json.dumps(options)
return request_form
def process(self, request_form):
async def process(self, request_form):
# if the dvm supports individual results, recalculate it every time for the request
if self.personalized:
return self.calculate_result(request_form)
@ -106,7 +103,7 @@ class DicoverContentCurrentlyPopularZaps(DVMTaskInterface):
else:
return self.result
def calculate_result(self, request_form):
async def calculate_result(self, request_form):
from nostr_sdk import Filter
from types import SimpleNamespace
ns = SimpleNamespace()
@ -118,10 +115,10 @@ class DicoverContentCurrentlyPopularZaps(DVMTaskInterface):
keys = Keys.parse(sk.to_hex())
signer = NostrSigner.keys(keys)
database = NostrDatabase.sqlite(self.db_name)
database = await NostrDatabase.sqlite(self.db_name)
cli = ClientBuilder().database(database).signer(signer).opts(opts).build()
cli.connect()
await cli.connect()
# Negentropy reconciliation
# Query events from database
@ -129,14 +126,14 @@ class DicoverContentCurrentlyPopularZaps(DVMTaskInterface):
since = Timestamp.from_secs(timestamp_hour_ago)
filter1 = Filter().kind(definitions.EventDefinitions.KIND_NOTE).since(since)
events = cli.database().query([filter1])
events = await cli.database().query([filter1])
print("[" + self.dvm_config.NIP89.NAME + "] Considering " + str(len(events)) + " Events")
ns.finallist = {}
for event in events:
if event.created_at().as_secs() > timestamp_hour_ago:
filt = Filter().kinds([definitions.EventDefinitions.KIND_ZAP]).event(event.id()).since(since)
reactions = cli.database().query([filt])
reactions = await cli.database().query([filt])
invoice_amount = 0
haspreimage = False
if len(reactions) >= self.min_reactions:
@ -162,11 +159,11 @@ class DicoverContentCurrentlyPopularZaps(DVMTaskInterface):
e_tag = Tag.parse(["e", entry[0]])
result_list.append(e_tag.as_vec())
print("[" + self.dvm_config.NIP89.NAME+ "] Filtered " + str(
print("[" + self.dvm_config.NIP89.NAME + "] Filtered " + str(
len(result_list)) + " fitting events.")
cli.disconnect()
cli.shutdown()
await cli.disconnect()
await cli.shutdown()
return json.dumps(result_list)
@ -181,33 +178,33 @@ class DicoverContentCurrentlyPopularZaps(DVMTaskInterface):
# if not text/plain, don't post-process
return result
def schedule(self, dvm_config):
async def schedule(self, dvm_config):
if dvm_config.SCHEDULE_UPDATES_SECONDS == 0:
return 0
else:
if Timestamp.now().as_secs() >= self.last_schedule + dvm_config.SCHEDULE_UPDATES_SECONDS:
try:
if self.dvm_config.UPDATE_DATABASE:
self.sync_db()
await self.sync_db()
self.last_schedule = Timestamp.now().as_secs()
if not self.personalized:
self.result = self.calculate_result(self.request_form)
self.result = await self.calculate_result(self.request_form)
except Exception as e:
print(e)
return 1
def sync_db(self):
async def sync_db(self):
opts = (Options().wait_for_send(False).send_timeout(timedelta(seconds=self.dvm_config.RELAY_LONG_TIMEOUT)))
sk = SecretKey.from_hex(self.dvm_config.PRIVATE_KEY)
keys = Keys.parse(sk.to_hex())
signer = NostrSigner.keys(keys)
database = NostrDatabase.sqlite(self.db_name)
database = await NostrDatabase.sqlite(self.db_name)
cli = ClientBuilder().signer(signer).database(database).opts(opts).build()
for relay in self.dvm_config.RECONCILE_DB_RELAY_LIST:
cli.add_relay(relay)
await cli.add_relay(relay)
cli.connect()
await cli.connect()
timestamp_since = Timestamp.now().as_secs() - self.db_since
since = Timestamp.from_secs(timestamp_since)
@ -219,10 +216,10 @@ class DicoverContentCurrentlyPopularZaps(DVMTaskInterface):
print("[" + self.dvm_config.NIP89.NAME + "] Syncing notes of the last " + str(
self.db_since) + " seconds.. this might take a while..")
dbopts = NegentropyOptions().direction(NegentropyDirection.DOWN)
cli.reconcile(filter1, dbopts)
cli.database().delete(Filter().until(Timestamp.from_secs(
await cli.reconcile(filter1, dbopts)
await cli.database().delete(Filter().until(Timestamp.from_secs(
Timestamp.now().as_secs() - self.db_since))) # Clear old events so db doesn't get too full.
cli.shutdown()
await cli.shutdown()
print(
"[" + self.dvm_config.NIP89.NAME + "] Done Syncing Notes of the last " + str(self.db_since) + " seconds..")

50
nostr_dvm/tasks/content_discovery_currently_popular_followers.py
View File

@ -1,3 +1,4 @@
import asyncio
import json
import os
from datetime import timedelta
@ -32,11 +33,8 @@ class DicoverContentCurrentlyPopularFollowers(DVMTaskInterface):
db_name = "db/nostr_recent_notes2.db"
min_reactions = 2
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
self.last_schedule = Timestamp.now().as_secs()
@ -50,7 +48,7 @@ class DicoverContentCurrentlyPopularFollowers(DVMTaskInterface):
init_logger(LogLevel.DEBUG)
if self.dvm_config.UPDATE_DATABASE:
self.sync_db()
await self.sync_db()
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -90,7 +88,7 @@ class DicoverContentCurrentlyPopularFollowers(DVMTaskInterface):
request_form['options'] = json.dumps(options)
return request_form
def process(self, request_form):
async def process(self, request_form):
from nostr_sdk import Filter
from types import SimpleNamespace
ns = SimpleNamespace()
@ -102,20 +100,20 @@ class DicoverContentCurrentlyPopularFollowers(DVMTaskInterface):
keys = Keys.parse(sk.to_hex())
signer = NostrSigner.keys(keys)
database = NostrDatabase.sqlite(self.db_name)
database = await NostrDatabase.sqlite(self.db_name)
cli = ClientBuilder().database(database).signer(signer).opts(opts).build()
cli.add_relay("wss://relay.damus.io")
cli.add_relay("wss://nostr.oxtr.dev")
cli.add_relay("wss://nostr.mom")
await cli.add_relay("wss://relay.damus.io")
await cli.add_relay("wss://nostr.oxtr.dev")
await cli.add_relay("wss://nostr.mom")
#ropts = RelayOptions().ping(False)
#cli.add_relay_with_opts("wss://nostr.band", ropts)
cli.connect()
await cli.connect()
user = PublicKey.parse(options["user"])
followers_filter = Filter().author(user).kinds([Kind(3)])
followers = cli.get_events_of([followers_filter], timedelta(seconds=self.dvm_config.RELAY_TIMEOUT))
followers = await cli.get_events_of([followers_filter], timedelta(seconds=self.dvm_config.RELAY_TIMEOUT))
#print(followers)
# Negentropy reconciliation
@ -142,7 +140,7 @@ class DicoverContentCurrentlyPopularFollowers(DVMTaskInterface):
followings.append(following)
filter1 = Filter().kind(definitions.EventDefinitions.KIND_NOTE).authors(followings).since(since)
events = cli.database().query([filter1])
events = await cli.database().query([filter1])
print("[" + self.dvm_config.NIP89.NAME + "] Considering " + str(len(events)) + " Events")
ns.finallist = {}
@ -151,7 +149,7 @@ class DicoverContentCurrentlyPopularFollowers(DVMTaskInterface):
filt = Filter().kinds(
[definitions.EventDefinitions.KIND_ZAP, definitions.EventDefinitions.KIND_REACTION, definitions.EventDefinitions.KIND_REPOST,
definitions.EventDefinitions.KIND_NOTE]).event(event.id()).since(since)
reactions = cli.database().query([filt])
reactions = await cli.database().query([filt])
if len(reactions) >= self.min_reactions:
ns.finallist[event.id().to_hex()] = len(reactions)
@ -162,8 +160,8 @@ class DicoverContentCurrentlyPopularFollowers(DVMTaskInterface):
# print(EventId.parse(entry[0]).to_bech32() + "/" + EventId.parse(entry[0]).to_hex() + ": " + str(entry[1]))
e_tag = Tag.parse(["e", entry[0]])
result_list.append(e_tag.as_vec())
cli.connect()
cli.shutdown()
await cli.connect()
await cli.shutdown()
print("[" + self.dvm_config.NIP89.NAME + "] Filtered " + str(
len(result_list)) + " fitting events.")
@ -180,7 +178,7 @@ class DicoverContentCurrentlyPopularFollowers(DVMTaskInterface):
# if not text/plain, don't post-process
return result
def schedule(self, dvm_config):
async def schedule(self, dvm_config):
if dvm_config.SCHEDULE_UPDATES_SECONDS == 0:
return 0
# We simply use the db from the other dvm that contains all notes
@ -188,23 +186,23 @@ class DicoverContentCurrentlyPopularFollowers(DVMTaskInterface):
else:
if Timestamp.now().as_secs() >= self.last_schedule + dvm_config.SCHEDULE_UPDATES_SECONDS:
if self.dvm_config.UPDATE_DATABASE:
self.sync_db()
await self.sync_db()
self.last_schedule = Timestamp.now().as_secs()
return 1
def sync_db(self):
async def sync_db(self):
opts = (Options().wait_for_send(False).send_timeout(timedelta(seconds=self.dvm_config.RELAY_LONG_TIMEOUT)))
sk = SecretKey.from_hex(self.dvm_config.PRIVATE_KEY)
keys = Keys.parse(sk.to_hex())
signer = NostrSigner.keys(keys)
database = NostrDatabase.sqlite(self.db_name)
database = await NostrDatabase.sqlite(self.db_name)
cli = ClientBuilder().signer(signer).database(database).opts(opts).build()
for relay in self.dvm_config.RECONCILE_DB_RELAY_LIST:
cli.add_relay(relay)
await cli.add_relay(relay)
cli.connect()
await cli.connect()
timestamp_since = Timestamp.now().as_secs() - self.db_since
since = Timestamp.from_secs(timestamp_since)
@ -217,10 +215,10 @@ class DicoverContentCurrentlyPopularFollowers(DVMTaskInterface):
print("[" + self.dvm_config.NIP89.NAME + "] Syncing notes of the last " + str(
self.db_since) + " seconds.. this might take a while..")
dbopts = NegentropyOptions().direction(NegentropyDirection.DOWN)
cli.reconcile(filter1, dbopts)
cli.database().delete(Filter().until(Timestamp.from_secs(
await cli.reconcile(filter1, dbopts)
await cli.database().delete(Filter().until(Timestamp.from_secs(
Timestamp.now().as_secs() - self.db_since))) # Clear old events so db doesn't get too full.
cli.shutdown()
await cli.shutdown()
print("[" + self.dvm_config.NIP89.NAME + "] Done Syncing Notes of the last " + str(
self.db_since) + " seconds..")

99
nostr_dvm/tasks/content_discovery_currently_popular_topic.py
View File

@ -1,3 +1,4 @@
import asyncio
import json
import os
from datetime import timedelta
@ -26,9 +27,10 @@ class DicoverContentCurrentlyPopularbyTopic(DVMTaskInterface):
TASK: str = "discover-content"
FIX_COST: float = 0
dvm_config: DVMConfig
request_form = None
last_schedule: int
min_reactions = 2
db_since = 10*3600
db_since = 10 * 3600
db_name = "db/nostr_default_recent_notes.db"
search_list = []
avoid_list = []
@ -36,15 +38,9 @@ class DicoverContentCurrentlyPopularbyTopic(DVMTaskInterface):
personalized = False
result = ""
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
# Generate Generic request form for dvms that provide generic results (e.g only a calculation per update,
# not per call)
self.request_form = {"jobID": "generic"}
opts = {
"max_results": 200,
@ -58,7 +54,7 @@ class DicoverContentCurrentlyPopularbyTopic(DVMTaskInterface):
self.last_schedule = Timestamp.now().as_secs()
if self.options.get("search_list"):
self.search_list = self.options.get("search_list")
#print(self.search_list)
# print(self.search_list)
if self.options.get("avoid_list"):
self.avoid_list = self.options.get("avoid_list")
if self.options.get("must_list"):
@ -68,15 +64,14 @@ class DicoverContentCurrentlyPopularbyTopic(DVMTaskInterface):
if self.options.get("db_since"):
self.db_since = int(self.options.get("db_since"))
use_logger = False
if use_logger:
init_logger(LogLevel.DEBUG)
if self.dvm_config.UPDATE_DATABASE:
self.sync_db()
await self.sync_db()
if not self.personalized:
self.result = self.calculate_result(self.request_form)
self.result = await self.calculate_result(self.request_form)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -110,15 +105,14 @@ class DicoverContentCurrentlyPopularbyTopic(DVMTaskInterface):
self.request_form = request_form
return request_form
def process(self, request_form):
async def process(self, request_form):
# if the dvm supports individual results, recalculate it every time for the request
if self.personalized:
return self.calculate_result(request_form)
#else return the result that gets updated once every schenduled update. In this case on database update.
return await self.calculate_result(request_form)
# else return the result that gets updated once every schenduled update. In this case on database update.
else:
return self.result
def post_process(self, result, event):
"""Overwrite the interface function to return a social client readable format, if requested"""
for tag in event.tags():
@ -129,7 +123,8 @@ class DicoverContentCurrentlyPopularbyTopic(DVMTaskInterface):
# if not text/plain, don't post-process
return result
def calculate_result(self, request_form):
async def calculate_result(self, request_form):
from nostr_sdk import Filter
from types import SimpleNamespace
ns = SimpleNamespace()
@ -141,10 +136,10 @@ class DicoverContentCurrentlyPopularbyTopic(DVMTaskInterface):
keys = Keys.parse(sk.to_hex())
signer = NostrSigner.keys(keys)
database = NostrDatabase.sqlite(self.db_name)
database = await NostrDatabase.sqlite(self.db_name)
cli = ClientBuilder().database(database).signer(signer).opts(opts).build()
cli.connect()
await cli.connect()
# Negentropy reconciliation
# Query events from database
@ -153,7 +148,7 @@ class DicoverContentCurrentlyPopularbyTopic(DVMTaskInterface):
filter1 = Filter().kind(definitions.EventDefinitions.KIND_NOTE).since(since)
events = cli.database().query([filter1])
events = await cli.database().query([filter1])
print("[" + self.dvm_config.NIP89.NAME + "] Considering " + str(len(events)) + " Events")
ns.finallist = {}
@ -166,7 +161,7 @@ class DicoverContentCurrentlyPopularbyTopic(DVMTaskInterface):
[definitions.EventDefinitions.KIND_ZAP, definitions.EventDefinitions.KIND_REACTION,
definitions.EventDefinitions.KIND_REPOST,
definitions.EventDefinitions.KIND_NOTE]).event(event.id()).since(since)
reactions = cli.database().query([filt])
reactions = await cli.database().query([filt])
if len(reactions) >= self.min_reactions:
ns.finallist[event.id().to_hex()] = len(reactions)
@ -179,56 +174,57 @@ class DicoverContentCurrentlyPopularbyTopic(DVMTaskInterface):
print("[" + self.dvm_config.NIP89.NAME + "] Filtered " + str(
len(result_list)) + " fitting events.")
cli.shutdown()
await cli.shutdown()
return json.dumps(result_list)
def schedule(self, dvm_config):
async def schedule(self, dvm_config):
if dvm_config.SCHEDULE_UPDATES_SECONDS == 0:
return 0
else:
if Timestamp.now().as_secs() >= self.last_schedule + dvm_config.SCHEDULE_UPDATES_SECONDS:
if self.dvm_config.UPDATE_DATABASE:
self.sync_db()
await self.sync_db()
self.last_schedule = Timestamp.now().as_secs()
self.result = self.calculate_result(self.request_form)
self.result = await self.calculate_result(self.request_form)
return 1
def sync_db(self):
async def sync_db(self):
opts = (Options().wait_for_send(False).send_timeout(timedelta(seconds=self.dvm_config.RELAY_LONG_TIMEOUT)))
sk = SecretKey.from_hex(self.dvm_config.PRIVATE_KEY)
keys = Keys.parse(sk.to_hex())
signer = NostrSigner.keys(keys)
database = NostrDatabase.sqlite(self.db_name)
database = await NostrDatabase.sqlite(self.db_name)
cli = ClientBuilder().signer(signer).database(database).opts(opts).build()
for relay in self.dvm_config.RECONCILE_DB_RELAY_LIST:
cli.add_relay(relay)
await cli.add_relay(relay)
cli.connect()
await cli.connect()
timestamp_since = Timestamp.now().as_secs() - self.db_since
since = Timestamp.from_secs(timestamp_since)
filter1 = Filter().kinds([definitions.EventDefinitions.KIND_NOTE, definitions.EventDefinitions.KIND_REACTION, definitions.EventDefinitions.KIND_ZAP]).since(since) # Notes, reactions, zaps
filter1 = Filter().kinds([definitions.EventDefinitions.KIND_NOTE, definitions.EventDefinitions.KIND_REACTION,
definitions.EventDefinitions.KIND_ZAP]).since(since) # Notes, reactions, zaps
# filter = Filter().author(keys.public_key())
print("[" + self.dvm_config.NIP89.NAME + "] Syncing notes of the last " + str(self.db_since) + " seconds.. this might take a while..")
print("[" + self.dvm_config.NIP89.NAME + "] Syncing notes of the last " + str(
self.db_since) + " seconds.. this might take a while..")
dbopts = NegentropyOptions().direction(NegentropyDirection.DOWN)
cli.reconcile(filter1, dbopts)
cli.database().delete(Filter().until(Timestamp.from_secs(
await cli.reconcile(filter1, dbopts)
await cli.database().delete(Filter().until(Timestamp.from_secs(
Timestamp.now().as_secs() - self.db_since))) # Clear old events so db doesn't get too full.
cli.shutdown()
print("[" + self.dvm_config.NIP89.NAME + "] Done Syncing Notes of the last " + str(self.db_since) + " seconds..")
await cli.shutdown()
print(
"[" + self.dvm_config.NIP89.NAME + "] Done Syncing Notes of the last " + str(self.db_since) + " seconds..")
# We build an example here that we can call by either calling this file directly from the main directory,
# or by adding it to our playground. You can call the example and adjust it to your needs or redefine it in the
# playground or elsewhere
def build_example(name, identifier, admin_config, options, image, description, update_rate=600, cost=0, processing_msg=None, update_db=True):
def build_example(name, identifier, admin_config, options, image, description, update_rate=600, cost=0,
processing_msg=None, update_db=True):
dvm_config = build_default_config(identifier)
dvm_config.USE_OWN_VENV = False
dvm_config.SHOWLOG = True
@ -241,7 +237,6 @@ def build_example(name, identifier, admin_config, options, image, description, u
dvm_config.CUSTOM_PROCESSING_MESSAGE = processing_msg
admin_config.LUD16 = dvm_config.LN_ADDRESS
# Add NIP89
nip89info = {
"name": name,
@ -267,10 +262,11 @@ def build_example(name, identifier, admin_config, options, image, description, u
nip89config.CONTENT = json.dumps(nip89info)
return DicoverContentCurrentlyPopularbyTopic(name=name, dvm_config=dvm_config, nip89config=nip89config,
admin_config=admin_config, options=options)
admin_config=admin_config, options=options)
def build_example_subscription(name, identifier, admin_config, options, image, description, processing_msg=None, update_db=True):
def build_example_subscription(name, identifier, admin_config, options, image, description, processing_msg=None,
update_db=True):
dvm_config = build_default_config(identifier)
dvm_config.USE_OWN_VENV = False
dvm_config.SHOWLOG = True
@ -281,7 +277,6 @@ def build_example_subscription(name, identifier, admin_config, options, image, d
dvm_config.CUSTOM_PROCESSING_MESSAGE = processing_msg
admin_config.LUD16 = dvm_config.LN_ADDRESS
# Add NIP89
nip89info = {
"name": name,
@ -319,17 +314,15 @@ def build_example_subscription(name, identifier, admin_config, options, image, d
nip88config.PERK2DESC = "Support NostrDVM & NostrSDK development"
nip88config.PAYMENT_VERIFIER_PUBKEY = "5b5c045ecdf66fb540bdf2049fe0ef7f1a566fa427a4fe50d400a011b65a3a7e"
# admin_config.FETCH_NIP88 = True
# admin_config.EVENTID = "63a791cdc7bf78c14031616963105fce5793f532bb231687665b14fb6d805fdb"
# admin_config.PRIVKEY = dvm_config.PRIVATE_KEY
# admin_config.FETCH_NIP88 = True
# admin_config.EVENTID = "63a791cdc7bf78c14031616963105fce5793f532bb231687665b14fb6d805fdb"
# admin_config.PRIVKEY = dvm_config.PRIVATE_KEY
return DicoverContentCurrentlyPopularbyTopic(name=name, dvm_config=dvm_config, nip89config=nip89config,
nip88config=nip88config,
admin_config=admin_config,
options=options)
nip88config=nip88config,
admin_config=admin_config,
options=options)
if __name__ == '__main__':
process_venv(DicoverContentCurrentlyPopularbyTopic)
process_venv(DicoverContentCurrentlyPopularbyTopic)

31
nostr_dvm/tasks/content_discovery_update_db_only.py
View File

@ -1,3 +1,4 @@
import asyncio
import json
import os
from datetime import timedelta
@ -26,6 +27,7 @@ class DicoverContentDBUpdateScheduler(DVMTaskInterface):
TASK: str = "update-db-on-schedule"
FIX_COST: float = 0
dvm_config: DVMConfig
request_form = None
last_schedule: int
min_reactions = 2
db_since = 10 * 3600
@ -36,11 +38,8 @@ class DicoverContentDBUpdateScheduler(DVMTaskInterface):
personalized = False
result = ""
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
# Generate Generic request form for dvms that provide generic results (e.g only a calculation per update,
# not per call)
@ -72,7 +71,7 @@ class DicoverContentDBUpdateScheduler(DVMTaskInterface):
init_logger(LogLevel.DEBUG)
if self.dvm_config.UPDATE_DATABASE:
self.sync_db()
await self.sync_db()
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -106,7 +105,7 @@ class DicoverContentDBUpdateScheduler(DVMTaskInterface):
self.request_form = request_form
return request_form
def process(self, request_form):
async def process(self, request_form):
return "I don't return results, I just update the DB."
def post_process(self, result, event):
@ -120,28 +119,28 @@ class DicoverContentDBUpdateScheduler(DVMTaskInterface):
# if not text/plain, don't post-process
return result
def schedule(self, dvm_config):
async def schedule(self, dvm_config):
if dvm_config.SCHEDULE_UPDATES_SECONDS == 0:
return 0
else:
if Timestamp.now().as_secs() >= self.last_schedule + dvm_config.SCHEDULE_UPDATES_SECONDS:
if self.dvm_config.UPDATE_DATABASE:
self.sync_db()
await self.sync_db()
self.last_schedule = Timestamp.now().as_secs()
return 1
def sync_db(self):
async def sync_db(self):
opts = (Options().wait_for_send(False).send_timeout(timedelta(seconds=self.dvm_config.RELAY_LONG_TIMEOUT)))
sk = SecretKey.from_hex(self.dvm_config.PRIVATE_KEY)
keys = Keys.parse(sk.to_hex())
signer = NostrSigner.keys(keys)
database = NostrDatabase.sqlite(self.db_name)
database = await NostrDatabase.sqlite(self.db_name)
cli = ClientBuilder().signer(signer).database(database).opts(opts).build()
for relay in self.dvm_config.RECONCILE_DB_RELAY_LIST:
cli.add_relay(relay)
await cli.add_relay(relay)
cli.connect()
await cli.connect()
timestamp_since = Timestamp.now().as_secs() - self.db_since
since = Timestamp.from_secs(timestamp_since)
@ -153,10 +152,10 @@ class DicoverContentDBUpdateScheduler(DVMTaskInterface):
print("[" + self.dvm_config.IDENTIFIER + "] Syncing notes of the last " + str(
self.db_since) + " seconds.. this might take a while..")
dbopts = NegentropyOptions().direction(NegentropyDirection.DOWN)
cli.reconcile(filter1, dbopts)
cli.database().delete(Filter().until(Timestamp.from_secs(
await cli.reconcile(filter1, dbopts)
await cli.database().delete(Filter().until(Timestamp.from_secs(
Timestamp.now().as_secs() - self.db_since))) # Clear old events so db doesn't get too full.
cli.shutdown()
await cli.shutdown()
print(
"[" + self.dvm_config.IDENTIFIER + "] Done Syncing Notes of the last " + str(self.db_since) + " seconds..")

8
nostr_dvm/tasks/convert_media.py
View File

@ -27,11 +27,9 @@ class MediaConverter(DVMTaskInterface):
FIX_COST = 20
PER_UNIT_COST = 0.1
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -72,7 +70,7 @@ class MediaConverter(DVMTaskInterface):
request_form['options'] = json.dumps(options)
return request_form
def process(self, request_form):
async def process(self, request_form):
options = self.set_options(request_form)
url = upload_media_to_hoster(options["filepath"])

45
nostr_dvm/tasks/discovery_bot_farms.py
View File

@ -27,17 +27,10 @@ class DiscoveryBotFarms(DVMTaskInterface):
dvm_config: DVMConfig
last_schedule: int = 0
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
use_logger = False
if use_logger:
init_logger(LogLevel.DEBUG)
self.sync_db()
await self.sync_db()
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -55,7 +48,7 @@ class DiscoveryBotFarms(DVMTaskInterface):
request_form = {"jobID": event.id().to_hex()}
# default values
search = "airdrop;just your average nostr enjoyer" #;@nostrich.house;
search = "airdrop;just your average nostr enjoyer" # ;@nostrich.house;
max_results = 500
for tag in event.tags():
@ -75,7 +68,7 @@ class DiscoveryBotFarms(DVMTaskInterface):
request_form['options'] = json.dumps(options)
return request_form
def process(self, request_form):
async def process(self, request_form):
from nostr_sdk import Filter
options = self.set_options(request_form)
@ -84,19 +77,19 @@ class DiscoveryBotFarms(DVMTaskInterface):
keys = Keys.parse(sk.to_hex())
signer = NostrSigner.keys(keys)
database = NostrDatabase.sqlite("db/nostr_profiles.db")
database = await NostrDatabase.sqlite("db/nostr_profiles.db")
cli = ClientBuilder().database(database).signer(signer).opts(opts).build()
cli.add_relay("wss://relay.damus.io")
await cli.add_relay("wss://relay.damus.io")
# cli.add_relay("wss://atl.purplerelay.com")
cli.connect()
await cli.connect()
# Negentropy reconciliation
# Query events from database
filter1 = Filter().kind(Kind(0))
events = cli.database().query([filter1])
events = await cli.database().query([filter1])
# for event in events:
# print(event.as_json())
@ -121,7 +114,7 @@ class DiscoveryBotFarms(DVMTaskInterface):
print(str(exp) + " " + event.author().to_hex())
else:
break
await cli.shutdown()
return json.dumps(result_list)
def post_process(self, result, event):
@ -135,33 +128,33 @@ class DiscoveryBotFarms(DVMTaskInterface):
# if not text/plain, don't post-process
return result
def schedule(self, dvm_config):
async def schedule(self, dvm_config):
if dvm_config.SCHEDULE_UPDATES_SECONDS == 0:
return 0
else:
if Timestamp.now().as_secs() >= self.last_schedule + dvm_config.SCHEDULE_UPDATES_SECONDS:
self.sync_db()
await self.sync_db()
self.last_schedule = Timestamp.now().as_secs()
return 1
def sync_db(self):
async def sync_db(self):
opts = (Options().wait_for_send(False).send_timeout(timedelta(seconds=self.dvm_config.RELAY_TIMEOUT)))
sk = SecretKey.from_hex(self.dvm_config.PRIVATE_KEY)
keys = Keys.parse(sk.to_hex())
signer = NostrSigner.keys(keys)
database = NostrDatabase.sqlite("db/nostr_profiles.db")
database = await NostrDatabase.sqlite("db/nostr_profiles.db")
cli = ClientBuilder().signer(signer).database(database).opts(opts).build()
cli.add_relay("wss://relay.damus.io")
cli.add_relay("wss://nostr21.com")
cli.connect()
await cli.add_relay("wss://relay.damus.io")
await cli.add_relay("wss://nostr21.com")
await cli.connect()
filter1 = Filter().kind(Kind(0))
# filter = Filter().author(keys.public_key())
print("Syncing Profile Database.. this might take a while..")
dbopts = NegentropyOptions().direction(NegentropyDirection.DOWN)
cli.reconcile(filter1, dbopts)
await cli.reconcile(filter1, dbopts)
print("Done Syncing Profile Database.")
@ -197,7 +190,7 @@ def build_example(name, identifier, admin_config):
options = {"relay": "wss://relay.damus.io"}
return DiscoveryBotFarms(name=name, dvm_config=dvm_config, nip89config=nip89config,
admin_config=admin_config, options=options)
admin_config=admin_config, options=options)
if __name__ == '__main__':

18
nostr_dvm/tasks/discovery_censor_wot.py
View File

@ -30,11 +30,10 @@ class DiscoverReports(DVMTaskInterface):
client: Client
dvm_config: DVMConfig
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
def is_input_supported(self, tags, client=None, dvm_config=None):
return True
@ -68,7 +67,7 @@ class DiscoverReports(DVMTaskInterface):
request_form['options'] = json.dumps(options)
return request_form
def process(self, request_form):
async def process(self, request_form):
from nostr_sdk import Filter
from types import SimpleNamespace
ns = SimpleNamespace()
@ -85,9 +84,9 @@ class DiscoverReports(DVMTaskInterface):
cli.add_relay(relay)
# add nostr band, too.
ropts = RelayOptions().ping(False)
cli.add_relay_with_opts("wss://nostr.band", ropts)
await cli.add_relay_with_opts("wss://nostr.band", ropts)
cli.connect()
await cli.connect()
options = self.set_options(request_form)
step = 20
@ -99,7 +98,7 @@ class DiscoverReports(DVMTaskInterface):
# if we don't add users, e.g. by a wot, we check all our followers.
if len(pubkeys) == 0:
followers_filter = Filter().author(PublicKey.parse(options["sender"])).kind(Kind(3))
followers = cli.get_events_of([followers_filter], timedelta(seconds=5))
followers = await cli.get_events_of([followers_filter], timedelta(seconds=5))
if len(followers) > 0:
result_list = []
@ -119,7 +118,7 @@ class DiscoverReports(DVMTaskInterface):
ago = Timestamp.now().as_secs() - 60*60*24*int(options["since_days"]) #TODO make this an option, 180 days for now
since = Timestamp.from_secs(ago)
kind1984_filter = Filter().authors(pubkeys).kind(Kind(1984)).since(since)
reports = cli.get_events_of([kind1984_filter], timedelta(seconds=self.dvm_config.RELAY_TIMEOUT))
reports = await cli.get_events_of([kind1984_filter], timedelta(seconds=self.dvm_config.RELAY_TIMEOUT))
bad_actors = []
ns.dic = {}
@ -149,6 +148,7 @@ class DiscoverReports(DVMTaskInterface):
bad_actors.append(p_tag.as_vec())
print(json.dumps(bad_actors))
await cli.shutdown()
return json.dumps(bad_actors)
def post_process(self, result, event):

22
nostr_dvm/tasks/discovery_inactive_follows.py
View File

@ -30,11 +30,9 @@ class DiscoverInactiveFollows(DVMTaskInterface):
client: Client
dvm_config: DVMConfig
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
def is_input_supported(self, tags, client=None, dvm_config=None):
# no input required
@ -64,7 +62,7 @@ class DiscoverInactiveFollows(DVMTaskInterface):
request_form['options'] = json.dumps(options)
return request_form
def process(self, request_form):
async def process(self, request_form):
from nostr_sdk import Filter
from types import SimpleNamespace
ns = SimpleNamespace()
@ -82,17 +80,17 @@ class DiscoverInactiveFollows(DVMTaskInterface):
cli = Client.with_opts(signer, opts)
for relay in self.dvm_config.RELAY_LIST:
cli.add_relay(relay)
await cli.add_relay(relay)
ropts = RelayOptions().ping(False)
cli.add_relay_with_opts("wss://nostr.band", ropts)
await cli.add_relay_with_opts("wss://nostr.band", ropts)
cli.connect()
await cli.connect()
options = self.set_options(request_form)
step = 20
followers_filter = Filter().author(PublicKey.parse(options["user"])).kind(Kind(3))
followers = cli.get_events_of([followers_filter], timedelta(seconds=5))
followers = await cli.get_events_of([followers_filter], timedelta(seconds=5))
if len(followers) > 0:
@ -126,7 +124,7 @@ class DiscoverInactiveFollows(DVMTaskInterface):
dif = Timestamp.now().as_secs() - not_active_since_seconds
not_active_since = Timestamp.from_secs(dif)
def scanList(users, instance, i, st, notactivesince):
async def scanList(users, instance, i, st, notactivesince):
from nostr_sdk import Filter
keys = Keys.parse(self.dvm_config.PRIVATE_KEY)
@ -135,8 +133,8 @@ class DiscoverInactiveFollows(DVMTaskInterface):
signer = NostrSigner.keys(keys)
cli = Client.with_opts(signer, opts)
for relay in self.dvm_config.RELAY_LIST:
cli.add_relay(relay)
cli.connect()
await cli.add_relay(relay)
await cli.connect()
filters = []
for i in range(i, i + st):

20
nostr_dvm/tasks/discovery_nonfollowers.py
View File

@ -30,11 +30,9 @@ class DiscoverNonFollowers(DVMTaskInterface):
client: Client
dvm_config: DVMConfig
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
def is_input_supported(self, tags, client=None, dvm_config=None):
# no input required
@ -59,7 +57,7 @@ class DiscoverNonFollowers(DVMTaskInterface):
request_form['options'] = json.dumps(options)
return request_form
def process(self, request_form):
async def process(self, request_form):
from nostr_sdk import Filter
from types import SimpleNamespace
ns = SimpleNamespace()
@ -71,12 +69,12 @@ class DiscoverNonFollowers(DVMTaskInterface):
cli = Client.with_opts(signer, opts)
# cli.add_relay("wss://relay.nostr.band")
for relay in self.dvm_config.RELAY_LIST:
cli.add_relay(relay)
await cli.add_relay(relay)
#add nostr band, too.
ropts = RelayOptions().ping(False)
cli.add_relay_with_opts("wss://nostr.band", ropts)
await cli.add_relay_with_opts("wss://nostr.band", ropts)
cli.connect()
await cli.connect()
options = self.set_options(request_form)
step = 20
@ -103,7 +101,7 @@ class DiscoverNonFollowers(DVMTaskInterface):
ns.dic[following] = "True"
print("Followings: " + str(len(followings)))
def scanList(users, instance, i, st):
async def scanList(users, instance, i, st):
from nostr_sdk import Filter
keys = Keys.parse(self.dvm_config.PRIVATE_KEY)
@ -112,14 +110,14 @@ class DiscoverNonFollowers(DVMTaskInterface):
signer = NostrSigner.keys(keys)
cli = Client.with_opts(signer, opts)
for relay in self.dvm_config.RELAY_LIST:
cli.add_relay(relay)
await cli.add_relay(relay)
cli.connect()
for i in range(i, i + st):
filters = []
filter1 = Filter().author(PublicKey.from_hex(users[i])).kind(Kind(3))
filters.append(filter1)
followers = cli.get_events_of(filters, timedelta(seconds=3))
followers = await cli.get_events_of(filters, timedelta(seconds=3))
if len(followers) > 0:
result_list = []

8
nostr_dvm/tasks/discovery_trending_notes_nostrband.py
View File

@ -25,11 +25,9 @@ class TrendingNotesNostrBand(DVMTaskInterface):
dvm_config: DVMConfig
logger = False
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
if self.options is not None:
if self.options.get("logger"):
@ -67,7 +65,7 @@ class TrendingNotesNostrBand(DVMTaskInterface):
request_form['options'] = json.dumps(options)
return request_form
def process(self, request_form):
async def process(self, request_form):
options = self.set_options(request_form)
import requests

8
nostr_dvm/tasks/imagegeneration_openai_dalle.py
View File

@ -31,11 +31,9 @@ class ImageGenerationDALLE(DVMTaskInterface):
dependencies = [("nostr-dvm", "nostr-dvm"),
("openai", "openai==1.3.5")]
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -94,7 +92,7 @@ class ImageGenerationDALLE(DVMTaskInterface):
return request_form
def process(self, request_form):
async def process(self, request_form):
try:
options = self.set_options(request_form)

8
nostr_dvm/tasks/imagegeneration_replicate_sdxl.py
View File

@ -30,11 +30,9 @@ class ImageGenerationReplicateSDXL(DVMTaskInterface):
dependencies = [("nostr-dvm", "nostr-dvm"),
("replicate", "replicate")]
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -91,7 +89,7 @@ class ImageGenerationReplicateSDXL(DVMTaskInterface):
return request_form
def process(self, request_form):
async def process(self, request_form):
try:
options = self.set_options(request_form)

8
nostr_dvm/tasks/imagegeneration_sd21_mlx.py
View File

@ -34,11 +34,9 @@ class ImageGenerationMLX(DVMTaskInterface):
("tqdm", "tqdm"),
]
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -95,7 +93,7 @@ class ImageGenerationMLX(DVMTaskInterface):
return request_form
def process(self, request_form):
async def process(self, request_form):
try:
import mlx.core as mx
from nostr_dvm.backends.mlx.modules.stable_diffusion import StableDiffusion

10
nostr_dvm/tasks/imagegeneration_sdxl.py
View File

@ -1,4 +1,5 @@
import json
import os
from multiprocessing.pool import ThreadPool
from nostr_sdk import Kind
@ -26,10 +27,9 @@ class ImageGenerationSDXL(DVMTaskInterface):
TASK: str = "text-to-image"
FIX_COST: float = 50
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -144,7 +144,7 @@ class ImageGenerationSDXL(DVMTaskInterface):
return request_form
def process(self, request_form):
async def process(self, request_form):
try:
# Call the process route of n-server with our request form.
response = send_request_to_server(request_form, self.options['server'])

10
nostr_dvm/tasks/imagegeneration_sdxlimg2img.py
View File

@ -1,4 +1,5 @@
import json
import os
from multiprocessing.pool import ThreadPool
from nostr_sdk import Kind
@ -26,10 +27,9 @@ class ImageGenerationSDXLIMG2IMG(DVMTaskInterface):
TASK: str = "image-to-image"
FIX_COST: float = 70
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
def is_input_supported(self, tags, client=None, dvm_config=None):
hasurl = False
@ -170,7 +170,7 @@ class ImageGenerationSDXLIMG2IMG(DVMTaskInterface):
return request_form
def process(self, request_form):
async def process(self, request_form):
try:
# Call the process route of NOVA-Server with our request form.
response = send_request_to_server(request_form, self.options['server'])

10
nostr_dvm/tasks/imageinterrogator.py
View File

@ -1,4 +1,5 @@
import json
import os
from multiprocessing.pool import ThreadPool
from nostr_sdk import Kind
@ -25,10 +26,9 @@ class ImageInterrogator(DVMTaskInterface):
TASK: str = "image-to-text"
FIX_COST: float = 80
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
def is_input_supported(self, tags, client=None, dvm_config=None):
hasurl = False
@ -87,7 +87,7 @@ class ImageInterrogator(DVMTaskInterface):
return request_form
def process(self, request_form):
async def process(self, request_form):
try:
# Call the process route of NOVA-Server with our request form.
response = send_request_to_server(request_form, self.options['server'])

10
nostr_dvm/tasks/imageupscale.py
View File

@ -1,4 +1,5 @@
import json
import os
from multiprocessing.pool import ThreadPool
from nostr_sdk import Kind
@ -25,10 +26,9 @@ class ImageUpscale(DVMTaskInterface):
TASK: str = "image-to-image"
FIX_COST: float = 20
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
def is_input_supported(self, tags, client=None, dvm_config=None):
hasurl = False
@ -89,7 +89,7 @@ class ImageUpscale(DVMTaskInterface):
return request_form
def process(self, request_form):
async def process(self, request_form):
try:
# Call the process route of NOVA-Server with our request form.
response = send_request_to_server(request_form, self.options['server'])

39
nostr_dvm/tasks/search_users.py
View File

@ -28,17 +28,10 @@ class SearchUser(DVMTaskInterface):
last_schedule: int = 0
db_name = "db/nostr_profiles.db"
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
use_logger = False
if use_logger:
init_logger(LogLevel.DEBUG)
self.sync_db()
await self.sync_db()
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -76,7 +69,7 @@ class SearchUser(DVMTaskInterface):
request_form['options'] = json.dumps(options)
return request_form
def process(self, request_form):
async def process(self, request_form):
from nostr_sdk import Filter
options = self.set_options(request_form)
@ -85,19 +78,19 @@ class SearchUser(DVMTaskInterface):
keys = Keys.parse(sk.to_hex())
signer = NostrSigner.keys(keys)
database = NostrDatabase.sqlite(self.db_name)
database = await NostrDatabase.sqlite(self.db_name)
cli = ClientBuilder().database(database).signer(signer).opts(opts).build()
cli.add_relay("wss://relay.damus.io")
await cli.add_relay("wss://relay.damus.io")
# cli.add_relay("wss://atl.purplerelay.com")
cli.connect()
await cli.connect()
# Negentropy reconciliation
# Query events from database
filter1 = Filter().kind(Kind(0))
events = cli.database().query([filter1])
events = await cli.database().query([filter1])
# for event in events:
# print(event.as_json())
@ -121,6 +114,7 @@ class SearchUser(DVMTaskInterface):
else:
break
await cli.disconnect()
return json.dumps(result_list)
def post_process(self, result, event):
@ -134,33 +128,34 @@ class SearchUser(DVMTaskInterface):
# if not text/plain, don't post-process
return result
def schedule(self, dvm_config):
async def schedule(self, dvm_config):
if dvm_config.SCHEDULE_UPDATES_SECONDS == 0:
return 0
else:
if Timestamp.now().as_secs() >= self.last_schedule + dvm_config.SCHEDULE_UPDATES_SECONDS:
self.sync_db()
await self.sync_db()
self.last_schedule = Timestamp.now().as_secs()
return 1
def sync_db(self):
async def sync_db(self):
opts = (Options().wait_for_send(False).send_timeout(timedelta(seconds=self.dvm_config.RELAY_TIMEOUT)))
sk = SecretKey.from_hex(self.dvm_config.PRIVATE_KEY)
keys = Keys.parse(sk.to_hex())
signer = NostrSigner.keys(keys)
database = NostrDatabase.sqlite(self.db_name)
database = await NostrDatabase.sqlite(self.db_name)
cli = ClientBuilder().signer(signer).database(database).opts(opts).build()
cli.add_relay("wss://relay.damus.io")
cli.connect()
await cli.add_relay("wss://relay.damus.io")
await cli.connect()
filter1 = Filter().kind(Kind(0))
# filter = Filter().author(keys.public_key())
print("Syncing Profile Database.. this might take a while..")
dbopts = NegentropyOptions().direction(NegentropyDirection.DOWN)
cli.reconcile(filter1, dbopts)
await cli.reconcile(filter1, dbopts)
print("Done Syncing Profile Database.")
await cli.shutdown()
# We build an example here that we can call by either calling this file directly from the main directory,

9
nostr_dvm/tasks/summarization_huggingchat.py
View File

@ -26,11 +26,10 @@ class TextSummarizationHuggingChat(DVMTaskInterface):
dependencies = [("nostr-dvm", "nostr-dvm"),
("hugchat", "hugchat")]
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -92,7 +91,7 @@ class TextSummarizationHuggingChat(DVMTaskInterface):
return request_form
def process(self, request_form):
async def process(self, request_form):
from hugchat import hugchat
from hugchat.login import Login
sign = Login(os.getenv("HUGGINGFACE_EMAIL"), os.getenv("HUGGINGFACE_PASSWORD"))

8
nostr_dvm/tasks/summarization_unleashed_chat.py
View File

@ -25,11 +25,9 @@ class SummarizationUnleashedChat(DVMTaskInterface):
dependencies = [("nostr-dvm", "nostr-dvm"),
("openai", "openai")]
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -93,7 +91,7 @@ class SummarizationUnleashedChat(DVMTaskInterface):
return request_form
def process(self, request_form):
async def process(self, request_form):
from openai import OpenAI
temp_open_ai_api_key = os.environ["OPENAI_API_KEY"]
os.environ["OPENAI_API_KEY"] = os.getenv("UNLEASHED_API_KEY")

11
nostr_dvm/tasks/textextraction_google.py
View File

@ -29,13 +29,10 @@ class SpeechToTextGoogle(DVMTaskInterface):
dependencies = [("nostr-dvm", "nostr-dvm"),
("speech_recognition", "SpeechRecognition==3.10.0")]
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
if options is None:
self.options = {}
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -106,7 +103,7 @@ class SpeechToTextGoogle(DVMTaskInterface):
request_form['options'] = json.dumps(options)
return request_form
def process(self, request_form):
async def process(self, request_form):
import speech_recognition as sr
if self.options.get("api_key"):
api_key = self.options['api_key']

8
nostr_dvm/tasks/textextraction_pdf.py
View File

@ -28,11 +28,9 @@ class TextExtractionPDF(DVMTaskInterface):
dependencies = [("nostr-dvm", "nostr-dvm"),
("pypdf", "pypdf==3.17.1")]
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -69,7 +67,7 @@ class TextExtractionPDF(DVMTaskInterface):
request_form['options'] = json.dumps(options)
return request_form
def process(self, request_form):
async def process(self, request_form):
from pypdf import PdfReader
from pathlib import Path
import requests

8
nostr_dvm/tasks/textextraction_whisperx.py
View File

@ -29,11 +29,9 @@ class SpeechToTextWhisperX(DVMTaskInterface):
FIX_COST: float = 10
PER_UNIT_COST: float = 0.1
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -132,7 +130,7 @@ class SpeechToTextWhisperX(DVMTaskInterface):
request_form['options'] = json.dumps(options)
return request_form
def process(self, request_form):
async def process(self, request_form):
try:
# Call the process route of NOVA-Server with our request form.
response = send_request_to_server(request_form, self.options['server'])

8
nostr_dvm/tasks/textgeneration_huggingchat.py
View File

@ -25,11 +25,9 @@ class TextGenerationHuggingChat(DVMTaskInterface):
dependencies = [("nostr-dvm", "nostr-dvm"),
("hugchat", "hugchat")]
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -58,7 +56,7 @@ class TextGenerationHuggingChat(DVMTaskInterface):
return request_form
def process(self, request_form):
async def process(self, request_form):
from hugchat import hugchat
from hugchat.login import Login
sign = Login(os.getenv("HUGGINGFACE_EMAIL"), os.getenv("HUGGINGFACE_PASSWORD"))

8
nostr_dvm/tasks/textgeneration_llmlite.py
View File

@ -25,11 +25,9 @@ class TextGenerationLLMLite(DVMTaskInterface):
dependencies = [("nostr-dvm", "nostr-dvm"),
("litellm", "litellm==1.12.3")]
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -68,7 +66,7 @@ class TextGenerationLLMLite(DVMTaskInterface):
return request_form
def process(self, request_form):
async def process(self, request_form):
from litellm import completion
options = self.set_options(request_form)

8
nostr_dvm/tasks/textgeneration_unleashed_chat.py
View File

@ -25,11 +25,9 @@ class TextGenerationUnleashedChat(DVMTaskInterface):
dependencies = [("nostr-dvm", "nostr-dvm"),
("openai", "openai")]
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -60,7 +58,7 @@ class TextGenerationUnleashedChat(DVMTaskInterface):
return request_form
def process(self, request_form):
async def process(self, request_form):
from openai import OpenAI
temp_open_ai_api_key = os.environ["OPENAI_API_KEY"]
os.environ["OPENAI_API_KEY"] = os.getenv("UNLEASHED_API_KEY")

8
nostr_dvm/tasks/texttospeech.py
View File

@ -33,11 +33,9 @@ class TextToSpeech(DVMTaskInterface):
dependencies = [("nostr-dvm", "nostr-dvm"),
("TTS", "TTS==0.22.0")]
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -103,7 +101,7 @@ class TextToSpeech(DVMTaskInterface):
return request_form
def process(self, request_form):
async def process(self, request_form):
import torch
from TTS.api import TTS
options = self.set_options(request_form)

8
nostr_dvm/tasks/translation_google.py
View File

@ -27,11 +27,9 @@ class TranslationGoogle(DVMTaskInterface):
dependencies = [("nostr-dvm", "nostr-dvm"),
("translatepy", "translatepy==2.3")]
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -77,7 +75,7 @@ class TranslationGoogle(DVMTaskInterface):
request_form['options'] = json.dumps(options)
return request_form
def process(self, request_form):
async def process(self, request_form):
from translatepy.translators.google import GoogleTranslate
options = self.set_options(request_form)

8
nostr_dvm/tasks/translation_libretranslate.py
View File

@ -27,11 +27,9 @@ class TranslationLibre(DVMTaskInterface):
TASK: str = "translation"
FIX_COST: float = 0
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -77,7 +75,7 @@ class TranslationLibre(DVMTaskInterface):
request_form['options'] = json.dumps(options)
return request_form
def process(self, request_form):
async def process(self, request_form):
options = self.set_options(request_form)
request = {
"q": options["text"],

8
nostr_dvm/tasks/videogeneration_replicate_svd.py
View File

@ -31,11 +31,9 @@ class VideoGenerationReplicateSVD(DVMTaskInterface):
dependencies = [("nostr-dvm", "nostr-dvm"),
("replicate", "replicate")]
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -79,7 +77,7 @@ class VideoGenerationReplicateSVD(DVMTaskInterface):
return request_form
def process(self, request_form):
async def process(self, request_form):
try:
options = self.set_options(request_form)
print(options["url"])

9
nostr_dvm/tasks/videogeneration_svd.py
View File

@ -26,10 +26,9 @@ class VideoGenerationSVD(DVMTaskInterface):
TASK: str = "image-to-video"
FIX_COST: float = 120
def __init__(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
super().__init__(name=name, dvm_config=dvm_config, nip89config=nip89config, nip88config=nip88config,
admin_config=admin_config, options=options)
async def init_dvm(self, name, dvm_config: DVMConfig, nip89config: NIP89Config, nip88config: NIP88Config = None,
admin_config: AdminConfig = None, options=None):
dvm_config.SCRIPT = os.path.abspath(__file__)
def is_input_supported(self, tags, client=None, dvm_config=None):
for tag in tags:
@ -80,7 +79,7 @@ class VideoGenerationSVD(DVMTaskInterface):
return request_form
def process(self, request_form):
async def process(self, request_form):
try:
# Call the process route of n-server with our request form.
response = send_request_to_server(request_form, self.options['server'])

4
nostr_dvm/utils/admin_utils.py
View File

@ -36,7 +36,7 @@ class AdminConfig:
PRIVKEY: str = ""
def admin_make_database_updates(adminconfig: AdminConfig = None, dvmconfig: DVMConfig = None, client: Client = None):
async def admin_make_database_updates(adminconfig: AdminConfig = None, dvmconfig: DVMConfig = None, client: Client = None):
# This is called on start of Server, Admin function to manually whitelist/blacklist/add balance/delete users
if adminconfig is None or dvmconfig is None:
return
@ -64,7 +64,7 @@ def admin_make_database_updates(adminconfig: AdminConfig = None, dvmconfig: DVMC
publickey = npub
if adminconfig.WHITELISTUSER:
user = get_or_add_user(db, publickey, client=client, config=dvmconfig)
user = await get_or_add_user(db, publickey, client=client, config=dvmconfig)
update_sql_table(db, user.npub, user.balance, True, False, user.nip05, user.lud16, user.name, user.lastactive, user.subscribed)
user = get_from_sql_table(db, publickey)
print(str(user.name) + " is whitelisted: " + str(user.iswhitelisted))

6
nostr_dvm/utils/cashu_utils.py
View File

@ -107,7 +107,7 @@ def parse_cashu(cashu_token: str):
return None, None, None, "Cashu Parser: " + str(e)
def redeem_cashu(cashu, config, client, required_amount=0, update_self=False) -> (bool, str, int, int):
async def redeem_cashu(cashu, config, client, required_amount=0, update_self=False) -> (bool, str, int, int):
proofs, mint, total_amount, message = parse_cashu(cashu)
if message is not None:
return False, message, 0, 0
@ -121,7 +121,7 @@ def redeem_cashu(cashu, config, client, required_amount=0, update_self=False) ->
invoice, paymenthash = create_bolt11_ln_bits(estimated_redeem_invoice_amount, config)
else:
user = get_or_add_user(db=config.DB, npub=config.PUBLIC_KEY,
user = await get_or_add_user(db=config.DB, npub=config.PUBLIC_KEY,
client=client, config=config, update=update_self)
invoice = create_bolt11_lud16(user.lud16, estimated_redeem_invoice_amount)
print(invoice)
@ -148,7 +148,7 @@ def redeem_cashu(cashu, config, client, required_amount=0, update_self=False) ->
invoice, paymenthash = create_bolt11_ln_bits(redeem_invoice_amount, config)
else:
user = get_or_add_user(db=config.DB, npub=config.PUBLIC_KEY,
user = await get_or_add_user(db=config.DB, npub=config.PUBLIC_KEY,
client=client, config=config, update=update_self)
invoice = create_bolt11_lud16(user.lud16, redeem_invoice_amount)
print(invoice)

12
nostr_dvm/utils/database_utils.py
View File

@ -110,7 +110,6 @@ def get_from_sql_table(db, npub):
add_sql_table_column(db)
# Migrate

user = User
user.npub = row[0]
user.balance = row[1]
@ -217,8 +216,7 @@ def update_user_subscription(npub, subscribed_until, client, dvm_config):
print("Updated user subscription for: " + str(user.name))
def get_or_add_user(db, npub, client, config, update=False, skip_meta = False):
async def get_or_add_user(db, npub, client, config, update=False, skip_meta=False):
user = get_from_sql_table(db, npub)
if user is None:
try:
@ -227,7 +225,7 @@ def get_or_add_user(db, npub, client, config, update=False, skip_meta = False):
nip05 = ""
lud16 = ""
else:
name, nip05, lud16 = fetch_user_metadata(npub, client)
name, nip05, lud16 = await fetch_user_metadata(npub, client)
print("Adding User: " + npub + " (" + npub + ")")
add_to_sql_table(db, npub, config.NEW_USER_BALANCE, False, False, nip05,
lud16, name, Timestamp.now().as_secs(), 0)
@ -237,7 +235,7 @@ def get_or_add_user(db, npub, client, config, update=False, skip_meta = False):
print("Error Adding User to DB: " + str(e))
elif update:
try:
name, nip05, lud16 = fetch_user_metadata(npub, client)
name, nip05, lud16 = await fetch_user_metadata(npub, client)
print("Updating User: " + npub + " (" + npub + ")")
update_sql_table(db, user.npub, user.balance, user.iswhitelisted, user.isblacklisted, nip05,
lud16, name, Timestamp.now().as_secs(), user.subscribed)
@ -249,14 +247,14 @@ def get_or_add_user(db, npub, client, config, update=False, skip_meta = False):
return user
def fetch_user_metadata(npub, client):
async def fetch_user_metadata(npub, client):
name = ""
nip05 = ""
lud16 = ""
pk = PublicKey.parse(npub)
print(f"\nGetting profile metadata for {pk.to_bech32()}...")
profile_filter = Filter().kind(Kind(0)).author(pk).limit(1)
events = client.get_events_of([profile_filter], timedelta(seconds=1))
events = await client.get_events_of([profile_filter], timedelta(seconds=1))
if len(events) > 0:
latest_entry = events[0]
latest_time = 0

56
nostr_dvm/utils/nostr_utils.py
View File

@ -1,3 +1,4 @@
import asyncio
import json
import os
from datetime import timedelta
@ -31,22 +32,30 @@ def get_event_by_id(event_id: str, client: Client, config=None) -> Event | None:
event_id = EventId.from_hex(event_id)
id_filter = Filter().id(event_id).limit(1)
events = client.get_events_of([id_filter], timedelta(seconds=config.RELAY_TIMEOUT))
#events = client.get_events_of([id_filter], timedelta(seconds=config.RELAY_TIMEOUT))
events = asyncio.run(get_events_async(client, id_filter, config.RELAY_TIMEOUT))
if len(events) > 0:
return events[0]
else:
return None
async def get_events_async(client, filter, timeout):
events = client.get_events_of([filter], timedelta(seconds=timeout)
return events
def get_events_by_ids(event_ids, client: Client, config=None) -> List | None:
search_ids = []
events = []
for event_id in event_ids:
split = event_id.split(":")
if len(split) == 3:
pk = PublicKey.from_hex(split[1])
id_filter = Filter().author(pk).custom_tag(SingleLetterTag.lowercase(Alphabet.D), [split[2]])
events = client.get_events_of([id_filter], timedelta(seconds=config.RELAY_TIMEOUT))
events = asyncio.run(get_events_async(client, id_filter, config.RELAY_TIMEOUT))
#events = client.get_events_of([id_filter], timedelta(seconds=config.RELAY_TIMEOUT))
else:
if str(event_id).startswith('note'):
event_id = EventId.from_bech32(event_id)
@ -59,12 +68,13 @@ def get_events_by_ids(event_ids, client: Client, config=None) -> List | None:
else:
event_id = EventId.from_hex(event_id)
search_ids.append(event_id)
search_ids.append(event_id)
id_filter = Filter().ids(search_ids)
events = client.get_events_of([id_filter], timedelta(seconds=config.RELAY_TIMEOUT))
id_filter = Filter().ids(search_ids)
events = asyncio.run(get_events_async(client, id_filter, config.RELAY_TIMEOUT))
#events = client.get_events_of([id_filter], timedelta(seconds=config.RELAY_TIMEOUT))
if len(events) > 0:
return events
else:
return None
@ -72,7 +82,8 @@ def get_events_by_ids(event_ids, client: Client, config=None) -> List | None:
def get_events_by_id(event_ids: list, client: Client, config=None) -> list[Event] | None:
id_filter = Filter().ids(event_ids)
events = client.get_events_of([id_filter], timedelta(seconds=config.RELAY_TIMEOUT))
events = asyncio.run(get_events_async(client, id_filter, config.RELAY_TIMEOUT))
#events = client.get_events_of([id_filter], timedelta(seconds=config.RELAY_TIMEOUT))
if len(events) > 0:
return events
else:
@ -98,7 +109,8 @@ def get_referenced_event_by_id(event_id, client, dvm_config, kinds) -> Event | N
else:
job_id_filter = Filter().event(event_id).limit(1)
events = client.get_events_of([job_id_filter], timedelta(seconds=dvm_config.RELAY_TIMEOUT))
events = asyncio.run(get_events_async(client, job_id_filter, dvm_config.RELAY_TIMEOUT))
#events = client.get_events_of([job_id_filter], timedelta(seconds=dvm_config.RELAY_TIMEOUT))
if len(events) > 0:
return events[0]
@ -106,7 +118,7 @@ def get_referenced_event_by_id(event_id, client, dvm_config, kinds) -> Event | N
return None
def get_inbox_relays(event_to_send: Event, client: Client, dvm_config):
async def get_inbox_relays(event_to_send: Event, client: Client, dvm_config):
ptags = []
for tag in event_to_send.tags():
if tag.as_vec()[0] == 'p':
@ -114,7 +126,7 @@ def get_inbox_relays(event_to_send: Event, client: Client, dvm_config):
ptags.append(ptag)
filter = Filter().kinds([EventDefinitions.KIND_RELAY_ANNOUNCEMENT]).authors(ptags)
events = client.get_events_of([filter], timedelta(dvm_config.RELAY_TIMEOUT))
events = await client.get_events_of([filter], timedelta(dvm_config.RELAY_TIMEOUT))
if len(events) == 0:
return []
else:
@ -130,7 +142,7 @@ def get_inbox_relays(event_to_send: Event, client: Client, dvm_config):
return relays
def send_event_outbox(event: Event, client, dvm_config) -> EventId:
async def send_event_outbox(event: Event, client, dvm_config) -> EventId:
# 1. OK, Let's overcomplicate things.
# 2. If our event has a relays tag, we just send the event to these relay in the classical way.
@ -149,12 +161,12 @@ def send_event_outbox(event: Event, client, dvm_config) -> EventId:
# 3. If we couldn't find relays, we look in the receivers inbox
if len(relays) == 0:
relays = get_inbox_relays(event, client, dvm_config)
relays = await get_inbox_relays(event, client, dvm_config)
# 4. If we don't find inbox relays (e.g. because the user didn't announce them, we just send to our default relays
if len(relays) == 0:
print("[" + dvm_config.NIP89.NAME + "] No Inbox found, replying to generic relays")
eventid = send_event(event, client, dvm_config)
eventid = await send_event(event, client, dvm_config)
return eventid
# 5. Otherwise, we create a new Outbox client with the inbox relays and send the event there
@ -172,22 +184,22 @@ def send_event_outbox(event: Event, client, dvm_config) -> EventId:
for relay in relays:
opts = RelayOptions().ping(False)
try:
outboxclient.add_relay_with_opts(relay, opts)
await outboxclient.add_relay_with_opts(relay, opts)
except:
print("[" + dvm_config.NIP89.NAME + "] " + relay + " couldn't be added to outbox relays")
outboxclient.connect()
await outboxclient.connect()
try:
event_id = outboxclient.send_event(event)
event_id = await outboxclient.send_event(event)
except:
event_id = send_event(event, client, dvm_config)
event_id = await send_event(event, client, dvm_config)
outboxclient.shutdown()
await outboxclient.shutdown()
return event_id
def send_event(event: Event, client: Client, dvm_config, blastr=False) -> EventId:
async def send_event(event: Event, client: Client, dvm_config, blastr=False) -> EventId:
try:
relays = []
for tag in event.tags():
@ -198,16 +210,16 @@ def send_event(event: Event, client: Client, dvm_config, blastr=False) -> EventI
for relay in relays:
if relay not in dvm_config.RELAY_LIST:
client.add_relay(relay)
await client.add_relay(relay)
#if blastr:
# client.add_relay("wss://nostr.mutinywallet.com")
event_id = client.send_event(event)
event_id = await client.send_event(event)
for relay in relays:
if relay not in dvm_config.RELAY_LIST:
client.remove_relay(relay)
await client.remove_relay(relay)
#if blastr:
# client.remove_relay("wss://nostr.mutinywallet.com")
return event_id