From ec44cd645b9e690522a0ec1b353ff4ff969024d2 Mon Sep 17 00:00:00 2001 From: jerick Date: Sun, 23 Nov 2025 03:12:03 -0500 Subject: [PATCH] First code implementation --- .gitignore | 1 + main.py | 487 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 488 insertions(+) create mode 100644 .gitignore create mode 100644 main.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..266c348 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +*.venv \ No newline at end of file diff --git a/main.py b/main.py new file mode 100644 index 0000000..babd918 --- /dev/null +++ b/main.py @@ -0,0 +1,487 @@ +#!/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 + +# --------------------------- +# 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 + +# --------------------------- +# Logging +# --------------------------- +logging.basicConfig(level=logging.INFO) +log = logging.getLogger("torn_dispatch") + +# --------------------------- +# 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)) + + 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 + + +# --------------------------- +# 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() + +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") + + +# --------------------------- +# 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 + +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}>" + +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 + +async def unassign_enemy(enemy_id: int): + async with state_lock: + if enemy_id in state.assignments: + del state.assignments[enemy_id] + save_state() + + +# --------------------------- +# 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"] + + # Wait first interval + await asyncio.sleep(ASSIGNMENT_TIMEOUT) + + # 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 hit: + # mark resolved + await channel.send(f"Enemy **{enemy.get('name','?')}** was hit — resolving assignment.") + await unassign_enemy(enemy_id) + return + + # 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") + + +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 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) + + 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.") + + +# --------------------------- +# Bot commands +# --------------------------- +@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)) + +@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.") + +@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 @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())