139 lines
4.6 KiB
Python
139 lines
4.6 KiB
Python
"""
|
|
RMI Community Forensics Worker
|
|
Background processing for:
|
|
- Report verification pipeline
|
|
- Graph generation for forensic reports
|
|
- Cross-chain correlation batch jobs
|
|
- Community reputation updates
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
from datetime import UTC, datetime
|
|
|
|
import httpx
|
|
|
|
from app.core.logging import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
# Config
|
|
REDIS_HOST = os.getenv("REDIS_HOST", "rmi-redis")
|
|
REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
|
|
REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", "")
|
|
REDIS_DB = int(os.getenv("REDIS_DB", 0))
|
|
BLOCKSCOUT_URL = os.getenv("BLOCKSCOUT_URL", "http://rmi-blockscout-backend:4000")
|
|
SUPABASE_URL = os.getenv("SUPABASE_URL", "")
|
|
SUPABASE_KEY = os.getenv("SUPABASE_KEY", "")
|
|
|
|
|
|
class CommunityForensicsWorker:
|
|
"""Background worker for community forensics tasks."""
|
|
|
|
def __init__(self):
|
|
self.blockscout = httpx.AsyncClient(base_url=BLOCKSCOUT_URL, timeout=30.0)
|
|
self.running = True
|
|
|
|
async def verify_report_graph(self, report_id: str, graph_data: dict) -> dict:
|
|
"""Verify and enrich a community report's graph data."""
|
|
# Validate graph structure
|
|
nodes = graph_data.get("nodes", [])
|
|
edges = graph_data.get("edges", [])
|
|
|
|
# Check for known scam addresses via Blockscout
|
|
enriched_nodes = []
|
|
for node in nodes:
|
|
addr = node.get("id", "")
|
|
if len(addr) >= 20: # Likely an address
|
|
try:
|
|
resp = await self.blockscout.get(f"/api/v2/addresses/{addr}")
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
node["verified_name"] = data.get("name", "")
|
|
node["tx_count"] = data.get("transactions_count", 0)
|
|
except Exception:
|
|
pass
|
|
enriched_nodes.append(node)
|
|
|
|
return {
|
|
"report_id": report_id,
|
|
"nodes_verified": len(enriched_nodes),
|
|
"edges_count": len(edges),
|
|
"enriched": True,
|
|
"timestamp": datetime.now(UTC).isoformat(),
|
|
}
|
|
|
|
async def batch_cross_chain_correlate(self, addresses: list[str]) -> list[dict]:
|
|
"""Batch cross-chain correlation for community investigations."""
|
|
results = []
|
|
|
|
for addr in addresses:
|
|
# Check multiple chains via Blockscout instances
|
|
chains = ["eth", "base", "bsc", "arbitrum"]
|
|
chain_hits = []
|
|
|
|
for chain in chains:
|
|
try:
|
|
# In production, each chain has its own Blockscout
|
|
# For now, query the main one
|
|
resp = await self.blockscout.get(f"/api/v2/addresses/{addr}")
|
|
if resp.status_code == 200:
|
|
chain_hits.append(chain)
|
|
except Exception:
|
|
pass
|
|
|
|
results.append(
|
|
{
|
|
"address": addr,
|
|
"chains_found": chain_hits,
|
|
"multi_chain": len(chain_hits) > 1,
|
|
}
|
|
)
|
|
|
|
return results
|
|
|
|
async def generate_leaderboard_snapshot(self) -> dict:
|
|
"""Generate daily leaderboard snapshot."""
|
|
# In production, query Supabase for real data
|
|
return {
|
|
"timestamp": datetime.now(UTC).isoformat(),
|
|
"period": "daily",
|
|
"top_sleuths": [],
|
|
}
|
|
|
|
async def run(self):
|
|
"""Main worker loop."""
|
|
logger.info("[OK] Community Forensics Worker started")
|
|
while self.running:
|
|
try:
|
|
# In production, read from Redis queue
|
|
# For now, health check loop
|
|
await asyncio.sleep(60)
|
|
|
|
# Health check Blockscout
|
|
try:
|
|
resp = await self.blockscout.get("/api/v2/main-page/indexing-status")
|
|
if resp.status_code == 200:
|
|
logger.info(f"[OK] Blockscout healthy: {resp.json().get('finished_indexing_blocks', False)}")
|
|
except Exception as e:
|
|
logger.warning(f"[WARN] Blockscout health check failed: {e}")
|
|
except Exception as e:
|
|
logger.warning(f"[ERROR] Worker loop: {e}")
|
|
await asyncio.sleep(5)
|
|
|
|
async def shutdown(self):
|
|
"""Graceful shutdown."""
|
|
self.running = False
|
|
await self.blockscout.aclose()
|
|
logger.info("[OK] Community Forensics Worker stopped")
|
|
async def main():
|
|
worker = CommunityForensicsWorker()
|
|
|
|
try:
|
|
await worker.run()
|
|
except KeyboardInterrupt:
|
|
await worker.shutdown()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|