Files
onlyscavs/import_quests.py
serversdwn 68005b1cb0 v0.1.1 -
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.
2026-02-21 09:41:06 +00:00

141 lines
3.0 KiB
Python

import requests
import sqlite3
DB_PATH = "tarkov.db"
API_URL = "https://api.tarkov.dev/graphql"
QUERY = """
{
tasks {
id
name
wikiLink
trader {
name
}
taskRequirements {
task {
id
}
}
}
}
"""
def fetch_quests():
print("Fetching quests from Tarkov.dev...")
response = requests.post(API_URL, json={"query": QUERY})
response.raise_for_status()
result = response.json()
if "errors" in result:
print("GraphQL Errors:")
print(result["errors"])
raise Exception("GraphQL query failed.")
return result["data"]["tasks"]
def setup_db(conn):
cur = conn.cursor()
cur.execute("""
CREATE TABLE IF NOT EXISTS quests (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
trader TEXT NOT NULL,
wiki_link TEXT
)
""")
cur.execute("""
CREATE TABLE IF NOT EXISTS quest_deps (
quest_id TEXT NOT NULL,
depends_on TEXT NOT NULL,
PRIMARY KEY (quest_id, depends_on)
)
""")
conn.commit()
def store_quests(conn, quests):
cur = conn.cursor()
for q in quests:
cur.execute(
"INSERT OR REPLACE INTO quests VALUES (?, ?, ?, ?)",
(q["id"], q["name"], q["trader"]["name"], q.get("wikiLink"))
)
for req in q.get("taskRequirements") or []:
cur.execute(
"INSERT OR IGNORE INTO quest_deps VALUES (?, ?)",
(q["id"], req["task"]["id"])
)
conn.commit()
def get_collector_id(conn):
cur = conn.cursor()
cur.execute("SELECT id FROM quests WHERE name = 'Collector'")
row = cur.fetchone()
if not row:
raise Exception("Collector quest not found in DB.")
return row[0]
def get_all_prereqs(conn, quest_id, seen=None):
if seen is None:
seen = set()
cur = conn.cursor()
cur.execute(
"SELECT depends_on FROM quest_deps WHERE quest_id = ?",
(quest_id,)
)
deps = [r[0] for r in cur.fetchall()]
for dep in deps:
if dep not in seen:
seen.add(dep)
get_all_prereqs(conn, dep, seen)
return seen
def main():
quests = fetch_quests()
conn = sqlite3.connect(DB_PATH)
setup_db(conn)
store_quests(conn, quests)
collector_id = get_collector_id(conn)
prereqs = get_all_prereqs(conn, collector_id)
print("\n=== Quests Required for Collector ===\n")
if prereqs:
cur = conn.cursor()
placeholders = ",".join("?" * len(prereqs))
cur.execute(
f"""
SELECT name, trader, wiki_link
FROM quests
WHERE id IN ({placeholders})
ORDER BY trader, name
""",
list(prereqs)
)
rows = cur.fetchall()
for name, trader, _ in rows:
print(f"[ ] {trader}: {name}")
print(f"\nTotal required quests: {len(rows)}")
else:
print("Collector has no prerequisites (unexpected).")
conn.close()
if __name__ == "__main__":
main()