mikx/MxPyE2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Classes/Bases/Skills/Uniques to DB

Browse Source
This commit is contained in:
mikx
2026-06-23 09:12:49 -04:00
commit ba5a438374
7 changed files with 820 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
.env.example Normal file
View File

@@ -0,0 +1,12 @@
# MySQL Database Configuration
DB_HOST=192.168.1.130
DB_PORT=3306
DB_USER=mxauth
DB_PASSWORD=9WjfC9kETfpFMaL8QYTnjgvDuYJN6sbZ
DB_NAME=mxauth
# JSON URL Configuration
JSON_URL=https://repoe-fork.github.io/poe2/base_items.json
SKILL_GEMS_URL=https://repoe-fork.github.io/poe2/skill_gems.json
ITEM_CLASSES_URL=https://repoe-fork.github.io/poe2/item_classes.json
UNIQUES_URL=https://repoe-fork.github.io/poe2/uniques.json

122
.gitignore vendored Normal file
View File

@@ -0,0 +1,122 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.pyc
*.pyo
*.pyd
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
/*.manifest
/*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nosenv/
.pytest_cache/
.mypy_cache/
.dmypy.json
dmypy.json
.hypothesis/
.run/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if Pipfile.lock cannot be shared, ignore it:
#Pipfile.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
#Poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm build stage
.pdm-build/
# PEP 582; project packages directory (used by pdm)
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# VS Code settings
.vscode/
# JetBrains IDEs
.idea/

348
db.py Normal file
View File

@@ -0,0 +1,348 @@
import os
import json
import mysql.connector
from mysql.connector import errorcode
def get_db_connection(host, port, user, password, db_name=None):
"""Establishes connection to MySQL. Optionally connects to a specific database."""
config = {
'host': host,
'port': port,
'user': user,
'password': password,
}
if db_name:
config['database'] = db_name
return mysql.connector.connect(**config)
def init_db(host, port, user, password, db_name):
"""Initializes the database and the item_bases table."""
try:
# Check if database exists by attempting connection
conn = get_db_connection(host, port, user, password, db_name)
conn.close()
except mysql.connector.Error as err:
if err.errno == errorcode.ER_BAD_DB_ERROR:
# Database doesn't exist, connect without database name to create it
print(f"Database '{db_name}' not found. Creating database...")
conn = get_db_connection(host, port, user, password)
cursor = conn.cursor()
cursor.execute(f"CREATE DATABASE `{db_name}` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci")
cursor.close()
conn.close()
else:
raise err
# Connect to the target database to verify / create table
conn = get_db_connection(host, port, user, password, db_name)
cursor = conn.cursor()
# Create main item_bases table with JSON support
table_query = """
CREATE TABLE IF NOT EXISTS item_bases (
id VARCHAR(255) PRIMARY KEY,
name VARCHAR(255) NOT NULL,
item_class VARCHAR(100),
inherits_from VARCHAR(255),
domain VARCHAR(50),
drop_level INT,
inventory_width INT,
inventory_height INT,
release_state VARCHAR(50),
req_level INT,
req_strength INT,
req_dexterity INT,
req_intelligence INT,
visual_identity_id VARCHAR(255),
visual_identity_dds VARCHAR(255),
properties JSON,
tags JSON,
implicits JSON,
skills_granted JSON,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB CHARACTER SET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
"""
cursor.execute(table_query)
# Create skill_gems table with JSON support
skill_table_query = """
CREATE TABLE IF NOT EXISTS skill_gems (
id VARCHAR(255) PRIMARY KEY,
display_name VARCHAR(255),
release_state VARCHAR(50),
color VARCHAR(50),
crafting_level INT,
gem_type VARCHAR(50),
icon_dds_file VARCHAR(255),
skill_name VARCHAR(255),
support_name VARCHAR(255),
support_text TEXT,
tutorial_video VARCHAR(255),
ui_image VARCHAR(255),
req_weight_dexterity INT,
req_weight_intelligence INT,
req_weight_strength INT,
crafting_types JSON,
grants_skills JSON,
recommended_supports JSON,
tags JSON,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB CHARACTER SET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
"""
cursor.execute(skill_table_query)
# Create item_classes table with JSON support
classes_table_query = """
CREATE TABLE IF NOT EXISTS item_classes (
id VARCHAR(255) PRIMARY KEY,
category VARCHAR(100),
category_id VARCHAR(100),
name VARCHAR(255) NOT NULL,
influence_tags JSON,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB CHARACTER SET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
"""
cursor.execute(classes_table_query)
# Create item_uniques table
uniques_table_query = """
CREATE TABLE IF NOT EXISTS item_uniques (
id VARCHAR(255) PRIMARY KEY,
name VARCHAR(255) NOT NULL,
unique_id VARCHAR(255),
item_class VARCHAR(100),
inventory_width INT,
inventory_height INT,
is_alternate_art BOOLEAN,
visual_identity_id VARCHAR(255),
visual_identity_dds VARCHAR(255),
renamed_version VARCHAR(255),
base_version VARCHAR(255),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB CHARACTER SET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
"""
cursor.execute(uniques_table_query)
conn.commit()
cursor.close()
print("Database tables 'item_bases', 'skill_gems', 'item_classes', and 'item_uniques' verified/created successfully.")
return conn
def bulk_upsert_items(conn, items_data):
"""Upserts list of items into database using batch processing."""
if not items_data:
return 0
query = """
INSERT INTO item_bases (
id, name, item_class, inherits_from, domain, drop_level,
inventory_width, inventory_height, release_state,
req_level, req_strength, req_dexterity, req_intelligence,
visual_identity_id, visual_identity_dds,
properties, tags, implicits, skills_granted
) VALUES (
%s, %s, %s, %s, %s, %s,
%s, %s, %s,
%s, %s, %s, %s,
%s, %s,
%s, %s, %s, %s
) ON DUPLICATE KEY UPDATE
name = VALUES(name),
item_class = VALUES(item_class),
inherits_from = VALUES(inherits_from),
domain = VALUES(domain),
drop_level = VALUES(drop_level),
inventory_width = VALUES(inventory_width),
inventory_height = VALUES(inventory_height),
release_state = VALUES(release_state),
req_level = VALUES(req_level),
req_strength = VALUES(req_strength),
req_dexterity = VALUES(req_dexterity),
req_intelligence = VALUES(req_intelligence),
visual_identity_id = VALUES(visual_identity_id),
visual_identity_dds = VALUES(visual_identity_dds),
properties = VALUES(properties),
tags = VALUES(tags),
implicits = VALUES(implicits),
skills_granted = VALUES(skills_granted)
"""
cursor = conn.cursor()
batch_size = 500
total_upserted = len(items_data)
print(f"Upserting {total_upserted} items in batches of {batch_size}...")
for i in range(0, total_upserted, batch_size):
chunk = items_data[i:i + batch_size]
cursor.executemany(query, chunk)
conn.commit()
print(f"Upserted {min(i + batch_size, total_upserted)} / {total_upserted} items...")
cursor.close()
return total_upserted
def clear_items_table(conn):
"""Truncates the item_bases table to empty all existing items."""
cursor = conn.cursor()
cursor.execute("TRUNCATE TABLE item_bases")
conn.commit()
cursor.close()
print("Table 'item_bases' cleared successfully.")
def clear_skills_table(conn):
"""Truncates the skill_gems table to empty all existing skill gems."""
cursor = conn.cursor()
cursor.execute("TRUNCATE TABLE skill_gems")
conn.commit()
cursor.close()
print("Table 'skill_gems' cleared successfully.")
def bulk_upsert_skills(conn, skills_data):
"""Upserts list of skill gems into database using batch processing."""
if not skills_data:
return 0
query = """
INSERT INTO skill_gems (
id, display_name, release_state, color, crafting_level, gem_type,
icon_dds_file, skill_name, support_name, support_text,
tutorial_video, ui_image,
req_weight_dexterity, req_weight_intelligence, req_weight_strength,
crafting_types, grants_skills, recommended_supports, tags
) VALUES (
%s, %s, %s, %s, %s, %s,
%s, %s, %s, %s,
%s, %s,
%s, %s, %s,
%s, %s, %s, %s
) ON DUPLICATE KEY UPDATE
display_name = VALUES(display_name),
release_state = VALUES(release_state),
color = VALUES(color),
crafting_level = VALUES(crafting_level),
gem_type = VALUES(gem_type),
icon_dds_file = VALUES(icon_dds_file),
skill_name = VALUES(skill_name),
support_name = VALUES(support_name),
support_text = VALUES(support_text),
tutorial_video = VALUES(tutorial_video),
ui_image = VALUES(ui_image),
req_weight_dexterity = VALUES(req_weight_dexterity),
req_weight_intelligence = VALUES(req_weight_intelligence),
req_weight_strength = VALUES(req_weight_strength),
crafting_types = VALUES(crafting_types),
grants_skills = VALUES(grants_skills),
recommended_supports = VALUES(recommended_supports),
tags = VALUES(tags)
"""
cursor = conn.cursor()
batch_size = 500
total_upserted = len(skills_data)
print(f"Upserting {total_upserted} skill gems in batches of {batch_size}...")
for i in range(0, total_upserted, batch_size):
chunk = skills_data[i:i + batch_size]
cursor.executemany(query, chunk)
conn.commit()
print(f"Upserted {min(i + batch_size, total_upserted)} / {total_upserted} skill gems...")
cursor.close()
return total_upserted
def clear_item_classes_table(conn):
"""Truncates the item_classes table to empty all existing item classes."""
cursor = conn.cursor()
cursor.execute("TRUNCATE TABLE item_classes")
conn.commit()
cursor.close()
print("Table 'item_classes' cleared successfully.")
def bulk_upsert_item_classes(conn, classes_data):
"""Upserts list of item classes into database using batch processing."""
if not classes_data:
return 0
query = """
INSERT INTO item_classes (
id, category, category_id, name, influence_tags
) VALUES (
%s, %s, %s, %s, %s
) ON DUPLICATE KEY UPDATE
category = VALUES(category),
category_id = VALUES(category_id),
name = VALUES(name),
influence_tags = VALUES(influence_tags)
"""
cursor = conn.cursor()
batch_size = 500
total_upserted = len(classes_data)
print(f"Upserting {total_upserted} item classes in batches of {batch_size}...")
for i in range(0, total_upserted, batch_size):
chunk = classes_data[i:i + batch_size]
cursor.executemany(query, chunk)
conn.commit()
print(f"Upserted {min(i + batch_size, total_upserted)} / {total_upserted} item classes...")
cursor.close()
return total_upserted
def clear_uniques_table(conn):
"""Truncates the item_uniques table to empty all existing uniques."""
cursor = conn.cursor()
cursor.execute("TRUNCATE TABLE item_uniques")
conn.commit()
cursor.close()
print("Table 'item_uniques' cleared successfully.")
def bulk_upsert_uniques(conn, uniques_data):
"""Upserts list of unique items into database using batch processing."""
if not uniques_data:
return 0
query = """
INSERT INTO item_uniques (
id, name, unique_id, item_class,
inventory_width, inventory_height, is_alternate_art,
visual_identity_id, visual_identity_dds,
renamed_version, base_version
) VALUES (
%s, %s, %s, %s,
%s, %s, %s,
%s, %s,
%s, %s
) ON DUPLICATE KEY UPDATE
name = VALUES(name),
unique_id = VALUES(unique_id),
item_class = VALUES(item_class),
inventory_width = VALUES(inventory_width),
inventory_height = VALUES(inventory_height),
is_alternate_art = VALUES(is_alternate_art),
visual_identity_id = VALUES(visual_identity_id),
visual_identity_dds = VALUES(visual_identity_dds),
renamed_version = VALUES(renamed_version),
base_version = VALUES(base_version)
"""
cursor = conn.cursor()
batch_size = 500
total_upserted = len(uniques_data)
print(f"Upserting {total_upserted} uniques in batches of {batch_size}...")
for i in range(0, total_upserted, batch_size):
chunk = uniques_data[i:i + batch_size]
cursor.executemany(query, chunk)
conn.commit()
print(f"Upserted {min(i + batch_size, total_upserted)} / {total_upserted} uniques...")
cursor.close()
return total_upserted

221
importer.py Normal file
View File

@@ -0,0 +1,221 @@
import json
import requests
def download_and_parse_items(url):
"""
Downloads item bases JSON from the given URL and parses it
into a flat structure suitable for MySQL insertion.
"""
print(f"Downloading base items JSON from: {url}")
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
}
response = requests.get(url, headers=headers, timeout=60)
response.raise_for_status()
print("Download completed. Parsing JSON structure...")
raw_data = response.json()
parsed_items = []
for item_id, item_val in raw_data.items():
# Extrapolate requirements
reqs = item_val.get("requirements") or {}
req_level = reqs.get("level")
req_strength = reqs.get("strength")
req_dexterity = reqs.get("dexterity")
req_intelligence = reqs.get("intelligence")
# Extrapolate visual identity
vis = item_val.get("visual_identity") or {}
vis_id = vis.get("id")
vis_dds = vis.get("dds_file")
# Serialize nested lists/dicts to JSON strings
properties = json.dumps(item_val.get("properties")) if item_val.get("properties") is not None else None
tags = json.dumps(item_val.get("tags")) if item_val.get("tags") is not None else None
implicits = json.dumps(item_val.get("implicits")) if item_val.get("implicits") is not None else None
skills_granted = json.dumps(item_val.get("skills_granted")) if item_val.get("skills_granted") is not None else None
# Prepare tuple matching schema order:
# id, name, item_class, inherits_from, domain, drop_level,
# inventory_width, inventory_height, release_state,
# req_level, req_strength, req_dexterity, req_intelligence,
# visual_identity_id, visual_identity_dds,
# properties, tags, implicits, skills_granted
parsed_item = (
item_id,
item_val.get("name", "Unknown Name"),
item_val.get("item_class"),
item_val.get("inherits_from"),
item_val.get("domain", "undefined"),
item_val.get("drop_level"),
item_val.get("inventory_width"),
item_val.get("inventory_height"),
item_val.get("release_state"),
req_level,
req_strength,
req_dexterity,
req_intelligence,
vis_id,
vis_dds,
properties,
tags,
implicits,
skills_granted
)
parsed_items.append(parsed_item)
print(f"Successfully parsed {len(parsed_items)} items from JSON.")
return parsed_items
def download_and_parse_skills(url):
"""
Downloads skill_gems.json from the given URL and parses it
into a flat structure suitable for MySQL insertion.
"""
print(f"Downloading skill gems JSON from: {url}")
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
}
response = requests.get(url, headers=headers, timeout=60)
response.raise_for_status()
print("Download completed. Parsing JSON structure...")
raw_data = response.json()
parsed_skills = []
for skill_id, skill_val in raw_data.items():
base_item = skill_val.get("base_item") or {}
display_name = base_item.get("display_name", "Unknown Name")
release_state = base_item.get("release_state")
req_weights = skill_val.get("requirement_weights") or {}
req_dex = req_weights.get("dexterity", 0)
req_int = req_weights.get("intelligence", 0)
req_str = req_weights.get("strength", 0)
# List structures to be JSON serialized
crafting_types = json.dumps(skill_val.get("crafting_types")) if skill_val.get("crafting_types") is not None else None
grants_skills = json.dumps(skill_val.get("grants_skills")) if skill_val.get("grants_skills") is not None else None
recommended_supports = json.dumps(skill_val.get("recommended_supports")) if skill_val.get("recommended_supports") is not None else None
tags = json.dumps(skill_val.get("tags")) if skill_val.get("tags") is not None else None
# Prepare tuple matching schema order:
# id, display_name, release_state, color, crafting_level, gem_type,
# icon_dds_file, skill_name, support_name, support_text,
# tutorial_video, ui_image,
# req_weight_dexterity, req_weight_intelligence, req_weight_strength,
# crafting_types, grants_skills, recommended_supports, tags
parsed_skill = (
skill_id,
display_name,
release_state,
skill_val.get("color"),
skill_val.get("crafting_level"),
skill_val.get("gem_type"),
skill_val.get("icon_dds_file"),
skill_val.get("skill_name"),
skill_val.get("support_name"),
skill_val.get("support_text"),
skill_val.get("tutorial_video"),
skill_val.get("ui_image"),
req_dex,
req_int,
req_str,
crafting_types,
grants_skills,
recommended_supports,
tags
)
parsed_skills.append(parsed_skill)
print(f"Successfully parsed {len(parsed_skills)} skill gems from JSON.")
return parsed_skills
def download_and_parse_item_classes(url):
"""
Downloads item_classes.json from the given URL and parses it
into a flat structure suitable for MySQL insertion.
"""
print(f"Downloading item classes JSON from: {url}")
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
}
response = requests.get(url, headers=headers, timeout=60)
response.raise_for_status()
print("Download completed. Parsing JSON structure...")
raw_data = response.json()
parsed_classes = []
for class_id, class_val in raw_data.items():
influence_tags = json.dumps(class_val.get("influence_tags")) if class_val.get("influence_tags") is not None else None
# Prepare tuple matching schema order:
# id, category, category_id, name, influence_tags
parsed_class = (
class_id,
class_val.get("category"),
class_val.get("category_id"),
class_val.get("name", "Unknown Name"),
influence_tags
)
parsed_classes.append(parsed_class)
print(f"Successfully parsed {len(parsed_classes)} item classes from JSON.")
return parsed_classes
def download_and_parse_uniques(url):
"""
Downloads uniques.json from the given URL and parses it
into a flat structure suitable for MySQL insertion.
"""
print(f"Downloading uniques JSON from: {url}")
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
}
response = requests.get(url, headers=headers, timeout=60)
response.raise_for_status()
print("Download completed. Parsing JSON structure...")
raw_data = response.json()
parsed_uniques = []
for item_id, item_val in raw_data.items():
# Extrapolate visual identity
vis = item_val.get("visual_identity") or {}
vis_id = vis.get("id")
vis_dds = vis.get("dds_file")
# Prepare tuple matching schema order:
# id, name, unique_id, item_class,
# inventory_width, inventory_height, is_alternate_art,
# visual_identity_id, visual_identity_dds,
# renamed_version, base_version
parsed_unique = (
item_id,
item_val.get("name", "Unknown Name"),
item_val.get("id"),
item_val.get("item_class"),
item_val.get("inventory_width"),
item_val.get("inventory_height"),
item_val.get("is_alternate_art"),
vis_id,
vis_dds,
item_val.get("renamed_version"),
item_val.get("base_version")
)
parsed_uniques.append(parsed_unique)
print(f"Successfully parsed {len(parsed_uniques)} uniques from JSON.")
return parsed_uniques

111
main.py Normal file
View File

@@ -0,0 +1,111 @@
import os
import sys
import time
from dotenv import load_dotenv
import db
import importer
def main():
# Load configuration
load_dotenv()
db_host = os.getenv("DB_HOST", "localhost")
db_port = int(os.getenv("DB_PORT", "3306"))
db_user = os.getenv("DB_USER", "root")
db_password = os.getenv("DB_PASSWORD", "")
db_name = os.getenv("DB_NAME", "poe2_items")
item_bases_url = os.getenv("ITEM_BASES_URL", "https://repoe-fork.github.io/poe2/base_items.json")
skill_gems_url = os.getenv("SKILL_GEMS_URL", "https://repoe-fork.github.io/poe2/skill_gems.json")
item_classes_url = os.getenv("ITEM_CLASSES_URL", "https://repoe-fork.github.io/poe2/item_classes.json")
uniques_url = os.getenv("UNIQUES_URL", "https://repoe-fork.github.io/poe2/uniques.json")
print("=" * 60)
print("Path of Exile 2 Base Items MySQL Importer Starting")
print("=" * 60)
print(f"Target DB Host: {db_host}:{db_port}")
print(f"Target Database: {db_name}")
print(f"Target User: {db_user}")
print("-" * 60)
start_time = time.time()
conn = None
try:
# Step 1: Initialize Database & Table
print("Step 1: Connecting to MySQL & Initializing Schema...")
conn = db.init_db(
host=db_host,
port=db_port,
user=db_user,
password=db_password,
db_name=db_name
)
# Step 2: Download & Parse base items
print("\nStep 2: Downloading & Parsing remote base items JSON...")
items_data = importer.download_and_parse_items(item_bases_url)
# Step 3: Clear existing table contents
print("\nStep 3: Clearing existing records from 'item_bases' table...")
db.clear_items_table(conn)
# Step 4: Insert into MySQL
print("\nStep 4: Upserting base items records into 'item_bases' table...")
items_upserted_count = db.bulk_upsert_items(conn, items_data)
# Step 5: Download & Parse skill gems
print("\nStep 5: Downloading & Parsing remote skill gems JSON...")
skills_data = importer.download_and_parse_skills(skill_gems_url)
# Step 6: Clear existing skill gems table contents
print("\nStep 6: Clearing existing records from 'skill_gems' table...")
db.clear_skills_table(conn)
# Step 7: Insert skill gems into MySQL
print("\nStep 7: Upserting skill gems records into 'skill_gems' table...")
skills_upserted_count = db.bulk_upsert_skills(conn, skills_data)
# Step 8: Download & Parse item classes
print("\nStep 8: Downloading & Parsing remote item classes JSON...")
classes_data = importer.download_and_parse_item_classes(item_classes_url)
# Step 9: Clear existing item classes table contents
print("\nStep 9: Clearing existing records from 'item_classes' table...")
db.clear_item_classes_table(conn)
# Step 10: Insert item classes into MySQL
print("\nStep 10: Upserting item classes records into 'item_classes' table...")
classes_upserted_count = db.bulk_upsert_item_classes(conn, classes_data)
# Step 11: Download & Parse uniques
print("\nStep 11: Downloading & Parsing remote uniques JSON...")
uniques_data = importer.download_and_parse_uniques(uniques_url)
# Step 12: Clear existing uniques table contents
print("\nStep 12: Clearing existing records from 'item_uniques' table...")
db.clear_uniques_table(conn)
# Step 13: Insert uniques into MySQL
print("\nStep 13: Upserting uniques records into 'item_uniques' table...")
uniques_upserted_count = db.bulk_upsert_uniques(conn, uniques_data)
elapsed = time.time() - start_time
print("\n" + "=" * 60)
print("IMPORT PROCESS COMPLETED SUCCESSFULLY")
print(f"Total time elapsed: {elapsed:.2f} seconds")
print(f"Base items upserted: {items_upserted_count}")
print(f"Skill gems upserted: {skills_upserted_count}")
print(f"Item classes upserted: {classes_upserted_count}")
print(f"Uniques upserted: {uniques_upserted_count}")
print("=" * 60)
except Exception as e:
print(f"\n[ERROR] An error occurred during the import process: {e}", file=sys.stderr)
sys.exit(1)
finally:
if conn and conn.is_connected():
conn.close()
print("Database connection closed cleanly.")
if __name__ == "__main__":
main()

3
requirements.txt Normal file
View File

@@ -0,0 +1,3 @@
mysql-connector-python>=9.0.0
requests>=2.31.0
python-dotenv>=1.0.1

3
run.bat Normal file
View File

@@ -0,0 +1,3 @@
@echo off
.\venv\Scripts\python.exe main.py