127 lines
3.8 KiB
Python
127 lines
3.8 KiB
Python
# services/torn_api.py
|
|
import aiohttp
|
|
import json
|
|
import asyncio
|
|
from pathlib import Path
|
|
from config import TORN_API_KEY, ENEMY_FACTION_ID, YOUR_FACTION_ID
|
|
from .ffscouter import fetch_batch_stats
|
|
|
|
ENEMY_FILE = Path("data/enemy_faction.json")
|
|
FRIENDLY_FILE = Path("data/friendly_faction.json")
|
|
|
|
# Track running tasks + current faction IDs
|
|
enemy_task = None
|
|
friendly_task = None
|
|
|
|
current_enemy_id = None
|
|
current_friendly_id = None
|
|
|
|
|
|
async def fetch_and_save_faction(faction_id: int, file_path: Path) -> bool:
|
|
"""
|
|
Fetches faction members from Torn, fetches their estimated BS from FFScouter,
|
|
and saves everything to a JSON file.
|
|
"""
|
|
url = f"https://api.torn.com/v2/faction/{faction_id}?selections=members&key={TORN_API_KEY}"
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(url) as resp:
|
|
if resp.status != 200:
|
|
print(f"Torn faction fetch error: {resp.status}")
|
|
return False
|
|
data = await resp.json()
|
|
|
|
members_list = data.get("members", [])
|
|
if not members_list:
|
|
return False
|
|
|
|
# Build list of IDs (Torn uses 'id', not 'player_id')
|
|
member_ids = [info.get("id") for info in members_list if "id" in info]
|
|
if not member_ids:
|
|
return False
|
|
|
|
# Fetch batch FFScouter stats
|
|
ff_data = await fetch_batch_stats(member_ids) # returns dict keyed by player_id
|
|
|
|
# Build final faction data
|
|
faction_data = []
|
|
for info in members_list:
|
|
pid = info.get("id")
|
|
if pid is None:
|
|
continue
|
|
|
|
est = ff_data.get(str(pid), {}).get("bs_estimate_human", "?")
|
|
member = {
|
|
"id": pid,
|
|
"name": info.get("name", "Unknown"),
|
|
"level": info.get("level", 0),
|
|
"status": info.get("status", {}).get("state", "Unknown"),
|
|
"estimate": est
|
|
}
|
|
faction_data.append(member)
|
|
|
|
# Save to file
|
|
file_path.parent.mkdir(exist_ok=True, parents=True) # ensure folder exists
|
|
with open(file_path, "w", encoding="utf-8") as f:
|
|
json.dump(faction_data, f, indent=2)
|
|
|
|
return True
|
|
|
|
|
|
|
|
|
|
#Loop for the constant update of members and the stop function
|
|
|
|
async def stop_task_if_running(task: asyncio.Task | None):
|
|
"""Cancel an existing running task safely."""
|
|
if task and not task.done():
|
|
task.cancel()
|
|
try:
|
|
await task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
async def faction_loop(faction_id: int, file_path: Path, interval: int):
|
|
"""
|
|
Runs fetch_and_save_faction() in a loop forever, waiting `interval`
|
|
seconds between iterations.
|
|
"""
|
|
while True:
|
|
try:
|
|
await fetch_and_save_faction(faction_id, file_path)
|
|
except Exception as e:
|
|
print(f"Error during faction loop for {faction_id}: {e}")
|
|
|
|
await asyncio.sleep(interval)
|
|
|
|
|
|
|
|
#Functions to call the loop, maybe add one to just call once?
|
|
async def update_enemy_faction(new_faction_id: int, interval: int):
|
|
global enemy_task, current_enemy_id
|
|
|
|
# If faction ID changes → stop old loop
|
|
if new_faction_id != current_enemy_id:
|
|
print(f"[ENEMY] Changing faction from {current_enemy_id} → {new_faction_id}")
|
|
await stop_task_if_running(enemy_task)
|
|
current_enemy_id = new_faction_id
|
|
|
|
# Start new loop
|
|
enemy_task = asyncio.create_task(
|
|
faction_loop(new_faction_id, ENEMY_FILE, interval)
|
|
)
|
|
|
|
|
|
async def update_friendly_faction(new_faction_id: int, interval: int):
|
|
global friendly_task, current_friendly_id
|
|
|
|
if new_faction_id != current_friendly_id:
|
|
print(f"[FRIENDLY] Changing faction from {current_friendly_id} → {new_faction_id}")
|
|
await stop_task_if_running(friendly_task)
|
|
current_friendly_id = new_faction_id
|
|
|
|
friendly_task = asyncio.create_task(
|
|
faction_loop(new_faction_id, FRIENDLY_FILE, interval)
|
|
)
|
|
|