#!/usr/bin/env python3 """ Minecraft Username Sniper - Snipe History Database (Phase 5 Feature #5) SQLite-based persistent logging for all snipe attempts with export and stats. Database location: ~/.hermes/minecraft-sniper/snipe_history.db """ import sqlite3 import os import csv import json from datetime import datetime from pathlib import Path DB_DIR = Path.home() / ".hermes" / "minecraft-sniper" DB_PATH = DB_DIR / "snipe_history.db" def _get_connection(): """Get a database connection, creating the database if needed.""" DB_DIR.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(str(DB_PATH)) conn.row_factory = sqlite3.Row _ensure_schema(conn) return conn def _ensure_schema(conn): """Create tables if they don't exist.""" conn.execute(""" CREATE TABLE IF NOT EXISTS snipes ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, username TEXT NOT NULL, success INTEGER NOT NULL DEFAULT 0, account_hash TEXT, response_time_ms REAL, error_message TEXT, drop_time_utc TEXT, mode TEXT DEFAULT 'exact', threads INTEGER DEFAULT 10, timing_mode TEXT DEFAULT 'exact' ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS my_names ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT NOT NULL UNIQUE, claimed_date TEXT NOT NULL, released_date TEXT, active INTEGER NOT NULL DEFAULT 1, notes TEXT ) """) conn.commit() # --- Public API --- def record_snipe(username, success, account_hash=None, response_time_ms=None, error_message=None, drop_time_utc=None, mode="exact", threads=10, timing_mode="exact"): """Record a snipe attempt to the database.""" conn = _get_connection() ts = datetime.utcnow().isoformat() + "Z" conn.execute(""" INSERT INTO snipes (timestamp, username, success, account_hash, response_time_ms, error_message, drop_time_utc, mode, threads, timing_mode) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (ts, username, int(success), account_hash, response_time_ms, error_message, drop_time_utc, mode, threads, timing_mode)) conn.commit() conn.close() return ts def get_history(limit=50, username=None, success=None): """Get snipe history with optional filters. Args: limit: Max rows to return username: Filter by username success: Filter by success (True/False) Returns: List of dicts """ conn = _get_connection() query = "SELECT * FROM snipes WHERE 1=1" params = [] if username: query += " AND username = ?" params.append(username) if success is not None: query += " AND success = ?" params.append(int(success)) query += " ORDER BY timestamp DESC LIMIT ?" params.append(limit) rows = conn.execute(query, params).fetchall() conn.close() return [dict(row) for row in rows] def get_stats(): """Get aggregate statistics. Returns: Dict with total_snipes, success_count, failure_count, success_rate, fastest_response_ms, slowest_response_ms, avg_response_ms, most_sniped_names (list of (name, count)) """ conn = _get_connection() total = conn.execute("SELECT COUNT(*) FROM snipes").fetchone()[0] if total == 0: conn.close() return {"total_snipes": 0} stats = conn.execute(""" SELECT COUNT(*) as total, SUM(success) as successes, COUNT(*) - SUM(success) as failures, ROUND(SUM(success) * 100.0 / COUNT(*), 1) as success_rate, MIN(response_time_ms) as fastest_ms, MAX(response_time_ms) as slowest_ms, ROUND(AVG(response_time_ms), 1) as avg_ms FROM snipes WHERE response_time_ms IS NOT NULL """).fetchone() top_names = conn.execute(""" SELECT username, COUNT(*) as cnt FROM snipes GROUP BY username ORDER BY cnt DESC LIMIT 10 """).fetchall() conn.close() return { "total_snipes": total, "success_count": stats["successes"], "failure_count": stats["failures"], "success_rate": stats["success_rate"], "fastest_response_ms": stats["fastest_ms"], "slowest_response_ms": stats["slowest_ms"], "avg_response_ms": stats["avg_ms"], "most_sniped_names": [(r["username"], r["cnt"]) for r in top_names], } def export_csv(path="snipe_history.csv"): """Export snipe history to CSV. Returns the file path.""" rows = get_history(limit=10000) if not rows: return None with open(path, "w", newline="") as f: writer = csv.DictWriter(f, fieldnames=rows[0].keys()) writer.writeheader() writer.writerows(rows) return path def export_json(path="snipe_history.json"): """Export snipe history to JSON. Returns the file path.""" rows = get_history(limit=10000) if not rows: return None with open(path, "w") as f: json.dump(rows, f, indent=2) return path # --- My Names (for auto-reclaim) --- def register_my_name(username, claimed_date=None, notes=""): """Register a name as one you own.""" if claimed_date is None: claimed_date = datetime.utcnow().isoformat() + "Z" conn = _get_connection() conn.execute(""" INSERT OR REPLACE INTO my_names (username, claimed_date, active, notes) VALUES (?, ?, 1, ?) """, (username, claimed_date, notes)) conn.commit() conn.close() def release_my_name(username): """Mark a name as released (available for reclaiming).""" released = datetime.utcnow().isoformat() + "Z" conn = _get_connection() conn.execute(""" UPDATE my_names SET released_date = ?, active = 0 WHERE username = ? """, (released, username)) conn.commit() conn.close() def get_released_names(): """Get names that were released and should be monitored for reclaiming.""" conn = _get_connection() rows = conn.execute(""" SELECT * FROM my_names WHERE active = 0 AND released_date IS NOT NULL ORDER BY released_date DESC """).fetchall() conn.close() return [dict(r) for r in rows] def get_all_my_names(): """Get all registered names.""" conn = _get_connection() rows = conn.execute("SELECT * FROM my_names ORDER BY claimed_date DESC").fetchall() conn.close() return [dict(r) for r in rows] if __name__ == "__main__": # Quick test print(f"Database: {DB_PATH}") record_snipe("TestName", True, account_hash="abc123", response_time_ms=150.5) record_snipe("TestName2", False, error_message="Name taken") print("History:", get_history()) print("Stats:", get_stats()) print("CSV exported:", export_csv("/tmp/test_snipe_history.csv"))