62 lines
1.5 KiB
Python
62 lines
1.5 KiB
Python
import base64
|
|
import json
|
|
import logging
|
|
import os
|
|
import urllib.request
|
|
from datetime import datetime
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _surreal_headers() -> dict[str, str]:
|
|
headers = {
|
|
"Content-Type": "text/plain",
|
|
"Accept": "application/json",
|
|
}
|
|
|
|
ns = os.getenv("SURREALDB_NS", "optovia")
|
|
db = os.getenv("SURREALDB_DB", "events")
|
|
user = os.getenv("SURREALDB_USER")
|
|
password = os.getenv("SURREALDB_PASS")
|
|
|
|
headers["NS"] = ns
|
|
headers["DB"] = db
|
|
|
|
if user and password:
|
|
token = base64.b64encode(f"{user}:{password}".encode("utf-8")).decode("utf-8")
|
|
headers["Authorization"] = f"Basic {token}"
|
|
|
|
return headers
|
|
|
|
|
|
def log_kyc_event(kyc_id: str, user_id: str, event: str, description: str) -> bool:
|
|
url = os.getenv("SURREALDB_URL")
|
|
if not url:
|
|
logger.warning("SURREALDB_URL is not set; skipping KYC event log")
|
|
return False
|
|
|
|
payload = {
|
|
"kyc_id": kyc_id,
|
|
"user_id": user_id,
|
|
"event": event,
|
|
"description": description,
|
|
"created_at": datetime.utcnow().isoformat() + "Z",
|
|
}
|
|
|
|
query = f"CREATE kyc_event CONTENT {json.dumps(payload)};"
|
|
|
|
req = urllib.request.Request(
|
|
f"{url.rstrip('/')}/sql",
|
|
data=query.encode("utf-8"),
|
|
method="POST",
|
|
headers=_surreal_headers(),
|
|
)
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as response:
|
|
response.read()
|
|
return True
|
|
except Exception as exc:
|
|
logger.error("Failed to log KYC event: %s", exc)
|
|
return False
|