141 lines
5.2 KiB
Python
141 lines
5.2 KiB
Python
import json
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
|
|
|
|
CONFIG_PATH = Path(__file__).resolve().parent.parent / "config.json"
|
|
|
|
|
|
def load_config():
|
|
|
|
with CONFIG_PATH.open("r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def get_product_config(product_name, config=None):
|
|
"""Resolve a product's setup flags and session defaults without mutating input."""
|
|
config = deepcopy(load_config() if config is None else config)
|
|
if "products" not in config:
|
|
if product_name != "IDAS":
|
|
raise ValueError(f"Unknown setup product: {product_name}")
|
|
config["token"] = config.get("trading_token_idas", {})
|
|
else:
|
|
product = config["products"][product_name]
|
|
setup = config.setdefault("setup", {})
|
|
for flag in ("run_session", "run_token"):
|
|
setup[flag] = product.get(flag, setup.get(flag, True))
|
|
if not isinstance(setup[flag], bool):
|
|
raise ValueError(f"products.{product_name}.{flag} must be true or false")
|
|
setup["session"] = {**setup.get("session", {}), **product.get("session", {})}
|
|
config["product"] = {"me_instance": product.get("me_instance")}
|
|
config["token"] = deepcopy(product.get("token", {}))
|
|
config["product_name"] = product_name
|
|
return config
|
|
|
|
|
|
def get_setup_configs(mode, config=None):
|
|
"""Partition products so a full setup run never creates a product twice."""
|
|
if mode not in ("combined", "session", "token"):
|
|
raise ValueError(f"Unknown setup mode: {mode}")
|
|
config = load_config() if config is None else config
|
|
products = config.get("products", {"IDAS": {}})
|
|
selected = []
|
|
for name, product in products.items():
|
|
if not product.get("enabled", True):
|
|
continue
|
|
variants = product.get("session_variants")
|
|
if variants is not None:
|
|
resolved = get_product_config(name, config)
|
|
# Expand variants when either session or token execution is
|
|
# enabled. Token-only UPC runs must still reach their variants.
|
|
setup = resolved.get("setup", {})
|
|
if not setup.get("run_session", True) and not setup.get("run_token", True):
|
|
continue
|
|
variant_config = deepcopy(config)
|
|
variant_config["products"] = {}
|
|
for variant_name, variant in variants.items():
|
|
settings = deepcopy(product)
|
|
settings.pop("session_variants")
|
|
settings.update(variant)
|
|
settings["session"] = {**product.get("session", {}), **variant.get("session", {})}
|
|
settings["token"] = {**product.get("token", {}), **variant.get("token", {})}
|
|
variant_config["products"][f"{name}_{variant_name}"] = settings
|
|
selected.extend(get_setup_configs(mode, variant_config))
|
|
continue
|
|
resolved = get_product_config(name, config)
|
|
setup = resolved.get("setup", {})
|
|
session = setup.get("run_session", True)
|
|
token = setup.get("run_token", True)
|
|
matches = {
|
|
# The full setup run owns every enabled product that has work to do.
|
|
# Products without tokens (for example RA) still belong in this
|
|
# shared browser flow; the combined test skips only their token step.
|
|
"combined": session or token,
|
|
"session": session and not token,
|
|
"token": token and not session,
|
|
}
|
|
if not matches[mode]:
|
|
continue
|
|
if session and not resolved.get("product", {}).get("me_instance"):
|
|
raise ValueError(f"products.{name}.me_instance is required for session creation")
|
|
if token:
|
|
settings = resolved["token"]
|
|
if not settings.get("market_code") or not (
|
|
parse_product_codes(settings.get("create_product_codes", settings.get("product_codes")))
|
|
or parse_product_codes(settings.get("product_code"))
|
|
):
|
|
raise ValueError(f"products.{name}.token requires market_code and product_code(s)")
|
|
selected.append(resolved)
|
|
return selected
|
|
|
|
|
|
def get_step_delay_seconds(default=5):
|
|
|
|
config = load_config()
|
|
return config.get("step_delay_seconds", default)
|
|
|
|
|
|
def get_login_captcha_wait_seconds(default=20):
|
|
|
|
config = load_config()
|
|
return config.get("login_captcha_wait_seconds", default)
|
|
|
|
|
|
def get_login_post_submit_wait_seconds(default=10):
|
|
|
|
config = load_config()
|
|
return config.get("login_post_submit_wait_seconds", default)
|
|
|
|
|
|
def is_setup_session_enabled(default=True):
|
|
|
|
config = load_config()
|
|
return config.get("setup", {}).get("run_session", default)
|
|
|
|
|
|
def is_setup_token_enabled(default=True):
|
|
|
|
config = load_config()
|
|
return config.get("setup", {}).get("run_token", default)
|
|
|
|
|
|
def is_combined_setup_enabled():
|
|
|
|
config = load_config()
|
|
setup = config.get("setup", {})
|
|
return setup.get("run_session", True) and setup.get("run_token", True)
|
|
|
|
|
|
def parse_product_codes(value):
|
|
|
|
if value is None:
|
|
return []
|
|
|
|
if isinstance(value, str):
|
|
return [code.strip() for code in value.split(",") if code.strip()]
|
|
|
|
if isinstance(value, list):
|
|
return [str(code).strip() for code in value if str(code).strip()]
|
|
|
|
return [str(value).strip()]
|