File size: 4,556 Bytes
6993919 | 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 | #!/usr/bin/env python3
"""
Ingest CryptoScamDB entries into RAG known_scams collection.
API: https://api.cryptoscamdb.org/v1/scams
"""
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_cryptoscamdb")
API_URLS = [
"https://api.cryptoscamdb.org/v1/scams",
"https://api.cryptoscamdb.org/v1/all",
]
RAG_API = "http://localhost:8000/api/v1/rag/ingest"
COLLECTION = "known_scams"
def make_doc_id(address: str, source: str) -> str:
return hashlib.sha256(f"known_scams:{address.lower()}:{source}".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 each API URL
data = None
for url in API_URLS:
try:
logger.info(f"Fetching from {url}...")
async with httpx.AsyncClient(timeout=30) as client:
resp = await client.get(url)
if resp.status_code == 200:
data = resp.json()
logger.info(f"Got response from {url}")
break
else:
logger.warning(f"{url} returned {resp.status_code}")
except Exception as e:
logger.warning(f"Failed to fetch {url}: {e}")
if not data:
logger.error("All CryptoScamDB API endpoints failed")
return
# Extract scam entries (API returns nested structure)
scams = []
if isinstance(data, dict):
scams = data.get("result", data.get("scams", data.get("data", [])))
elif isinstance(data, list):
scams = data
if not scams:
logger.error(
f"No scam entries found in response. Keys: {list(data.keys()) if isinstance(data, dict) else 'list'}"
)
return
logger.info(f"Found {len(scams)} scam entries")
ingested = 0
errors = 0
for i, scam in enumerate(scams):
if not isinstance(scam, dict):
continue
name = scam.get("name", "")
category = scam.get("category", "unknown")
status = scam.get("status", "unknown")
description = scam.get("description", "")
url = scam.get("url", "")
addresses = scam.get("addresses", [])
if not addresses:
# Create one doc for URL-based scam
content = f"CryptoScamDB entry: {name}. Category: {category}. Status: {status}. URL: {url}. Description: {description}"
metadata = {
"source": "cryptoscamdb",
"category": category,
"status": status,
"name": name,
"url": url,
"severity": "high" if status == "Active" else "medium",
}
make_doc_id(name or url or str(i), "cryptoscamdb")
if await ingest_via_api(content, metadata):
ingested += 1
else:
errors += 1
else:
# Create one doc per blockchain address
for addr in addresses:
if not addr or len(addr) < 10:
continue
content = f"CryptoScamDB flagged address: {addr}. Project: {name}. Category: {category}. Status: {status}. Description: {description}"
metadata = {
"source": "cryptoscamdb",
"address": addr.lower(),
"category": category,
"status": status,
"name": name,
"severity": "high" if status == "Active" else "medium",
}
if await ingest_via_api(content, metadata):
ingested += 1
else:
errors += 1
if (i + 1) % 100 == 0:
logger.info(f"Progress: {i + 1}/{len(scams)} | ingested: {ingested}, errors: {errors}")
await asyncio.sleep(0.1) # Yield to event loop
logger.info(f"COMPLETE: ingested={ingested}, errors={errors}")
if __name__ == "__main__":
asyncio.run(main())
|