Wroking stat retrieval, inaccurate stat estimate
This commit is contained in:
793
main.py
793
main.py
@@ -1,487 +1,374 @@
|
||||
#!/usr/bin/env python3
|
||||
# torn_hit_dispatch.py
|
||||
# Requirements: discord.py (2.x), aiohttp
|
||||
# pip install -U "discord.py" aiohttp
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
from typing import Dict, List, Optional
|
||||
import aiohttp
|
||||
import discord
|
||||
from discord.ext import commands, tasks
|
||||
import asyncio
|
||||
import aiohttp
|
||||
from discord.ext import commands
|
||||
import re
|
||||
|
||||
# ---------------------------
|
||||
# Configuration (edit me)
|
||||
# ---------------------------
|
||||
DISCORD_TOKEN = "YOUR_DISCORD_BOT_TOKEN"
|
||||
TORN_API_KEY = "YOUR_TORN_API_KEY"
|
||||
FACTION_ID = 123456 # the enemy faction id to poll for enemies
|
||||
DATA_FILE = "dispatch_state.json"
|
||||
ASSIGNMENT_TIMEOUT = 60 # seconds to wait before pinging assigned player
|
||||
REASSIGNMENT_TIMEOUT = 60 # additional seconds to wait before reassigning
|
||||
MAX_REASSIGN_ATTEMPTS = 5 # stop after this many reassignments to avoid loops
|
||||
PING_ROLE_ID = None # optional: role id to mention when broadcasting assignments
|
||||
# ==============================
|
||||
# CONFIGURATION
|
||||
# ==============================
|
||||
|
||||
# ---------------------------
|
||||
# Logging
|
||||
# ---------------------------
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
log = logging.getLogger("torn_dispatch")
|
||||
TORN_API_KEY = "9VLK0Wte1BwXOheB"
|
||||
ENEMY_FACTION_ID = 52935
|
||||
YOUR_FACTION_ID = 654321
|
||||
ALLOWED_CHANNEL_ID = 1442876328536707316
|
||||
|
||||
# ---------------------------
|
||||
# Torn API helper (implement your endpoints here)
|
||||
# ---------------------------
|
||||
class TornAPI:
|
||||
"""
|
||||
A small wrapper to communicate with Torn API. Replace the TODOs in the methods
|
||||
with the exact endpoints & logic you want. The rest of the bot depends only on these functions.
|
||||
"""
|
||||
def __init__(self, api_key: str):
|
||||
self.api_key = api_key
|
||||
self.base = "https://api.torn.com" # change if you have other host
|
||||
self.session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=15))
|
||||
POLL_INTERVAL = 30
|
||||
HIT_CHECK_INTERVAL = 60
|
||||
REASSIGN_DELAY = 120
|
||||
|
||||
async def close(self):
|
||||
await self.session.close()
|
||||
|
||||
async def get_enemy_members(self, faction_id: int) -> List[Dict]:
|
||||
"""
|
||||
Return a list of enemy members as dicts with at least:
|
||||
- 'player_id' (int)
|
||||
- 'name' (str)
|
||||
- 'status' keys to determine hospital/abroad (could be 'hospital' bool or status string)
|
||||
Implement this using the Torn API endpoint that returns faction members.
|
||||
"""
|
||||
# TODO: Replace with actual Torn API call for faction roster or selection
|
||||
# Example (pseudo):
|
||||
# url = f"{self.base}/faction/{faction_id}?selections=members&key={self.api_key}"
|
||||
# resp = await self.session.get(url)
|
||||
# data = await resp.json()
|
||||
# return data["members"] # adapt to real response shape
|
||||
raise NotImplementedError("Implement get_enemy_members() with Torn API")
|
||||
|
||||
async def is_player_available(self, player_id: int) -> bool:
|
||||
"""
|
||||
Check if a given player (enemy) is hittable:
|
||||
- not in hospital
|
||||
- not abroad/out of country
|
||||
Implement using Torn API player endpoint.
|
||||
"""
|
||||
# TODO: implement actual check using Torn API
|
||||
raise NotImplementedError("Implement is_player_available() with Torn API")
|
||||
|
||||
async def was_player_hit_recently(self, enemy_player_id: int, since_seconds: int = 120) -> bool:
|
||||
"""
|
||||
Check if the enemy player has been hit in the recent window.
|
||||
You can check API endpoints for attacks or health changes; returns True if recently hit.
|
||||
"""
|
||||
# TODO: implement by checking attack history or health changes via Torn API
|
||||
raise NotImplementedError("Implement was_player_hit_recently() with Torn API")
|
||||
# ---------------------------
|
||||
|
||||
|
||||
# ---------------------------
|
||||
# Bot state classes
|
||||
# ---------------------------
|
||||
class Participant:
|
||||
def __init__(self, discord_id: int, display_name: str):
|
||||
self.discord_id = discord_id
|
||||
self.display_name = display_name
|
||||
|
||||
def to_dict(self):
|
||||
return {"discord_id": self.discord_id, "display_name": self.display_name}
|
||||
|
||||
@staticmethod
|
||||
def from_dict(d):
|
||||
return Participant(d["discord_id"], d.get("display_name", "Unknown"))
|
||||
|
||||
|
||||
class DispatchState:
|
||||
def __init__(self):
|
||||
# participants enrolled (round-robin)
|
||||
self.participants: List[Participant] = []
|
||||
# queue of enemy player ids to assign (list of dicts with id + name)
|
||||
self.enemy_list: List[Dict] = []
|
||||
# current assignments: enemy_id -> assignment record
|
||||
self.assignments: Dict[int, Dict] = {}
|
||||
# round-robin pointer
|
||||
self.rr_index = 0
|
||||
|
||||
def to_dict(self):
|
||||
return {
|
||||
"participants": [p.to_dict() for p in self.participants],
|
||||
"enemy_list": self.enemy_list,
|
||||
"assignments": self.assignments,
|
||||
"rr_index": self.rr_index,
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def from_dict(d):
|
||||
s = DispatchState()
|
||||
s.participants = [Participant.from_dict(x) for x in d.get("participants", [])]
|
||||
s.enemy_list = d.get("enemy_list", [])
|
||||
s.assignments = d.get("assignments", {})
|
||||
s.rr_index = d.get("rr_index", 0)
|
||||
return s
|
||||
|
||||
|
||||
# ---------------------------
|
||||
# Discord bot
|
||||
# ---------------------------
|
||||
intents = discord.Intents.default()
|
||||
intents.message_content = True
|
||||
intents.members = True
|
||||
|
||||
bot = commands.Bot(command_prefix="!", intents=intents)
|
||||
torn = TornAPI(TORN_API_KEY)
|
||||
state = DispatchState()
|
||||
monitor_tasks: Dict[int, asyncio.Task] = {} # enemy_id -> monitor task
|
||||
state_lock = asyncio.Lock() # protect shared state writes
|
||||
# ==============================
|
||||
# STATE STORAGE
|
||||
# ==============================
|
||||
|
||||
enrolled_attackers = []
|
||||
enemy_queue = []
|
||||
active_assignments = {}
|
||||
round_robin_index = 0
|
||||
|
||||
# ==============================
|
||||
# BOT SUBCLASS
|
||||
# ==============================
|
||||
|
||||
class HitDispatchBot(commands.Bot):
|
||||
async def setup_hook(self):
|
||||
# Start background loops
|
||||
self.bg_tasks = []
|
||||
self.bg_tasks.append(asyncio.create_task(update_enemy_queue_loop()))
|
||||
self.bg_tasks.append(asyncio.create_task(monitor_assignments_loop()))
|
||||
|
||||
async def cog_check(self, ctx):
|
||||
return ctx.channel.id == ALLOWED_CHANNEL_ID
|
||||
|
||||
|
||||
# ---------------------------
|
||||
# Persistence helpers
|
||||
# ---------------------------
|
||||
def load_state():
|
||||
try:
|
||||
with open(DATA_FILE, "r", encoding="utf-8") as f:
|
||||
obj = json.load(f)
|
||||
return DispatchState.from_dict(obj)
|
||||
except FileNotFoundError:
|
||||
return DispatchState()
|
||||
except Exception as e:
|
||||
log.exception("Failed to load state: %s", e)
|
||||
return DispatchState()
|
||||
# Create bot now (ONE bot only)
|
||||
bot = HitDispatchBot(command_prefix="!", intents=intents)
|
||||
|
||||
def save_state():
|
||||
try:
|
||||
with open(DATA_FILE, "w", encoding="utf-8") as f:
|
||||
json.dump(state.to_dict(), f, indent=2)
|
||||
except Exception:
|
||||
log.exception("Failed to save state")
|
||||
# ------------------------------------------------------------------
|
||||
# Helper – get a single member's crimes & networth
|
||||
# ------------------------------------------------------------------
|
||||
async def _get_user_stats(session: aiohttp.ClientSession, user_id: int):
|
||||
url = f"https://api.torn.com/v2/user/{user_id}/personalstats?stat=criminaloffenses,networth&key={TORN_API_KEY}"
|
||||
async with session.get(url) as resp:
|
||||
data = await resp.json()
|
||||
|
||||
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():
|
||||
"""
|
||||
Pulls all active members from the target faction and enriches each
|
||||
member with level, crimes and networth (in millions).
|
||||
"""
|
||||
url = (
|
||||
f"https://api.torn.com/v2/faction/{ENEMY_FACTION_ID}"
|
||||
"?selections=members"
|
||||
f"&key={TORN_API_KEY}"
|
||||
)
|
||||
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(url) as resp:
|
||||
data = await resp.json()
|
||||
|
||||
# The response is a dict → {"members": [ {...}, ... ]}
|
||||
members_list = data.get("members", [])
|
||||
|
||||
enemies: list[dict] = []
|
||||
|
||||
for info in members_list:
|
||||
status_state = info["status"]["state"]
|
||||
if status_state not in ["Okay", "Idle"]:
|
||||
continue
|
||||
|
||||
level = int(info.get("level", 0))
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Pull the per‑user stats (crimes & networth)
|
||||
# ------------------------------------------------------------------
|
||||
crimes, networth_mill = await _get_user_stats(session, info["id"])
|
||||
|
||||
enemies.append({
|
||||
"id": int(info["id"]),
|
||||
"name": info["name"],
|
||||
"status": status_state,
|
||||
"level": level,
|
||||
"crimes": crimes,
|
||||
"networth": networth_mill, # millions
|
||||
})
|
||||
|
||||
return enemies
|
||||
|
||||
async def update_enemy_queue_loop():
|
||||
global enemy_queue
|
||||
|
||||
await bot.wait_until_ready()
|
||||
|
||||
while not bot.is_closed():
|
||||
print("Refreshing enemy list...")
|
||||
enemy_queue = await fetch_enemy_faction()
|
||||
print(f"Enemy queue updated: {len(enemy_queue)} valid targets.")
|
||||
await asyncio.sleep(POLL_INTERVAL)
|
||||
|
||||
# ==============================
|
||||
# STAT SYSTEM
|
||||
# ==============================
|
||||
|
||||
def estimate_battlestats(level: int, crimes: int, networth: float) -> str:
|
||||
"""
|
||||
Returns the score‑range string that matches the chart.
|
||||
"""
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Bucket definitions (same as in the original code)
|
||||
# ------------------------------------------------------------------
|
||||
level_buckets = [
|
||||
(6, "2k‑25k"),
|
||||
(11, "20k‑25k"),
|
||||
(26, "200k‑250k"),
|
||||
(31, "2m‑2.5m"),
|
||||
(50, "20m‑35m"),
|
||||
(float('inf'), "200m‑250m")
|
||||
]
|
||||
|
||||
crimes_buckets = [
|
||||
(5000, "2k‑25k"),
|
||||
(10000, "20k‑25k"),
|
||||
(20000, "200k‑250k"),
|
||||
(30000, "2m‑2.5m"),
|
||||
(50000, "20m‑35m"),
|
||||
(float('inf'), "200m‑250m")
|
||||
]
|
||||
|
||||
networth_buckets = [
|
||||
(50, "2k‑25k"), # < 50 m
|
||||
(500, "20k‑25k"), # < 500 m
|
||||
(5_000, "200k‑250k"), # < 5 b
|
||||
(50_000, "2m‑2.5m"), # < 50 b
|
||||
(200_000, "20m‑35m"), # < 200 b
|
||||
(float('inf'), "200m‑250m")
|
||||
]
|
||||
|
||||
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 '200k‑250k' 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 '200k‑250k' or '20m‑35' 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
|
||||
|
||||
|
||||
# ---------------------------
|
||||
# Utility functions
|
||||
# ---------------------------
|
||||
def next_participant() -> Optional[Participant]:
|
||||
if not state.participants:
|
||||
return None
|
||||
# round-robin selection
|
||||
p = state.participants[state.rr_index % len(state.participants)]
|
||||
state.rr_index = (state.rr_index + 1) % max(1, len(state.participants))
|
||||
return p
|
||||
# ==============================
|
||||
# ASSIGNMENT SYSTEM
|
||||
# ==============================
|
||||
|
||||
async def mention_user(ctx_or_member, discord_id: int) -> str:
|
||||
# helper to build mention text (works in-channel and in messages)
|
||||
return f"<@{discord_id}>"
|
||||
def get_next_attacker():
|
||||
global round_robin_index
|
||||
|
||||
async def assign_enemy_to_next(enemy: Dict, reason: str = "initial") -> Optional[Dict]:
|
||||
"""
|
||||
Assigns the enemy to the next available participant and creates an assignment record.
|
||||
Returns the assignment record or None if no participants.
|
||||
"""
|
||||
async with state_lock:
|
||||
p = next_participant()
|
||||
if p is None:
|
||||
return None
|
||||
enemy_id = int(enemy["player_id"])
|
||||
record = {
|
||||
"enemy": enemy,
|
||||
"assigned_to": p.to_dict(),
|
||||
"assigned_at": asyncio.get_event_loop().time(),
|
||||
"attempts": 0,
|
||||
"last_pinged": None,
|
||||
"reason": reason,
|
||||
}
|
||||
state.assignments[enemy_id] = record
|
||||
save_state()
|
||||
return record
|
||||
if not enrolled_attackers:
|
||||
return None
|
||||
|
||||
async def unassign_enemy(enemy_id: int):
|
||||
async with state_lock:
|
||||
if enemy_id in state.assignments:
|
||||
del state.assignments[enemy_id]
|
||||
save_state()
|
||||
attacker = enrolled_attackers[round_robin_index]
|
||||
round_robin_index = (round_robin_index + 1) % len(enrolled_attackers)
|
||||
return attacker
|
||||
|
||||
|
||||
# ---------------------------
|
||||
# Monitor & assignment loop
|
||||
# ---------------------------
|
||||
async def monitor_assignment(guild: discord.Guild, channel: discord.TextChannel, enemy_id: int):
|
||||
"""
|
||||
Monitors a single assignment: wait ASSIGNMENT_TIMEOUT, check if hit, ping if not.
|
||||
If still not hit, reassign after REASSIGNMENT_TIMEOUT. Repeat until MAX_REASSIGN_ATTEMPTS.
|
||||
"""
|
||||
try:
|
||||
for attempt in range(1, MAX_REASSIGN_ATTEMPTS + 1):
|
||||
async with state_lock:
|
||||
record = state.assignments.get(enemy_id)
|
||||
if not record:
|
||||
# assignment removed or resolved
|
||||
return
|
||||
assigned_to = record["assigned_to"]
|
||||
enemy = record["enemy"]
|
||||
async def assign_next_target():
|
||||
if not enemy_queue:
|
||||
return None, None
|
||||
|
||||
# Wait first interval
|
||||
await asyncio.sleep(ASSIGNMENT_TIMEOUT)
|
||||
enemy = enemy_queue.pop(0)
|
||||
attacker = get_next_attacker()
|
||||
|
||||
# check if enemy was hit in the meantime
|
||||
try:
|
||||
hit = await torn.was_player_hit_recently(enemy_id, since_seconds=(ASSIGNMENT_TIMEOUT + 5))
|
||||
except NotImplementedError:
|
||||
# default fallback: assume not hit (so the bot will ping participants)
|
||||
hit = False
|
||||
except Exception:
|
||||
log.exception("Error checking hit status")
|
||||
hit = False
|
||||
if attacker is None:
|
||||
return None, None
|
||||
|
||||
if hit:
|
||||
# mark resolved
|
||||
await channel.send(f"Enemy **{enemy.get('name','?')}** was hit — resolving assignment.")
|
||||
await unassign_enemy(enemy_id)
|
||||
return
|
||||
active_assignments[enemy["id"]] = {
|
||||
"enemy": enemy,
|
||||
"attacker": attacker,
|
||||
"time_assigned": asyncio.get_event_loop().time()
|
||||
}
|
||||
|
||||
# ping the assigned participant
|
||||
mention = f"<@{assigned_to['discord_id']}>"
|
||||
role_mention = f"<@&{PING_ROLE_ID}>" if PING_ROLE_ID else ""
|
||||
await channel.send(f"{role_mention} {mention} — you were assigned to hit **{enemy.get('name','?')}** (ID {enemy_id}). Attempt {attempt}/{MAX_REASSIGN_ATTEMPTS}.")
|
||||
|
||||
# update attempt count and last_pinged
|
||||
async with state_lock:
|
||||
rec = state.assignments.get(enemy_id)
|
||||
if rec:
|
||||
rec["attempts"] = attempt
|
||||
rec["last_pinged"] = asyncio.get_event_loop().time()
|
||||
save_state()
|
||||
|
||||
# wait second interval
|
||||
await asyncio.sleep(REASSIGNMENT_TIMEOUT)
|
||||
|
||||
# check again if hit
|
||||
try:
|
||||
hit2 = await torn.was_player_hit_recently(enemy_id, since_seconds=(REASSIGNMENT_TIMEOUT + 5))
|
||||
except Exception:
|
||||
log.exception("Error checking hit status (2)")
|
||||
hit2 = False
|
||||
|
||||
if hit2:
|
||||
await channel.send(f"Enemy **{enemy.get('name','?')}** was hit — resolving assignment.")
|
||||
await unassign_enemy(enemy_id)
|
||||
return
|
||||
|
||||
# if not hit, reassign to next participant
|
||||
async with state_lock:
|
||||
# pick next participant (advance rr pointer)
|
||||
next_p = next_participant()
|
||||
if not next_p:
|
||||
await channel.send(f"No participants available to reassign **{enemy.get('name','?')}**.")
|
||||
# keep the existing assignment but increment attempts
|
||||
rec = state.assignments.get(enemy_id)
|
||||
if rec:
|
||||
rec["attempts"] += 1
|
||||
save_state()
|
||||
continue
|
||||
|
||||
# update assignment record
|
||||
rec = state.assignments.get(enemy_id)
|
||||
if rec:
|
||||
rec["assigned_to"] = next_p.to_dict()
|
||||
rec["assigned_at"] = asyncio.get_event_loop().time()
|
||||
rec["reason"] = f"reassign_attempt_{attempt}"
|
||||
save_state()
|
||||
|
||||
await channel.send(f"Reassigned **{enemy.get('name','?')}** to {next_p.display_name} (attempt {attempt}).")
|
||||
|
||||
# if reached here, exhausted attempts
|
||||
await channel.send(f"Exhausted attempts assigning **{state.assignments.get(enemy_id, {}).get('enemy', {}).get('name','?')}** — removing from assignments.")
|
||||
await unassign_enemy(enemy_id)
|
||||
|
||||
except asyncio.CancelledError:
|
||||
# task was cancelled (probably due to stop)
|
||||
log.info("Monitor for enemy %s cancelled", enemy_id)
|
||||
return
|
||||
except Exception:
|
||||
log.exception("Error in monitor_assignment")
|
||||
return enemy, attacker
|
||||
|
||||
|
||||
async def refresh_enemy_list():
|
||||
"""
|
||||
Pull the latest enemy list from Torn and populate state.enemy_list with enemies that are hittable.
|
||||
"""
|
||||
try:
|
||||
raw = await torn.get_enemy_members(FACTION_ID)
|
||||
available = []
|
||||
for r in raw:
|
||||
pid = int(r["player_id"])
|
||||
# check availability
|
||||
try:
|
||||
ok = await torn.is_player_available(pid)
|
||||
except NotImplementedError:
|
||||
# if not implemented, default to including them
|
||||
ok = True
|
||||
if ok:
|
||||
available.append({"player_id": pid, "name": r.get("name", "Unknown")})
|
||||
async with state_lock:
|
||||
state.enemy_list = available
|
||||
save_state()
|
||||
return available
|
||||
except Exception:
|
||||
log.exception("Failed to refresh enemy list")
|
||||
return []
|
||||
async def monitor_assignments_loop():
|
||||
await bot.wait_until_ready()
|
||||
|
||||
while not bot.is_closed():
|
||||
now = asyncio.get_event_loop().time()
|
||||
reassign_list = []
|
||||
|
||||
for enemy_id, data in list(active_assignments.items()):
|
||||
elapsed = now - data["time_assigned"]
|
||||
|
||||
if elapsed >= HIT_CHECK_INTERVAL and elapsed < HIT_CHECK_INTERVAL + 5:
|
||||
attacker_user = bot.get_user(data["attacker"])
|
||||
if attacker_user:
|
||||
try:
|
||||
await attacker_user.send(
|
||||
f"Reminder: You were assigned **{data['enemy']['name']}** and they are not down yet!"
|
||||
)
|
||||
except:
|
||||
pass
|
||||
|
||||
if elapsed >= REASSIGN_DELAY:
|
||||
reassign_list.append(enemy_id)
|
||||
|
||||
for enemy_id in reassign_list:
|
||||
info = active_assignments.pop(enemy_id)
|
||||
await reassign_target(info["enemy"])
|
||||
|
||||
await asyncio.sleep(5)
|
||||
|
||||
|
||||
async def assign_all_enemies(channel: discord.TextChannel):
|
||||
"""
|
||||
Iterate enemy_list and create assignments for unassigned enemies.
|
||||
"""
|
||||
async with state_lock:
|
||||
enemy_copy = list(state.enemy_list)
|
||||
async def reassign_target(enemy):
|
||||
attacker = get_next_attacker()
|
||||
|
||||
for enemy in enemy_copy:
|
||||
enemy_id = int(enemy["player_id"])
|
||||
async with state_lock:
|
||||
if enemy_id in state.assignments:
|
||||
continue
|
||||
rec = await assign_enemy_to_next(enemy, reason="batch_assign")
|
||||
if rec:
|
||||
# start monitor task for this enemy
|
||||
task = asyncio.create_task(monitor_assignment(channel.guild, channel, enemy_id))
|
||||
monitor_tasks[enemy_id] = task
|
||||
await channel.send(f"Assigned **{enemy.get('name','?')}** to {rec['assigned_to']['display_name']}.")
|
||||
else:
|
||||
await channel.send(f"No participants to assign **{enemy.get('name','?')}** — skipping.")
|
||||
if attacker is None:
|
||||
return
|
||||
|
||||
active_assignments[enemy["id"]] = {
|
||||
"enemy": enemy,
|
||||
"attacker": attacker,
|
||||
"time_assigned": asyncio.get_event_loop().time()
|
||||
}
|
||||
|
||||
attacker_user = bot.get_user(attacker)
|
||||
if attacker_user:
|
||||
try:
|
||||
await attacker_user.send(
|
||||
f"Target reassigned to you: **{enemy['name']}**!"
|
||||
)
|
||||
except:
|
||||
pass
|
||||
|
||||
|
||||
# ---------------------------
|
||||
# Bot commands
|
||||
# ---------------------------
|
||||
# ==============================
|
||||
# COMMANDS
|
||||
# ==============================
|
||||
|
||||
@bot.command()
|
||||
async def enroll(ctx):
|
||||
user_id = ctx.author.id
|
||||
|
||||
if user_id in enrolled_attackers:
|
||||
await ctx.send("You are already enrolled.")
|
||||
return
|
||||
|
||||
enrolled_attackers.append(user_id)
|
||||
await ctx.send(f"{ctx.author.mention} has been enrolled in the hit rotation!")
|
||||
|
||||
|
||||
@bot.command()
|
||||
async def drop(ctx):
|
||||
user_id = ctx.author.id
|
||||
|
||||
if user_id not in enrolled_attackers:
|
||||
await ctx.send("You are not enrolled.")
|
||||
return
|
||||
|
||||
enrolled_attackers.remove(user_id)
|
||||
await ctx.send(f"{ctx.author.mention} has been removed from the rotation.")
|
||||
|
||||
|
||||
@bot.command()
|
||||
async def next(ctx):
|
||||
enemy, attacker = await assign_next_target()
|
||||
|
||||
if enemy is None:
|
||||
await ctx.send("No targets available or no attackers enrolled.")
|
||||
return
|
||||
|
||||
attacker_user = bot.get_user(attacker)
|
||||
await ctx.send(f"Assigned **{enemy['name']}** → <@{attacker}>")
|
||||
|
||||
if attacker_user:
|
||||
try:
|
||||
await attacker_user.send(
|
||||
f"Hit assignment: **{enemy['name']}**"
|
||||
)
|
||||
except:
|
||||
pass
|
||||
|
||||
@bot.command()
|
||||
async def stats(ctx):
|
||||
enemies = await fetch_enemy_faction()
|
||||
|
||||
if not enemies:
|
||||
await ctx.send("No active members found.")
|
||||
return
|
||||
|
||||
for e in enemies:
|
||||
score_range = estimate_battlestats(e["level"], e["crimes"], e["networth"])
|
||||
# await ctx.send(
|
||||
# f'{e["name"]} (ID:{e["id"]}) | '
|
||||
# f'Lv {e["level"]}, Crimes {e["crimes"]}, NetWorth ${e["networth"]:,.1f}m → '
|
||||
# f'Score: {score_range}'
|
||||
# )
|
||||
print(
|
||||
f'{e["name"]} (ID:{e["id"]}) | '
|
||||
f'Lv {e["level"]}, Crimes {e["crimes"]}, NetWorth ${e["networth"]:,.1f}m → '
|
||||
f'Score: {score_range}'
|
||||
)
|
||||
# ==============================
|
||||
# BOT READY
|
||||
# ==============================
|
||||
|
||||
@bot.event
|
||||
async def on_ready():
|
||||
log.info(f"Logged in as {bot.user} (id: {bot.user.id})")
|
||||
global state
|
||||
state = load_state()
|
||||
log.info("State loaded: %d participants, %d enemies, %d assignments",
|
||||
len(state.participants), len(state.enemy_list), len(state.assignments))
|
||||
print(f"Logged in as {bot.user.name}")
|
||||
|
||||
@bot.command(name="enroll")
|
||||
async def enroll(ctx):
|
||||
"""Enroll yourself into the round-robin pool. Usage: !enroll"""
|
||||
async with state_lock:
|
||||
existing = [p for p in state.participants if p.discord_id == ctx.author.id]
|
||||
if existing:
|
||||
await ctx.send("You're already enrolled.")
|
||||
return
|
||||
p = Participant(ctx.author.id, ctx.author.display_name)
|
||||
state.participants.append(p)
|
||||
save_state()
|
||||
await ctx.send(f"Enrolled {ctx.author.display_name} in the hit queue.")
|
||||
|
||||
@bot.command(name="leave")
|
||||
async def leave(ctx):
|
||||
"""Leave the round-robin pool. Usage: !leave"""
|
||||
async with state_lock:
|
||||
before = len(state.participants)
|
||||
state.participants = [p for p in state.participants if p.discord_id != ctx.author.id]
|
||||
if len(state.participants) < before:
|
||||
save_state()
|
||||
await ctx.send("You have been removed from the pool.")
|
||||
else:
|
||||
await ctx.send("You were not enrolled.")
|
||||
# ==============================
|
||||
# RUN BOT
|
||||
# ==============================
|
||||
|
||||
@bot.command(name="list_participants")
|
||||
async def list_participants(ctx):
|
||||
"""List enrolled participants."""
|
||||
if not state.participants:
|
||||
await ctx.send("No participants enrolled.")
|
||||
return
|
||||
msg = "Enrolled participants:\n" + "\n".join(f"- {p.display_name} (discord:{p.discord_id})" for p in state.participants)
|
||||
await ctx.send(msg)
|
||||
|
||||
@bot.command(name="refresh_enemies")
|
||||
async def cmd_refresh_enemies(ctx):
|
||||
"""Refresh enemy list from Torn and show the count. Usage: !refresh_enemies"""
|
||||
enemies = await refresh_enemy_list()
|
||||
await ctx.send(f"Refreshed enemy list — {len(enemies)} hittable enemies loaded.")
|
||||
|
||||
@bot.command(name="start_round")
|
||||
async def start_round(ctx):
|
||||
"""Assign enemies to participants and start monitors. Usage: !start_round"""
|
||||
await refresh_enemy_list()
|
||||
await assign_all_enemies(ctx.channel)
|
||||
await ctx.send("Round started — assignments made and monitors running.")
|
||||
|
||||
@bot.command(name="stop_round")
|
||||
async def stop_round(ctx):
|
||||
"""Stop all monitors and clear assignments. Usage: !stop_round"""
|
||||
# Cancel monitor tasks
|
||||
for t in list(monitor_tasks.values()):
|
||||
t.cancel()
|
||||
monitor_tasks.clear()
|
||||
async with state_lock:
|
||||
state.assignments.clear()
|
||||
save_state()
|
||||
await ctx.send("Stopped round: monitors cancelled and assignments cleared.")
|
||||
|
||||
@bot.command(name="status")
|
||||
async def status(ctx):
|
||||
"""Show current assignments and queue status."""
|
||||
async with state_lock:
|
||||
parts = state.participants
|
||||
enemies = state.enemy_list
|
||||
assigns = state.assignments.copy()
|
||||
msg = f"Participants: {len(parts)}\nEnemies queued: {len(enemies)}\nAssignments: {len(assigns)}\n"
|
||||
if assigns:
|
||||
for eid, rec in assigns.items():
|
||||
enemy_name = rec["enemy"].get("name", "?")
|
||||
assignee = rec["assigned_to"].get("display_name", "?")
|
||||
attempts = rec.get("attempts", 0)
|
||||
msg += f"- {enemy_name} (ID {eid}) -> {assignee} (attempts {attempts})\n"
|
||||
await ctx.send(msg)
|
||||
|
||||
@bot.command(name="manual_assign")
|
||||
@commands.has_permissions(administrator=True)
|
||||
async def manual_assign(ctx, enemy_id: int, member: discord.Member):
|
||||
"""Manually assign a specific enemy to a participant. Usage: !manual_assign <enemy_id> @member"""
|
||||
enemy = None
|
||||
async with state_lock:
|
||||
for e in state.enemy_list:
|
||||
if int(e["player_id"]) == int(enemy_id):
|
||||
enemy = e
|
||||
break
|
||||
if enemy is None:
|
||||
await ctx.send("Enemy not found in current enemy list — try refreshing with !refresh_enemies")
|
||||
return
|
||||
# make assignment record
|
||||
rec = {
|
||||
"enemy": enemy,
|
||||
"assigned_to": Participant(member.id, member.display_name).to_dict(),
|
||||
"assigned_at": asyncio.get_event_loop().time(),
|
||||
"attempts": 0,
|
||||
"last_pinged": None,
|
||||
"reason": "manual",
|
||||
}
|
||||
state.assignments[int(enemy_id)] = rec
|
||||
save_state()
|
||||
# start monitor
|
||||
task = asyncio.create_task(monitor_assignment(ctx.guild, ctx.channel, int(enemy_id)))
|
||||
monitor_tasks[int(enemy_id)] = task
|
||||
await ctx.send(f"Manually assigned enemy {enemy.get('name','?')} to {member.display_name}.")
|
||||
|
||||
# ---------------------------
|
||||
# Clean shutdown
|
||||
# ---------------------------
|
||||
async def shutdown():
|
||||
await torn.close()
|
||||
await bot.close()
|
||||
|
||||
# ---------------------------
|
||||
# Run the bot
|
||||
# ---------------------------
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
bot.run(DISCORD_TOKEN)
|
||||
except KeyboardInterrupt:
|
||||
log.info("Interrupted — shutting down")
|
||||
asyncio.run(shutdown())
|
||||
bot.run("MTQ0Mjg3NjU3NTUzMDg3NzAxMQ.GNuHPr.UreuYD1B7YYjfsbfRcEbhFyjyqvhQDepRCN4kk")
|
||||
|
||||
Reference in New Issue
Block a user