File size: 5,484 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 | #!/usr/bin/env python3
"""
Ingest CryptoScamDB blacklist data into RAG known_scams collection.
Source: https://github.com/CryptoScamDB/blacklist
"""
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_blacklist")
RAG_API = "http://localhost:8000/api/v1/rag/ingest"
COLLECTION = "known_scams"
DATA_DIR = os.path.join(os.path.dirname(__file__), "..", "data")
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:
return False
async def ingest_yaml():
"""Parse the YAML blacklist and ingest entries with addresses."""
yaml_path = os.path.join(DATA_DIR, "cryptoscamdb_urls.yaml")
if not os.path.exists(yaml_path):
logger.error(f"YAML file not found: {yaml_path}")
return {"ingested": 0, "no_address": 0}
# Simple YAML parser (avoid pyyaml dependency for this structure)
entries = []
current = {}
with open(yaml_path, encoding="utf-8", errors="replace") as f:
for line in f:
stripped = line.rstrip()
if not stripped or stripped.startswith("#"):
continue
if stripped.startswith("- name:"):
if current:
entries.append(current)
current = {"name": stripped[7:].strip().strip('"')}
elif stripped.startswith(" url:"):
current["url"] = stripped[6:].strip().strip('"')
elif stripped.startswith(" category:"):
current["category"] = stripped[11:].strip().strip('"')
elif stripped.startswith(" subcategory:"):
current["subcategory"] = stripped[14:].strip().strip('"')
elif stripped.startswith(" description:"):
current["description"] = stripped[14:].strip().strip('"')
elif stripped.startswith(" reporter:"):
current["reporter"] = stripped[11:].strip().strip('"')
elif stripped.startswith(" ETH:"):
current["chain"] = "ethereum"
elif stripped.startswith(' - "0x') or stripped.startswith(" - 0x"):
addr = stripped.split('"')[-2] if '"' in stripped else stripped.split("-")[-1].strip()
current.setdefault("addresses", []).append(addr)
elif stripped.startswith(" BTC:"):
current["chain"] = "bitcoin"
elif stripped.startswith(" BSC:") or stripped.startswith(" BNB:"):
current["chain"] = "bsc"
if current:
entries.append(current)
logger.info(f"Parsed {len(entries)} entries from YAML")
ingested = 0
no_address = 0
for i, entry in enumerate(entries):
name = entry.get("name", "")
category = entry.get("category", "unknown")
subcategory = entry.get("subcategory", "")
description = entry.get("description", "")
url = entry.get("url", "")
addresses = entry.get("addresses", [])
if not addresses:
# Ingest URL-based entry without blockchain address
content = f"CryptoScamDB: {name}. Category: {category}/{subcategory}. URL: {url}. {description}"
metadata = {
"source": "cryptoscamdb",
"category": category.lower(),
"subcategory": subcategory.lower(),
"name": name,
"url": url,
"severity": "high" if category == "Phishing" else "medium",
}
if await ingest_via_api(content, metadata):
no_address += 1
else:
for addr in addresses:
if not addr or len(addr) < 10:
continue
chain = entry.get("chain", "ethereum")
content = f"CryptoScamDB flagged address: {addr} on {chain}. Project: {name}. Category: {category}/{subcategory}. {description}"
metadata = {
"source": "cryptoscamdb",
"address": addr.lower(),
"chain": chain,
"category": category.lower(),
"subcategory": subcategory.lower(),
"name": name,
"url": url,
"severity": "high" if category in ("Phishing", "Scam") else "medium",
}
if await ingest_via_api(content, metadata):
ingested += 1
if (i + 1) % 500 == 0:
logger.info(f"Progress: {i + 1}/{len(entries)} | addr_entries: {ingested}, url_only: {no_address}")
await asyncio.sleep(0.1)
logger.info(f"COMPLETE: addr_entries={ingested}, url_only={no_address}")
return {"ingested": ingested, "no_address": no_address}
async def main():
result = await ingest_yaml()
logger.info(f"Final: {result}")
if __name__ == "__main__":
asyncio.run(main())
|