68005b1cb0
feat: collector task checklist. chore: docs updated, gitignore updated. Docs: TARKOV_DEV_API.md fully explains tarkov.dev's api for future coding agents and forgetful people llike me.
126 lines
2.8 KiB
Python
126 lines
2.8 KiB
Python
import requests
|
|
import sqlite3
|
|
import sys
|
|
|
|
DB_PATH = "tarkov.db"
|
|
API_URL = "https://api.tarkov.dev/graphql"
|
|
|
|
GRAPHQL_QUERY = """
|
|
query {
|
|
items(types: [keys]) {
|
|
id
|
|
name
|
|
shortName
|
|
weight
|
|
wikiLink
|
|
gridImageLink
|
|
properties {
|
|
... on ItemPropertiesKey {
|
|
uses
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
def fetch_keys():
|
|
response = requests.post(
|
|
API_URL,
|
|
json={"query": GRAPHQL_QUERY},
|
|
timeout=30
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
if "errors" in data:
|
|
raise RuntimeError(data["errors"])
|
|
|
|
return data["data"]["items"]
|
|
|
|
def upsert_keys(conn, keys):
|
|
inserted = 0
|
|
updated = 0
|
|
skipped = 0
|
|
|
|
cursor = conn.cursor()
|
|
|
|
for k in keys:
|
|
api_id = k.get("id")
|
|
name = k.get("name")
|
|
short_name = k.get("shortName")
|
|
weight = k.get("weight")
|
|
wiki_url = k.get("wikiLink")
|
|
grid_image_url = k.get("gridImageLink")
|
|
uses = None
|
|
|
|
props = k.get("properties")
|
|
if props and "uses" in props:
|
|
uses = props["uses"]
|
|
|
|
if not api_id or not name:
|
|
skipped += 1
|
|
continue
|
|
|
|
cursor.execute(
|
|
"""
|
|
SELECT id FROM keys WHERE api_id = ?
|
|
""",
|
|
(api_id,)
|
|
)
|
|
row = cursor.fetchone()
|
|
|
|
if row:
|
|
cursor.execute(
|
|
"""
|
|
UPDATE keys
|
|
SET name = ?, short_name = ?, weight_kg = ?, uses = ?, wiki_url = ?, grid_image_url = ?
|
|
WHERE api_id = ?
|
|
""",
|
|
(name, short_name, weight, uses, wiki_url, grid_image_url, api_id)
|
|
)
|
|
updated += 1
|
|
else:
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO keys (api_id, name, short_name, weight_kg, uses, wiki_url, grid_image_url)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(api_id, name, short_name, weight, uses, wiki_url, grid_image_url)
|
|
)
|
|
inserted += 1
|
|
|
|
conn.commit()
|
|
return inserted, updated, skipped
|
|
|
|
def main():
|
|
print("Fetching keys from Tarkov.dev...")
|
|
try:
|
|
keys = fetch_keys()
|
|
except Exception as e:
|
|
print("ERROR: Failed to fetch keys")
|
|
print(e)
|
|
sys.exit(1)
|
|
|
|
print(f"Fetched {len(keys)} keys")
|
|
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.execute("PRAGMA foreign_keys = ON")
|
|
|
|
try:
|
|
inserted, updated, skipped = upsert_keys(conn, keys)
|
|
except Exception as e:
|
|
conn.rollback()
|
|
print("ERROR: Database operation failed")
|
|
print(e)
|
|
sys.exit(1)
|
|
finally:
|
|
conn.close()
|
|
|
|
print("Import complete")
|
|
print(f"Inserted: {inserted}")
|
|
print(f"Updated: {updated}")
|
|
print(f"Skipped: {skipped}")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|