File size: 4,718 Bytes
bde2f3a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | """
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())
|