File size: 4,842 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 | #!/usr/bin/env python3
"""
Ingest ChainAbuse reports into RAG known_scams collection.
API: https://api.chainabuse.com/v1/reports
"""
import asyncio
import hashlib
import logging
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")
logger = logging.getLogger("ingest_chainabuse")
CHAINABUSE_API = "https://api.chainabuse.com/v1/reports"
RAG_API = "http://localhost:8000/api/v1/rag/ingest"
COLLECTION = "known_scams"
def make_doc_id(address: str) -> str:
return hashlib.sha256(f"known_scams:{address.lower()}:chainabuse".encode()).hexdigest()[:16]
async def ingest_via_api(content: str, metadata: dict) -> bool:
try:
import httpx
async with httpx.AsyncClient(timeout=15) as client:
resp = await client.post(RAG_API, json={"collection": COLLECTION, "content": content, "metadata": metadata})
return resp.status_code == 200
except Exception as e:
logger.debug(f"API ingest failed: {e}")
return False
async def main():
import httpx
# Try multiple endpoints and pagination
all_reports = []
for page in range(0, 5): # Up to 5 pages of 100
url = f"{CHAINABUSE_API}?limit=100&offset={page * 100}"
try:
logger.info(f"Fetching page {page + 1} from ChainAbuse...")
async with httpx.AsyncClient(timeout=30, headers={"Accept": "application/json"}) as client:
resp = await client.get(url)
if resp.status_code == 200:
data = resp.json()
reports = data if isinstance(data, list) else data.get("data", data.get("reports", []))
if not reports:
logger.info(f"No more reports at page {page + 1}")
break
all_reports.extend(reports)
logger.info(f"Got {len(reports)} reports (total: {len(all_reports)})")
elif resp.status_code == 401:
logger.warning("ChainAbuse API requires authentication — trying public feed")
break
else:
logger.warning(f"ChainAbuse returned {resp.status_code}")
break
except Exception as e:
logger.warning(f"ChainAbuse fetch failed: {e}")
break
if not all_reports:
# Try alternative public endpoints
alt_urls = [
"https://api.chainabuse.com/v0/reports?limit=100",
"https://api.chainabuse.com/reports?limit=100",
"https://chainabuse.com/api/reports",
]
for url in alt_urls:
try:
async with httpx.AsyncClient(timeout=15) as client:
resp = await client.get(url)
if resp.status_code == 200:
data = resp.json()
reports = data if isinstance(data, list) else data.get("data", data.get("reports", []))
if reports:
all_reports.extend(reports)
logger.info(f"Got {len(reports)} reports from alt endpoint")
break
except Exception:
pass
if not all_reports:
logger.warning("No ChainAbuse reports accessible via API. Ingestion skipped.")
logger.info("ChainAbuse may require API key — visit https://chainabuse.com for access")
return
logger.info(f"Processing {len(all_reports)} ChainAbuse reports")
ingested = 0
errors = 0
for i, report in enumerate(all_reports):
if not isinstance(report, dict):
continue
address = report.get("address", report.get("targetAddress", ""))
category = report.get("category", report.get("type", "unknown"))
description = report.get("description", report.get("report", ""))
status = report.get("status", "unknown")
reporter = report.get("reporter", "")
if not address:
continue
content = f"ChainAbuse report: {address}. Category: {category}. Status: {status}. Description: {description}. Reporter: {reporter}."
metadata = {
"source": "chainabuse",
"address": address.lower(),
"category": category,
"status": status,
"severity": "high",
}
if await ingest_via_api(content, metadata):
ingested += 1
else:
errors += 1
if (i + 1) % 50 == 0:
logger.info(f"Progress: {i + 1}/{len(all_reports)} | ingested: {ingested}")
logger.info(f"COMPLETE: ingested={ingested}, errors={errors}")
if __name__ == "__main__":
asyncio.run(main())
|