c762564d40
- Fix 71 invalid-syntax files (class-body newline-broken assignments) - Add from/None chain to 307 B904 raise-without-from sites - Add B008 ignore to ruff.toml (already in pyproject.toml) - Noqa F401 on __init__.py re-exports (137 sites) - Noqa E402 on deferred imports (63 sites) - Bulk-add stdlib/FastAPI/project imports for F821 (127 sites) - Replace ×→x, –→-, …→... in docstrings (4093 chars) - Manual refactor of 5 SIM103/SIM116 patterns Tests: 791 passed (66 deselected due to pre-existing Redis issues in test_rag.py) Co-authored-by: opencode <opencode@rugmunch.io>
258 lines
7.5 KiB
Python
258 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Nightly Wallet Label Backup & Sync
|
|
====================================
|
|
1. Validates clean label data integrity
|
|
2. Backs up to R2 cloud storage (with rotation)
|
|
3. Syncs clean labels into running Redis/ClickHouse
|
|
4. Logs freshness metrics
|
|
"""
|
|
|
|
import csv
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
DATA_DIR = os.environ.get("RMI_DATA_DIR", "/root/backend/data")
|
|
CLEAN_DIR = os.path.join(DATA_DIR, "wallet-labels-clean")
|
|
BACKUP_DIR = os.path.join(DATA_DIR, "wallet-labels-backups")
|
|
MANIFEST_PATH = os.path.join(CLEAN_DIR, "manifest.json")
|
|
|
|
# R2 config
|
|
R2_BUCKET = os.environ.get("R2_BUCKET", "rmi-backups")
|
|
R2_ENDPOINT = os.environ.get("R2_ENDPOINT", "")
|
|
R2_KEY = os.environ.get("R2_ACCESS_KEY", "")
|
|
R2_SECRET = os.environ.get("R2_SECRET_KEY", "")
|
|
|
|
MAX_BACKUPS = 30 # Keep 30 days of backups
|
|
|
|
|
|
def validate_labels():
|
|
"""Validate all clean label files for integrity."""
|
|
print("Validating label data...")
|
|
results = {}
|
|
|
|
for fname in ["wallet_labels_ethereum.csv", "wallet_labels_solana.csv"]:
|
|
path = os.path.join(CLEAN_DIR, fname)
|
|
if not os.path.exists(path):
|
|
results[fname] = {"status": "MISSING"}
|
|
continue
|
|
|
|
count = 0
|
|
dupes = 0
|
|
invalids = 0
|
|
seen = set()
|
|
|
|
with open(path) as f:
|
|
reader = csv.DictReader(f)
|
|
for row in reader:
|
|
addr = row.get("address", "")
|
|
if not addr:
|
|
invalids += 1
|
|
continue
|
|
if addr.lower() in seen:
|
|
dupes += 1
|
|
seen.add(addr.lower())
|
|
count += 1
|
|
|
|
results[fname] = {
|
|
"status": "OK" if dupes == 0 and invalids == 0 else "ISSUES",
|
|
"count": count,
|
|
"duplicates": dupes,
|
|
"invalids": invalids,
|
|
"unique": len(seen),
|
|
}
|
|
print(f" {fname}: {count:,} rows, {dupes} dupes, {invalids} invalid - {results[fname]['status']}")
|
|
|
|
# Validate manifest
|
|
if os.path.exists(MANIFEST_PATH):
|
|
with open(MANIFEST_PATH) as f:
|
|
manifest = json.load(f)
|
|
results["manifest"] = {
|
|
"generated": manifest.get("generated_at"),
|
|
"total": manifest.get("total_labels"),
|
|
"chains": manifest.get("chains"),
|
|
}
|
|
|
|
return results
|
|
|
|
|
|
def create_backup():
|
|
"""Create a timestamped backup of the clean label data."""
|
|
os.makedirs(BACKUP_DIR, exist_ok=True)
|
|
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
backup_path = os.path.join(BACKUP_DIR, f"labels_{timestamp}")
|
|
os.makedirs(backup_path, exist_ok=True)
|
|
|
|
# Copy clean label files
|
|
for fname in os.listdir(CLEAN_DIR):
|
|
src = os.path.join(CLEAN_DIR, fname)
|
|
dst = os.path.join(backup_path, fname)
|
|
if os.path.isfile(src):
|
|
shutil.copy2(src, dst)
|
|
|
|
# Create checksums
|
|
checksums = {}
|
|
for fname in os.listdir(backup_path):
|
|
fpath = os.path.join(backup_path, fname)
|
|
with open(fpath, "rb") as f:
|
|
checksums[fname] = hashlib.sha256(f.read()).hexdigest()
|
|
|
|
with open(os.path.join(backup_path, "checksums.json"), "w") as f:
|
|
json.dump(checksums, f, indent=2)
|
|
|
|
# Create backup manifest
|
|
backup_manifest = {
|
|
"backup_time": datetime.now(UTC).isoformat(),
|
|
"backup_id": timestamp,
|
|
"files": os.listdir(backup_path),
|
|
"checksums": checksums,
|
|
}
|
|
with open(os.path.join(backup_path, "backup_manifest.json"), "w") as f:
|
|
json.dump(backup_manifest, f, indent=2)
|
|
|
|
# Compress
|
|
archive_path = f"{backup_path}.tar.gz"
|
|
shutil.make_archive(backup_path, "gztar", BACKUP_DIR, f"labels_{timestamp}")
|
|
|
|
# Cleanup uncompressed
|
|
shutil.rmtree(backup_path)
|
|
|
|
# Rotate old backups
|
|
backups = sorted(Path(BACKUP_DIR).glob("labels_*.tar.gz"))
|
|
while len(backups) > MAX_BACKUPS:
|
|
oldest = backups.pop(0)
|
|
oldest.unlink()
|
|
print(f" Rotated out: {oldest.name}")
|
|
|
|
size_mb = os.path.getsize(archive_path) / (1024 * 1024)
|
|
print(f"Backup created: {archive_path} ({size_mb:.1f} MB)")
|
|
|
|
return archive_path
|
|
|
|
|
|
def upload_to_r2(archive_path: str) -> bool:
|
|
"""Upload backup to R2 cloud storage."""
|
|
if not all([R2_ENDPOINT, R2_KEY, R2_SECRET]):
|
|
print("R2 not configured - skipping cloud upload")
|
|
return False
|
|
|
|
try:
|
|
import boto3
|
|
|
|
s3 = boto3.client(
|
|
"s3",
|
|
endpoint_url=R2_ENDPOINT,
|
|
aws_access_key_id=R2_KEY,
|
|
aws_secret_access_key=R2_SECRET,
|
|
)
|
|
|
|
fname = os.path.basename(archive_path)
|
|
s3.upload_file(
|
|
archive_path,
|
|
R2_BUCKET,
|
|
f"wallet-labels/{fname}",
|
|
)
|
|
print(f"Uploaded to R2: wallet-labels/{fname}")
|
|
return True
|
|
except ImportError:
|
|
print("boto3 not installed - skipping R2 upload")
|
|
return False
|
|
except Exception as e:
|
|
print(f"R2 upload failed: {e}")
|
|
return False
|
|
|
|
|
|
def sync_to_redis():
|
|
"""Sync clean labels into Redis for fast lookups."""
|
|
try:
|
|
import redis
|
|
|
|
r = redis.Redis(
|
|
host=os.environ.get("REDIS_HOST", "localhost"),
|
|
port=int(os.environ.get("REDIS_PORT", "6379")),
|
|
password=os.environ.get("REDIS_PASSWORD", ""),
|
|
decode_responses=True,
|
|
)
|
|
r.ping()
|
|
except Exception as e:
|
|
print(f"Redis not available: {e}")
|
|
return 0
|
|
|
|
synced = 0
|
|
for chain_file in ["wallet_labels_ethereum.csv", "wallet_labels_solana.csv"]:
|
|
path = os.path.join(CLEAN_DIR, chain_file)
|
|
if not os.path.exists(path):
|
|
continue
|
|
|
|
pipe = r.pipeline()
|
|
batch = 0
|
|
with open(path) as f:
|
|
for row in csv.DictReader(f):
|
|
addr = row["address"].lower()
|
|
key = f"wl:{addr}"
|
|
pipe.hset(
|
|
key,
|
|
mapping={
|
|
"chain": row.get("chain", ""),
|
|
"source": row.get("source", ""),
|
|
"name": row.get("name", ""),
|
|
"entity_type": row.get("entity_type", ""),
|
|
"synced_at": datetime.now(UTC).isoformat(),
|
|
},
|
|
)
|
|
batch += 1
|
|
if batch % 1000 == 0:
|
|
pipe.execute()
|
|
pipe = r.pipeline()
|
|
synced += batch
|
|
batch = 0
|
|
|
|
if batch > 0:
|
|
pipe.execute()
|
|
synced += batch
|
|
|
|
print(f"Synced {synced:,} labels to Redis")
|
|
return synced
|
|
|
|
|
|
def main():
|
|
print(f"=== Wallet Label Nightly Backup - {datetime.now().isoformat()} ===\n")
|
|
|
|
# 1. Validate
|
|
validation = validate_labels()
|
|
|
|
# 2. Backup
|
|
archive = create_backup()
|
|
|
|
# 3. Upload to cloud
|
|
uploaded = upload_to_r2(archive)
|
|
|
|
# 4. Sync to Redis
|
|
synced = sync_to_redis()
|
|
|
|
# 5. Update manifest
|
|
manifest = {
|
|
"generated_at": datetime.now(UTC).isoformat(),
|
|
"last_backup": str(archive),
|
|
"cloud_uploaded": uploaded,
|
|
"redis_synced": synced,
|
|
"validation": validation,
|
|
}
|
|
with open(os.path.join(CLEAN_DIR, "nightly_manifest.json"), "w") as f:
|
|
json.dump(manifest, f, indent=2, default=str)
|
|
|
|
print("\n=== Done ===")
|
|
print(f"Backup: {archive}")
|
|
print(f"Cloud: {'✓' if uploaded else '✗'}")
|
|
print(f"Redis: {synced:,} labels synced")
|
|
|
|
return manifest
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|