{detail["wallet_balance"] if not detail.get("error") else "error"}
{detail["user_balance"] if not detail.get("error") else "-"}
0 else ""}">{detail["owner_balance"] if not detail.get("error") else "-"}
"""
for detail in balance_details
if detail.get("wallet_balance", 0) > 0 or detail.get("error")
]
)
return f"""
Cashu Wallet Balance
Your Balance (Total){owner_balance} sats
Total Wallet{total_wallet_balance_sats} sats
User Balance{total_user_balance_sats} sats
Your balance = Total wallet - User balance
Mint / Unit
Wallet
Users
Owner
{rows}
"""
@admin_router.get(
"/partials/apikeys",
dependencies=[Depends(require_admin_api)],
response_class=HTMLResponse,
)
async def partial_apikeys(request: Request) -> str:
async with create_session() as session:
result = await session.exec(select(ApiKey))
api_keys = result.all()
def fmt_time(ts: int | None) -> str:
if ts is None:
return ""
dt = datetime.fromtimestamp(ts, tz=timezone.utc)
return f"{ts} ({dt.strftime('%Y-%m-%d %H:%M:%S')} UTC)"
rows = "".join(
[
f"
{key.hashed_key}
{key.balance}
{key.total_spent}
{key.total_requests}
{key.refund_address}
{fmt_time(key.key_expiry_time)}
"
for key in api_keys
]
)
return f"""
Temporary Balances
Hashed Key
Balance (mSats)
Total Spent (mSats)
Total Requests
Refund Address
Refund Time
{rows}
"""
@admin_router.get("/api/balances", dependencies=[Depends(require_admin_api)])
async def get_balances_api(request: Request) -> list[dict[str, object]]:
balance_details, _tw, _tu, _ow = await fetch_all_balances()
return [dict(d) for d in balance_details]
@admin_router.get("/api/settings", dependencies=[Depends(require_admin_api)])
async def get_settings(request: Request) -> dict:
data = settings.dict()
if "upstream_api_key" in data:
data["upstream_api_key"] = "[REDACTED]" if data["upstream_api_key"] else ""
if "admin_password" in data:
data["admin_password"] = "[REDACTED]" if data["admin_password"] else ""
if "nsec" in data:
data["nsec"] = "[REDACTED]" if data["nsec"] else ""
return data
class SettingsUpdate(BaseModel):
__root__: dict[str, object]
@admin_router.patch("/api/settings", dependencies=[Depends(require_admin_api)])
async def update_settings(request: Request, update: SettingsUpdate) -> dict:
async with create_session() as session:
new_settings = await SettingsService.update(update.__root__, session)
data = new_settings.dict()
if "upstream_api_key" in data:
data["upstream_api_key"] = "[REDACTED]" if data["upstream_api_key"] else ""
if "admin_password" in data:
data["admin_password"] = "[REDACTED]" if data["admin_password"] else ""
if "nsec" in data:
data["nsec"] = "[REDACTED]" if data["nsec"] else ""
return data
class WithdrawRequest(BaseModel):
amount: int
mint_url: str | None = None
unit: str = "sat"
def login_form() -> str:
return """
🔐 Admin Login
"""
def info(content: str) -> str:
return f"""
{content}
"""
def admin_auth() -> str:
try:
settings = SettingsService.get()
admin_pw = settings.admin_password
except Exception:
admin_pw = os.getenv("ADMIN_PASSWORD", "")
if admin_pw == "":
return info("Please set a secure ADMIN_PASSWORD= in your ENV variables.")
else:
return login_form()
async def dashboard(request: Request) -> str:
return (
f"""
"""
+ """
"""
+ """
Admin Dashboard
Loading balances…
×
Withdraw Balance
Select mint and currency:
Enter amount to withdraw:
Maximum: -
Your recommended balance: -
⚠️ Warning: Withdrawing more than your balance will use user funds!
×
Edit Settings (JSON)
Values shown as "[REDACTED]" will remain unchanged if left as-is.
×
Investigate Logs
Enter Request ID to investigate:
Withdrawal Token:
Save this token! It represents your withdrawn balance.
Temporary Balances
Loading API keys…
"""
)
@admin_router.get("/", response_class=HTMLResponse)
async def admin(request: Request) -> str:
if is_admin_authenticated(request):
return await dashboard(request)
return admin_auth()
@admin_router.get("/logs/{request_id}", response_class=HTMLResponse)
async def view_logs(request: Request, request_id: str) -> str:
if not is_admin_authenticated(request):
return admin_auth()
logger.info(f"Investigating logs for request_id: {request_id}")
# Search for log entries with this request_id
log_entries = []
logs_dir = Path("logs")
if logs_dir.exists():
# Get all log files sorted by modification time (most recent first)
log_files = sorted(
logs_dir.glob("*.log"), key=lambda x: x.stat().st_mtime, reverse=True
)
for log_file in log_files[:7]: # Check last 7 days of logs
try:
with open(log_file, "r") as f:
for line in f:
if request_id in line:
try:
# Parse JSON log entry
log_data = json.loads(line.strip())
log_entries.append(log_data)
except json.JSONDecodeError:
# If not JSON, include raw line
log_entries.append({"raw": line.strip()})
except Exception as e:
logger.error(f"Error reading log file {log_file}: {e}")
# Sort entries by timestamp if available
log_entries.sort(key=lambda x: x.get("asctime", ""), reverse=False)
# Format log entries for display
formatted_logs = []
for entry in log_entries:
if "raw" in entry:
formatted_logs.append(f'
{entry["raw"]}
')
else:
# Format JSON log entry
timestamp = entry.get("asctime", "Unknown time")
level = entry.get("levelname", "INFO")
message = entry.get("message", "")
pathname = entry.get("pathname", "")
lineno = entry.get("lineno", "")
# Extract additional fields
extra_fields = {
k: v
for k, v in entry.items()
if k
not in [
"asctime",
"levelname",
"message",
"pathname",
"lineno",
"name",
"version",
"request_id",
]
}
level_class = level.lower()
formatted_entry = f"""
{timestamp}[{level}]{pathname}:{lineno}
{message}
"""
if extra_fields:
formatted_entry += '
'
for key, value in extra_fields.items():
formatted_entry += f'
{key}: {json.dumps(value) if isinstance(value, (dict, list)) else value}
'
formatted_entry += "
"
formatted_entry += "
"
formatted_logs.append(formatted_entry)
return (
f"""
"""
+ f"""
← Back to Dashboard
Log Investigation
Request ID: {request_id}
{"".join(formatted_logs) if formatted_logs else '
No log entries found for this Request ID
'}
Found {len(log_entries)} log entries • Searched last 7 days of logs