Working stat estimates

This commit is contained in:
2025-11-25 14:22:36 -05:00
parent e4bb274ca3
commit 93a5da75eb
2 changed files with 85 additions and 134 deletions

1
.gitignore vendored
View File

@@ -1 +1,2 @@
*.venv *.venv
temp.md

214
main.py
View File

@@ -12,6 +12,7 @@ TORN_API_KEY = "9VLK0Wte1BwXOheB"
ENEMY_FACTION_ID = 52935 ENEMY_FACTION_ID = 52935
YOUR_FACTION_ID = 654321 YOUR_FACTION_ID = 654321
ALLOWED_CHANNEL_ID = 1442876328536707316 ALLOWED_CHANNEL_ID = 1442876328536707316
FFSCOUTER_KEY = "XYmWPO9ZYkLqnv3v"
POLL_INTERVAL = 30 POLL_INTERVAL = 30
HIT_CHECK_INTERVAL = 60 HIT_CHECK_INTERVAL = 60
@@ -47,71 +48,96 @@ class HitDispatchBot(commands.Bot):
# Create bot now (ONE bot only) # Create bot now (ONE bot only)
bot = HitDispatchBot(command_prefix="!", intents=intents) bot = HitDispatchBot(command_prefix="!", intents=intents)
# ------------------------------------------------------------------ async def fetch_ffscouter_stats(session: aiohttp.ClientSession, torn_id: int):
# Helper get a single member's crimes & networth """
# ------------------------------------------------------------------ Calls FFScouter and returns predicted battle stats.
async def _get_user_stats(session: aiohttp.ClientSession, user_id: int): Uses an existing aiohttp session (caller must provide).
url = f"https://api.torn.com/v2/user/{user_id}/personalstats?stat=criminaloffenses,networth&key={TORN_API_KEY}" """
url = f"https://ffscouter.com/api/v1/get-stats?key={FFSCOUTER_KEY}&targets={torn_id}"
#print(url)
async with session.get(url) as resp: async with session.get(url) as resp:
if resp.status != 200:
print(f"FFScouter Error for {torn_id}:", resp.status)
return None
data = await resp.json() data = await resp.json()
# FFScouter v1 returns: {"code":200,"message":"OK","data": {"<id>": {...}}}
if "data" not in data:
return None
inner = data["data"]
return inner.get(str(torn_id))
crimes = 0
networth_raw = 0
for stat in data.get("personalstats", []):
if stat.get("name") == "criminaloffenses":
crimes = int(stat.get("value", 0))
elif stat.get("name") == "networth":
networth_raw = float(stat.get("value", 0))
networth_mill = networth_raw / 1_000_000
return crimes, networth_mill
# ------------------------------------------------------------------
# Updated fetch_enemy_faction now pulls crimes & networth
# ------------------------------------------------------------------
async def fetch_enemy_faction(): async def fetch_enemy_faction():
""" """
Pulls all active members from the target faction and enriches each Pulls faction members from Torn (selections=members),
member with level, crimes and networth (in millions). then fetches FFScouter stats in a single batch request.
Returns a list of enemies with name, id, level, and estimated BS.
""" """
url = ( url = (
f"https://api.torn.com/v2/faction/{ENEMY_FACTION_ID}" f"https://api.torn.com/v2/faction/{ENEMY_FACTION_ID}"
"?selections=members" f"?selections=members&key={TORN_API_KEY}"
f"&key={TORN_API_KEY}"
) )
enemies = []
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
# --- Fetch faction members ---
async with session.get(url) as resp: async with session.get(url) as resp:
if resp.status != 200:
print("Torn faction fetch error:", resp.status)
return enemies
data = await resp.json() data = await resp.json()
# The response is a dict → {"members": [ {...}, ... ]}
members_list = data.get("members", []) members_list = data.get("members", [])
if not members_list:
return enemies
enemies: list[dict] = [] # --- Build comma-separated list of IDs ---
member_ids = [
str(info.get("player_id", info.get("id", 0)))
for info in members_list
if info.get("status", {}).get("state", "Unknown") in ("Okay", "Idle")
]
if not member_ids:
return enemies
ids_str = ",".join(member_ids)
# --- Single FFScouter batch request ---
ff_url = f"https://ffscouter.com/api/v1/get-stats?key={FFSCOUTER_KEY}&targets={ids_str}"
async with session.get(ff_url) as resp:
if resp.status != 200:
print("FFScouter batch error:", resp.status)
ff_data_list = []
else:
ff_data_list = await resp.json()
# --- Map FFScouter data by player_id for quick lookup ---
ff_map = {str(d["player_id"]): d for d in ff_data_list}
# --- Build final enemies list ---
for info in members_list: for info in members_list:
status_state = info["status"]["state"] pid = str(info.get("player_id", info.get("id", 0)))
if status_state not in ["Okay", "Idle"]: state = info.get("status", {}).get("state", "Unknown")
if state not in ("Okay", "Idle"):
continue continue
name = info.get("name", "Unknown")
level = int(info.get("level", 0)) level = int(info.get("level", 0))
est = ff_map.get(pid, {}).get("bs_estimate_human", "?")
# ------------------------------------------------------------------
# Pull the peruser stats (crimes & networth)
# ------------------------------------------------------------------
crimes, networth_mill = await _get_user_stats(session, info["id"])
enemies.append({ enemies.append({
"id": int(info["id"]), "id": int(pid),
"name": info["name"], "name": name,
"status": status_state, "level": level,
"level": level, "estimate": est
"crimes": crimes,
"networth": networth_mill, # millions
}) })
return enemies return enemies
async def update_enemy_queue_loop(): async def update_enemy_queue_loop():
global enemy_queue global enemy_queue
@@ -124,86 +150,6 @@ async def update_enemy_queue_loop():
print(f"Enemy queue updated: {len(enemy_queue)} valid targets.") print(f"Enemy queue updated: {len(enemy_queue)} valid targets.")
await asyncio.sleep(POLL_INTERVAL) await asyncio.sleep(POLL_INTERVAL)
# ==============================
# STAT SYSTEM
# ==============================
def estimate_battlestats(level: int, crimes: int, networth: float) -> str:
"""
Returns the scorerange string that matches the chart.
"""
# ------------------------------------------------------------------
# Bucket definitions (same as in the original code)
# ------------------------------------------------------------------
level_buckets = [
(6, "2k25k"),
(11, "20k25k"),
(26, "200k250k"),
(31, "2m2.5m"),
(50, "20m35m"),
(float('inf'), "200m250m")
]
crimes_buckets = [
(5000, "2k25k"),
(10000, "20k25k"),
(20000, "200k250k"),
(30000, "2m2.5m"),
(50000, "20m35m"),
(float('inf'), "200m250m")
]
networth_buckets = [
(50, "2k25k"), # <50m
(500, "20k25k"), # <500m
(5_000, "200k250k"), # <5b
(50_000, "2m2.5m"), # <50b
(200_000, "20m35m"), # <200b
(float('inf'), "200m250m")
]
def find_bucket(value: float, buckets):
for limit, label in buckets:
if value < limit:
return label
return "Unknown"
lvl_label = find_bucket(level, level_buckets)
crime_label = find_bucket(crimes, crimes_buckets)
net_label = find_bucket(networth, networth_buckets)
# ------------------------------------------------------------------
# Convert a label like '200k250k' into its lower bound in points.
# We only need the *first* number everything after the dash is
# ignored. The unit (k/m/b) tells us how to scale it.
# ------------------------------------------------------------------
def score_value(label: str) -> float:
"""
Convert a label such as '200k250k' or '20m35' into the numeric
lower bound in points.
"""
m = re.match(r"(\d+(?:\.\d*)?)([kmb]?)", label, flags=re.IGNORECASE)
if not m:
return float('inf') # shouldn't happen
number_str, unit = m.groups()
try:
num = float(number_str)
except ValueError:
return float('inf')
unit = unit.lower()
if unit == 'm':
return num * 1_000_000
if unit == 'k':
return num * 1_000
# No unit → plain points
return num
chosen_label = min([lvl_label, crime_label, net_label], key=score_value)
return chosen_label
# ============================== # ==============================
# ASSIGNMENT SYSTEM # ASSIGNMENT SYSTEM
@@ -224,6 +170,7 @@ async def assign_next_target():
if not enemy_queue: if not enemy_queue:
return None, None return None, None
# Pop the weakest enemy (already sorted by score_value)
enemy = enemy_queue.pop(0) enemy = enemy_queue.pop(0)
attacker = get_next_attacker() attacker = get_next_attacker()
@@ -238,7 +185,6 @@ async def assign_next_target():
return enemy, attacker return enemy, attacker
async def monitor_assignments_loop(): async def monitor_assignments_loop():
await bot.wait_until_ready() await bot.wait_until_ready()
@@ -346,18 +292,22 @@ async def stats(ctx):
await ctx.send("No active members found.") await ctx.send("No active members found.")
return return
for e in enemies: lines = [
score_range = estimate_battlestats(e["level"], e["crimes"], e["networth"]) f"**{e['name']}** (ID:{e['id']}) | Lv {e['level']} | Estimated BS: {e['estimate']}"
# await ctx.send( for e in enemies
# f'{e["name"]} (ID:{e["id"]}) | ' ]
# f'Lv {e["level"]}, Crimes {e["crimes"]}, NetWorth ${e["networth"]:,.1f}m → '
# f'Score: {score_range}' # Discord chunking
# ) chunk = ""
print( for line in lines:
f'{e["name"]} (ID:{e["id"]}) | ' if len(chunk) + len(line) > 1900:
f'Lv {e["level"]}, Crimes {e["crimes"]}, NetWorth ${e["networth"]:,.1f}m → ' await ctx.send(chunk)
f'Score: {score_range}' chunk = ""
) chunk += line + "\n"
if chunk:
await ctx.send(chunk)
# ============================== # ==============================
# BOT READY # BOT READY
# ============================== # ==============================