967 lines
42 KiB
Python
967 lines
42 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
traderx_build.py - Maintain TraderX trader configuration from ONE human-readable
|
|
source file (class names, words instead of bitfields, reusable attachment groups)
|
|
and compile it into the mod's legacy Products/ Categories/ Presets files.
|
|
|
|
Commands
|
|
--------
|
|
import-traderx <TraderX dir> reverse-import an existing config
|
|
import-traderplus <TraderPlusPriceConfig.json> TraderPlus v1 export
|
|
import-expansion <expansion market .json or dir> Expansion market export
|
|
import-plain <classnames.txt> --price N plain class list
|
|
build <source.yaml|.json> --out <TraderX dir> [--dry-run] [--clean]
|
|
check <source.yaml|.json> validate only
|
|
|
|
Every import command writes a source file (-o) or appends a category into an
|
|
existing one (--into source.yaml). Use --category / --trader to name the
|
|
category and the trader(s) it belongs to.
|
|
|
|
Source schema (YAML shown; JSON with the same keys works without pyyaml)
|
|
-----------------------------------------------------------------------
|
|
defaults: # every key optional
|
|
stock: unlimited # unlimited | <int>
|
|
coefficient: 1.0
|
|
sell_percent: 25 # sell = buy * 25% when 'sell' not given
|
|
buy: full # empty | full | static N | coef N% (item quantity on purchase)
|
|
sell: full # any | empty | full | static N | coef N%
|
|
restart: keep # keep | reset | random
|
|
destock: 0 # 0..1 fraction removed on restart
|
|
attachment_price: 50 # price for attachments that are not defined anywhere else
|
|
attachment_sell_percent: 25
|
|
|
|
groups: # reusable lists of class names (or other group names)
|
|
ak_optics: [SNAFU_AKAimpoint_M5, SNAFU_AKKahles]
|
|
|
|
items: # optional: products not shown in any category (attachments etc.)
|
|
- SNAFU_AKKahles: {price: 800}
|
|
|
|
categories:
|
|
- name: Assault Rifles
|
|
id: cat_assault_rifles_001 # optional, derived from name if absent
|
|
traders: [Boris, 3] # givenName or npcId from TraderXGeneralSettings.json
|
|
licenses: [Weapons License] # names; created in GeneralSettings if missing
|
|
visible: true
|
|
icon: ""
|
|
price: 1000 # category default buy price
|
|
stock / coefficient / sell_percent / buy / sell / restart / destock (category defaults)
|
|
items:
|
|
- SNAFU_AK12A # bare class -> category defaults
|
|
- SNAFU_AK103A: {price: 1200, sell: 300, stock: 5, coefficient: 0.95}
|
|
- SNAFU_AK19:
|
|
id: prod_snafu_ak19_001 # optional, keeps an existing product id / stock file
|
|
attachments: [ak_optics, SNAFU_Standardized_Suppressor_Black]
|
|
- SNAFU_AK12A:
|
|
attachments: ["@compat", "@compat_mags"] # needs build --compat FILE:
|
|
# @compat = every compatible attachment that is a product in this source
|
|
# @compat_mags = same for magazines
|
|
# @compat_all = everything compatible, auto-creating missing products
|
|
variants: [SNAFU_AK19_Tan] # variants inherit attachments unless they define their own
|
|
presets:
|
|
"CQB*": [SNAFU_AKAimpoint_M5, SNAFU_Standardized_Suppressor_Black] # * = default preset
|
|
"DMR": [SNAFU_AKKahles]
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import sys
|
|
from collections import OrderedDict
|
|
from datetime import datetime
|
|
|
|
try:
|
|
import yaml # type: ignore
|
|
except ImportError: # pragma: no cover
|
|
yaml = None
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# constants mirrored from the mod
|
|
# ---------------------------------------------------------------------------
|
|
SELL_MODES = {"any": 0, "no_matter": 0, "empty": 1, "full": 2, "coef": 3, "static": 4}
|
|
BUY_MODES = {"empty": 0, "full": 2, "coef": 3, "static": 4}
|
|
SELL_MODE_NAMES = {0: "any", 1: "empty", 2: "full", 3: "coef", 4: "static"}
|
|
BUY_MODE_NAMES = {0: "empty", 1: "empty", 2: "full", 3: "coef", 4: "static"}
|
|
RESTART = {"keep": 0, "reset": 1, "random": 2}
|
|
RESTART_NAMES = {0: "keep", 1: "reset", 2: "random"}
|
|
|
|
PRODUCT_KEYS = ["className", "coefficient", "maxStock", "tradeQuantity", "buyPrice",
|
|
"sellPrice", "stockSettings", "attachments", "variants"]
|
|
|
|
DEFAULTS = {
|
|
"stock": "unlimited", "coefficient": 1.0, "sell_percent": 25,
|
|
"buy": "full", "sell": "full", "restart": "keep", "destock": 0,
|
|
"attachment_price": 50, "attachment_sell_percent": 25,
|
|
}
|
|
ITEM_LEVEL_KEYS = ("price", "sell", "stock", "coefficient", "buy", "sell_mode",
|
|
"restart", "destock", "attachments", "variants", "presets", "id", "sell_percent")
|
|
|
|
KNOWN_SURVIVORS = {
|
|
"M": "Boris Cyril Denis Elias Francis Guo Hassan Indar Jose Kaito Lewis Lucas Malik Manua Mirek Niki Oliver Peter Quinn Rolf Seth Taiki".split(),
|
|
"F": "Eva Frida Gabi Helga Irena Judy Keiko Linda Maria Naomi Baty".split(),
|
|
}
|
|
|
|
|
|
class BuildError(Exception):
|
|
pass
|
|
|
|
|
|
def is_abstract(cls):
|
|
return cls.endswith("_Base") or cls.endswith("Base") and cls[:-4].endswith("_") or cls.startswith("Test_") or cls.endswith("_TEST")
|
|
|
|
|
|
def load_compat(paths):
|
|
"""Load one or more 'All classes and their attachments' files.
|
|
Returns {lower_classname: {"attachments": [...], "magazines": [...], "bullets": [...]}}"""
|
|
out = {}
|
|
for p in paths or []:
|
|
with open(p, encoding="utf-8-sig") as f:
|
|
d = json.load(f)
|
|
entries = []
|
|
if isinstance(d, dict):
|
|
for v in d.values():
|
|
if isinstance(v, list):
|
|
entries.extend(v)
|
|
elif isinstance(d, list):
|
|
entries = d
|
|
for e in entries:
|
|
if not isinstance(e, dict) or not e.get("name"):
|
|
continue
|
|
rec = out.setdefault(e["name"].lower(), {"name": e["name"], "attachments": [], "magazines": [], "bullets": []})
|
|
for k in ("attachments", "magazines", "bullets"):
|
|
for c in e.get(k, []) or []:
|
|
if c and not is_abstract(c) and c not in rec[k]:
|
|
rec[k].append(c)
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# small helpers
|
|
# ---------------------------------------------------------------------------
|
|
def sanitize(s, lowercase=True):
|
|
s = s.lower() if lowercase else s
|
|
s = re.sub(r"[^A-Za-z0-9_]", "_", s)
|
|
s = re.sub(r"_+", "_", s).strip("_")
|
|
return s or "x"
|
|
|
|
|
|
def load_source(path):
|
|
with open(path, "r", encoding="utf-8-sig") as f:
|
|
text = f.read()
|
|
if path.lower().endswith((".yaml", ".yml")):
|
|
if not yaml:
|
|
sys.exit("pyyaml is not installed: pip install pyyaml (or use a .json source)")
|
|
return yaml.safe_load(text) or {}
|
|
return json.loads(text)
|
|
|
|
|
|
def dump_source(data, path):
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
if path.lower().endswith((".yaml", ".yml")) and yaml:
|
|
yaml.safe_dump(data, f, sort_keys=False, allow_unicode=True, width=120)
|
|
else:
|
|
json.dump(data, f, indent=2, ensure_ascii=False)
|
|
f.write("\n")
|
|
|
|
|
|
def load_json(path):
|
|
with open(path, "r", encoding="utf-8-sig") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def save_json(path, data):
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, indent=2, ensure_ascii=False)
|
|
f.write("\n")
|
|
|
|
|
|
def find_traderx_root(path):
|
|
"""Return the TraderX profile folder (parent of TraderXConfig)."""
|
|
path = os.path.abspath(path)
|
|
for c in (path, os.path.join(path, "TraderX")):
|
|
if os.path.isdir(os.path.join(c, "TraderXConfig")):
|
|
return c
|
|
return path # will be created on build
|
|
|
|
|
|
def parse_qty(spec, is_buy):
|
|
"""'full' | 'empty' | 'any' | 'static 22' | 'coef 50%' | 'coef 0.5' -> (mode, value)"""
|
|
if isinstance(spec, (int, float)) and not isinstance(spec, bool):
|
|
return ("static", int(spec))
|
|
s = str(spec).strip().lower()
|
|
parts = s.split()
|
|
mode = parts[0]
|
|
modes = BUY_MODES if is_buy else SELL_MODES
|
|
if mode not in modes:
|
|
raise BuildError(f"unknown {'buy' if is_buy else 'sell'} quantity '{spec}' "
|
|
f"(use {', '.join(sorted(set(modes)))})")
|
|
val = 0
|
|
if mode in ("static", "coef"):
|
|
if len(parts) < 2:
|
|
raise BuildError(f"'{spec}' needs a value, e.g. 'static 22' or 'coef 50%'")
|
|
v = parts[1].rstrip("%")
|
|
val = float(v)
|
|
if mode == "coef":
|
|
if parts[1].endswith("%") or val > 1:
|
|
val = val # already percent
|
|
else:
|
|
val = val * 100
|
|
val = int(round(val))
|
|
return (mode, val)
|
|
|
|
|
|
def pack_trade_quantity(buy_spec, sell_spec):
|
|
bm, bv = parse_qty(buy_spec, True)
|
|
sm, sv = parse_qty(sell_spec, False)
|
|
if not 0 <= sv <= 0x1FFF:
|
|
raise BuildError(f"sell value {sv} out of range 0-8191")
|
|
return SELL_MODES[sm] | (BUY_MODES[bm] << 3) | (sv << 6) | (bv << 19)
|
|
|
|
|
|
def unpack_trade_quantity(tq):
|
|
sm, bm, sv, bv = tq & 7, (tq >> 3) & 7, (tq >> 6) & 0x1FFF, tq >> 19
|
|
sell = SELL_MODE_NAMES.get(sm, "full")
|
|
buy = BUY_MODE_NAMES.get(bm, "full")
|
|
if sell == "static":
|
|
sell = f"static {sv}"
|
|
elif sell == "coef":
|
|
sell = f"coef {sv}%"
|
|
if buy == "static":
|
|
buy = f"static {bv}"
|
|
elif buy == "coef":
|
|
buy = f"coef {bv}%"
|
|
return buy, sell
|
|
|
|
|
|
def pack_stock_settings(restart, destock):
|
|
if restart not in RESTART:
|
|
raise BuildError(f"restart must be keep|reset|random, got '{restart}'")
|
|
d = float(destock)
|
|
if not 0 <= d <= 1:
|
|
raise BuildError(f"destock must be 0..1, got {destock}")
|
|
return int(round(d * 100)) | (RESTART[restart] << 7)
|
|
|
|
|
|
def unpack_stock_settings(ss):
|
|
return RESTART_NAMES.get((ss >> 7) & 3, "keep"), (ss & 0x7F) / 100
|
|
|
|
|
|
def parse_stock(v):
|
|
if v in (None, "unlimited", "inf", -1, "-1"):
|
|
return -1
|
|
return int(v)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# BUILD
|
|
# ---------------------------------------------------------------------------
|
|
class Builder:
|
|
def __init__(self, src, root, dry_run=False, clean=False, strict=False, compat=None):
|
|
self.compat = compat or {}
|
|
self.src = src
|
|
self.root = root
|
|
self.cfg = os.path.join(root, "TraderXConfig")
|
|
self.dry = dry_run
|
|
self.clean = clean
|
|
self.strict = strict
|
|
self.defaults = dict(DEFAULTS, **(src.get("defaults") or {}))
|
|
self.groups = src.get("groups") or {}
|
|
self.errors, self.warnings, self.notes = [], [], []
|
|
self.products = OrderedDict() # productId -> dict (mod format)
|
|
self.class_to_id = {} # lower classname -> productId
|
|
self.categories = OrderedDict() # categoryId -> dict
|
|
self.presets = OrderedDict() # productId -> TraderXPresets dict
|
|
self.cat_traders = OrderedDict() # categoryId -> [trader refs]
|
|
self.cat_licenses = {} # categoryId -> [license names]
|
|
self.used_ids = set()
|
|
self.gs = None
|
|
|
|
# -- id allocation -----------------------------------------------------
|
|
def alloc_id(self, prefix, name, wanted=None):
|
|
if wanted:
|
|
if wanted in self.used_ids:
|
|
raise BuildError(f"duplicate id {wanted}")
|
|
self.used_ids.add(wanted)
|
|
return wanted
|
|
base = f"{prefix}{sanitize(name)}"
|
|
n = 1
|
|
while f"{base}_{n:03d}" in self.used_ids:
|
|
n += 1
|
|
if n > 999:
|
|
raise BuildError(f"no free id for {name}")
|
|
pid = f"{base}_{n:03d}"
|
|
self.used_ids.add(pid)
|
|
return pid
|
|
|
|
# -- normalisation -----------------------------------------------------
|
|
@staticmethod
|
|
def norm_item(entry):
|
|
"""Accept 'Class' | {'Class': {...}} | {'class': 'Class', ...} -> (class, opts)"""
|
|
if isinstance(entry, str):
|
|
return entry.strip(), {}
|
|
if isinstance(entry, dict):
|
|
if "class" in entry:
|
|
d = dict(entry)
|
|
return d.pop("class").strip(), d
|
|
if len(entry) == 1:
|
|
k, v = next(iter(entry.items()))
|
|
return k.strip(), (v or {})
|
|
raise BuildError(f"cannot understand item entry {entry!r}")
|
|
|
|
def compat_refs(self, ref, owner):
|
|
"""@compat / @compat_mags / @compat_all -> class list from the compat file."""
|
|
rec = self.compat.get(owner.lower())
|
|
if not rec:
|
|
if not self.compat:
|
|
raise BuildError(f"{owner}: '{ref}' used but no --compat file was given")
|
|
self.warnings.append(f"{owner}: '{ref}' used but {owner} is not in the compat file")
|
|
return []
|
|
if ref == "@compat_mags":
|
|
pool = rec["magazines"]
|
|
elif ref == "@compat_all":
|
|
pool = rec["attachments"] + rec["magazines"]
|
|
else:
|
|
pool = rec["attachments"]
|
|
if ref == "@compat_all":
|
|
return list(pool)
|
|
# only classes that are defined as products somewhere in the source
|
|
return [c for c in pool if c.lower() in self._declared]
|
|
|
|
def expand_refs(self, lst, where, seen=None):
|
|
"""Resolve group names recursively -> list of class names."""
|
|
seen = seen or set()
|
|
out = []
|
|
for ref in lst or []:
|
|
if isinstance(ref, str) and ref.startswith("@compat"):
|
|
out.extend(self.compat_refs(ref, where))
|
|
elif isinstance(ref, str) and ref in self.groups:
|
|
if ref in seen:
|
|
raise BuildError(f"{where}: group '{ref}' references itself")
|
|
out.extend(self.expand_refs(self.groups[ref], where, seen | {ref}))
|
|
elif isinstance(ref, str):
|
|
out.append(ref.strip())
|
|
else:
|
|
raise BuildError(f"{where}: bad reference {ref!r}")
|
|
res = []
|
|
for c in out:
|
|
if c not in res:
|
|
res.append(c)
|
|
return res
|
|
|
|
# -- pass 1: register every product ------------------------------------
|
|
def register(self, cls, opts, ctx, category_id=None):
|
|
key = cls.lower()
|
|
if key in self.class_to_id:
|
|
pid = self.class_to_id[key]
|
|
if category_id and opts and any(k in opts for k in ITEM_LEVEL_KEYS if k not in ("attachments", "variants", "presets")):
|
|
self.warnings.append(f"{cls}: defined twice with options; first definition wins")
|
|
return pid
|
|
pid = self.alloc_id("prod_", cls, opts.get("id"))
|
|
self.class_to_id[key] = pid
|
|
price = opts.get("price", ctx.get("price"))
|
|
if price is None:
|
|
price = -1
|
|
if category_id:
|
|
self.errors.append(f"{cls}: no price (set 'price' on the item, the category, or defaults)")
|
|
price = int(price)
|
|
if "sell" in opts and not isinstance(opts["sell"], str):
|
|
sell = int(opts["sell"])
|
|
elif "sell" in opts and str(opts["sell"]).lstrip("-").isdigit():
|
|
sell = int(opts["sell"])
|
|
else:
|
|
pct = float(opts.get("sell_percent", ctx.get("sell_percent", self.defaults["sell_percent"])))
|
|
sell = -1 if price < 0 else int(round(price * pct / 100))
|
|
# 'sell' may be a quantity word ("full") rather than a price -> handle sell_mode
|
|
sell_mode = opts.get("sell_mode")
|
|
if isinstance(opts.get("sell"), str) and not str(opts["sell"]).lstrip("-").isdigit():
|
|
sell_mode = opts["sell"]
|
|
sell_mode = sell_mode or ctx.get("sell_mode", ctx.get("sell", self.defaults["sell"]))
|
|
if not isinstance(sell_mode, str) or sell_mode.lstrip("-").isdigit():
|
|
sell_mode = self.defaults["sell"]
|
|
buy_mode = opts.get("buy", ctx.get("buy", self.defaults["buy"]))
|
|
stock = parse_stock(opts.get("stock", ctx.get("stock", self.defaults["stock"])))
|
|
coef = float(opts.get("coefficient", ctx.get("coefficient", self.defaults["coefficient"])))
|
|
restart = opts.get("restart", ctx.get("restart", self.defaults["restart"]))
|
|
destock = opts.get("destock", ctx.get("destock", self.defaults["destock"]))
|
|
try:
|
|
tq = pack_trade_quantity(buy_mode, sell_mode)
|
|
ss = pack_stock_settings(restart, destock)
|
|
except BuildError as e:
|
|
self.errors.append(f"{cls}: {e}")
|
|
tq, ss = 18, 0
|
|
if coef != 1.0 and stock == -1:
|
|
self.warnings.append(f"{cls}: coefficient {coef} ignored by the mod because stock is unlimited")
|
|
if price >= 0 and sell > price:
|
|
self.warnings.append(f"{cls}: sell {sell} > buy {price}")
|
|
self.products[pid] = {
|
|
"className": cls, "coefficient": coef, "maxStock": stock, "tradeQuantity": tq,
|
|
"buyPrice": price, "sellPrice": sell, "stockSettings": ss,
|
|
"attachments": [], "variants": [],
|
|
"_att": self.expand_refs(opts.get("attachments"), cls),
|
|
"_var": self.expand_refs(opts.get("variants"), cls),
|
|
"_presets": opts.get("presets") or {},
|
|
}
|
|
return pid
|
|
|
|
def collect_declared(self):
|
|
"""All class names written anywhere in the source (items, categories, groups)."""
|
|
decl = set()
|
|
def add_list(lst):
|
|
for r in lst or []:
|
|
if isinstance(r, str) and not r.startswith("@") and r not in self.groups:
|
|
decl.add(r.lower())
|
|
for entry in self.src.get("items") or []:
|
|
cls, opts = self.norm_item(entry)
|
|
decl.add(cls.lower())
|
|
add_list(opts.get("attachments")); add_list(opts.get("variants"))
|
|
for cat in self.src.get("categories") or []:
|
|
for entry in cat.get("items") or []:
|
|
cls, opts = self.norm_item(entry)
|
|
decl.add(cls.lower())
|
|
add_list(opts.get("attachments")); add_list(opts.get("variants"))
|
|
for g in self.groups.values():
|
|
add_list(g)
|
|
self._declared = decl
|
|
|
|
def walk(self):
|
|
self.collect_declared()
|
|
ctx0 = {k: self.defaults[k] for k in ("sell_percent", "buy", "sell", "stock", "coefficient", "restart", "destock")}
|
|
for entry in self.src.get("items") or []:
|
|
cls, opts = self.norm_item(entry)
|
|
self.register(cls, opts, ctx0)
|
|
for cat in self.src.get("categories") or []:
|
|
name = cat.get("name")
|
|
if not name:
|
|
self.errors.append("category without 'name'")
|
|
continue
|
|
cid = self.alloc_id("cat_", name, cat.get("id"))
|
|
ctx = dict(ctx0)
|
|
for k in ("price", "sell_percent", "buy", "sell", "stock", "coefficient", "restart", "destock"):
|
|
if k in cat:
|
|
ctx[k] = cat[k]
|
|
pids = []
|
|
for entry in cat.get("items") or []:
|
|
cls, opts = self.norm_item(entry)
|
|
pid = self.register(cls, opts, ctx, cid)
|
|
if pid not in pids:
|
|
pids.append(pid)
|
|
else:
|
|
self.warnings.append(f"{cls}: listed twice in category '{name}'")
|
|
self.categories[cid] = {
|
|
"isVisible": bool(cat.get("visible", True)),
|
|
"icon": cat.get("icon", "") or "",
|
|
"categoryName": name,
|
|
"licensesRequired": [],
|
|
"productIds": pids,
|
|
}
|
|
self.cat_traders[cid] = cat.get("traders") or ([cat["trader"]] if cat.get("trader") else [])
|
|
self.cat_licenses[cid] = cat.get("licenses") or []
|
|
if not pids:
|
|
self.warnings.append(f"category '{name}' has no items")
|
|
|
|
# -- pass 2: resolve attachments / variants / presets ------------------
|
|
def ensure_attachment(self, cls, owner):
|
|
key = cls.lower()
|
|
if key in self.class_to_id:
|
|
return self.class_to_id[key]
|
|
if self.strict:
|
|
self.errors.append(f"{owner}: attachment/variant '{cls}' is not defined anywhere (strict mode)")
|
|
return None
|
|
self.notes.append(f"{owner}: '{cls}' auto-created as hidden product at attachment_price")
|
|
pid = self.register(cls, {"price": self.defaults["attachment_price"],
|
|
"sell_percent": self.defaults["attachment_sell_percent"]},
|
|
{"sell_percent": self.defaults["attachment_sell_percent"]})
|
|
return pid
|
|
|
|
def resolve(self):
|
|
for pid in list(self.products):
|
|
p = self.products[pid]
|
|
cls = p["className"]
|
|
rec = self.compat.get(cls.lower())
|
|
if rec and p["_att"]:
|
|
ok = {c.lower() for c in rec["attachments"] + rec["magazines"]}
|
|
bad = [c for c in p["_att"] if c.lower() not in ok]
|
|
if bad:
|
|
self.warnings.append(f"{cls}: not attachable according to compat file: {', '.join(bad[:6])}"
|
|
+ (f" (+{len(bad)-6} more)" if len(bad) > 6 else ""))
|
|
p["attachments"] = [x for x in (self.ensure_attachment(c, cls) for c in p["_att"]) if x]
|
|
for pid in list(self.products):
|
|
p = self.products[pid]
|
|
cls = p["className"]
|
|
vids = []
|
|
for vc in p["_var"]:
|
|
vid = self.ensure_attachment(vc, cls)
|
|
if not vid or vid == pid:
|
|
continue
|
|
vids.append(vid)
|
|
v = self.products[vid]
|
|
if not v["attachments"] and not v["_att"]:
|
|
v["attachments"] = list(p["attachments"])
|
|
p["variants"] = vids
|
|
for pid, p in self.products.items():
|
|
if not p["_presets"]:
|
|
continue
|
|
cls = p["className"]
|
|
presets = {"productId": pid, "defaultPresetId": "", "presets": []}
|
|
for pname, atts in p["_presets"].items():
|
|
is_default = pname.endswith("*")
|
|
clean = pname.rstrip("*").strip()
|
|
preset_id = f"preset_{sanitize(cls)}_{sanitize(clean)}_001"
|
|
att_ids = []
|
|
for c in self.expand_refs(atts, f"{cls}/preset {clean}"):
|
|
aid = self.class_to_id.get(c.lower())
|
|
if not aid:
|
|
self.errors.append(f"{cls}: preset '{clean}' uses undefined class '{c}'")
|
|
continue
|
|
if aid not in p["attachments"]:
|
|
self.errors.append(f"{cls}: preset '{clean}' uses '{c}' which is not in the weapon's attachments list")
|
|
continue
|
|
att_ids.append(aid)
|
|
presets["presets"].append({"presetName": clean, "presetId": preset_id,
|
|
"productId": pid, "attachments": att_ids})
|
|
if is_default:
|
|
presets["defaultPresetId"] = preset_id
|
|
if not presets["defaultPresetId"] and presets["presets"]:
|
|
presets["defaultPresetId"] = presets["presets"][0]["presetId"]
|
|
self.presets[pid] = presets
|
|
for p in self.products.values():
|
|
for k in ("_att", "_var", "_presets"):
|
|
p.pop(k, None)
|
|
|
|
# -- pass 3: traders & licenses in GeneralSettings -----------------------
|
|
def resolve_settings(self):
|
|
gs_path = os.path.join(self.cfg, "TraderXGeneralSettings.json")
|
|
if not os.path.exists(gs_path):
|
|
self.warnings.append("TraderXGeneralSettings.json not found - traders/licenses not updated")
|
|
return
|
|
gs = load_json(gs_path)
|
|
self.gs = gs
|
|
traders = gs.setdefault("traders", [])
|
|
licenses = gs.setdefault("licenses", [])
|
|
by_name = {str(t.get("givenName", "")).lower(): t for t in traders}
|
|
by_id = {str(t.get("npcId")): t for t in traders}
|
|
lic_by_name = {str(l.get("licenseName", "")).lower(): l for l in licenses}
|
|
managed = set()
|
|
for cid, refs in self.cat_traders.items():
|
|
for ref in refs:
|
|
t = by_id.get(str(ref)) or by_name.get(str(ref).lower())
|
|
if not t:
|
|
self.errors.append(f"category {cid}: trader '{ref}' not found in TraderXGeneralSettings.json")
|
|
continue
|
|
key = str(t.get("npcId"))
|
|
if key not in managed:
|
|
t["categoriesId"] = []
|
|
managed.add(key)
|
|
if cid not in t["categoriesId"]:
|
|
t["categoriesId"].append(cid)
|
|
for cid, names in self.cat_licenses.items():
|
|
ids = []
|
|
for n in names:
|
|
l = lic_by_name.get(str(n).lower())
|
|
if not l:
|
|
import uuid
|
|
l = {"licenseId": str(uuid.uuid4()), "licenseName": n, "description": ""}
|
|
licenses.append(l)
|
|
lic_by_name[n.lower()] = l
|
|
self.notes.append(f"license '{n}' created in TraderXGeneralSettings.json")
|
|
ids.append(l["licenseId"])
|
|
self.categories[cid]["licensesRequired"] = ids
|
|
for t in traders:
|
|
if str(t.get("npcId")) not in managed:
|
|
self.warnings.append(f"trader {t.get('npcId')} ({t.get('givenName')}) is not referenced by any category - left untouched")
|
|
cn = str(t.get("className", ""))
|
|
m = re.match(r"Survivor([MF])_(\w+)$", cn)
|
|
if m:
|
|
sex, nm = m.groups()
|
|
other = "F" if sex == "M" else "M"
|
|
if nm in KNOWN_SURVIVORS[other] and nm not in KNOWN_SURVIVORS[sex]:
|
|
self.errors.append(f"trader {t.get('npcId')}: {cn} does not exist, use Survivor{other}_{nm}")
|
|
|
|
# -- write --------------------------------------------------------------
|
|
def write(self):
|
|
prod_dir = os.path.join(self.cfg, "Products")
|
|
cat_dir = os.path.join(self.cfg, "Categories")
|
|
pre_dir = os.path.join(self.root, "TraderXPresets")
|
|
changed = added = 0
|
|
stale = []
|
|
plan = []
|
|
|
|
def emit(path, data):
|
|
nonlocal changed, added
|
|
new = json.dumps(data, indent=2, ensure_ascii=False) + "\n"
|
|
old = None
|
|
if os.path.exists(path):
|
|
with open(path, "r", encoding="utf-8-sig") as f:
|
|
old = f.read()
|
|
if old == new:
|
|
return
|
|
if old is None:
|
|
added += 1
|
|
plan.append(f" + {os.path.relpath(path, self.root)}")
|
|
else:
|
|
changed += 1
|
|
plan.append(f" ~ {os.path.relpath(path, self.root)}")
|
|
if not self.dry:
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
f.write(new)
|
|
|
|
for pid, p in self.products.items():
|
|
emit(os.path.join(prod_dir, pid + ".json"), {k: p[k] for k in PRODUCT_KEYS})
|
|
for cid, c in self.categories.items():
|
|
emit(os.path.join(cat_dir, cid + ".json"), c)
|
|
for pid, pr in self.presets.items():
|
|
emit(os.path.join(pre_dir, f"presets_{pid}.json"), pr)
|
|
if self.gs is not None:
|
|
emit(os.path.join(self.cfg, "TraderXGeneralSettings.json"), self.gs)
|
|
|
|
for d, ids in ((prod_dir, self.products), (cat_dir, self.categories)):
|
|
if os.path.isdir(d):
|
|
for f in os.listdir(d):
|
|
if f.endswith(".json") and f[:-5] not in ids:
|
|
stale.append(os.path.join(d, f))
|
|
for f in stale:
|
|
plan.append(f" {'- ' if self.clean else '? '}{os.path.relpath(f, self.root)}"
|
|
+ ("" if self.clean else " (not in source; --clean removes)"))
|
|
if self.clean and not self.dry:
|
|
os.remove(f)
|
|
return added, changed, stale, plan
|
|
|
|
def run(self):
|
|
try:
|
|
self.walk()
|
|
self.resolve()
|
|
self.resolve_settings()
|
|
except BuildError as e:
|
|
self.errors.append(str(e))
|
|
return self
|
|
|
|
|
|
def cmd_build(args, check_only=False):
|
|
src = load_source(args.source)
|
|
root = find_traderx_root(args.out) if not check_only else "/nonexistent"
|
|
compat = load_compat(getattr(args, "compat", None))
|
|
if compat:
|
|
print(f"compat data: {len(compat)} weapons")
|
|
b = Builder(src, root, dry_run=args.dry_run or check_only, clean=getattr(args, "clean", False),
|
|
strict=args.strict, compat=compat).run()
|
|
for n in b.notes:
|
|
print(f"note: {n}")
|
|
for w in b.warnings:
|
|
print(f"warning: {w}")
|
|
for e in b.errors:
|
|
print(f"ERROR: {e}")
|
|
print(f"\n{len(b.products)} products, {len(b.categories)} categories, {len(b.presets)} preset files")
|
|
if b.errors:
|
|
print(f"{len(b.errors)} error(s) - nothing written")
|
|
return 1
|
|
if check_only:
|
|
print("OK")
|
|
return 0
|
|
if not args.dry_run and not args.no_backup:
|
|
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
bdir = os.path.join(root, f"Backup_build_{stamp}")
|
|
for sub in ("TraderXConfig/Products", "TraderXConfig/Categories", "TraderXPresets"):
|
|
s = os.path.join(root, sub)
|
|
if os.path.isdir(s):
|
|
shutil.copytree(s, os.path.join(bdir, sub))
|
|
gsp = os.path.join(b.cfg, "TraderXGeneralSettings.json")
|
|
if os.path.exists(gsp):
|
|
os.makedirs(os.path.join(bdir, "TraderXConfig"), exist_ok=True)
|
|
shutil.copy2(gsp, os.path.join(bdir, "TraderXConfig"))
|
|
print(f"backup: {bdir}")
|
|
added, changed, stale, plan = b.write()
|
|
if plan:
|
|
print("\n".join(plan))
|
|
print(f"\n{'would write' if args.dry_run else 'wrote'}: {added} new, {changed} changed, "
|
|
f"{len(stale)} stale file(s){' removed' if args.clean and not args.dry_run else ''}")
|
|
comp = os.path.join(b.cfg, "Compiled")
|
|
if os.path.isdir(comp) and any(f.endswith(".json") for f in os.listdir(comp)):
|
|
print("WARNING: TraderXConfig/Compiled/ exists - the mod will load THAT instead of these files. Delete it.")
|
|
return 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# IMPORTERS (all produce the source schema)
|
|
# ---------------------------------------------------------------------------
|
|
def item_entry(cls, **opts):
|
|
opts = {k: v for k, v in opts.items() if v not in (None, [], {}, "")}
|
|
return cls if not opts else {cls: opts}
|
|
|
|
|
|
def merge_into(new_src, args):
|
|
"""Write new_src to -o, or append its categories/items/groups into --into."""
|
|
if args.into:
|
|
base = load_source(args.into) if os.path.exists(args.into) else {}
|
|
base.setdefault("defaults", DEFAULTS.copy())
|
|
base.setdefault("groups", {}).update(new_src.get("groups", {}))
|
|
if new_src.get("items"):
|
|
base.setdefault("items", []).extend(new_src["items"])
|
|
existing = {c["name"].lower() for c in base.get("categories", [])}
|
|
for c in new_src.get("categories", []):
|
|
if c["name"].lower() in existing:
|
|
c["name"] += " (imported)"
|
|
base.setdefault("categories", []).append(c)
|
|
dump_source(base, args.into)
|
|
print(f"appended to {args.into}")
|
|
else:
|
|
out = args.output or ("source.yaml" if yaml else "source.json")
|
|
new_src = dict({"defaults": DEFAULTS.copy()}, **new_src) if "defaults" not in new_src else new_src
|
|
dump_source(new_src, out)
|
|
print(f"written {out}")
|
|
|
|
|
|
def cat_block(name, args, items, **extra):
|
|
c = {"name": name}
|
|
if args.trader:
|
|
c["traders"] = [int(t) if t.isdigit() else t for t in args.trader]
|
|
c.update({k: v for k, v in extra.items() if v not in (None, "", [])})
|
|
c["items"] = items
|
|
return c
|
|
|
|
|
|
def cmd_import_plain(args):
|
|
with open(args.file, encoding="utf-8-sig") as f:
|
|
lines = [l.strip() for l in f]
|
|
items = []
|
|
for l in lines:
|
|
if not l or l.startswith("#"):
|
|
continue
|
|
parts = [p.strip() for p in re.split(r"[,\t;]", l)]
|
|
cls = parts[0]
|
|
if len(parts) > 1 and parts[1].lstrip("-").isdigit():
|
|
items.append(item_entry(cls, price=int(parts[1]), sell=int(parts[2]) if len(parts) > 2 and parts[2].lstrip("-").isdigit() else None))
|
|
else:
|
|
items.append(cls)
|
|
cat = cat_block(args.category or os.path.splitext(os.path.basename(args.file))[0], args, items, price=args.price)
|
|
merge_into({"categories": [cat]}, args)
|
|
|
|
|
|
def cmd_import_traderplus(args):
|
|
data = load_json(args.file)
|
|
cats_in = data.get("TraderCategories") or data.get("Categories") or ([data] if "Products" in data else data)
|
|
cats = []
|
|
for c in cats_in:
|
|
items = []
|
|
for line in c.get("Products", []):
|
|
parts = [p.strip() for p in line.split(",")]
|
|
if len(parts) < 6:
|
|
print(f"skip malformed line: {line}")
|
|
continue
|
|
cls, coef, maxstock, tq, buy, sell = parts[:6]
|
|
try:
|
|
coef, maxstock, tq, buy, sell = float(coef), int(maxstock), int(tq), int(buy), int(sell)
|
|
except ValueError:
|
|
print(f"skip malformed line: {line}")
|
|
continue
|
|
bmode, smode = unpack_trade_quantity(tq)
|
|
items.append(item_entry(cls, price=buy, sell=sell,
|
|
stock=None if maxstock == -1 else maxstock,
|
|
coefficient=None if coef == 1.0 else coef,
|
|
buy=None if bmode == DEFAULTS["buy"] else bmode,
|
|
sell_mode=None if smode == DEFAULTS["sell"] else smode))
|
|
cats.append(cat_block(args.category or c.get("CategoryName", "Imported"), args, items))
|
|
merge_into({"categories": cats}, args)
|
|
|
|
|
|
def cmd_import_expansion(args):
|
|
paths = []
|
|
if os.path.isdir(args.file):
|
|
paths = [os.path.join(args.file, f) for f in sorted(os.listdir(args.file)) if f.lower().endswith(".json")]
|
|
else:
|
|
paths = [args.file]
|
|
cats = []
|
|
for p in paths:
|
|
d = load_json(p)
|
|
items = []
|
|
for it in d.get("Items", []):
|
|
cls = it.get("ClassName")
|
|
if not cls:
|
|
continue
|
|
buy = int(it.get("MaxPriceThreshold", -1))
|
|
minp = int(it.get("MinPriceThreshold", buy))
|
|
spp = float(it.get("SellPricePercent", -1))
|
|
sell = int(round(buy * spp / 100)) if spp >= 0 and buy >= 0 else None
|
|
maxs, mins = int(it.get("MaxStockThreshold", 1)), int(it.get("MinStockThreshold", 1))
|
|
opts = {"price": buy, "sell": sell}
|
|
if maxs > 1 and buy != minp:
|
|
# Expansion dynamic price between min..max over stock range -> approximate coefficient
|
|
opts["stock"] = maxs
|
|
if buy > 0 and minp > 0 and maxs > 1:
|
|
opts["coefficient"] = round((minp / buy) ** (1.0 / (maxs - 1)), 4)
|
|
qp = it.get("QuantityPercent", -1)
|
|
if isinstance(qp, (int, float)) and qp >= 0:
|
|
opts["buy"] = "empty" if qp == 0 else ("full" if qp >= 100 else f"coef {int(qp)}%")
|
|
opts["attachments"] = [a for a in it.get("SpawnAttachments", []) if a]
|
|
opts["variants"] = [v for v in it.get("Variants", []) if v]
|
|
items.append(item_entry(cls, **opts))
|
|
cats.append(cat_block(args.category or d.get("DisplayName", os.path.splitext(os.path.basename(p))[0]).replace("_", " ").title(),
|
|
args, items))
|
|
merge_into({"categories": cats}, args)
|
|
|
|
|
|
def cmd_import_traderx(args):
|
|
root = find_traderx_root(args.path)
|
|
cfg = os.path.join(root, "TraderXConfig")
|
|
prods, cats, presets = {}, {}, {}
|
|
pdir, cdir, prdir = os.path.join(cfg, "Products"), os.path.join(cfg, "Categories"), os.path.join(root, "TraderXPresets")
|
|
for f in sorted(os.listdir(pdir)) if os.path.isdir(pdir) else []:
|
|
if f.endswith(".json"):
|
|
try:
|
|
d = load_json(os.path.join(pdir, f))
|
|
if isinstance(d, dict) and d.get("className"):
|
|
prods[f[:-5]] = d
|
|
except Exception as e:
|
|
print(f"skip {f}: {e}")
|
|
for f in sorted(os.listdir(cdir)) if os.path.isdir(cdir) else []:
|
|
if f.endswith(".json"):
|
|
try:
|
|
cats[f[:-5]] = load_json(os.path.join(cdir, f))
|
|
except Exception as e:
|
|
print(f"skip {f}: {e}")
|
|
for f in sorted(os.listdir(prdir)) if os.path.isdir(prdir) else []:
|
|
if f.startswith("presets_") and f.endswith(".json"):
|
|
try:
|
|
d = load_json(os.path.join(prdir, f))
|
|
presets[d.get("productId", f[8:-5])] = d
|
|
except Exception as e:
|
|
print(f"skip {f}: {e}")
|
|
gs = load_json(os.path.join(cfg, "TraderXGeneralSettings.json")) if os.path.exists(os.path.join(cfg, "TraderXGeneralSettings.json")) else {}
|
|
lic_names = {l.get("licenseId"): l.get("licenseName") for l in gs.get("licenses", [])}
|
|
cat_traders = {}
|
|
for t in gs.get("traders", []):
|
|
for cid in t.get("categoriesId", []):
|
|
cat_traders.setdefault(cid, []).append(t.get("givenName") or t.get("npcId"))
|
|
|
|
cls_of = {pid: p["className"] for pid, p in prods.items()}
|
|
|
|
def entry(pid):
|
|
p = prods[pid]
|
|
bmode, smode = unpack_trade_quantity(int(p.get("tradeQuantity", 18)))
|
|
restart, destock = unpack_stock_settings(int(p.get("stockSettings", 0)))
|
|
opts = {"id": pid, "price": p.get("buyPrice", -1), "sell": p.get("sellPrice", -1)}
|
|
if p.get("maxStock", -1) != -1:
|
|
opts["stock"] = p["maxStock"]
|
|
if float(p.get("coefficient", 1.0)) != 1.0:
|
|
opts["coefficient"] = p["coefficient"]
|
|
if bmode != DEFAULTS["buy"]:
|
|
opts["buy"] = bmode
|
|
if smode != DEFAULTS["sell"]:
|
|
opts["sell_mode"] = smode
|
|
if restart != "keep":
|
|
opts["restart"] = restart
|
|
if destock:
|
|
opts["destock"] = destock
|
|
opts["attachments"] = [cls_of[a] for a in p.get("attachments", []) if a in cls_of]
|
|
opts["variants"] = [cls_of[v] for v in p.get("variants", []) if v in cls_of]
|
|
pr = presets.get(pid)
|
|
if pr:
|
|
pm = {}
|
|
for ps in pr.get("presets", []):
|
|
nm = ps.get("presetName", "preset") + ("*" if ps.get("presetId") == pr.get("defaultPresetId") else "")
|
|
pm[nm] = [cls_of[a] for a in ps.get("attachments", []) if a in cls_of]
|
|
opts["presets"] = pm
|
|
return item_entry(p["className"], **opts)
|
|
|
|
used = set()
|
|
out_cats = []
|
|
for cid, c in cats.items():
|
|
items = []
|
|
for pid in c.get("productIds", []):
|
|
if pid in prods:
|
|
items.append(entry(pid))
|
|
used.add(pid)
|
|
cat = {"name": c.get("categoryName", cid), "id": cid}
|
|
if cid in cat_traders:
|
|
cat["traders"] = cat_traders[cid]
|
|
lic = [lic_names.get(l, l) for l in c.get("licensesRequired", [])]
|
|
if lic:
|
|
cat["licenses"] = lic
|
|
if not c.get("isVisible", True):
|
|
cat["visible"] = False
|
|
if c.get("icon"):
|
|
cat["icon"] = c["icon"]
|
|
cat["items"] = items
|
|
out_cats.append(cat)
|
|
hidden = [entry(pid) for pid in prods if pid not in used]
|
|
src = {"defaults": DEFAULTS.copy(), "groups": {}, "items": hidden, "categories": out_cats}
|
|
out = args.output or ("source.yaml" if yaml else "source.json")
|
|
dump_source(src, out)
|
|
print(f"written {out}: {len(prods)} products ({len(hidden)} hidden), {len(cats)} categories, {len(presets)} preset files")
|
|
|
|
|
|
def cmd_compat(args):
|
|
compat = load_compat(args.files)
|
|
rec = compat.get(args.classname.lower())
|
|
if not rec:
|
|
print(f"{args.classname} not found in compat data ({len(compat)} weapons known)")
|
|
return 1
|
|
for k in ("attachments", "magazines", "bullets"):
|
|
print(f"{k} ({len(rec[k])}):")
|
|
for c in rec[k]:
|
|
print(f" {c}")
|
|
return 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
def main():
|
|
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
sub = ap.add_subparsers(dest="cmd", required=True)
|
|
|
|
def add_import_common(p):
|
|
p.add_argument("-o", "--output", help="write a new source file (default source.yaml/.json)")
|
|
p.add_argument("--into", help="append into an existing source file instead")
|
|
p.add_argument("--category", help="category name (default: from the export)")
|
|
p.add_argument("--trader", action="append", default=[], help="trader givenName or npcId (repeatable)")
|
|
|
|
p = sub.add_parser("import-traderx", help="reverse-import an existing TraderX config")
|
|
p.add_argument("path")
|
|
p.add_argument("-o", "--output")
|
|
p.set_defaults(fn=cmd_import_traderx)
|
|
|
|
p = sub.add_parser("import-traderplus", help="TraderPlus v1 PriceConfig export")
|
|
p.add_argument("file")
|
|
add_import_common(p)
|
|
p.set_defaults(fn=cmd_import_traderplus)
|
|
|
|
p = sub.add_parser("import-expansion", help="Expansion market json (file or folder)")
|
|
p.add_argument("file")
|
|
add_import_common(p)
|
|
p.set_defaults(fn=cmd_import_expansion)
|
|
|
|
p = sub.add_parser("import-plain", help="plain class list, optionally 'class,buy[,sell]'")
|
|
p.add_argument("file")
|
|
p.add_argument("--price", type=int, help="default buy price for the category")
|
|
add_import_common(p)
|
|
p.set_defaults(fn=cmd_import_plain)
|
|
|
|
p = sub.add_parser("build", help="compile source -> TraderX files")
|
|
p.add_argument("source")
|
|
p.add_argument("--out", required=True, help="TraderX profile folder (parent of TraderXConfig)")
|
|
p.add_argument("--dry-run", action="store_true")
|
|
p.add_argument("--clean", action="store_true", help="delete product/category files not in the source")
|
|
p.add_argument("--strict", action="store_true", help="do not auto-create undefined attachments")
|
|
p.add_argument("--compat", action="append", default=[], metavar="FILE",
|
|
help="'All classes and their attachments' json (repeatable); enables @compat refs and compatibility warnings")
|
|
p.add_argument("--no-backup", action="store_true")
|
|
p.set_defaults(fn=lambda a: cmd_build(a))
|
|
|
|
p = sub.add_parser("check", help="validate the source only")
|
|
p.add_argument("source")
|
|
p.add_argument("--strict", action="store_true")
|
|
p.add_argument("--compat", action="append", default=[], metavar="FILE")
|
|
p.set_defaults(fn=lambda a: cmd_build(argparse.Namespace(**vars(a), out=".", dry_run=True, clean=False, no_backup=True), check_only=True))
|
|
|
|
p = sub.add_parser("compat", help="show what a weapon accepts according to a compat file")
|
|
p.add_argument("classname")
|
|
p.add_argument("files", nargs="+", help="one or more compat json files")
|
|
p.set_defaults(fn=cmd_compat)
|
|
|
|
args = ap.parse_args()
|
|
return args.fn(args) or 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|