git / moonbite-network / forum_bot.py · raw
"""Automatic network reports on the community forum.Posts as the verified MoonBite account (forum.OFFICIAL_NAME) into the Networkcategory, and every post is tagged "Automated report". Every number is readfrom a MoonBite node at posting time; nothing is typed in by hand, and if thenode can't be reached the run posts nothing (it never uses the explorer'sdemo sample data).Posts: daily one report per UTC day covering the previous 24 hours: blocks found, average block time, hashrate, supply, next halving, and how long a single computer can expect to wait for a block milestone a thread when the chain passes each 10,000-block height weekly Monday "Show your rig", Thursday "Hot take" (from forum_prompts.py), Saturday "Predictions" plus the real numbers on last week's threadRun it on a timer (it is idempotent; already-posted keys are skipped): */15 * * * * cd /srv/moonbite-network && .venv/bin/python forum_bot.py >> /var/log/moonbite-forum-bot.log 2>&1with the same BIGCOIN_RPC_* and MOONBITE_FORUM_DB environment as the web app.`python forum_bot.py --dry-run` prints what it would post and writes nothing."""from __future__ import annotationsimport argparseimport datetimeimport hashlibimport jsonimport osimport sysimport timeimport forumfrom live_explorer import HALVING_INTERVAL, UNITS_PER_COIN, issued_supply_unitsMILESTONE_EVERY = 10_000TARGET_BLOCK_SECONDS = 120DAY = 86_400# Example machines for the "how long will I wait" table, in H/s.EXAMPLE_RATES = (1_000, 5_000, 10_000)SITE = "https://moonbite.org"class NodeUnavailable(RuntimeError): pass# ---------------------------------------------------------------- formattingdef _num(n) -> str: return f"{n:,}"def _rate(hps: float) -> str: for unit, div in (("GH/s", 1e9), ("MH/s", 1e6), ("kH/s", 1e3)): if hps >= div: return f"{hps / div:.2f} {unit}" return f"{hps:.0f} H/s"def _duration(sec: float) -> str: sec = max(0, sec) if sec < 90: return f"{sec:.0f} seconds" if sec < 90 * 60: return f"{sec / 60:.0f} minutes" if sec < 48 * 3600: return f"{sec / 3600:.1f} hours" if sec < 365 * DAY: return f"{sec / DAY:.0f} days" return f"{sec / (365.25 * DAY):.1f} years"def expected_wait(block_seconds: float, network_hps: float, my_hps: float) -> float: """Average seconds between blocks for one miner once difficulty has retargeted to include them: the network still makes a block every `block_seconds`, and this miner's share of them is my / (network + my).""" return block_seconds * (network_hps + my_hps) / my_hps# ---------------------------------------------------------------- chain readsdef _header(rpc, height: int) -> dict: return rpc.call("getblockheader", rpc.getblockhash(height))def collect(rpc, until: int) -> dict: """Read everything a report needs: the 24 hours of blocks before `until`, plus the current tip, hashrate and supply. Header-only reads (cheap on the seed box: getblock re-verifies RandomX).""" height = int(rpc.getblockcount()) tip = _header(rpc, height) try: hashps = float(rpc.call("getnetworkhashps", 120)) except NodeUnavailable: raise except Exception: # noqa: BLE001 — reported as unknown, never guessed hashps = None since = until - DAY times, h, walked = [], height, 0 # Capped walk: a day is ~720 blocks; stop well past that whatever happens. while h > 0 and walked < 4 * DAY // TARGET_BLOCK_SECONDS: hdr = tip if h == height else _header(rpc, h) walked += 1 h -= 1 t = int(hdr["time"]) if t >= until: continue if t < since: break times.append(t) found = len(times) avg = (max(times) - min(times)) / (found - 1) if found > 1 else None return { "height": height, "tip_time": int(tip["time"]), "hashps": hashps, "blocks_24h": found, "avg_block_seconds": avg, "supply_coins": issued_supply_units(height) / UNITS_PER_COIN, }# ---------------------------------------------------------------- postsdef daily_post(stats: dict, day_label: str, max_supply: float) -> tuple[str, str]: avg = stats["avg_block_seconds"] hps = stats["hashps"] next_halving = (stats["height"] // HALVING_INTERVAL + 1) * HALVING_INTERVAL to_halving = next_halving - stats["height"] pace = avg or TARGET_BLOCK_SECONDS lines = [ f"Automatic report for the 24 hours to {day_label}. Every number here was read from a MoonBite node when this was posted, and you can check each one in the explorer: {SITE}/explorer", "", f"Blocks found: {_num(stats['blocks_24h'])} (target {_num(DAY // TARGET_BLOCK_SECONDS)})", f"Average time between blocks: {_duration(avg) if avg else 'not enough blocks to measure'}", f"Network hashrate: {_rate(hps) if hps else 'not reported by the node'} (estimated over the last 120 blocks)", f"Chain height: {_num(stats['height'])}", f"Coins issued so far: {stats['supply_coins']:,.0f} MBITE ({stats['supply_coins'] / max_supply * 100:.3f}% of the {max_supply:,.2f} cap)", f"Next halving: block {_num(next_halving)}, {_num(to_halving)} blocks away (about {_duration(to_halving * pace)} at the current pace)", ] if hps: lines += [ "", "How long would one computer wait for a block?", f"The network aims for one block every {_duration(TARGET_BLOCK_SECONDS)} and retargets difficulty every 60 blocks to get there (over the last day it averaged {_duration(pace)}). A single machine's share depends on its hashrate against everyone else's. Once difficulty has adjusted to include it, the average wait is:", "", ] lines += [f"`{_rate(r)}` about {_duration(expected_wait(TARGET_BLOCK_SECONDS, hps, r))} per block" for r in EXAMPLE_RATES] lines += [ "", "These are averages. Mining is luck, so single waits can be much longer or shorter, and they grow as more people join.", ] lines += ["", f"Start mining: {SITE}/mine"] return f"Network report: {day_label}", "\n".join(lines)def milestone_post(stats: dict, milestone: int, reached_at: int, genesis_time: int) -> tuple[str, str]: avg = (reached_at - genesis_time) / milestone if milestone else None when = time.strftime("%d %b %Y, %H:%M UTC", time.gmtime(reached_at)) issued = issued_supply_units(milestone) / UNITS_PER_COIN body = "\n".join([ f"Block {_num(milestone)} was found on {when}.", "", f"Average time per block since genesis: {_duration(avg) if avg else 'n/a'}", f"Coins issued up to this block: {issued:,.0f} MBITE", f"Next milestone: block {_num(milestone + MILESTONE_EVERY)}", "", f"See it in the explorer: {SITE}/explorer#block/{milestone}", ]) return f"Block {_num(milestone)} reached", body# ---------------------------------------------------------------- rundef _publish(key: str, title: str, body: str, dry_run: bool, out, category: str = "network") -> bool: if dry_run: print(f"--- would post [{key}] {title}\n{body}\n", file=out) return True if not forum.bot_claim(key): return False try: t = forum.create_thread(title, None, body, category, official=True, automated=True) except Exception: forum.bot_release(key) raise forum.bot_record(key, t["id"]) print(f"posted [{key}] thread {t['id']}: {title}", file=out) return Truedef run(rpc, now: int | None = None, *, max_supply: float, dry_run: bool = False, out=sys.stdout) -> list[str]: """One pass: post whatever is due and not yet posted. Returns posted keys.""" now = int(now if now is not None else time.time()) posted = [] day_start = now - now % DAY day_label = time.strftime("00:00 UTC, %d %b %Y", time.gmtime(day_start)) stats = None daily_key = "daily:" + time.strftime("%Y-%m-%d", time.gmtime(day_start)) if dry_run or (not _already(daily_key) and _due(daily_key, now, *DAILY_WINDOW)): stats = collect(rpc, day_start) title, body = daily_post(stats, day_label, max_supply) if _publish(daily_key, title, body, dry_run, out): posted.append(daily_key) height = stats["height"] if stats else int(rpc.getblockcount()) milestone = height - height % MILESTONE_EVERY if milestone > 0: key = f"milestone:{milestone}" if dry_run or not _already(key): reached = int(_header(rpc, milestone)["time"]) # Only announce recent milestones, so a first run doesn't backfill. if now - reached <= 2 * DAY: title, body = milestone_post(stats or {}, milestone, reached, int(_header(rpc, 0)["time"])) if _publish(key, title, body, dry_run, out): posted.append(key) posted += _weekly(rpc, now, stats, dry_run, out) return posted# Posting windows (UTC hours: start, length). Each post gets its own minute# inside the window, fixed by hashing its key, so posts are spread out and# land at a different time each day instead of all at the stroke of midnight.DAILY_WINDOW = (6, 3)WEEKLY_WINDOW = (9, 9)def slot(key: str, start_hour: int, span_hours: int) -> int: """Seconds after 00:00 UTC at which the post with this key becomes due.""" h = int(hashlib.sha256(key.encode()).hexdigest(), 16) return start_hour * 3600 + h % (span_hours * 3600)def _due(key: str, now: int, start_hour: int, span_hours: int) -> bool: return now % DAY >= slot(key, start_hour, span_hours)def _week(ts: int) -> str: """ISO week label in UTC, e.g. '2026-W43'.""" iso = datetime.datetime.fromtimestamp(ts, datetime.timezone.utc).isocalendar() return f"{iso[0]}-W{iso[1]:02d}"def _weekly(rpc, now: int, stats: dict | None, dry_run: bool, out) -> list[str]: """Scheduled conversation starters (UTC): Monday rig thread, Thursday hot take, Saturday predictions plus last Saturday's real numbers.""" import forum_prompts as fp posted = [] wd = time.gmtime(now).tm_wday # Monday = 0 week = _week(now) date = time.strftime("%d %b %Y", time.gmtime(now)) if wd == 0: key = f"weekly:rig:{week}" if dry_run or (not _already(key) and _due(key, now, *WEEKLY_WINDOW)): if _publish(key, f"Show your rig: week of {date}", fp.RIG_PROMPT, dry_run, out, "mining"): posted.append(key) elif wd == 3: key = f"weekly:hottake:{week}" if dry_run or (not _already(key) and _due(key, now, *WEEKLY_WINDOW)): take = fp.HOT_TAKES[int(week.split("W")[1]) % len(fp.HOT_TAKES)] title = "Hot take: " + (take if len(take) <= 100 else take[:99].rstrip() + "…") if _publish(key, title, fp.HOT_TAKE_PROMPT.format(take=take), dry_run, out, "general"): posted.append(key) elif wd == 5: key = f"weekly:predict:{week}" if dry_run or (not _already(key) and _due(key, now, *WEEKLY_WINDOW)): height = stats["height"] if stats else int(rpc.getblockcount()) try: hps = stats["hashps"] if stats else float(rpc.call("getnetworkhashps", 120)) except NodeUnavailable: raise except Exception: # noqa: BLE001 hps = None rate = _rate(hps) if hps else "not reported by the node" # Last week's thread gets the real numbers first. prev = _thread_for(f"weekly:predict:{_week(now - 7 * DAY)}") if prev and not dry_run: forum.add_reply(prev, None, fp.PREDICTION_RESULT.format(height=_num(height), rate=rate), official=True) body = fp.PREDICTION_PROMPT.format(height=_num(height), rate=rate) if _publish(key, f"Predictions: where will the chain be on {time.strftime('%d %b', time.gmtime(now + 7 * DAY))}?", body, dry_run, out, "general"): posted.append(key) return posteddef _thread_for(key: str): with forum._connect() as conn: row = conn.execute("SELECT thread_id FROM bot_state WHERE key = ?", (key,)).fetchone() return row["thread_id"] if row and row["thread_id"] else Nonedef _already(key: str) -> bool: with forum._connect() as conn: return conn.execute("SELECT 1 FROM bot_state WHERE key = ?", (key,)).fetchone() is not None# ---------------------------------------------------------------- queueQUEUE_PATH = os.environ.get("MOONBITE_FORUM_QUEUE", "").strip() or "forum_queue.json"_UNRESOLVED = "[NEEDS FOUNDER INPUT"def _parse_time(value: str) -> int: return int(datetime.datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp())def publish_queue(now: int | None = None, path: str = QUEUE_PATH, dry_run: bool = False, out=sys.stdout) -> list[str]: """Publish founder-written posts whose time has come. The queue is a JSON list of posts, each published once as the verified MoonBite account (NOT tagged automated: a person wrote them): {"key": "ama", "publish_at": "2026-10-25T10:20:00Z", "category": "general", "title": "...", "body": "...", "pin": true} {"key": "ama-2", "publish_at": "2026-10-26T15:05:00Z", "reply_to": "ama", "body": "..."} A post that still contains an unresolved [NEEDS FOUNDER INPUT ...] marker is never published.""" now = int(now if now is not None else time.time()) try: with open(path, encoding="utf-8") as fh: items = json.load(fh) except FileNotFoundError: return [] posted = [] for item in items: key = "queue:" + str(item["key"]) if _parse_time(item["publish_at"]) > now or (not dry_run and _already(key)): continue text = item.get("title", "") + item.get("body", "") if _UNRESOLVED in text: print(f"held [{key}]: still has {_UNRESOLVED} ...]", file=out) continue if dry_run: print(f"--- would post [{key}] {item.get('title') or 'reply to ' + item['reply_to']}", file=out) posted.append(key) continue parent = None if item.get("reply_to"): parent = _thread_for("queue:" + item["reply_to"]) if parent is None: continue # its thread isn't up yet; try again next run if not forum.bot_claim(key): continue try: if parent: forum.add_reply(parent, None, item["body"], official=True) thread_id = parent else: t = forum.create_thread(item["title"], None, item["body"], item.get("category", "general"), official=True) thread_id = t["id"] if item.get("pin"): forum.set_flag("t", thread_id, "pinned", True) except Exception: forum.bot_release(key) raise forum.bot_record(key, thread_id) print(f"posted [{key}] thread {thread_id}", file=out) posted.append(key) return postedclass StrictRpc: """The explorer's RpcClient without its demo fallback: any connection problem raises NodeUnavailable instead of returning sample data, because these numbers go out as public posts.""" def __init__(self, inner): self._inner = inner def call(self, method, *params): from explorer.rpc import RPCConnectionError try: return self._inner._http_call(method, list(params)) except RPCConnectionError as e: raise NodeUnavailable(f"node unreachable: {e}") from e def getblockcount(self): return self.call("getblockcount") def getblockhash(self, height): return self.call("getblockhash", height)def _rpc(): if not (os.environ.get("BIGCOIN_RPC_URL") or os.environ.get("BIGCOIN_RPC_USER") or os.environ.get("BIGCOIN_RPC_PASSWORD")): raise NodeUnavailable("no node configured (set BIGCOIN_RPC_URL or BIGCOIN_RPC_USER/PASSWORD)") here = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, os.path.join(here, "explorer")) from explorer.rpc import RpcClient return StrictRpc(RpcClient())def main(argv=None) -> int: ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) ap.add_argument("--dry-run", action="store_true", help="print posts, write nothing") ap.add_argument("--slots", action="store_true", help="print today's posting times and exit") args = ap.parse_args(argv) from params import MAX_SUPPLY, CENTS_PER_COIN try: if args.slots: now = int(time.time()) day = time.strftime("%Y-%m-%d", time.gmtime(now)) for key, win in ((f"daily:{day}", DAILY_WINDOW), (f"weekly:rig:{_week(now)}", WEEKLY_WINDOW), (f"weekly:hottake:{_week(now)}", WEEKLY_WINDOW), (f"weekly:predict:{_week(now)}", WEEKLY_WINDOW)): sec = slot(key, *win) print(f"{key:<28} {sec // 3600:02d}:{sec % 3600 // 60:02d} UTC") return 0 publish_queue(dry_run=args.dry_run) # founder posts don't need the node run(_rpc(), max_supply=MAX_SUPPLY / CENTS_PER_COIN, dry_run=args.dry_run) except NodeUnavailable as e: print(f"skipped: {e}", file=sys.stderr) return 2 return 0if __name__ == "__main__": sys.exit(main())