18040180_《雌堕企划》——《成为女店主之奴》第七人 赵研博

admin

import json
import time
import threading
import urllib.request
import urllib.error

BASE = "http://localhost:5472/api"


def call(method, path, token=None, body=None, timeout=180):
    data = None
    headers = {"Content-Type": "application/json"}
    if body is not None:
        data = json.dumps(body, ensure_ascii=False).encode("utf-8")
    if token:
        headers["Authorization"] = "Bearer " + token
    req = urllib.request.Request(BASE + path, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return r.status, json.loads(r.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        raw = e.read().decode("utf-8", "replace")
        try:
            return e.code, json.loads(raw)
        except Exception:
            return e.code, {"__nonjson__": raw[:200]}


st, admin = call("POST", "/auth/login", body={"username": "admin", "password": "admin123"})
atok = admin["access_token"]

uname = "race%d" % int(time.time())
st, reg = call("POST", "/auth/register", body={"username": uname, "password": "testpass123"})
utok = reg["access_token"]
uid = reg["user"]["id"]
print("fresh user uid=%d points=%d" % (uid, reg["user"]["points"]))

# Spend down to exactly 15 points so only ONE analysis should be affordable.
# 99 -> need 84 spent = wrong; instead directly assert: 99/15 -> 6 allowed.
# Simpler: fire 8 concurrent requests from 99 points. Honest max = 6 successes.
results = []
lock = threading.Lock()


def worker(n):
    st, body = call("POST", "/ai/analyze", token=utok,
                    body={"prompt": "Race probe %d: one sentence on AI." % n,
                          "category": "行业分析"})
    with lock:
        results.append((n, st, body.get("cost") if st == 200 else body))


threads = [threading.Thread(target=worker, args=(i,)) for i in range(8)]
print("\nfiring 8 concurrent /ai/analyze with 99 points (honest max = 6 paid analyses)")
t0 = time.time()
for t in threads:
    t.start()
for t in threads:
    t.join()
print("elapsed %.1fs" % (time.time() - t0))

ok = [r for r in results if r[1] == 200]
bad = [r for r in results if r[1] != 200]
print("  HTTP 200 (analysis delivered): %d" % len(ok))
print("  non-200: %d" % len(bad))
for n, st, body in sorted(bad):
    print("     req#%d -> %s %s" % (n, st, body))

st, me = call("GET", "/auth/me", token=utok)
print("\nfinal points = %s (expected 99 - 15*%d = %d)"
      % (me["points"], len(ok), 99 - 15 * len(ok)))
st, hist = call("GET", "/user/analyses", token=utok)
billed = sum(a["cost"] for a in hist)
print("stored analysis records = %d, sum(cost) = %s" % (len(hist), billed))
print("points actually consumed = %d" % (99 - me["points"]))
if billed != (99 - me["points"]):
    print("  *** MISMATCH: billed %s points on records but only %d deducted from wallet ***"
          % (billed, 99 - me["points"]))