73 lines
2.3 KiB
Python
73 lines
2.3 KiB
Python
"""
|
|
token_pool.py — GoRouter token pool with rotation
|
|
Usage:
|
|
python token_pool.py # show pool status
|
|
python token_pool.py next # get next token (round-robin)
|
|
python token_pool.py random # get random token
|
|
python token_pool.py refresh # reload from tokens.json
|
|
"""
|
|
import json, random, sys
|
|
from pathlib import Path
|
|
|
|
TOKENS_FILE = Path(__file__).parent / "tokens.json"
|
|
POOL_FILE = Path(__file__).parent / "pool_state.json"
|
|
|
|
class TokenPool:
|
|
def __init__(self):
|
|
self.tokens = []
|
|
self.index = 0
|
|
self._load()
|
|
|
|
def _load(self):
|
|
if TOKENS_FILE.exists():
|
|
raw = json.loads(TOKENS_FILE.read_text(encoding="utf-8"))
|
|
self.tokens = [t for t in raw if t.get("platform") == "gorouter"]
|
|
if POOL_FILE.exists():
|
|
state = json.loads(POOL_FILE.read_text(encoding="utf-8"))
|
|
self.index = state.get("index", 0)
|
|
|
|
def save_state(self):
|
|
POOL_FILE.write_text(json.dumps({"index": self.index, "total": len(self.tokens)}))
|
|
|
|
def refresh(self):
|
|
self._load()
|
|
return len(self.tokens)
|
|
|
|
def next(self) -> dict | None:
|
|
if not self.tokens:
|
|
return None
|
|
token = [REDACTED] % len(self.tokens)]
|
|
self.index = (self.index + 1) % len(self.tokens)
|
|
self.save_state()
|
|
return token
|
|
|
|
def random_token(self) -> dict | None:
|
|
if not self.tokens:
|
|
return None
|
|
return random.choice(self.tokens)
|
|
|
|
def status(self):
|
|
return {
|
|
"total": len(self.tokens),
|
|
"index": self.index,
|
|
"value": f"${len(self.tokens) * 70}",
|
|
"tokens": [{"email": t["github_email"], "token_preview": t["token"][:8] + "..."} for t in self.tokens]
|
|
}
|
|
|
|
if __name__ == "__main__":
|
|
pool = TokenPool()
|
|
cmd = sys.argv[1] if len(sys.argv) > 1 else "status"
|
|
|
|
if cmd == "next":
|
|
t = pool.next()
|
|
if t: print(t["token"])
|
|
else: print("NO_TOKENS")
|
|
elif cmd == "random":
|
|
t = pool.random_token()
|
|
if t: print(t["token"])
|
|
else: print("NO_TOKENS")
|
|
elif cmd == "refresh":
|
|
n = pool.refresh()
|
|
print(f"Pool refreshed: {n} tokens")
|
|
else:
|
|
print(json.dumps(pool.status(), indent=2, ensure_ascii=False)) |