108 lines
3.4 KiB
Python
108 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import requests
|
|
import sqlite3
|
|
import configparser
|
|
from collections import defaultdict
|
|
|
|
# Testing out the fleetyard API
|
|
|
|
class FleetyardAPI:
|
|
def __init__(self, base_url):
|
|
self.base_url = base_url
|
|
self.session = requests.Session()
|
|
|
|
def create_session(self, ini_path="fleetyard_login.ini"):
|
|
"""Create a new session using credentials from an INI file."""
|
|
config = configparser.ConfigParser()
|
|
config.read(ini_path)
|
|
login = config.get('login', 'username', fallback=None)
|
|
password = config.get('login', 'password', fallback=None)
|
|
remember_me = config.getboolean('login', 'rememberMe', fallback=True)
|
|
if not login or not password:
|
|
raise ValueError("Missing login or password in fleetyard_login.ini")
|
|
payload = {
|
|
"login": login,
|
|
"password": password,
|
|
"rememberMe": remember_me,
|
|
}
|
|
response = self.session.post(f"{self.base_url}/sessions", json=payload)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def get_igns_fleet(self):
|
|
"""Get the IGNs fleet."""
|
|
fleet_url = f"{self.base_url}/fleets/igns/vehicles/export"
|
|
response = self.session.get(fleet_url)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def process_fleet_data(fleet_data):
|
|
"""Groups fleet data by ship and aggregates owners."""
|
|
ship_owners = defaultdict(list)
|
|
for ship in fleet_data:
|
|
if "username" in ship:
|
|
key = (ship["manufacturerName"], ship["name"])
|
|
ship_owners[key].append(ship["username"])
|
|
return ship_owners
|
|
|
|
def store_in_database(ship_owners):
|
|
"""Stores the processed fleet data in a SQLite database."""
|
|
db_file = "fleet.db"
|
|
conn = sqlite3.connect(db_file)
|
|
cursor = conn.cursor()
|
|
|
|
# Create table
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS ship_owner_list (
|
|
manufacturerName TEXT,
|
|
name TEXT,
|
|
usernames TEXT,
|
|
PRIMARY KEY (manufacturerName, name)
|
|
)
|
|
''')
|
|
|
|
# Clear existing data to prevent duplicates on re-runs
|
|
cursor.execute('DELETE FROM ship_owner_list')
|
|
|
|
# Insert new data
|
|
for (manufacturer, ship_name), owners in ship_owners.items():
|
|
usernames_str = ", ".join(sorted(owners))
|
|
cursor.execute(
|
|
"INSERT INTO ship_owner_list (manufacturerName, name, usernames) VALUES (?, ?, ?)",
|
|
(manufacturer, ship_name, usernames_str)
|
|
)
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
print(f"Data successfully stored in {db_file}")
|
|
|
|
if __name__ == "__main__":
|
|
base_url = "https://api.fleetyards.net/v1"
|
|
api = FleetyardAPI(base_url)
|
|
|
|
try:
|
|
# Create a session
|
|
session_response = api.create_session()
|
|
print("Session created.")
|
|
|
|
# Get the IGNs fleet
|
|
if session_response.get("code") == "success":
|
|
fleet_json = api.get_igns_fleet()
|
|
print("IGNs fleet data retrieved.")
|
|
|
|
# Process the data
|
|
processed_data = process_fleet_data(fleet_json)
|
|
|
|
# Store in database
|
|
store_in_database(processed_data)
|
|
|
|
else:
|
|
print("Failed to create session, cannot retrieve fleet.")
|
|
|
|
except requests.exceptions.HTTPError as e:
|
|
print(f"An HTTP error occurred: {e}")
|
|
print(f"Response body: {e.response.text}")
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"A request error occurred: {e}")
|