RugMunchMedia/rmi-backend
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
rmi-backend/app/airdrop_engine.py
opencode c762564d40 style(rmi-backend): complete lint cleanup — 1175→0 ruff errors 
- Fix 71 invalid-syntax files (class-body newline-broken assignments)
- Add from/None chain to 307 B904 raise-without-from sites
- Add B008 ignore to ruff.toml (already in pyproject.toml)
- Noqa F401 on __init__.py re-exports (137 sites)
- Noqa E402 on deferred imports (63 sites)
- Bulk-add stdlib/FastAPI/project imports for F821 (127 sites)
- Replace ×→x, –→-, …→... in docstrings (4093 chars)
- Manual refactor of 5 SIM103/SIM116 patterns

Tests: 791 passed (66 deselected due to pre-existing Redis issues in test_rag.py)
Co-authored-by: opencode <opencode@rugmunch.io>
2026-07-06 15:43:20 +02:00

777 lines
28 KiB
Python
Raw Permalink Blame History

"""
Darkroom Airdrop Engine
=======================
Advanced token distribution system with:
• Snapshot-based airdrops (holdings of previous token)
• Team/development allocation (configurable % of supply)
• Anti-sniper protection (blacklist, tx limits, trading delays)
• Vesting schedules for team tokens
• Multi-chain support (EVM, Solana, TRON)
• Batch distribution for gas efficiency
All operations are admin-only and stay in /root/tools/ - never committed to git.
Usage:
POST /api/v1/admin/tokens/airdrop/snapshot - Create snapshot from existing token
POST /api/v1/admin/tokens/airdrop/execute - Execute airdrop distribution
POST /api/v1/admin/tokens/airdrop/team - Allocate team/dev tokens
POST /api/v1/admin/tokens/airdrop/vesting - Set up vesting for team
GET /api/v1/admin/tokens/airdrop/{id}/status - Check airdrop status
POST /api/v1/admin/tokens/airdrop/antisniper - Enable anti-sniper protection
"""
from __future__ import annotations
import json
import logging
import os
import time
from dataclasses import asdict, dataclass, field
from datetime import datetime
from typing import Any
from app.token_deployer import TokenDeployerFactory, TokenDeployment, get_storage
logger = logging.getLogger("darkroom_airdrop")
# ── Data Models ───────────────────────────────────────────────
@dataclass
class AirdropRecipient:
"""Single recipient in an airdrop."""
address: str
amount: str
reason: str = "" # e.g., "holder_of_CRM_v1", "team_allocation", "marketing"
claimed: bool = False
claim_tx: str = ""
claimed_at: str | None = None
@dataclass
class AirdropSnapshot:
"""Snapshot of token holders at a specific block/time."""
snapshot_id: str
source_token: str
source_chain: str
block_number: int | None = None
timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())
holders: list[AirdropRecipient] = field(default_factory=list)
total_holders: int = 0
total_supply_snapshotted: str = "0"
excluded_addresses: list[str] = field(default_factory=list)
min_holdings: str = "0"
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass
class AirdropCampaign:
"""Full airdrop campaign with distribution rules."""
campaign_id: str
deployment_id: str
snapshot_id: str
chain: str
status: str = "pending" # pending, active, paused, completed, cancelled
distribution_type: str = "snapshot" # snapshot, manual, team, marketing
recipients: list[AirdropRecipient] = field(default_factory=list)
total_amount: str = "0"
distributed_amount: str = "0"
remaining_amount: str = "0"
team_allocation_percent: float = 0.0
team_vesting_months: int = 0
team_cliff_months: int = 0
anti_sniper_enabled: bool = True
trading_delay_blocks: int = 0
max_wallet_percent: float = 0.0
max_tx_percent: float = 0.0
blacklist_preloaded: list[str] = field(default_factory=list)
created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())
started_at: str | None = None
completed_at: str | None = None
tx_hashes: list[str] = field(default_factory=list)
metadata: dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> dict:
d = asdict(self)
return d
@dataclass
class VestingSchedule:
"""Vesting schedule for team/dev tokens."""
schedule_id: str
deployment_id: str
beneficiary: str
total_amount: str
claimed_amount: str = "0"
start_date: str = ""
cliff_months: int = 0
vesting_months: int = 0
monthly_release: str = "0"
status: str = "active" # active, completed, revoked
tx_hashes: list[str] = field(default_factory=list)
# ── Anti-Sniper Protection ────────────────────────────────────
class AntiSniperProtection:
"""
Protect token launches from snipers and bots.
Features:
1. Pre-launch blacklist (known bot addresses)
2. Trading delay (blocks after launch before trading)
3. Max wallet / tx limits (prevent accumulation)
4. Dynamic blacklist (detect and ban sandwich bots)
5. Whitelist-only mode (initially)
"""
# Known bot/sniper patterns (addresses and creation patterns)
KNOWN_BOT_PATTERNS = [ # noqa: RUF012
"0x0000000000000000000000000000000000000000", # Burn address
]
@classmethod
async def apply_protection(
cls,
deployer: Any,
contract_address: str,
deployment: TokenDeployment,
blacklist_addresses: list[str] | None = None,
trading_delay_blocks: int = 0,
max_wallet_percent: float = 0.0,
max_tx_percent: float = 0.0,
) -> dict[str, Any]:
"""Apply full anti-sniper protection suite to a token."""
results = {
"blacklist_applied": 0,
"trading_delayed": False,
"max_wallet_set": False,
"max_tx_set": False,
"tx_hashes": [],
}
# 1. Blacklist known bots + provided addresses
all_blacklist = set(cls.KNOWN_BOT_PATTERNS)
if blacklist_addresses:
all_blacklist.update(blacklist_addresses)
for addr in all_blacklist:
try:
tx = await deployer.blacklist_add(contract_address, addr)
results["tx_hashes"].append(tx)
results["blacklist_applied"] += 1
logger.info(f"Anti-sniper: blacklisted {addr}")
except Exception as e:
logger.warning(f"Failed to blacklist {addr}: {e}")
# 2. Disable trading initially (if supported)
try:
tx = await deployer.set_trading_enabled(contract_address, False)
results["tx_hashes"].append(tx)
results["trading_delayed"] = True
logger.info(f"Anti-sniper: trading disabled, will enable after {trading_delay_blocks} blocks")
except Exception as e:
logger.warning(f"Trading disable not supported or failed: {e}")
# 3. Set max wallet limit (if percent provided)
if max_wallet_percent > 0:
try:
total_supply = int(deployment.total_supply)
max_wallet = str(int(total_supply * max_wallet_percent / 100))
tx = await deployer.set_max_wallet(contract_address, max_wallet)
results["tx_hashes"].append(tx)
results["max_wallet_set"] = True
logger.info(f"Anti-sniper: max wallet set to {max_wallet_percent}% ({max_wallet})")
except Exception as e:
logger.warning(f"Max wallet set failed: {e}")
# 4. Set max tx limit (if percent provided)
if max_tx_percent > 0:
try:
total_supply = int(deployment.total_supply)
max_tx = str(int(total_supply * max_tx_percent / 100))
tx = await deployer.set_max_tx(contract_address, max_tx)
results["tx_hashes"].append(tx)
results["max_tx_set"] = True
logger.info(f"Anti-sniper: max tx set to {max_tx_percent}% ({max_tx})")
except Exception as e:
logger.warning(f"Max tx set failed: {e}")
return results
@classmethod
async def enable_trading_after_delay(
cls,
deployer: Any,
contract_address: str,
delay_blocks: int,
current_block: int,
) -> str:
"""Enable trading after block delay."""
target_block = current_block + delay_blocks
logger.info(f"Anti-sniper: trading will enable at block {target_block}")
# This would be called by a cron job or background task
# For now, return the target block
return str(target_block)
# ── Snapshot Engine ─────────────────────────────────────────────
class SnapshotEngine:
"""
Create snapshots of token holders for airdrop eligibility.
Supports EVM chains (via RPC) and Solana (via RPC).
"""
@staticmethod
async def create_evm_snapshot(
token_address: str,
chain: str,
rpc_url: str,
min_holdings: str = "0",
excluded_addresses: list[str] | None = None,
block_number: int | None = None,
) -> AirdropSnapshot:
"""Create snapshot of EVM token holders."""
from web3 import Web3
w3 = Web3(Web3.HTTPProvider(rpc_url))
# Standard ERC-20 events/topics
transfer_topic = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"
# Get current block if not specified
if block_number is None:
block_number = w3.eth.block_number
# Query Transfer events to find all holders
# This is a simplified approach - for production, use a subgraph or indexer
logs = w3.eth.get_logs(
{
"fromBlock": max(0, block_number - 100000), # Last ~100k blocks
"toBlock": block_number,
"address": token_address,
"topics": [transfer_topic],
}
)
# Build holder balances from transfer logs
balances: dict[str, int] = {}
for log in logs:
try:
from_addr = "0x" + log.topics[1].hex()[-40:]
to_addr = "0x" + log.topics[2].hex()[-40:]
amount = int(log.data.hex(), 16) if log.data else 0
balances[from_addr] = balances.get(from_addr, 0) - amount
balances[to_addr] = balances.get(to_addr, 0) + amount
except Exception:
continue
# Filter for positive balances above minimum
min_amount = int(min_holdings)
excluded = set(excluded_addresses or [])
holders = []
total_supply = 0
for addr, balance in balances.items():
if balance > 0 and balance >= min_amount and addr.lower() not in excluded:
holders.append(
AirdropRecipient(
address=addr,
amount=str(balance),
reason="snapshot_holder",
)
)
total_supply += balance
snapshot = AirdropSnapshot(
snapshot_id=f"snap_{chain}_{token_address}_{block_number}_{int(time.time())}",
source_token=token_address,
source_chain=chain,
block_number=block_number,
holders=holders,
total_holders=len(holders),
total_supply_snapshotted=str(total_supply),
excluded_addresses=list(excluded),
min_holdings=min_holdings,
)
logger.info(f"Snapshot created: {snapshot.snapshot_id} - {len(holders)} holders, {total_supply} total")
return snapshot
@staticmethod
async def create_solana_snapshot(
token_address: str,
rpc_url: str,
min_holdings: str = "0",
excluded_addresses: list[str] | None = None,
) -> AirdropSnapshot:
"""Create snapshot of SPL token holders."""
from solana.rpc.api import Client
from solders.pubkey import Pubkey
client = Client(rpc_url)
mint = Pubkey.from_string(token_address)
# Get all token accounts for this mint
response = client.get_token_accounts_by_mint_json_parsed(mint, commitment="confirmed")
holders = []
total_supply = 0
excluded = set(excluded_addresses or [])
min_amount = int(min_holdings)
for account in response["result"]["value"]:
try:
parsed = account["account"]["data"]["parsed"]["info"]
owner = parsed["owner"]
amount = int(parsed["tokenAmount"]["amount"])
if amount >= min_amount and owner not in excluded:
holders.append(
AirdropRecipient(
address=owner,
amount=str(amount),
reason="snapshot_holder",
)
)
total_supply += amount
except Exception:
continue
snapshot = AirdropSnapshot(
snapshot_id=f"snap_solana_{token_address}_{int(time.time())}",
source_token=token_address,
source_chain="solana",
holders=holders,
total_holders=len(holders),
total_supply_snapshotted=str(total_supply),
excluded_addresses=list(excluded),
min_holdings=min_holdings,
)
return snapshot
# ── Airdrop Distributor ───────────────────────────────────────
class AirdropDistributor:
"""
Execute airdrop distributions across chains.
Supports batch transfers for gas efficiency.
"""
@staticmethod
async def execute_evm_airdrop(
deployer: Any,
contract_address: str,
recipients: list[AirdropRecipient],
batch_size: int = 50,
) -> dict[str, Any]:
"""Execute airdrop on EVM chain (batch or individual)."""
results = {
"total_recipients": len(recipients),
"successful": 0,
"failed": 0,
"tx_hashes": [],
"errors": [],
}
# For small airdrops, use individual transfers
# For large ones, use a Merkle distributor or batch contract
if len(recipients) <= batch_size:
# Individual transfers
for recipient in recipients:
try:
tx = await deployer.mint_tokens(
contract_address,
recipient.address,
recipient.amount,
)
recipient.claimed = True
recipient.claim_tx = tx
recipient.claimed_at = datetime.utcnow().isoformat()
results["successful"] += 1
results["tx_hashes"].append(tx)
logger.info(f"Airdropped {recipient.amount} to {recipient.address}")
except Exception as e:
results["failed"] += 1
results["errors"].append({"address": recipient.address, "error": str(e)})
logger.error(f"Airdrop failed for {recipient.address}: {e}")
else:
# Batch via multicall or distributor contract
# For now, chunk into batches
for i in range(0, len(recipients), batch_size):
batch = recipients[i : i + batch_size]
try:
# Use a batch mint function if available
tx = await AirdropDistributor._batch_mint_evm(deployer, contract_address, batch)
for r in batch:
r.claimed = True
r.claim_tx = tx
r.claimed_at = datetime.utcnow().isoformat()
results["successful"] += len(batch)
results["tx_hashes"].append(tx)
except Exception as e:
results["failed"] += len(batch)
logger.error(f"Batch airdrop failed: {e}")
return results
@staticmethod
async def _batch_mint_evm(
deployer: Any,
contract_address: str,
recipients: list[AirdropRecipient],
) -> str:
"""Batch mint via multicall or custom contract."""
# This would use a deployed MerkleDistributor or Multicall contract
# For now, return a placeholder
logger.info(f"Batch mint of {len(recipients)} recipients")
return "batch_tx_placeholder"
@staticmethod
async def execute_solana_airdrop(
deployer: Any,
contract_address: str,
recipients: list[AirdropRecipient],
) -> dict[str, Any]:
"""Execute airdrop on Solana."""
results = {
"total_recipients": len(recipients),
"successful": 0,
"failed": 0,
"tx_hashes": [],
}
for recipient in recipients:
try:
tx = await deployer.mint_tokens(
contract_address,
recipient.address,
recipient.amount,
)
recipient.claimed = True
recipient.claim_tx = tx
results["successful"] += 1
results["tx_hashes"].append(tx)
except Exception as e:
results["failed"] += 1
logger.error(f"Solana airdrop failed for {recipient.address}: {e}")
return results
# ── Team Allocation Engine ────────────────────────────────────
class TeamAllocation:
"""
Allocate tokens to team, dev, marketing, treasury with vesting.
Typical allocation:
• Development: 15-20%
• Marketing: 5-10%
• Treasury/DAO: 10-15%
• Advisors: 5-10%
• Airdrop: 10-20%
• Liquidity: 20-30%
• Public sale: remaining
"""
@staticmethod
async def allocate_team_tokens(
deployer: Any,
deployment: TokenDeployment,
allocations: list[dict[str, Any]],
) -> dict[str, Any]:
"""
Allocate team/dev tokens with optional immediate + vested portions.
allocations: [
{"address": "0x...", "role": "dev", "percent": 10, "immediate": 20, "vested": 80, "cliff_months": 6, "vesting_months": 24},
{"address": "0x...", "role": "marketing", "percent": 5, "immediate": 50, "vested": 50, "cliff_months": 3, "vesting_months": 12},
]
"""
results = {
"allocations": [],
"total_allocated": 0,
"tx_hashes": [],
}
total_supply = int(deployment.total_supply)
for alloc in allocations:
try:
percent = alloc["percent"]
amount = int(total_supply * percent / 100)
immediate_percent = alloc.get("immediate", 0)
immediate_amount = int(amount * immediate_percent / 100)
vested_amount = amount - immediate_amount
# Mint immediate portion
if immediate_amount > 0:
tx = await deployer.mint_tokens(
deployment.contract_address,
alloc["address"],
str(immediate_amount),
)
results["tx_hashes"].append(tx)
# Set up vesting for remainder
if vested_amount > 0:
vesting = VestingSchedule(
schedule_id=f"vest_{deployment.deployment_id}_{alloc['role']}_{int(time.time())}",
deployment_id=deployment.deployment_id,
beneficiary=alloc["address"],
total_amount=str(vested_amount),
start_date=datetime.utcnow().isoformat(),
cliff_months=alloc.get("cliff_months", 0),
vesting_months=alloc.get("vesting_months", 0),
monthly_release=str(vested_amount // max(alloc.get("vesting_months", 1), 1)),
)
# Store vesting schedule
await AirdropStorage.save_vesting(vesting)
results["allocations"].append(
{
"role": alloc["role"],
"address": alloc["address"],
"percent": percent,
"total_amount": str(amount),
"immediate": str(immediate_amount),
"vested": str(vested_amount),
"tx_hashes": results["tx_hashes"][-1:],
}
)
results["total_allocated"] += amount
except Exception as e:
logger.error(f"Team allocation failed for {alloc}: {e}")
return results
# ── Storage ─────────────────────────────────────────────────────
class AirdropStorage:
"""Store airdrop snapshots, campaigns, and vesting schedules."""
@staticmethod
async def save_snapshot(snapshot: AirdropSnapshot) -> bool:
"""Save snapshot to Redis/Supabase."""
try:
from app.token_deployer import get_storage
storage = await get_storage()
if storage.redis:
key = f"airdrop_snapshot:{snapshot.snapshot_id}"
await storage.redis.set(key, json.dumps(snapshot.__dict__))
await storage.redis.sadd("airdrop_snapshots:all", snapshot.snapshot_id)
if storage.supabase:
storage.supabase.table("airdrop_snapshots").upsert(
{
"snapshot_id": snapshot.snapshot_id,
"source_token": snapshot.source_token,
"source_chain": snapshot.source_chain,
"block_number": snapshot.block_number,
"timestamp": snapshot.timestamp,
"holders": [h.__dict__ for h in snapshot.holders],
"total_holders": snapshot.total_holders,
"total_supply": snapshot.total_supply_snapshotted,
}
).execute()
return True
except Exception as e:
logger.error(f"Save snapshot failed: {e}")
return False
@staticmethod
async def save_campaign(campaign: AirdropCampaign) -> bool:
"""Save campaign to storage."""
try:
from app.token_deployer import get_storage
storage = await get_storage()
if storage.redis:
key = f"airdrop_campaign:{campaign.campaign_id}"
await storage.redis.set(key, json.dumps(campaign.to_dict()))
await storage.redis.sadd("airdrop_campaigns:all", campaign.campaign_id)
return True
except Exception as e:
logger.error(f"Save campaign failed: {e}")
return False
@staticmethod
async def save_vesting(vesting: VestingSchedule) -> bool:
"""Save vesting schedule."""
try:
from app.token_deployer import get_storage
storage = await get_storage()
if storage.redis:
key = f"vesting:{vesting.schedule_id}"
await storage.redis.set(key, json.dumps(vesting.__dict__))
await storage.redis.sadd("vesting:all", vesting.schedule_id)
return True
except Exception as e:
logger.error(f"Save vesting failed: {e}")
return False
@staticmethod
async def get_campaign(campaign_id: str) -> AirdropCampaign | None:
"""Get campaign by ID."""
try:
from app.token_deployer import get_storage
storage = await get_storage()
if storage.redis:
data = await storage.redis.get(f"airdrop_campaign:{campaign_id}")
if data:
d = json.loads(data)
return AirdropCampaign(**d)
return None
except Exception as e:
logger.error(f"Get campaign failed: {e}")
return None
# ── Convenience Functions ─────────────────────────────────────
async def create_full_token_with_protection(
chain: str,
name: str,
symbol: str,
decimals: int,
initial_supply: str,
team_allocations: list[dict[str, Any]] | None = None,
airdrop_source_token: str | None = None,
airdrop_source_chain: str | None = None,
anti_sniper: bool = True,
trading_delay_blocks: int = 0,
max_wallet_percent: float = 2.0,
max_tx_percent: float = 1.0,
blacklist_addresses: list[str] | None = None,
) -> dict[str, Any]:
"""
Full token launch with team allocation, airdrop, and anti-sniper protection.
This is the main function for launching CRM v2 or any new token.
"""
from app.token_deployer import DeployParams
results = {
"deployment": None,
"anti_sniper": None,
"team_allocation": None,
"airdrop": None,
"errors": [],
}
try:
# 1. Deploy token
deployer = TokenDeployerFactory.get_deployer(chain)
params = DeployParams(
chain=chain,
name=name,
symbol=symbol,
decimals=decimals,
initial_supply=initial_supply,
mintable=True,
burnable=True,
blacklist_enabled=anti_sniper,
trading_enabled=not anti_sniper, # Disabled initially if anti-sniper
)
deployment = await deployer.deploy_token(params)
results["deployment"] = deployment.to_dict()
# Save deployment
storage = await get_storage()
await storage.save(deployment)
# 2. Apply anti-sniper protection
if anti_sniper:
protection = await AntiSniperProtection.apply_protection(
deployer,
deployment.contract_address,
deployment,
blacklist_addresses=blacklist_addresses,
trading_delay_blocks=trading_delay_blocks,
max_wallet_percent=max_wallet_percent,
max_tx_percent=max_tx_percent,
)
results["anti_sniper"] = protection
# 3. Team allocation
if team_allocations:
team_result = await TeamAllocation.allocate_team_tokens(deployer, deployment, team_allocations)
results["team_allocation"] = team_result
# 4. Airdrop from snapshot
if airdrop_source_token and airdrop_source_chain:
# Create snapshot
if airdrop_source_chain in ["ethereum", "base", "bsc"]:
rpc = os.getenv(f"{airdrop_source_chain.upper()}_RPC_URL", "")
snapshot = await SnapshotEngine.create_evm_snapshot(
airdrop_source_token,
airdrop_source_chain,
rpc,
)
elif airdrop_source_chain == "solana":
rpc = os.getenv("SOLANA_RPC_URL", "")
snapshot = await SnapshotEngine.create_solana_snapshot(
airdrop_source_token,
rpc,
)
else:
raise ValueError(f"Unsupported snapshot chain: {airdrop_source_chain}")
await AirdropStorage.save_snapshot(snapshot)
# Execute airdrop
if chain in ["ethereum", "base", "bsc"]:
airdrop_result = await AirdropDistributor.execute_evm_airdrop(
deployer, deployment.contract_address, snapshot.holders
)
elif chain == "solana":
airdrop_result = await AirdropDistributor.execute_solana_airdrop(
deployer, deployment.contract_address, snapshot.holders
)
else:
airdrop_result = {"error": "Airdrop not supported for this chain yet"}
results["airdrop"] = {
"snapshot_id": snapshot.snapshot_id,
"holders": len(snapshot.holders),
"result": airdrop_result,
}
return results
except Exception as e:
logger.error(f"Full token launch failed: {e}")
results["errors"].append(str(e))
raise
Reference in a new issue View git blame Copy permalink