All source code single file
This commit is contained in:
Executable
+191
@@ -0,0 +1,191 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Send Podcasting 2.0 boostagrams via AlbyHub."""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
import urllib.request
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
PODCAST_NS = "https://podcastindex.org/namespace/1.0"
|
||||
ALBY_CLI = ["npx", "@getalby/cli"]
|
||||
|
||||
|
||||
def fetch_feed(url):
|
||||
req = urllib.request.Request(url, headers={"User-Agent": "boostagram/1.0"})
|
||||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||||
return ET.fromstring(resp.read())
|
||||
|
||||
|
||||
def find_value_block(root, episode_query):
|
||||
channel = root.find("channel")
|
||||
if channel is None:
|
||||
return None, None, None
|
||||
|
||||
channel_title = channel.findtext("title", "Unknown Podcast")
|
||||
items = channel.findall("item")
|
||||
if not items:
|
||||
vb = channel.find(f"{{{PODCAST_NS}}}value")
|
||||
return vb, channel_title, None
|
||||
|
||||
if episode_query:
|
||||
match = None
|
||||
for item in items:
|
||||
title = item.findtext("title", "")
|
||||
if episode_query.lower() in title.lower():
|
||||
match = item
|
||||
break
|
||||
if match:
|
||||
vb = match.find(f"{{{PODCAST_NS}}}value")
|
||||
if vb is not None:
|
||||
return vb, channel_title, match
|
||||
print(f"warning: no episode matching '{episode_query}', using latest",
|
||||
file=sys.stderr)
|
||||
|
||||
latest = items[0]
|
||||
vb = latest.find(f"{{{PODCAST_NS}}}value")
|
||||
if vb is not None:
|
||||
return vb, channel_title, latest
|
||||
|
||||
vb = channel.find(f"{{{PODCAST_NS}}}value")
|
||||
return vb, channel_title, latest
|
||||
|
||||
|
||||
def parse_recipients(value_block):
|
||||
recipients = []
|
||||
for r in value_block.findall(f"{{{PODCAST_NS}}}valueRecipient"):
|
||||
addr = r.get("address", "")
|
||||
if not addr:
|
||||
continue
|
||||
recipients.append({
|
||||
"name": r.get("name", "Unknown"),
|
||||
"type": r.get("type", "node"),
|
||||
"address": addr,
|
||||
"split": int(r.get("split", "1")),
|
||||
"customKey": r.get("customKey"),
|
||||
"customValue": r.get("customValue"),
|
||||
})
|
||||
return recipients
|
||||
|
||||
|
||||
def build_boostagram(feed_url, channel_title, episode_item, recipient,
|
||||
amount_sats, message, sender, ts, app_name):
|
||||
msats = amount_sats * 1000
|
||||
boost = {
|
||||
"app_name": app_name,
|
||||
"action": "boost",
|
||||
"value_msat_total": msats,
|
||||
"value_msat": msats,
|
||||
"podcast": channel_title,
|
||||
"url": feed_url,
|
||||
"name": recipient["name"],
|
||||
"sender_name": sender,
|
||||
}
|
||||
if message:
|
||||
boost["message"] = message
|
||||
if ts is not None:
|
||||
boost["ts"] = ts
|
||||
if episode_item is not None:
|
||||
title = episode_item.findtext("title")
|
||||
if title:
|
||||
boost["episode"] = title
|
||||
guid = episode_item.findtext(f"{{{PODCAST_NS}}}guid")
|
||||
if guid:
|
||||
boost["episode_guid"] = guid
|
||||
elif episode_item.findtext("guid"):
|
||||
boost["episode_guid"] = episode_item.findtext("guid")
|
||||
return boost
|
||||
|
||||
|
||||
def send_keysend(pubkey, sats, tlv_records):
|
||||
cmd = ALBY_CLI + [
|
||||
"pay-keysend",
|
||||
"-p", pubkey,
|
||||
"--amount", str(sats),
|
||||
"--currency", "BTC",
|
||||
"--unit", "sats",
|
||||
"--network", "lightning",
|
||||
"--tlv-records", json.dumps(tlv_records),
|
||||
]
|
||||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||||
if result.returncode != 0:
|
||||
err = result.stderr.strip()
|
||||
try:
|
||||
err = json.loads(err).get("error", err)
|
||||
except (json.JSONDecodeError, AttributeError):
|
||||
pass
|
||||
raise RuntimeError(err)
|
||||
try:
|
||||
return json.loads(result.stdout)
|
||||
except json.JSONDecodeError:
|
||||
return {"raw": result.stdout}
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Send a Podcasting 2.0 boostagram")
|
||||
parser.add_argument("--feed", required=True, help="Podcast RSS feed URL")
|
||||
parser.add_argument("--amount", required=True, type=int, help="Total sats to boost")
|
||||
parser.add_argument("--message", default="", help="Boostagram message")
|
||||
parser.add_argument("--sender", required=True, help="Your display name")
|
||||
parser.add_argument("--episode", default=None, help="Episode title (default: latest)")
|
||||
parser.add_argument("--ts", type=int, default=None, help="Playback position in seconds")
|
||||
parser.add_argument("--app-name", default="boostagram", help="App name for boostagram")
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.amount <= 0:
|
||||
print("error: --amount must be positive", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
root = fetch_feed(args.feed)
|
||||
value_block, channel_title, episode_item = find_value_block(root, args.episode)
|
||||
|
||||
if value_block is None:
|
||||
print("error: no <podcast:value> block found in feed", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
recipients = parse_recipients(value_block)
|
||||
if not recipients:
|
||||
print("error: no recipients found in value block", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
total_shares = sum(r["split"] for r in recipients)
|
||||
if total_shares == 0:
|
||||
print("error: total splits sum to zero", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
episode_title = episode_item.findtext("title", "latest") if episode_item is not None else "latest"
|
||||
print(f"boosting {channel_title} - {episode_title}")
|
||||
print(f"total: {args.amount} sats across {len(recipients)} recipients\n")
|
||||
|
||||
ok = 0
|
||||
for r in recipients:
|
||||
split_sats = max(1, args.amount * r["split"] // total_shares)
|
||||
boostagram = build_boostagram(
|
||||
args.feed, channel_title, episode_item, r,
|
||||
split_sats, args.message, args.sender, args.ts, args.app_name,
|
||||
)
|
||||
boost_json = json.dumps(boostagram, ensure_ascii=False)
|
||||
boost_hex = boost_json.encode("utf-8").hex()
|
||||
|
||||
tlv = [{"type": 7629169, "value": boost_hex}]
|
||||
|
||||
if r["customKey"]:
|
||||
custom_hex = r["customValue"].encode("utf-8").hex() if r["customValue"] else ""
|
||||
tlv.append({"type": int(r["customKey"]), "value": custom_hex})
|
||||
|
||||
print(f" {r['name']}: {split_sats} sats")
|
||||
try:
|
||||
result = send_keysend(r["address"], split_sats, tlv)
|
||||
preimage = result.get("preimage", "?")
|
||||
print(f" -> ok (preimage: {preimage[:16]}...)")
|
||||
ok += 1
|
||||
except RuntimeError as e:
|
||||
print(f" -> failed: {e}")
|
||||
|
||||
print(f"\n{ok}/{len(recipients)} payments succeeded")
|
||||
sys.exit(0 if ok > 0 else 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user