"""Measures the LexGo API on a stated synthetic dataset. The numbers this prints describe one machine and one dataset; they are an observation, not a capacity promise. Everything runs against a database whose name contains ``lexgo`` and that the operator creates for the run, so no real instance is touched. """ from __future__ import annotations import argparse import hashlib import json import os import platform import statistics import subprocess import sys import time from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "scripts")) import ops # noqa: E402 WORDS = ("curiosity", "door", "step", "reading", "language", "garden", "window", "music", "river", "stone", "letter", "morning", "shadow", "bridge", "silence", "travel", "kitchen", "paper") def machine_facts() -> dict: facts = { "os": platform.platform(), "cpu_count": os.cpu_count(), "python": platform.python_version(), } try: import shutil if os.name == "nt": output = subprocess.run(["wmic", "computersystem", "get", "TotalPhysicalMemory"], capture_output=True, text=True, encoding="utf-8", errors="replace").stdout numbers = [line.strip() for line in output.splitlines() if line.strip().isdigit()] if numbers: facts["memory_gb"] = round(int(numbers[0]) / (1024 ** 3), 1) else: with open("/proc/meminfo") as handle: for line in handle: if line.startswith("MemTotal"): facts["memory_gb"] = round(int(line.split()[1]) / (1024 ** 2), 1) break facts["go"] = subprocess.run(["go", "version"], capture_output=True, text=True, encoding="utf-8", errors="replace").stdout.strip() del shutil except Exception: pass return facts def seed(database: str, chapters: int, chapter_words: int, terms: int, reviews: int) -> dict: """Fills a fresh benchmark database with a synthetic learner, book, terms and schedules. The account itself is left alone: it was created through the API beforehand, so its password hash is a real one and the benchmark measures the same login path a user takes. """ for table in ("lexgo_review_answers", "lexgo_term_reviews", "lexgo_terms", "lexgo_chapters", "lexgo_books", "lexgo_spaces", "lexgo_sessions"): ops.query(database, "DELETE FROM " + table) owner = int(ops.scalar(database, "SELECT user_id FROM sys_user ORDER BY user_id LIMIT 1", "0") or 0) if not owner: raise SystemExit("先创建一个账号再压测（例如 smoke 或 bootstrap）。") ops.query(database, "INSERT INTO lexgo_spaces (owner_id, language) VALUES (%d, 'en')" % owner) stamp = "2026-01-01 00:00:00.000" body = " ".join(WORDS[index % len(WORDS)] for index in range(chapter_words)) for chapter in range(1, chapters + 1): text = "%s chapter %d. " % (body, chapter) sha = hashlib.sha256(text.encode("utf-8")).hexdigest() ops.query(database, "INSERT INTO lexgo_books (id, owner_id, title, language, created_at, updated_at) " "VALUES (%d, %d, 'Bench book %d', 'en', '%s', '%s')" % (chapter, owner, chapter, stamp, stamp)) ops.query(database, "INSERT INTO lexgo_chapters (id, book_id, owner_id, ordinal, title, original_text, char_count, " "content_sha256, status, created_at, updated_at) VALUES (%d, %d, %d, 1, 'Bench %d', '%s', %d, " "'%s', 'ready', '%s', '%s')" % (chapter, chapter, owner, chapter, text.replace("'", "''"), len(text), sha, stamp, stamp)) rows = [] for term_id in range(1, terms + 1): word = "%s%d" % (WORDS[term_id % len(WORDS)], term_id) rows.append("(%d, %d, 'en', '%s', '%s', '虚构释义', '', 'new', 0, '%s', '%s')" % (term_id, owner, word, word, stamp, stamp)) if len(rows) == 500: ops.query(database, "INSERT INTO lexgo_terms (id, owner_id, language, term, original_form, definition, examples, " "status, level, created_at, updated_at) VALUES " + ",".join(rows)) rows = [] if rows: ops.query(database, "INSERT INTO lexgo_terms (id, owner_id, language, term, original_form, definition, examples, " "status, level, created_at, updated_at) VALUES " + ",".join(rows)) rows = [] for term_id in range(1, terms + 1): rows.append("(%d, %d, 'en', DATE_ADD('%s', INTERVAL %d SECOND), 0, 0, 0)" % (term_id, owner, stamp, term_id % 900000)) if len(rows) == 500: ops.query(database, "INSERT INTO lexgo_term_reviews (term_id, owner_id, language, due_at, review_count, " "correct_count, wrong_count) VALUES " + ",".join(rows)) rows = [] if rows: ops.query(database, "INSERT INTO lexgo_term_reviews (term_id, owner_id, language, due_at, review_count, " "correct_count, wrong_count) VALUES " + ",".join(rows)) if reviews: rows = [] for index in range(1, reviews + 1): rows.append("(%d, %d, 'b%063d', %d, 'correct', 'applied', 'new', 'learning', 0, 1, '%s', '%s', 0, '%s')" % (index, owner, index, (index % terms) + 1, stamp, stamp, stamp)) if len(rows) == 500: ops.query(database, "INSERT INTO lexgo_review_answers (id, owner_id, answer_key, term_id, grade, result, " "status_before, status_after, level_before, level_after, due_at_before, due_at_after, " "requeued, created_at) VALUES " + ",".join(rows)) rows = [] if rows: ops.query(database, "INSERT INTO lexgo_review_answers (id, owner_id, answer_key, term_id, grade, result, " "status_before, status_after, level_before, level_after, due_at_before, due_at_after, " "requeued, created_at) VALUES " + ",".join(rows)) return { "owner_id": owner, "chapters": chapters, "words_per_chapter": chapter_words, "terms": terms, "review_answers": reviews, "row_counts": ops.row_counts(database), } def measure(label: str, call, repeats: int) -> dict: timings = [] errors = 0 for index in range(repeats): started = time.perf_counter() status, _ = call(index) elapsed = (time.perf_counter() - started) * 1000 if status >= 400: errors += 1 timings.append(elapsed) timings.sort() return { "endpoint": label, "requests": repeats, "errors": errors, "p50_ms": round(statistics.median(timings), 1), "p95_ms": round(timings[min(len(timings) - 1, int(len(timings) * 0.95))], 1), "max_ms": round(timings[-1], 1), } def main() -> int: parser = argparse.ArgumentParser(description="在写明规模的人造数据集上测量接口耗时") parser.add_argument("--database", default="lexgo_bench") parser.add_argument("--api", default="http://127.0.0.1:8012") parser.add_argument("--user", default="bench_admin", help="压测账号") parser.add_argument("--chapters", type=int, default=20) parser.add_argument("--chapter-words", type=int, default=500) parser.add_argument("--terms", type=int, default=2000) parser.add_argument("--reviews", type=int, default=8000) parser.add_argument("--repeats", type=int, default=30) parser.add_argument("--out", default=str(ROOT / ".local" / "issue15-bench.json")) args = parser.parse_args() ops.load_env() database = ops.validate_database_name(args.database, "压测") dataset = seed(database, args.chapters, args.chapter_words, args.terms, args.reviews) print("数据集：" + json.dumps(dataset["row_counts"], ensure_ascii=False)) client = ops.Api(args.api) password = os.environ.get("LEXGO_BENCH_PASSWORD", os.environ.get("LEXGO_TRIAL_PASSWORD", "")) status, token = client.login(args.user, password) if status != 200 or not token: raise SystemExit(args.user + " 无法登录：请先在演练库上建号并设置 LEXGO_BENCH_PASSWORD。") # The login endpoint allows 30 attempts per minute per address, so a sustained loop would # measure the limiter instead of the login path. A short loop keeps the number meaningful. login_repeats = max(1, min(args.repeats, 10)) results = [measure("POST /login", lambda _: client.login(args.user, password), login_repeats)] results.append(measure("GET /books", lambda _: client.call("GET", "books", token), args.repeats)) results.append(measure("GET /chapters/1", lambda _: client.call("GET", "chapters/1", token), args.repeats)) results.append(measure("GET /chapters/1/tokens", lambda _: client.call("GET", "chapters/1/tokens", token), args.repeats)) results.append(measure("GET /terms (page 1)", lambda _: client.call("GET", "terms?page=1&limit=20", token), args.repeats)) results.append(measure("GET /terms?query (search)", lambda _: client.call("GET", "terms?query=curiosity&limit=20", token), args.repeats)) results.append(measure("GET /progress", lambda _: client.call("GET", "progress", token), args.repeats)) results.append(measure("GET /reviews/queue", lambda _: client.call("GET", "reviews/queue", token), args.repeats)) report = { "measured_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "api": args.api, "database": database, "dataset": dataset, "machine": machine_facts(), "mysql_server": ops.scalar("", "SELECT VERSION()", ""), "results": results, "note": "单机、单进程、无并发压力；数字是观察值，不是容量承诺。登录接口每分钟每地址限 30 次，" "因此登录只测 10 次，且该数字不代表登录吞吐。", } Path(args.out).write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") for row in results: print(" %-28s p50 %7.1f ms p95 %7.1f ms max %7.1f ms 错误 %d" % (row["endpoint"], row["p50_ms"], row["p95_ms"], row["max_ms"], row["errors"])) print("报告：" + args.out) return 0 if __name__ == "__main__": sys.exit(main())