jerick/faction_war_dispatch_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Smart hit assignation and chain monitoring

Browse Source
This commit is contained in:
jerick
2026-01-27 08:26:48 -05:00
parent 5ef8707122
commit 99ffe7f9e9
13 changed files with 188 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
services/__pycache__/bot_assignment.cpython-311.pyc
View File

Binary file not shown.

BIN
services/__pycache__/server_state.cpython-311.pyc
View File

Binary file not shown.

BIN
services/__pycache__/torn_api.cpython-311.pyc
View File

Binary file not shown.

182
services/bot_assignment.py
View File

@@ -1,11 +1,12 @@
# services/bot_assignment.py
import asyncio
import json
import aiohttp
from pathlib import Path
from typing import Dict, Optional
from datetime import datetime
from services.server_state import STATE
from config import ASSIGNMENT_TIMEOUT, ASSIGNMENT_REMINDER, ALLOWED_CHANNEL_ID
from config import ASSIGNMENT_TIMEOUT, ASSIGNMENT_REMINDER, ALLOWED_CHANNEL_ID, CHAIN_TIMER_THRESHOLD, TORN_API_KEY
class BotAssignmentManager:
def __init__(self, bot):
@@ -15,6 +16,14 @@ class BotAssignmentManager:
self.running = False
self.task = None
# Chain monitoring state
self.chain_timeout = 0 # Seconds remaining on chain
self.chain_active = False # Whether we're in assignment mode
self.assigned_friendlies = set() # Track who has been assigned in this chain session
self.current_group_index = 0 # For round-robin group assignment
self.chain_warning_sent = False # Track if 30-second warning was sent
self.last_chain_check = datetime.now()
# Load Discord mapping
self.load_discord_mapping()
@@ -38,6 +47,27 @@ class BotAssignmentManager:
"""Get Discord user ID for a Torn player ID"""
return self.discord_mapping.get(torn_id)
async def fetch_chain_timer(self) -> Optional[int]:
"""Fetch chain timeout from Torn API. Returns seconds remaining, or None if no chain or error."""
if not STATE.friendly_faction_id:
return None
try:
url = f"https://api.torn.com/v2/faction/{STATE.friendly_faction_id}/chain?key={TORN_API_KEY}"
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
if resp.status != 200:
return None
data = await resp.json()
# Get timeout value in seconds
chain_data = data.get("chain", {})
timeout = chain_data.get("timeout", 0)
return timeout
except Exception as e:
print(f"Error fetching chain timer: {e}")
return None
async def start(self):
"""Start the bot assignment loop"""
if self.running:
@@ -119,10 +149,125 @@ class BotAssignmentManager:
return eid
return None
async def check_chain_timer(self):
"""Check chain timer and update chain state"""
timeout = await self.fetch_chain_timer()
if timeout is None:
# No chain data or error
if self.chain_active:
print("Chain ended or API error - resetting chain state")
self.chain_active = False
self.assigned_friendlies.clear()
self.current_group_index = 0
self.chain_warning_sent = False
return
self.chain_timeout = timeout
threshold_seconds = CHAIN_TIMER_THRESHOLD * 60
# Check if we should enter chain mode
if timeout > 0 and timeout <= threshold_seconds and not self.chain_active:
print(f"Chain timer at {timeout}s (threshold: {threshold_seconds}s) - entering chain mode")
self.chain_active = True
self.assigned_friendlies.clear()
self.current_group_index = 0
self.chain_warning_sent = False
# Check if chain expired
elif timeout > threshold_seconds and self.chain_active:
print(f"Chain timer above threshold ({timeout}s > {threshold_seconds}s) - exiting chain mode")
self.chain_active = False
self.assigned_friendlies.clear()
self.current_group_index = 0
self.chain_warning_sent = False
# Check if chain expired (timeout = 0)
elif timeout == 0 and self.chain_active:
print("Chain expired - resetting chain state")
self.chain_active = False
self.assigned_friendlies.clear()
self.current_group_index = 0
self.chain_warning_sent = False
# Send 30-second warning
if self.chain_active and timeout <= 30 and not self.chain_warning_sent:
await self.send_chain_expiration_warning()
self.chain_warning_sent = True
async def send_chain_expiration_warning(self):
"""Send @here alert that chain is about to expire"""
try:
channel = self.bot.get_channel(ALLOWED_CHANNEL_ID)
if channel and STATE.friendly_faction_id:
faction_link = f"https://www.torn.com/factions.php?step=your#/tab=chain"
message = f"@here **CHAIN EXPIRING IN 30 SECONDS!** Attack a faction member to keep it alive!\n{faction_link}"
await channel.send(message)
print("Sent chain expiration warning")
except Exception as e:
print(f"Error sending chain warning: {e}")
async def assign_next_chain_hit(self):
"""Assign next hit in round-robin fashion through groups"""
# Only assign if there's no active assignment waiting
if self.active_targets:
return # Wait for current assignment to complete
# Get list of group IDs (sorted for consistency)
group_ids = sorted(STATE.groups.keys())
if not group_ids:
return
# Try to find a group with available friendly and enemy
attempts = 0
while attempts < len(group_ids):
# Get current group
group_id = group_ids[self.current_group_index % len(group_ids)]
async with STATE.lock:
friendly_ids = STATE.groups[group_id].get("friendly", [])
enemy_ids = STATE.groups[group_id].get("enemy", [])
# Find a friendly who hasn't been assigned yet
available_friendly = None
for fid in friendly_ids:
if fid not in self.assigned_friendlies:
available_friendly = fid
break
# Find an attackable enemy
enemy_id = self.get_next_enemy_in_group(group_id, enemy_ids)
if available_friendly and enemy_id:
# Make assignment
self.assigned_friendlies.add(available_friendly)
await self.assign_target(group_id, available_friendly, enemy_id)
print(f"Chain hit assigned: Group {group_id}, Friendly {available_friendly} -> Enemy {enemy_id}")
# Move to next group for next assignment
self.current_group_index += 1
return
# Move to next group and try again
self.current_group_index += 1
attempts += 1
# If we've tried all groups and no assignments possible
# Check if we need to reset assigned_friendlies (everyone has been assigned)
async with STATE.lock:
all_friendlies = set()
for assignments in STATE.groups.values():
all_friendlies.update(assignments.get("friendly", []))
if self.assigned_friendlies and self.assigned_friendlies >= all_friendlies:
print("All friendlies have been assigned - resetting for round-robin")
self.assigned_friendlies.clear()
self.current_group_index = 0
async def assignment_loop(self):
"""Main loop that assigns targets and monitors status"""
"""Main loop that monitors chain timer and assigns targets"""
await self.bot.wait_until_ready()
print("Bot is ready, assignment loop running")
print("Bot is ready, assignment loop running with chain timer monitoring")
first_run = True
while self.running:
@@ -136,31 +281,18 @@ class BotAssignmentManager:
continue
if first_run:
print("Bot activated - processing assignments")
print("Bot activated - chain timer monitoring enabled")
first_run = False
# Process each group
has_assignments = False
async with STATE.lock:
for group_id, assignments in STATE.groups.items():
friendly_ids = assignments.get("friendly", [])
enemy_ids = assignments.get("enemy", [])
# Check chain timer every 30 seconds
now = datetime.now()
if (now - self.last_chain_check).total_seconds() >= 30:
await self.check_chain_timer()
self.last_chain_check = now
if friendly_ids and enemy_ids:
has_assignments = True
if not friendly_ids or not enemy_ids:
continue
# Try to assign any unassigned enemies
enemy_id = self.get_next_enemy_in_group(group_id, enemy_ids)
if enemy_id:
friendly_id = self.get_next_friendly_in_group(group_id, friendly_ids)
if friendly_id:
await self.assign_target(group_id, friendly_id, enemy_id)
if not has_assignments and STATE.bot_running:
print("No group assignments found - drag members into groups first!")
# If chain is active, try to assign next hit
if self.chain_active:
await self.assign_next_chain_hit()
# Monitor active targets for status changes or timeouts
await self.monitor_active_targets()

3
services/server_state.py
View File

@@ -28,6 +28,9 @@ class ServerState:
# bot running flag
self.bot_running: bool = False
# faction IDs for API monitoring
self.friendly_faction_id: Optional[int] = None
# concurrency lock for async safety
self.lock = asyncio.Lock()

14
services/torn_api.py
View File

@@ -5,9 +5,8 @@ from config import TORN_API_KEY
from .ffscouter import fetch_batch_stats
from .server_state import STATE
# -----------------------------
# Tasks
# -----------------------------
friendly_status_task = None
enemy_status_task = None
@@ -15,9 +14,8 @@ enemy_status_task = None
friendly_lock = asyncio.Lock()
enemy_lock = asyncio.Lock()
# -----------------------------
# Populate faction (memory only)
# -----------------------------
async def populate_faction(faction_id: int, kind: str):
"""
Fetch members + FFScouter estimates once and store in STATE.
@@ -64,9 +62,8 @@ async def populate_faction(faction_id: int, kind: str):
return True
# -----------------------------
# Status refresh loop
# -----------------------------
async def refresh_status_loop(faction_id: int, kind: str, lock: asyncio.Lock, interval: int):
"""
Periodically refresh member statuses in STATE.
@@ -95,10 +92,11 @@ async def refresh_status_loop(faction_id: int, kind: str, lock: asyncio.Lock, in
await asyncio.sleep(interval)
# -----------------------------
# Public API helpers
# -----------------------------
async def populate_friendly(faction_id: int):
# Store friendly faction ID for chain monitoring
STATE.friendly_faction_id = faction_id
return await populate_faction(faction_id, "friendly")
async def populate_enemy(faction_id: int):