#!/usr/bin/env python3 """ Refresh buff_selection_request_item for one request from a JSON file. Default target: - user_id = 8 - request_id = 20 - json files = datarefresh/111.json, datarefresh/333.json, datarefresh/2222.json """ from __future__ import annotations import argparse import json from dataclasses import dataclass from decimal import Decimal, ROUND_HALF_UP from pathlib import Path from typing import Any import pymysql USD_RATE = Decimal("5.1") TWO_DEC = Decimal("0.01") @dataclass class DbConfig: host: str port: int user: str password: str database: str def parse_env_file(env_path: Path) -> dict[str, str]: if not env_path.exists(): raise FileNotFoundError(f".env file not found: {env_path}") result: dict[str, str] = {} for raw_line in env_path.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) result[key.strip()] = value.strip().strip('"').strip("'") return result def load_db_config(env_path: Path) -> DbConfig: env_map = parse_env_file(env_path) required_keys = ["DB_HOST", "DB_PORT", "DB_USER", "DB_PASSWORD", "DB_NAME"] missing = [k for k in required_keys if k not in env_map or not env_map[k]] if missing: raise ValueError(f"Missing DB config in {env_path}: {', '.join(missing)}") return DbConfig( host=env_map["DB_HOST"], port=int(env_map["DB_PORT"]), user=env_map["DB_USER"], password=env_map["DB_PASSWORD"], database=env_map["DB_NAME"], ) def normalize_price(price: Any) -> Decimal: cny = Decimal(str(price)).quantize(TWO_DEC, rounding=ROUND_HALF_UP) return cny def build_wear_range(min_paintwear: Any, max_paintwear: Any) -> str: return f"{min_paintwear}-{max_paintwear}" def placeholders(size: int) -> str: return ",".join(["%s"] * size) def load_payload_from_files(json_paths: list[Path]) -> dict[str, Any]: if not json_paths: raise ValueError("At least one JSON path is required") merged_payload: dict[str, Any] = {} for path in json_paths: if not path.exists(): raise FileNotFoundError(f"JSON file not found: {path}") payload = json.loads(path.read_text(encoding="utf-8")) if not isinstance(payload, dict): raise ValueError(f"JSON root must be an object in file: {path}") if not payload: raise ValueError(f"JSON has no data in file: {path}") merged_payload.update(payload) if not merged_payload: raise ValueError("Merged JSON payload has no data") return merged_payload def refresh_items( *, db_config: DbConfig, json_paths: list[Path], user_id: int, request_id: int, ) -> None: payload = load_payload_from_files(json_paths) goods_ids = [str(goods_id) for goods_id in payload.keys()] conn = pymysql.connect( host=db_config.host, port=db_config.port, user=db_config.user, password=db_config.password, database=db_config.database, charset="utf8mb4", autocommit=False, cursorclass=pymysql.cursors.DictCursor, ) try: with conn.cursor() as cursor: sql_goods = f""" SELECT gi.goods_id, gi.name AS goods_name, gi.steam_url AS steam_link, ga.icon_url FROM buff_goods_info gi LEFT JOIN buff_goods_attr ga ON gi.goods_id = ga.goods_id WHERE gi.goods_id IN ({placeholders(len(goods_ids))}) """ cursor.execute(sql_goods, goods_ids) goods_rows = cursor.fetchall() goods_map = {str(row["goods_id"]): row for row in goods_rows} missing_goods = [gid for gid in goods_ids if gid not in goods_map] if missing_goods: raise ValueError( "These goods_id are missing in buff_goods_info: " + ", ".join(missing_goods) ) delete_sql = """ DELETE FROM buff_selection_request_item WHERE request_id = %s AND user_id = %s """ cursor.execute(delete_sql, (request_id, user_id)) deleted_count = cursor.rowcount insert_sql = """ INSERT INTO buff_selection_request_item ( request_id, user_id, goods_id, goods_name, icon_url, steam_link, price_cny, price_usd, wear_range ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) """ rows_to_insert = [] for goods_id, value in payload.items(): if not isinstance(value, dict): raise ValueError(f"Invalid value for goods_id={goods_id}: must be object") if "min_paintwear" not in value or "max_paintwear" not in value or "price" not in value: raise ValueError( f"goods_id={goods_id} missing one of required keys: " "min_paintwear, max_paintwear, price" ) goods_meta = goods_map[str(goods_id)] price_cny = normalize_price(value["price"]) price_usd = (price_cny / USD_RATE).quantize(TWO_DEC, rounding=ROUND_HALF_UP) wear_range = build_wear_range(value["min_paintwear"], value["max_paintwear"]) rows_to_insert.append( ( request_id, user_id, str(goods_id), goods_meta.get("goods_name"), goods_meta.get("icon_url"), goods_meta.get("steam_link"), price_cny, price_usd, wear_range, ) ) cursor.executemany(insert_sql, rows_to_insert) inserted_count = cursor.rowcount conn.commit() print(f"Done. Deleted {deleted_count} rows, inserted {inserted_count} rows.") files = ", ".join(str(p) for p in json_paths) print(f"request_id={request_id}, user_id={user_id}, json_files=[{files}]") except Exception: conn.rollback() raise finally: conn.close() def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Refresh buff_selection_request_item from JSON data" ) parser.add_argument( "--json-paths", nargs="+", default=[ str(Path(__file__).resolve().parent / "111.json"), str(Path(__file__).resolve().parent / "333.json"), str(Path(__file__).resolve().parent / "2222.json"), ], help="Paths to source JSON files. Later files override duplicate goods_id", ) parser.add_argument( "--env-path", default=str(Path(__file__).resolve().parents[1] / "backend-admin" / ".env"), help="Path to .env with DB_HOST/DB_PORT/DB_USER/DB_PASSWORD/DB_NAME", ) parser.add_argument("--user-id", type=int, default=11, help="Target user_id") parser.add_argument("--request-id", type=int, default=21, help="Target request_id") return parser.parse_args() def main() -> None: args = parse_args() env_path = Path(args.env_path).resolve() json_paths = [Path(path).resolve() for path in args.json_paths] db_config = load_db_config(env_path) refresh_items( db_config=db_config, json_paths=json_paths, user_id=args.user_id, request_id=args.request_id, ) if __name__ == "__main__": main()