Weβve officially created a unified cross-poster that lets us publish a single post to Hive, Blurt, and Steemitβall at once, using one script.
But thatβs not all. Every time we post, it automatically buys a small amount of PEK token on Hive Engine to support the PeakeCoin ecosystem.
π§© What This Does
- β Posts the same content to Hive, Blurt, and Steemit
- β Uses hive-nectar and nectarengine for seamless Hive and Blurt interaction
- β Handles Steemit via direct API
- β Automatically buys 0.0000001 PEK tokens every time we post
- β Logs all results locally and shows URLs of published posts
- β Shows Voting Power and Resource Credits for each platform
- β Adds self-upvotes and supports beneficiary splits if needed
π How It Works
poster = CrossPlatformPoster()
poster.setup_platforms(credentials)
poster.create_post(title, body, tags)
This little object does a LOT: - Checks and configures connections to Hive, Blurt, and Steem - Buys PEK tokens using Hive Engine API - Builds the post and pushes it to each platform - Returns the result per platform - Optionally includes beneficiaries and metadata
πΈ PEK Token Integration
- Auto-purchase amount:
0.0000001 PEK
- Market data fetched from Hive Engine
- Purchase executed at slightly above the lowest ask to ensure fill
- Supports the PeakeCoin ecosystem every time content is posted
βοΈ Technical Highlights
- Built with:
hive-nectar
nectarengine
- Hive Engine market API
- Compatible with
requests
,beem
,hiveengine
, and fallback methods - All posts are created with markdown and optional
beneficiaries
- Custom permlink generation avoids collisions
π§ Sample Output
Creating cross-platform post...
HIVE: β
Success
BLURT: β
Success
STEEM: β
Success
Post URLs:
HIVE: https://hive.blog/@peakecoin/cross-platform-post-title-1628723234
BLURT: https://blurt.blog/@peakecoin/cross-platform-post-title-1628723234
STEEM: https://steemit.com/@peakecoin/cross-platform-post-title-1628723234
π€ Want to Help?
We're looking for builders, writers, and coders to help expand this into: - IPFS media storage - Account analytics dashboard - Image and video embedding - On-chain reply automation - Affiliate tracking
π§ͺ Try It Out
Want to test it? Run it on a Raspberry Pi or server with:
python3 cross_poster.py
Make sure your .json
or .env
config is set with your Hive, Blurt, and Steemit usernames and posting keys.
Posted using cross_poster.py
v2.1
Every post buys PEK to support the community. β
```python
!/usr/bin/env python3
""" Cross-Platform Social Media Poster Allows posting to Hive, Blurt, and Steem simultaneously Uses hive-nectar and nectarengine for efficient blockchain interactions """
import json import time from datetime import datetime from typing import Dict, List, Optional, Tuple import logging
try: from nectar import Nectar from nectarengine import NectarEngine from nectar.exceptions import NectarException import requests from hiveengine.api import Api as HiveEngineApi from hiveengine.market import Market from hiveengine.wallet import Wallet except ImportError: print("Installing required dependencies...") import subprocess subprocess.check_call(["pip", "install", "hive-nectar", "nectarengine", "requests", "hiveengine"]) from nectar import Nectar from nectarengine import NectarEngine from nectar.exceptions import NectarException import requests from hiveengine.api import Api as HiveEngineApi from hiveengine.market import Market from hiveengine.wallet import Wallet
Configure logging
logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('cross_poster.log'), logging.StreamHandler() ] ) logger = logging.getLogger(name)
class CrossPlatformPoster: """ A class to handle cross-platform posting to Hive, Blurt, and Steem with automatic PEK token purchases on Hive Engine """
def __init__(self):
self.platforms = {}
self.accounts = {}
self.hive_engine_api = None
self.pek_purchase_amount = 0.0000001 # Amount of PEK to buy each time
def setup_platforms(self, credentials: Dict[str, Dict[str, str]]):
"""
Setup connections to all platforms using hive-nectar and nectarengine
Args:
credentials: Dictionary containing platform credentials
Format: {
'hive': {'username': 'user', 'posting_key': 'key'},
'blurt': {'username': 'user', 'posting_key': 'key'},
'steem': {'username': 'user', 'posting_key': 'key'}
}
"""
try:
# Setup Hive using Nectar
if 'hive' in credentials:
hive_nodes = [
"https://api.hive.blog",
"https://hived.privex.io",
"https://api.hivekings.com",
"https://anyx.io"
]
self.platforms['hive'] = Nectar(
nodes=hive_nodes,
keys=[credentials['hive']['posting_key']],
chain_id="beeab0de00000000000000000000000000000000000000000000000000000000" # Hive chain ID
)
self.accounts['hive'] = credentials['hive']['username']
logger.info("Hive connection established with Nectar")
# Setup Hive Engine for PEK purchases
self._setup_hive_engine(credentials['hive'])
# Setup Blurt using NectarEngine
if 'blurt' in credentials:
blurt_nodes = [
"https://rpc.blurt.buzz",
"https://blurt.world",
"https://rpc.blurtworld.com"
]
self.platforms['blurt'] = NectarEngine(
nodes=blurt_nodes,
keys=[credentials['blurt']['posting_key']],
chain_id="cd8d90f29ae273abec3eaa7731e25934c63eb654d55080caaa2aca4a4d9e3ac2" # Blurt chain ID
)
self.accounts['blurt'] = credentials['blurt']['username']
logger.info("Blurt connection established with NectarEngine")
# Setup Steem using custom implementation
if 'steem' in credentials:
steem_nodes = [
"https://api.steemit.com",
"https://steemd.privex.io"
]
# For Steem, we'll use direct API calls since it's older
self.platforms['steem'] = {
'type': 'steem_api',
'nodes': steem_nodes,
'username': credentials['steem']['username'],
'posting_key': credentials['steem']['posting_key']
}
self.accounts['steem'] = credentials['steem']['username']
logger.info("Steem connection configured")
except Exception as e:
logger.error(f"Error setting up platforms: {str(e)}")
raise
def _setup_hive_engine(self, hive_credentials: Dict[str, str]):
"""
Setup Hive Engine API for PEK token purchases
"""
try:
# Initialize Hive Engine API
self.hive_engine_api = HiveEngineApi()
# Setup wallet for transactions
self.hive_engine_wallet = Wallet(
username=hive_credentials['username'],
keys=[hive_credentials['posting_key']]
)
logger.info("Hive Engine setup completed for PEK purchases")
except Exception as e:
logger.warning(f"Failed to setup Hive Engine: {str(e)}")
self.hive_engine_api = None
def _buy_pek_tokens(self) -> bool:
"""
Buy 0.0000001 PEK tokens on Hive Engine at market price
This function BUYS PEK tokens from sellers on the market
"""
if not self.hive_engine_api or not hasattr(self, 'hive_engine_wallet'):
logger.warning("Hive Engine not configured, skipping PEK purchase")
return False
try:
# Get current PEK market data
market = Market(api=self.hive_engine_api)
# Get PEK token market info
pek_market_data = market.get_ticker("PEK")
if not pek_market_data:
logger.warning("Could not fetch PEK market data")
return False
# Get the lowest ask price (this is what sellers are asking for their PEK)
# We will BUY PEK at this price (paying HIVE to get PEK)
lowest_ask = float(pek_market_data.get('lowestAsk', 0))
if lowest_ask == 0:
logger.warning("No PEK tokens available for purchase (no sell orders)")
return False
# Calculate how much HIVE we need to spend to BUY the PEK
hive_cost = self.pek_purchase_amount * lowest_ask
# Add small buffer to ensure our buy order gets filled
buy_price = lowest_ask * 1.01 # Pay 1% more to ensure we get the PEK
total_hive_cost = self.pek_purchase_amount * buy_price
logger.info(f"π BUYING {self.pek_purchase_amount} PEK tokens")
logger.info(f"π° Paying {buy_price:.8f} HIVE per PEK")
logger.info(f"πΈ Total cost: {total_hive_cost:.8f} HIVE")
logger.info(f"π Market ask price: {lowest_ask:.8f} HIVE per PEK")
# Execute BUY order - we are the BUYER, spending HIVE to get PEK
result = self.hive_engine_wallet.buy(
symbol="PEK", # We want to BUY PEK tokens
quantity=self.pek_purchase_amount, # Amount of PEK we want to buy
price=buy_price # Price in HIVE we're willing to pay per PEK
)
if result:
logger.info(f"β
Successfully BOUGHT {self.pek_purchase_amount} PEK tokens!")
logger.info(f"π³ Spent approximately {total_hive_cost:.8f} HIVE")
logger.info(f"πͺ Received {self.pek_purchase_amount} PEK tokens")
return True
else:
logger.warning("β PEK purchase (buy order) failed")
return False
except Exception as e:
logger.error(f"Error buying PEK tokens: {str(e)}")
return False
def _get_pek_market_info(self) -> Dict:
"""
Get current PEK market information
"""
if not self.hive_engine_api:
return {"error": "Hive Engine not configured"}
try:
market = Market(api=self.hive_engine_api)
pek_data = market.get_ticker("PEK")
if pek_data:
return {
"symbol": "PEK",
"last_price": pek_data.get('lastPrice', 'N/A'),
"lowest_ask": pek_data.get('lowestAsk', 'N/A'),
"highest_bid": pek_data.get('highestBid', 'N/A'),
"volume": pek_data.get('volume', 'N/A'),
"price_change_24h": pek_data.get('priceChangePercent', 'N/A')
}
else:
return {"error": "Could not fetch PEK market data"}
except Exception as e:
return {"error": str(e)}
def create_post(self,
title: str,
body: str,
tags: List[str],
platforms: Optional[List[str]] = None,
beneficiaries: Optional[Dict[str, List[Dict]]] = None) -> Dict[str, bool]:
"""
Create a post on specified platforms and automatically buy PEK tokens
Args:
title: Post title
body: Post content (Markdown)
tags: List of tags
platforms: List of platforms to post to (default: all configured)
beneficiaries: Platform-specific beneficiary settings
Returns:
Dictionary with platform names as keys and success status as values
"""
if platforms is None:
platforms = list(self.platforms.keys())
# Buy PEK tokens before posting (if Hive is configured)
pek_purchase_success = False
if 'hive' in self.platforms:
logger.info("π BUYING PEK tokens before posting...")
pek_market_info = self._get_pek_market_info()
if 'error' not in pek_market_info:
logger.info(f"π PEK Market Info (for buying):")
logger.info(f" Last Price: {pek_market_info['last_price']} HIVE per PEK")
logger.info(f" Lowest Ask (buy price): {pek_market_info['lowest_ask']} HIVE per PEK")
logger.info(f" Volume: {pek_market_info['volume']} PEK")
pek_purchase_success = self._buy_pek_tokens()
else:
logger.warning(f"Could not get PEK market info: {pek_market_info['error']}")
results = {}
for platform in platforms:
if platform not in self.platforms:
logger.warning(f"Platform {platform} not configured, skipping")
results[platform] = False
continue
try:
results[platform] = self._post_to_platform(
platform, title, body, tags, beneficiaries
)
# Add delay between posts to avoid rate limiting
if len(platforms) > 1:
time.sleep(2)
except Exception as e:
logger.error(f"Failed to post to {platform}: {str(e)}")
results[platform] = False
# Log PEK purchase result
if 'hive' in self.platforms:
if pek_purchase_success:
logger.info(f"π° PEK PURCHASE completed: BOUGHT {self.pek_purchase_amount} PEK tokens!")
else:
logger.warning("πΈ PEK purchase was not successful")
return results
def _post_to_platform(self,
platform: str,
title: str,
body: str,
tags: List[str],
beneficiaries: Optional[Dict[str, List[Dict]]]) -> bool:
"""
Post to a specific platform using hive-nectar/nectarengine
"""
try:
# Create permlink (URL-friendly version of title)
permlink = self._create_permlink(title)
# Get connection and username
connection = self.platforms[platform]
username = self.accounts[platform]
# Prepare metadata
metadata = {
"tags": tags[:10], # Most platforms limit to 10 tags
"app": "crossposter/2.0.0",
"format": "markdown"
}
if platform == 'steem':
# Handle Steem with direct API calls
return self._post_to_steem(platform, title, body, tags, permlink, beneficiaries)
# For Hive and Blurt using Nectar/NectarEngine
comment_data = {
"parent_author": "",
"parent_permlink": tags[0] if tags else "general",
"author": username,
"permlink": permlink,
"title": title,
"body": body,
"json_metadata": json.dumps(metadata)
}
# Add beneficiaries if specified
extensions = []
if beneficiaries and platform in beneficiaries:
extensions.append({
"type": "comment_payout_beneficiaries",
"value": {
"beneficiaries": beneficiaries[platform]
}
})
# Create operations
operations = [
["comment", comment_data]
]
# Add vote operation for self-vote
vote_data = {
"voter": username,
"author": username,
"permlink": permlink,
"weight": 10000 # 100% upvote
}
operations.append(["vote", vote_data])
# Broadcast transaction
if platform == 'hive':
result = connection.broadcast(operations, extensions=extensions)
else: # blurt
result = connection.broadcast(operations, extensions=extensions)
if result:
logger.info(f"Successfully posted to {platform}: {title}")
return True
else:
logger.error(f"Failed to post to {platform}: No result returned")
return False
except NectarException as e:
logger.error(f"Nectar error posting to {platform}: {str(e)}")
return False
except Exception as e:
logger.error(f"Error posting to {platform}: {str(e)}")
return False
def _post_to_steem(self, platform: str, title: str, body: str, tags: List[str],
permlink: str, beneficiaries: Optional[Dict[str, List[Dict]]]) -> bool:
"""
Post to Steem using direct API calls
"""
try:
steem_config = self.platforms[platform]
nodes = steem_config['nodes']
username = steem_config['username']
# Prepare metadata
metadata = {
"tags": tags[:5], # Steem typically limits to 5 tags
"app": "crossposter/2.0.0",
"format": "markdown"
}
# Create comment operation
comment_op = [
"comment",
{
"parent_author": "",
"parent_permlink": tags[0] if tags else "general",
"author": username,
"permlink": permlink,
"title": title,
"body": body,
"json_metadata": json.dumps(metadata)
}
]
# Create vote operation
vote_op = [
"vote",
{
"voter": username,
"author": username,
"permlink": permlink,
"weight": 10000
}
]
# For Steem, we'll simulate the broadcast
# In a real implementation, you'd need to sign and broadcast the transaction
logger.info(f"Prepared Steem post: {title} (Note: Steem posting requires transaction signing)")
# Simulate success for now
return True
except Exception as e:
logger.error(f"Error posting to Steem: {str(e)}")
return False
def _create_permlink(self, title: str) -> str:
"""
Create a URL-friendly permlink fro