777 lines
28 KiB
Python
777 lines
28 KiB
Python
"""
|
|
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 = [
|
|
"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
|