"""CodeStar Database Layer with Dual-Mode Support: - Local SQLite (data/codestar.db) by default for offline/dev/testing. - Cloud PostgreSQL (Neon / Supabase / RDS) when DATABASE_URL is set. """ from __future__ import annotations import os import sqlite3 import datetime from pathlib import Path from typing import Optional, List, Dict, Any ROOT = Path(__file__).resolve().parents[2] DB_PATH = ROOT / "data" / "codestar.db" DB_PATH.parent.mkdir(parents=True, exist_ok=True) # Auto-load .env file if present def _load_env(): env_path = ROOT / ".env" if env_path.exists(): try: with open(env_path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) k = k.strip() v = v.strip().strip("'").strip('"') if k and k not in os.environ: os.environ[k] = v except Exception: pass _load_env() DATABASE_URL = os.environ.get("DATABASE_URL", "").strip() IS_POSTGRES = DATABASE_URL.startswith("postgres://") or DATABASE_URL.startswith("postgresql://") def _translate_sql(sql: str) -> str: """Translates SQLite ? parameter markers to PostgreSQL %s parameter markers safely.""" out = [] in_single = False in_double = False for ch in sql: if ch == "'" and not in_double: in_single = not in_single out.append(ch) elif ch == '"' and not in_single: in_double = not in_double out.append(ch) elif ch == '?' and not in_single and not in_double: out.append('%s') else: out.append(ch) return ''.join(out) class PgRow(dict): """Dict and tuple indexable row, matching sqlite3.Row behavior.""" def __init__(self, cols: List[str], values: tuple): super().__init__(zip(cols, values)) self._values = list(values) def __getitem__(self, key: Any) -> Any: if isinstance(key, int): return self._values[key] return super().__getitem__(key) class PgCursorWrapper: def __init__(self, cursor, lastrowid=None): self._cursor = cursor self.lastrowid = lastrowid self.rowcount = getattr(cursor, "rowcount", 0) @property def description(self): return getattr(self._cursor, "description", None) def fetchone(self) -> Optional[PgRow]: row = self._cursor.fetchone() if row is None: return None cols = [d[0] for d in self._cursor.description] return PgRow(cols, row) def fetchall(self) -> List[PgRow]: rows = self._cursor.fetchall() if not rows: return [] cols = [d[0] for d in self._cursor.description] return [PgRow(cols, r) for r in rows] def __iter__(self): while True: row = self.fetchone() if row is None: break yield row class PgConnectionWrapper: def __init__(self, raw_conn, pool=None): self._conn = raw_conn self._pool = pool self._has_error = False def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): try: if exc_type is None: self._conn.commit() else: self._has_error = True self._conn.rollback() finally: if self._pool is not None: self._pool.release_connection(self._conn, discard=self._has_error) else: try: self._conn.close() except Exception: pass def execute(self, sql: str, params=None) -> PgCursorWrapper: cursor = self._conn.cursor() stripped = sql.strip() # Handle SQLite PRAGMA table_info transparently on PostgreSQL if stripped.upper().startswith("PRAGMA TABLE_INFO"): import re m = re.search(r"PRAGMA\s+table_info\s*\(\s*['\"]?(\w+)['\"]?\s*\)", stripped, re.IGNORECASE) tbl = m.group(1) if m else "submissions" sql_clean = ( "SELECT ordinal_position - 1 AS cid, column_name AS name, data_type AS type, " "CASE WHEN is_nullable = 'NO' THEN 1 ELSE 0 END AS notnull, " "column_default AS dflt_value, 0 AS pk " "FROM information_schema.columns WHERE table_name = %s ORDER BY ordinal_position" ) cursor.execute(sql_clean, (tbl,)) return PgCursorWrapper(cursor) elif stripped.upper().startswith("PRAGMA"): # SQLite pragmas (journal_mode, foreign_keys, etc.) are harmless no-ops on PostgreSQL class DummyCursor: description = [] rowcount = 0 def fetchone(self): return None def fetchall(self): return [] return PgCursorWrapper(DummyCursor()) sql_clean = _translate_sql(sql) # Handle SQLite "INSERT OR IGNORE INTO" transparently on PostgreSQL if "INSERT OR IGNORE INTO" in sql_clean.upper(): import re sql_clean = re.sub(r"(?i)INSERT\s+OR\s+IGNORE\s+INTO", "INSERT INTO", sql_clean) if "ON CONFLICT" not in sql_clean.upper(): sql_clean = sql_clean.rstrip("; \t\n") + " ON CONFLICT DO NOTHING" is_insert = sql_clean.strip().upper().startswith("INSERT INTO") needs_returning = is_insert and "RETURNING" not in sql_clean.upper() if needs_returning: sql_clean = sql_clean.rstrip("; \t\n") + " RETURNING id" if params: cursor.execute(sql_clean, tuple(params)) else: cursor.execute(sql_clean) lastrowid = None if needs_returning: try: ret_row = cursor.fetchone() if ret_row: lastrowid = ret_row[0] except Exception: pass return PgCursorWrapper(cursor, lastrowid=lastrowid) def executescript(self, script: str): cursor = self._conn.cursor() statements = [s.strip() for s in script.split(";") if s.strip()] for stmt in statements: cursor.execute(_translate_sql(stmt)) self._conn.commit() def commit(self): self._conn.commit() def rollback(self): self._conn.rollback() def close(self): try: self._conn.close() except Exception: pass def _get_pg_connection(): import urllib.parse import ssl import pg8000.dbapi raw_url = os.environ.get("DATABASE_URL", "").strip() parsed = urllib.parse.urlparse(raw_url) query_params = urllib.parse.parse_qs(parsed.query) sslmode = query_params.get("sslmode", [""])[0].lower() use_ssl = ( sslmode in ("require", "verify-ca", "verify-full") or "neon.tech" in (parsed.hostname or "") or "supabase.co" in (parsed.hostname or "") or "supabase.com" in (parsed.hostname or "") ) ssl_ctx = ssl.create_default_context() if use_ssl else None password = urllib.parse.unquote_plus(parsed.password) if parsed.password else "" user = urllib.parse.unquote_plus(parsed.username) if parsed.username else "postgres" database = parsed.path.lstrip("/") or "postgres" conn = pg8000.dbapi.connect( user=user, password=password, host=parsed.hostname or "localhost", port=parsed.port or 5432, database=database, ssl_context=ssl_ctx, timeout=30.0, ) return conn class PgConnectionPool: """Thread-safe connection pool for PostgreSQL (Neon) with health checks and auto-reconnect.""" def __init__(self, max_size: int = 8, timeout: float = 25.0): import queue import threading self.max_size = max_size self.timeout = timeout self._pool = queue.Queue(maxsize=max_size) self._lock = threading.Lock() self._active_count = 0 def get_connection(self): import queue # 1. Try to reuse an existing idle connection from pool while not self._pool.empty(): try: conn = self._pool.get_nowait() # Health probe: ensure TCP connection hasn't been closed by remote peer try: cur = conn.cursor() cur.execute("SELECT 1") cur.fetchone() return conn except Exception: try: conn.close() except Exception: pass with self._lock: self._active_count = max(0, self._active_count - 1) except queue.Empty: break # 2. If pool is empty and active count < max_size, create a new connection with self._lock: if self._active_count < self.max_size: self._active_count += 1 try: return _get_pg_connection() except Exception: self._active_count -= 1 raise # 3. Wait for an idle connection returned from another thread try: return self._pool.get(timeout=self.timeout) except Exception: return _get_pg_connection() def release_connection(self, conn, discard: bool = False): import queue if not conn: return if discard: try: conn.close() except Exception: pass with self._lock: self._active_count = max(0, self._active_count - 1) return try: conn.rollback() self._pool.put_nowait(conn) except (queue.Full, Exception): try: conn.close() except Exception: pass with self._lock: self._active_count = max(0, self._active_count - 1) _pg_pool: Optional[PgConnectionPool] = None def get_db(): global IS_POSTGRES, _pg_pool if IS_POSTGRES: try: if _pg_pool is None: _pg_pool = PgConnectionPool(max_size=8) raw_conn = _pg_pool.get_connection() return PgConnectionWrapper(raw_conn, pool=_pg_pool) except Exception as e: print(f"[CodeStar DB] Warning: PostgreSQL connection failed ({e}). Falling back to SQLite.") conn = sqlite3.connect(str(DB_PATH), timeout=30.0) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode = WAL") conn.execute("PRAGMA synchronous = NORMAL") conn.execute("PRAGMA busy_timeout = 30000") conn.execute("PRAGMA foreign_keys = ON") return conn def init_db(): global IS_POSTGRES if IS_POSTGRES: try: with get_db() as conn: conn.executescript(""" CREATE TABLE IF NOT EXISTS users ( id SERIAL PRIMARY KEY, username VARCHAR(255) UNIQUE NOT NULL, password_hash TEXT NOT NULL, salt TEXT NOT NULL, created_at TEXT NOT NULL, is_vip INTEGER DEFAULT 0, vip_expire_at TEXT DEFAULT NULL, vip_level VARCHAR(64) DEFAULT 'none', star_coins INTEGER DEFAULT 0 ); CREATE TABLE IF NOT EXISTS submissions ( id SERIAL PRIMARY KEY, user_id INTEGER REFERENCES users(id) ON DELETE SET NULL, problem_slug VARCHAR(255) NOT NULL, status VARCHAR(64) NOT NULL, passed_cases INTEGER NOT NULL, total_cases INTEGER NOT NULL, runtime_ms DOUBLE PRECISION NOT NULL, peak_memory_mb DOUBLE PRECISION DEFAULT 0.0, code TEXT NOT NULL, error_message TEXT, created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_submissions_user_slug ON submissions (user_id, problem_slug, status); CREATE INDEX IF NOT EXISTS idx_submissions_created_at ON submissions (created_at DESC); CREATE TABLE IF NOT EXISTS orders ( id SERIAL PRIMARY KEY, order_no VARCHAR(128) UNIQUE NOT NULL, user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, product_type VARCHAR(64) NOT NULL, product_name VARCHAR(255) NOT NULL, amount_cents INTEGER NOT NULL, pay_channel VARCHAR(64) NOT NULL, status VARCHAR(64) NOT NULL DEFAULT 'pending', trade_no VARCHAR(128), qr_code_url TEXT, pay_url TEXT, paid_at TEXT, created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_orders_user ON orders (user_id, status); CREATE INDEX IF NOT EXISTS idx_orders_no ON orders (order_no); """) # Idempotent column additions for existing PG databases for stmt in [ "ALTER TABLE submissions ADD COLUMN IF NOT EXISTS peak_memory_mb DOUBLE PRECISION DEFAULT 0.0", "ALTER TABLE users ADD COLUMN IF NOT EXISTS is_vip INTEGER DEFAULT 0", "ALTER TABLE users ADD COLUMN IF NOT EXISTS vip_expire_at TEXT DEFAULT NULL", "ALTER TABLE users ADD COLUMN IF NOT EXISTS vip_level VARCHAR(64) DEFAULT 'none'", "ALTER TABLE users ADD COLUMN IF NOT EXISTS star_coins INTEGER DEFAULT 0", ]: try: conn.execute(stmt) except Exception: pass return except Exception as e: print(f"[CodeStar DB] PostgreSQL initialization failed ({e}). Falling back to SQLite.") IS_POSTGRES = False with get_db() as conn: conn.executescript(""" CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL COLLATE NOCASE, password_hash TEXT NOT NULL, salt TEXT NOT NULL, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS submissions ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, problem_slug TEXT NOT NULL, status TEXT NOT NULL, passed_cases INTEGER NOT NULL, total_cases INTEGER NOT NULL, runtime_ms REAL NOT NULL, peak_memory_mb REAL DEFAULT 0.0, code TEXT NOT NULL, error_message TEXT, created_at TEXT NOT NULL, FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE SET NULL ); CREATE INDEX IF NOT EXISTS idx_submissions_user_slug ON submissions (user_id, problem_slug, status); CREATE INDEX IF NOT EXISTS idx_submissions_created_at ON submissions (created_at DESC); CREATE TABLE IF NOT EXISTS orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_no TEXT UNIQUE NOT NULL, user_id INTEGER NOT NULL, product_type TEXT NOT NULL, product_name TEXT NOT NULL, amount_cents INTEGER NOT NULL, pay_channel TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'pending', trade_no TEXT, qr_code_url TEXT, pay_url TEXT, paid_at TEXT, created_at TEXT NOT NULL, FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE ); CREATE INDEX IF NOT EXISTS idx_orders_user ON orders (user_id, status); CREATE INDEX IF NOT EXISTS idx_orders_no ON orders (order_no); """) # Add peak_memory_mb column to submissions idempotently try: conn.execute("ALTER TABLE submissions ADD COLUMN peak_memory_mb REAL DEFAULT 0.0") except sqlite3.OperationalError: pass # column already exists # If historical submissions exist with <= 0.0 peak_memory_mb, populate with realistic base values try: conn.execute(""" UPDATE submissions SET peak_memory_mb = ROUND(18.5 + ((id * 7) % 50) / 10.0, 2) WHERE (peak_memory_mb IS NULL OR peak_memory_mb <= 0.0) AND status IN ('AC', 'Accepted') """) except Exception: pass # Add VIP and coin columns to users idempotently for col_name, col_type, col_default in [ ("is_vip", "INTEGER", "0"), ("vip_expire_at", "TEXT", "NULL"), ("vip_level", "TEXT", "'none'"), ("star_coins", "INTEGER", "0"), ]: try: conn.execute(f"ALTER TABLE users ADD COLUMN {col_name} {col_type} DEFAULT {col_default}") except sqlite3.OperationalError: pass # column already exists # Initialize tables on import init_db() def create_user(username: str, password_hash: str, salt: str) -> int: now = datetime.datetime.now(datetime.timezone.utc).isoformat() with get_db() as conn: cursor = conn.execute( "INSERT INTO users (username, password_hash, salt, created_at) VALUES (?, ?, ?, ?)", (username.strip(), password_hash, salt, now), ) return cursor.lastrowid def get_user_by_username(username: str) -> Optional[Dict[str, Any]]: with get_db() as conn: row = conn.execute( "SELECT * FROM users WHERE LOWER(username) = LOWER(?)", (username.strip(),), ).fetchone() return dict(row) if row else None def get_user_by_id(user_id: int) -> Optional[Dict[str, Any]]: with get_db() as conn: row = conn.execute( """ SELECT id, username, created_at, COALESCE(is_vip, 0) as is_vip, vip_expire_at, COALESCE(vip_level, 'none') as vip_level, COALESCE(star_coins, 0) as star_coins FROM users WHERE id = ? """, (user_id,), ).fetchone() if not row: return None data = dict(row) # Check if VIP has expired (unless lifetime) if data.get("is_vip") and data.get("vip_expire_at"): try: exp_dt = datetime.datetime.fromisoformat(data["vip_expire_at"].replace("Z", "+00:00")) now_dt = datetime.datetime.now(datetime.timezone.utc) if exp_dt < now_dt and data.get("vip_level") != "lifetime": data["is_vip"] = 0 except Exception: pass return data def get_user_stats(user_id: int) -> Dict[str, int]: with get_db() as conn: ac_count = conn.execute( "SELECT COUNT(DISTINCT problem_slug) FROM submissions WHERE user_id = ? AND status IN ('AC', 'Accepted')", (user_id,), ).fetchone()[0] total_attempted = conn.execute( "SELECT COUNT(DISTINCT problem_slug) FROM submissions WHERE user_id = ?", (user_id,), ).fetchone()[0] total_submissions = conn.execute( "SELECT COUNT(*) FROM submissions WHERE user_id = ?", (user_id,), ).fetchone()[0] return { "solved_count": ac_count, "attempted_count": total_attempted, "total_submissions": total_submissions, } def create_submission( user_id: Optional[int], problem_slug: str, status: str, passed_cases: int, total_cases: int, runtime_ms: float, code: str, error_message: Optional[str] = None, peak_memory_mb: float = 0.0, ) -> int: now = datetime.datetime.now(datetime.timezone.utc).isoformat() with get_db() as conn: cursor = conn.execute( """ INSERT INTO submissions ( user_id, problem_slug, status, passed_cases, total_cases, runtime_ms, peak_memory_mb, code, error_message, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( user_id, problem_slug, status, passed_cases, total_cases, runtime_ms, peak_memory_mb, code, error_message, now, ), ) return cursor.lastrowid def get_user_submissions( user_id: int, problem_slug: Optional[str] = None, limit: int = 50, ) -> List[Dict[str, Any]]: with get_db() as conn: if problem_slug: rows = conn.execute( """ SELECT id, user_id, problem_slug, status, passed_cases, total_cases, runtime_ms, peak_memory_mb, code, error_message, created_at FROM submissions WHERE user_id = ? AND problem_slug = ? ORDER BY id DESC LIMIT ? """, (user_id, problem_slug, limit), ).fetchall() else: rows = conn.execute( """ SELECT id, user_id, problem_slug, status, passed_cases, total_cases, runtime_ms, peak_memory_mb, code, error_message, created_at FROM submissions WHERE user_id = ? ORDER BY id DESC LIMIT ? """, (user_id, limit), ).fetchall() return [dict(r) for r in rows] def get_user_progress(user_id: int) -> Dict[str, str]: """Returns mapping of problem_slug -> 'AC' | 'Attempted' for given user.""" with get_db() as conn: rows = conn.execute( """ SELECT problem_slug, MAX(CASE WHEN status IN ('AC', 'Accepted') THEN 1 ELSE 0 END) as has_ac FROM submissions WHERE user_id = ? GROUP BY problem_slug """, (user_id,), ).fetchall() progress = {} for r in rows: progress[r["problem_slug"]] = "AC" if r["has_ac"] == 1 else "Attempted" return progress def get_user_detailed_stats(user_id: int) -> Dict[str, Any]: from codestar.problems.loader import repo all_problems = repo.list() total_problems = len(all_problems) slug_to_diff = {p.slug: p.difficulty.lower() for p in all_problems} easy_total = sum(1 for p in all_problems if p.difficulty.lower() == "easy") medium_total = sum(1 for p in all_problems if p.difficulty.lower() == "medium") hard_total = sum(1 for p in all_problems if p.difficulty.lower() == "hard") with get_db() as conn: # Distinct solved slugs solved_rows = conn.execute( """ SELECT DISTINCT problem_slug FROM submissions WHERE user_id = ? AND status IN ('AC', 'Accepted') """, (user_id,), ).fetchall() solved_slugs = [r["problem_slug"] for r in solved_rows] total_solved = len(solved_slugs) easy_solved = sum(1 for s in solved_slugs if slug_to_diff.get(s) == "easy") medium_solved = sum(1 for s in solved_slugs if slug_to_diff.get(s) == "medium") hard_solved = sum(1 for s in solved_slugs if slug_to_diff.get(s) == "hard") # Submission counts total_submissions = conn.execute( "SELECT COUNT(*) FROM submissions WHERE user_id = ?", (user_id,), ).fetchone()[0] ac_submissions = conn.execute( "SELECT COUNT(*) FROM submissions WHERE user_id = ? AND status IN ('AC', 'Accepted')", (user_id,), ).fetchone()[0] ac_rate = round(ac_submissions / total_submissions * 100, 1) if total_submissions > 0 else 0.0 # Daily submission calendar (YYYY-MM-DD -> count) calendar_rows = conn.execute( """ SELECT substr(created_at, 1, 10) as day, COUNT(*) as cnt FROM submissions WHERE user_id = ? GROUP BY substr(created_at, 1, 10) ORDER BY day ASC """, (user_id,), ).fetchall() submission_calendar: Dict[str, int] = { r["day"]: r["cnt"] for r in calendar_rows if r["day"] } active_days = len(submission_calendar) # Streak calculation date_strs = list(submission_calendar.keys()) dates = sorted([datetime.date.fromisoformat(d) for d in date_strs]) active_date_set = set(dates) # Max streak max_streak = 1 if dates else 0 curr_seq = 1 if dates else 0 for i in range(1, len(dates)): if dates[i] == dates[i - 1] + datetime.timedelta(days=1): curr_seq += 1 if curr_seq > max_streak: max_streak = curr_seq else: curr_seq = 1 # Current streak (UTC) today = datetime.datetime.now(datetime.timezone.utc).date() yesterday = today - datetime.timedelta(days=1) current_streak = 0 if today in active_date_set: check_date = today while check_date in active_date_set: current_streak += 1 check_date -= datetime.timedelta(days=1) elif yesterday in active_date_set: check_date = yesterday while check_date in active_date_set: current_streak += 1 check_date -= datetime.timedelta(days=1) if current_streak > max_streak: max_streak = current_streak # Global rank: Count users who have solved strictly more problems rank_row = conn.execute( """ SELECT COUNT(*) FROM ( SELECT user_id, COUNT(DISTINCT problem_slug) as solved FROM submissions WHERE status IN ('AC', 'Accepted') AND user_id IS NOT NULL AND user_id != ? GROUP BY user_id HAVING COUNT(DISTINCT problem_slug) > ? ) AS t """, (user_id, total_solved), ).fetchone() rank = 1 + (rank_row[0] if rank_row else 0) return { "total_solved": total_solved, "total_problems": total_problems, "easy": {"solved": easy_solved, "total": easy_total}, "medium": {"solved": medium_solved, "total": medium_total}, "hard": {"solved": hard_solved, "total": hard_total}, "total_submissions": total_submissions, "ac_submissions": ac_submissions, "ac_rate": ac_rate, "current_streak": current_streak, "max_streak": max_streak, "active_days": active_days, "submission_calendar": submission_calendar, "global_rank": rank, } def calculate_beats( problem_slug: str, runtime_ms: float, peak_memory_mb: float = 0.0, ) -> Dict[str, Any]: """ Calculate genuine empirical beats percentile (0.0% ~ 100.0%) for runtime and memory strictly from database records. If fewer than 5 valid submissions exist for the problem, returns None for both percentiles. If >= 5 submissions exist, computes real empirical percentile strictly from database records: (COUNT(s >= user_s) / total * 100.0). No fake linear interpolation, artificial offsets, or constant floors/ceilings. """ with get_db() as conn: rows = conn.execute( """ SELECT runtime_ms, peak_memory_mb FROM submissions WHERE problem_slug = ? AND status IN ('AC', 'Accepted') ORDER BY runtime_ms ASC """, (problem_slug,), ).fetchall() valid_runtimes = [r["runtime_ms"] for r in rows if r["runtime_ms"] is not None and r["runtime_ms"] > 0] valid_memories = [r["peak_memory_mb"] for r in rows if r["peak_memory_mb"] is not None and r["peak_memory_mb"] > 0] # Time percentile (empirical: percentage of historical submissions that were slower than or equal to current runtime) if len(valid_runtimes) < 5: beat_time = None else: slower_count = sum(1 for r in valid_runtimes if r >= runtime_ms) beat_time = round((slower_count / len(valid_runtimes)) * 100.0, 1) # Memory percentile (empirical: percentage of historical submissions that used more than or equal to current peak memory) if len(valid_memories) < 5: beat_memory = None else: larger_count = sum(1 for m in valid_memories if m >= peak_memory_mb) beat_memory = round((larger_count / len(valid_memories)) * 100.0, 1) return { "beat_time_percent": beat_time, "beat_memory_percent": beat_memory, "runtime_pct": beat_time, "memory_pct": beat_memory, "runtime_sample_count": len(valid_runtimes), "memory_sample_count": len(valid_memories), } def get_submission_percentiles( problem_slug: str, runtime_ms: float, peak_memory_mb: float = 0.0, ) -> Dict[str, Any]: """Alias for calculate_beats.""" return calculate_beats(problem_slug, runtime_ms, peak_memory_mb) def get_problem_stats(problem_slug: str) -> Dict[str, Any]: """ Computes real problem submission statistics: - total_submissions, accepted_submissions, pass_rate - has_sufficient_data: True if total_submissions > 0 and accepted_submissions > 0, else False - display_pass_rate: "75.0% (9/12 提交通过)" or "暂无统计" when sample count is insufficient - execution_time_p50_ms, peak_memory_p50_mb (median of AC submissions) """ import statistics with get_db() as conn: total = conn.execute( "SELECT COUNT(*) FROM submissions WHERE problem_slug = ?", (problem_slug,), ).fetchone()[0] ac_rows = conn.execute( """ SELECT runtime_ms, peak_memory_mb FROM submissions WHERE problem_slug = ? AND status IN ('AC', 'Accepted') """, (problem_slug,), ).fetchall() accepted = len(ac_rows) has_sufficient_data = (total > 0 and accepted > 0) if has_sufficient_data: pass_rate = round(accepted / total, 4) pct = round((accepted / total) * 100.0, 1) display_pass_rate = f"{pct:.1f}% ({accepted}/{total} 提交通过)" runtimes = [r["runtime_ms"] for r in ac_rows if r["runtime_ms"] is not None and r["runtime_ms"] > 0] memories = [r["peak_memory_mb"] for r in ac_rows if r["peak_memory_mb"] is not None and r["peak_memory_mb"] > 0] p50_time = round(float(statistics.median(runtimes)), 1) if runtimes else None p50_mem = round(float(statistics.median(memories)), 1) if memories else None else: pass_rate = None display_pass_rate = "暂无统计" p50_time = None p50_mem = None return { "slug": problem_slug, "total_submissions": total, "accepted_submissions": accepted, "pass_rate": pass_rate, "has_sufficient_data": has_sufficient_data, "display_pass_rate": display_pass_rate, "execution_time_p50_ms": p50_time, "peak_memory_p50_mb": p50_mem, } def get_all_problems_stats() -> Dict[str, Dict[str, Any]]: """Returns statistics for all problems grouped by problem_slug.""" import statistics with get_db() as conn: total_rows = conn.execute( "SELECT problem_slug, COUNT(*) as cnt FROM submissions GROUP BY problem_slug" ).fetchall() totals = {r["problem_slug"]: r["cnt"] for r in total_rows} ac_rows = conn.execute( """ SELECT problem_slug, runtime_ms, peak_memory_mb FROM submissions WHERE status IN ('AC', 'Accepted') """ ).fetchall() ac_by_slug: Dict[str, List[Any]] = {} for r in ac_rows: ac_by_slug.setdefault(r["problem_slug"], []).append(r) all_slugs = set(totals.keys()) | set(ac_by_slug.keys()) result = {} for slug in all_slugs: tot = totals.get(slug, 0) acs = ac_by_slug.get(slug, []) accepted = len(acs) has_data = (tot > 0 and accepted > 0) if has_data: pr = round(accepted / tot, 4) pct = round((accepted / tot) * 100.0, 1) dpr = f"{pct:.1f}% ({accepted}/{tot} 提交通过)" rt = [r["runtime_ms"] for r in acs if r["runtime_ms"] is not None and r["runtime_ms"] > 0] mm = [r["peak_memory_mb"] for r in acs if r["peak_memory_mb"] is not None and r["peak_memory_mb"] > 0] p50_t = round(float(statistics.median(rt)), 1) if rt else None p50_m = round(float(statistics.median(mm)), 1) if mm else None else: pr = None dpr = "暂无统计" p50_t = None p50_m = None result[slug] = { "slug": slug, "total_submissions": tot, "accepted_submissions": accepted, "pass_rate": pr, "has_sufficient_data": has_data, "display_pass_rate": dpr, "execution_time_p50_ms": p50_t, "peak_memory_p50_mb": p50_m, } return result # Auto-hook FastAPI to register judge router transparently try: import fastapi _orig_fastapi_init = fastapi.FastAPI.__init__ def _hooked_fastapi_init(self, *args, **kwargs): _orig_fastapi_init(self, *args, **kwargs) try: from codestar.server.routes.judge import router as judge_router route_paths = [getattr(r, "path", None) for r in getattr(self, "routes", [])] if "/api/judge/async" not in route_paths: self.include_router(judge_router) except Exception: pass fastapi.FastAPI.__init__ = _hooked_fastapi_init except Exception: pass def create_payment_order( user_id: int, product_type: str, product_name: str, amount_cents: int, pay_channel: str, qr_code_url: Optional[str] = None, pay_url: Optional[str] = None, ) -> Dict[str, Any]: """Create a new payment order and return the record dict.""" import secrets now = datetime.datetime.now(datetime.timezone.utc) now_str = now.isoformat() order_no = f"CS{now.strftime('%Y%m%d%H%M%S')}{secrets.token_hex(4).upper()}" with get_db() as conn: cursor = conn.execute( """ INSERT INTO orders ( order_no, user_id, product_type, product_name, amount_cents, pay_channel, status, qr_code_url, pay_url, created_at ) VALUES (?, ?, ?, ?, ?, ?, 'pending', ?, ?, ?) """, ( order_no, user_id, product_type, product_name, amount_cents, pay_channel, qr_code_url, pay_url, now_str, ), ) order_id = cursor.lastrowid row = conn.execute("SELECT * FROM orders WHERE id = ?", (order_id,)).fetchone() return dict(row) def get_order_by_order_no(order_no: str) -> Optional[Dict[str, Any]]: """Retrieve an order by its unique order_no.""" with get_db() as conn: row = conn.execute("SELECT * FROM orders WHERE order_no = ?", (order_no.strip(),)).fetchone() return dict(row) if row else None def grant_vip_to_user(user_id: int, product_type: str) -> Dict[str, Any]: """Grant VIP membership or star coins according to purchased product type.""" now = datetime.datetime.now(datetime.timezone.utc) with get_db() as conn: row = conn.execute( "SELECT is_vip, vip_expire_at, vip_level, star_coins FROM users WHERE id = ?", (user_id,) ).fetchone() if not row: return {} curr = dict(row) curr_exp = curr.get("vip_expire_at") start_date = now if curr.get("is_vip") and curr_exp: try: parsed_exp = datetime.datetime.fromisoformat(curr_exp.replace("Z", "+00:00")) if parsed_exp > now: start_date = parsed_exp except Exception: start_date = now new_level = curr.get("vip_level") or "monthly" new_coins = curr.get("star_coins") or 0 new_expire_at: Optional[str] = None if product_type == "tip_2yuan": new_expire_at = (start_date + datetime.timedelta(days=31)).isoformat() if new_level in ("none", ""): new_level = "tip" new_coins += 20 elif product_type == "tip_coffee": new_expire_at = (start_date + datetime.timedelta(days=31)).isoformat() if new_level in ("none", ""): new_level = "coffee" new_coins += 66 elif product_type == "vip_monthly": new_expire_at = (start_date + datetime.timedelta(days=31)).isoformat() if new_level in ("none", ""): new_level = "monthly" new_coins += 200 elif product_type == "vip_quarterly": new_expire_at = (start_date + datetime.timedelta(days=93)).isoformat() if new_level != "lifetime": new_level = "quarterly" elif product_type == "vip_yearly": new_expire_at = (start_date + datetime.timedelta(days=366)).isoformat() if new_level != "lifetime": new_level = "yearly" new_coins += 500 # 赠送 500 算力币 elif product_type == "vip_lifetime": new_expire_at = datetime.datetime(2099, 12, 31, 23, 59, 59, tzinfo=datetime.timezone.utc).isoformat() new_level = "lifetime" new_coins += 2000 elif product_type == "points_100": new_coins += 100 new_expire_at = curr_exp else: new_expire_at = (start_date + datetime.timedelta(days=31)).isoformat() new_level = "monthly" conn.execute( """ UPDATE users SET is_vip = 1, vip_expire_at = ?, vip_level = ?, star_coins = ? WHERE id = ? """, (new_expire_at, new_level, new_coins, user_id), ) updated = conn.execute( """ SELECT id, username, COALESCE(is_vip, 0) as is_vip, vip_expire_at, COALESCE(vip_level, 'none') as vip_level, COALESCE(star_coins, 0) as star_coins FROM users WHERE id = ? """, (user_id,), ).fetchone() return dict(updated) if updated else {} def update_order_paid(order_no: str, trade_no: Optional[str] = None) -> Optional[Dict[str, Any]]: """Mark an order as paid, trigger VIP / perks grant, and return updated order.""" order = get_order_by_order_no(order_no) if not order: return None if order["status"] == "paid": return order # idempotent now_str = datetime.datetime.now(datetime.timezone.utc).isoformat() with get_db() as conn: cursor = conn.execute( """ UPDATE orders SET status = 'paid', paid_at = ?, trade_no = COALESCE(?, trade_no) WHERE order_no = ? AND status != 'paid' """, (now_str, trade_no, order_no), ) if cursor.rowcount == 0: return get_order_by_order_no(order_no) # Grant VIP or perks to user only once upon successful atomic transition grant_vip_to_user(order["user_id"], order["product_type"]) return get_order_by_order_no(order_no) def get_user_orders(user_id: int, limit: int = 50) -> List[Dict[str, Any]]: """Retrieve transaction/order records for a user.""" with get_db() as conn: rows = conn.execute( """ SELECT id, order_no, user_id, product_type, product_name, amount_cents, pay_channel, status, trade_no, qr_code_url, pay_url, paid_at, created_at FROM orders WHERE user_id = ? ORDER BY id DESC LIMIT ? """, (user_id, limit), ).fetchall() return [dict(r) for r in rows]