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

User Log and Persistent Faction Information

Browse Source
This commit is contained in:
jerick
2026-01-27 14:48:46 -05:00
parent 4ae3a9eb17
commit 4850c16b87
39 changed files with 782 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
services/__pycache__/activity_log.cpython-311.pyc Normal file
View File

Binary file not shown.

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

Binary file not shown.

BIN
services/__pycache__/ffscouter.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.

51
services/activity_log.py Normal file
View File

@@ -0,0 +1,51 @@
# services/activity_log.py
"""Activity logging and user tracking system."""
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from collections import deque
import asyncio
class ActivityLogger:
def __init__(self, max_logs: int = 1000):
self.logs: deque = deque(maxlen=max_logs) # Keep last 1000 logs
self.active_users: Dict[str, datetime] = {} # username -> last_activity_time
self.lock = asyncio.Lock()
async def log_action(self, username: str, action: str, details: str = ""):
"""Log a user action"""
async with self.lock:
timestamp = datetime.now()
log_entry = {
"timestamp": timestamp.isoformat(),
"username": username,
"action": action,
"details": details
}
self.logs.append(log_entry)
# Update user activity
self.active_users[username] = timestamp
async def get_logs(self, limit: int = 100) -> List[Dict]:
"""Get recent logs"""
async with self.lock:
# Return most recent logs first
return list(self.logs)[-limit:][::-1]
async def get_active_users(self, timeout_minutes: int = 30) -> List[str]:
"""Get list of users active in the last N minutes"""
async with self.lock:
cutoff = datetime.now() - timedelta(minutes=timeout_minutes)
active = [
username for username, last_activity in self.active_users.items()
if last_activity >= cutoff
]
return sorted(active)
async def update_user_activity(self, username: str):
"""Update user's last activity timestamp"""
async with self.lock:
self.active_users[username] = datetime.now()
# Global instance
activity_logger = ActivityLogger()

65
services/bot_assignment.py
View File

@@ -6,6 +6,7 @@ from pathlib import Path
from typing import Dict, Optional
from datetime import datetime
from services.server_state import STATE
from services.activity_log import activity_logger
from config import ASSIGNMENT_TIMEOUT, ASSIGNMENT_REMINDER, ALLOWED_CHANNEL_ID, CHAIN_TIMER_THRESHOLD, TORN_API_KEY
class BotAssignmentManager:
@@ -28,7 +29,7 @@ class BotAssignmentManager:
self.load_discord_mapping()
def load_discord_mapping(self):
"""Load Torn ID to Discord ID mapping from JSON file"""
#Load Torn ID to Discord ID mapping from JSON file
path = Path("data/discord_mapping.json")
if not path.exists():
print("No discord_mapping.json found")
@@ -44,11 +45,11 @@ class BotAssignmentManager:
print(f"Error loading discord mapping: {e}")
def get_discord_id(self, torn_id: int) -> Optional[int]:
"""Get Discord user ID for a Torn player ID"""
#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."""
#Fetch chain timeout from Torn API. Returns seconds remaining, or None if no chain or error.
if not STATE.friendly_faction_id:
return None
@@ -69,7 +70,7 @@ class BotAssignmentManager:
return None
async def start(self):
"""Start the bot assignment loop"""
#Start the bot assignment loop
if self.running:
print("WARNING: Bot assignment already running")
return
@@ -94,18 +95,16 @@ class BotAssignmentManager:
print("Bot assignment stopped")
def friendly_has_active_target(self, friendly_id: int) -> bool:
"""Check if a friendly player already has an active target assigned"""
#Check if a friendly player already has an active target assigned
for target_data in self.active_targets.values():
if target_data["friendly_id"] == friendly_id:
return True
return False
def get_next_friendly_in_group(self, group_id: str, friendly_ids: list) -> Optional[int]:
"""
Get the next friendly in the group who should receive a target.
Prioritizes members with fewer hits.
Only returns friendlies who DON'T already have an active assignment.
"""
#Get the next friendly in the group who should receive a target.
#Prioritizes members with fewer hits.
#Only returns friendlies who DON'T already have an active assignment.
if not friendly_ids:
return None
@@ -130,10 +129,8 @@ class BotAssignmentManager:
return friendly_hits[0][0]
def get_next_enemy_in_group(self, group_id: str, enemy_ids: list) -> Optional[int]:
"""
Get the next enemy in the group who needs to be assigned.
Returns None if all enemies are already assigned or not attackable.
"""
#Get the next enemy in the group who needs to be assigned.
#Returns None if all enemies are already assigned or not attackable.
for eid in enemy_ids:
key = f"{group_id}:{eid}"
# If enemy is already assigned, skip them
@@ -150,7 +147,7 @@ class BotAssignmentManager:
return None
async def check_chain_timer(self):
"""Check chain timer and update chain state"""
#Check chain timer and update chain state
timeout = await self.fetch_chain_timer()
if timeout is None:
@@ -173,6 +170,8 @@ class BotAssignmentManager:
self.assigned_friendlies.clear()
self.current_group_index = 0
self.chain_warning_sent = False
# Log to activity
await activity_logger.log_action("System", "Chain Mode Activated", f"Timer: {timeout}s, Threshold: {threshold_seconds}s")
# Check if chain expired
elif timeout > threshold_seconds and self.chain_active:
@@ -181,6 +180,8 @@ class BotAssignmentManager:
self.assigned_friendlies.clear()
self.current_group_index = 0
self.chain_warning_sent = False
# Log to activity
await activity_logger.log_action("System", "Chain Mode Deactivated", f"Timer: {timeout}s exceeded threshold")
# Check if chain expired (timeout = 0)
elif timeout == 0 and self.chain_active:
@@ -189,14 +190,18 @@ class BotAssignmentManager:
self.assigned_friendlies.clear()
self.current_group_index = 0
self.chain_warning_sent = False
# Log to activity
await activity_logger.log_action("System", "Chain Expired", "Chain timer reached 0")
# 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
# Log to activity
await activity_logger.log_action("System", "Chain Expiration Warning", "30 seconds remaining")
async def send_chain_expiration_warning(self):
"""Send @here alert that chain is about to expire"""
#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:
@@ -208,7 +213,7 @@ class BotAssignmentManager:
print(f"Error sending chain warning: {e}")
async def assign_next_chain_hit(self):
"""Assign next hit in round-robin fashion through groups"""
#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
@@ -265,7 +270,7 @@ class BotAssignmentManager:
self.current_group_index = 0
async def assignment_loop(self):
"""Main loop that monitors chain timer and assigns targets"""
#Main loop that monitors chain timer and assigns targets
await self.bot.wait_until_ready()
print("Bot is ready, assignment loop running with chain timer monitoring")
@@ -307,7 +312,7 @@ class BotAssignmentManager:
await asyncio.sleep(5)
async def assign_target(self, group_id: str, friendly_id: int, enemy_id: int):
"""Assign an enemy target to a friendly player"""
#Assign an enemy target to a friendly player
# Get member data
friendly = STATE.friendly.get(friendly_id)
enemy = STATE.enemy.get(enemy_id)
@@ -367,6 +372,8 @@ class BotAssignmentManager:
if channel:
await channel.send(message)
print(f"Assigned {enemy.name} to {friendly.name} (Discord: {discord_user.name})")
# Log to activity
await activity_logger.log_action("System", "Hit Assigned", f"{friendly.name} -> {enemy.name} (Level {enemy.level})")
else:
print(f"Assignment channel {ALLOWED_CHANNEL_ID} not found")
self.active_targets[key]["failed"] = True
@@ -375,7 +382,7 @@ class BotAssignmentManager:
self.active_targets[key]["failed"] = True
async def monitor_active_targets(self):
"""Monitor active targets for status changes or timeouts"""
#Monitor active targets for status changes or timeouts
now = datetime.now()
to_reassign = []
@@ -404,7 +411,11 @@ class BotAssignmentManager:
if friendly_id in STATE.friendly:
# Increment hit count
STATE.friendly[friendly_id].hits += 1
print(f"SUCCESS: {STATE.friendly[friendly_id].name} successfully hospitalized {enemy.name}")
friendly_name = STATE.friendly[friendly_id].name
enemy_name = enemy.name
print(f"SUCCESS: {friendly_name} successfully hospitalized {enemy_name}")
# Log to activity
await activity_logger.log_action("System", "Hit Completed", f"{friendly_name} hospitalized {enemy_name}")
# Remove from active targets
del self.active_targets[key]
@@ -412,7 +423,13 @@ class BotAssignmentManager:
# Check if enemy is no longer attackable (traveling, etc.)
if enemy.status.lower() != "okay":
print(f"Target {enemy.name} is now '{enemy.status}' - removing assignment")
friendly_id = data["friendly_id"]
friendly_name = STATE.friendly.get(friendly_id).name if friendly_id in STATE.friendly else "Unknown"
enemy_name = enemy.name
enemy_status = enemy.status
print(f"Target {enemy_name} is now '{enemy_status}' - removing assignment")
# Log to activity
await activity_logger.log_action("System", "Hit Dropped", f"{friendly_name}'s target {enemy_name} became {enemy_status}")
del self.active_targets[key]
continue
@@ -445,7 +462,11 @@ class BotAssignmentManager:
friendly_ids = STATE.groups[group_id].get("friendly", [])
friendly_id = self.get_next_friendly_in_group(group_id, friendly_ids)
if friendly_id:
enemy_name = STATE.enemy.get(enemy_id).name if enemy_id in STATE.enemy else f"Enemy {enemy_id}"
friendly_name = STATE.friendly.get(friendly_id).name if friendly_id in STATE.friendly else f"Friendly {friendly_id}"
print(f"Reassigning enemy {enemy_id} (timeout)")
# Log to activity
await activity_logger.log_action("System", "Hit Timed Out", f"Reassigning {enemy_name} to {friendly_name} (previous assignee timed out)")
await self.assign_target(group_id, friendly_id, enemy_id)
# Global instance (will be initialized with bot in main.py)

7
services/ffscouter.py
View File

@@ -2,10 +2,9 @@ import aiohttp
from config import FFSCOUTER_KEY
async def fetch_batch_stats(ids: list[int]):
"""
Fetches predicted stats for a list of Torn IDs in a single FFScouter request.
Returns dict keyed by player_id.
"""
#Fetches predicted stats for a list of Torn IDs in a single FFScouter request.
#Returns dict keyed by player_id.
if not ids:
return {}

7
services/server_state.py
View File

@@ -30,6 +30,13 @@ class ServerState:
# faction IDs for API monitoring
self.friendly_faction_id: Optional[int] = None
self.enemy_faction_id: Optional[int] = None
# status refresh state
self.friendly_status_interval: int = 10
self.friendly_status_running: bool = False
self.enemy_status_interval: int = 10
self.enemy_status_running: bool = False
# concurrency lock for async safety
self.lock = asyncio.Lock()

40
services/torn_api.py
View File

@@ -17,10 +17,10 @@ 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.
kind: "friendly" or "enemy"
"""
#Fetch members + FFScouter estimates once and store in STATE.
#kind: "friendly" or "enemy"
url = f"https://api.torn.com/v2/faction/{faction_id}?selections=members&key={TORN_API_KEY}"
async with aiohttp.ClientSession() as session:
@@ -65,9 +65,7 @@ async def populate_faction(faction_id: int, kind: str):
# Status refresh loop
async def refresh_status_loop(faction_id: int, kind: str, lock: asyncio.Lock, interval: int):
"""
Periodically refresh member statuses in STATE.
"""
#Periodically refresh member statuses in STATE.
while True:
try:
url = f"https://api.torn.com/v2/faction/{faction_id}?selections=members&key={TORN_API_KEY}"
@@ -100,6 +98,8 @@ async def populate_friendly(faction_id: int):
return await populate_faction(faction_id, "friendly")
async def populate_enemy(faction_id: int):
# Store enemy faction ID
STATE.enemy_faction_id = faction_id
return await populate_faction(faction_id, "enemy")
async def start_friendly_status_loop(faction_id: int, interval: int):
@@ -109,6 +109,9 @@ async def start_friendly_status_loop(faction_id: int, interval: int):
friendly_status_task = asyncio.create_task(
refresh_status_loop(faction_id, "friendly", friendly_lock, interval)
)
# Save state
STATE.friendly_status_interval = interval
STATE.friendly_status_running = True
async def start_enemy_status_loop(faction_id: int, interval: int):
global enemy_status_task
@@ -117,3 +120,26 @@ async def start_enemy_status_loop(faction_id: int, interval: int):
enemy_status_task = asyncio.create_task(
refresh_status_loop(faction_id, "enemy", enemy_lock, interval)
)
# Save state
STATE.enemy_status_interval = interval
STATE.enemy_status_running = True
async def stop_friendly_status_loop():
global friendly_status_task
if friendly_status_task and not friendly_status_task.done():
friendly_status_task.cancel()
try:
await friendly_status_task
except asyncio.CancelledError:
pass
STATE.friendly_status_running = False
async def stop_enemy_status_loop():
global enemy_status_task
if enemy_status_task and not enemy_status_task.done():
enemy_status_task.cancel()
try:
await enemy_status_task
except asyncio.CancelledError:
pass
STATE.enemy_status_running = False