# -*- coding: utf-8 -*- """SQLite data-access layer for the DNF90 GM tool. The server stores almost everything as JSON blobs in dnf_repository_records (record_kind + record_key -> payload). This module reads/writes those blobs with a small amount of schema-aware convenience on top. """ import hashlib import json import os import sqlite3 import subprocess import sys import threading import time def _resource_dir(): """资源目录（PyInstaller 打包时资源在 _MEIPASS，否则脚本目录）。""" if getattr(sys, "frozen", False): return getattr(sys, "_MEIPASS", os.path.dirname(os.path.abspath(__file__))) return os.path.dirname(os.path.abspath(__file__)) # 物品名称映射缓存（item_id -> name），从外部 GM配置 目录的 dnf_items.json 惰性加载。 # 该文件由 pvfexport.exe 从 Script.pvf 导出，离线也能显示物品名称。 # 服务端(SERVER_ROOT)与 GM配置 目录由 server.py 启动时通过 set_gm_config 注入。 _ITEM_NAMES = None _ITEM_NAMES_LOADED = False _GM_CONFIG_DIR = "" _PVF_PATH = "" _EXPORT_LOCK = threading.Lock() def set_gm_config(gm_config_dir, pvf_path): """由 server.py 启动时注入 GM配置 目录与 Script.pvf 路径（可多次调用重置）。""" global _GM_CONFIG_DIR, _PVF_PATH, _ITEM_NAMES, _ITEM_NAMES_LOADED _GM_CONFIG_DIR = gm_config_dir or "" _PVF_PATH = pvf_path or "" _ITEM_NAMES = None _ITEM_NAMES_LOADED = False def _items_json_path(): return os.path.join(_GM_CONFIG_DIR, "dnf_items.json") if _GM_CONFIG_DIR else "" def _find_pvf_path(): """返回 Script.pvf 路径；优先 _PVF_PATH，否则在 GM配置 上级的 Client/ 下查找。""" if _PVF_PATH and os.path.exists(_PVF_PATH): return _PVF_PATH # 兜底：GM配置 目录上级的 Client/Script.pvf 或 Server/deploy/assets/Script.pvf if _GM_CONFIG_DIR: root = os.path.dirname(_GM_CONFIG_DIR) for cand in ( os.path.join(root, "Client", "Script.pvf"), os.path.join(root, "Server", "deploy", "assets", "Script.pvf"), ): if os.path.exists(cand): return cand return _PVF_PATH def _run_pvfexport(items_path, pvf_path): """调用 pvfexport.exe 重新导出名称表，成功返回 True。""" exe = os.path.join(_GM_CONFIG_DIR, "pvfexport.exe") if not os.path.exists(exe): return False try: # 用 subprocess 调 Windows 程序，路径须为 Windows 风格（反斜杠/盘符）。 exe_win = exe pvf_win = pvf_path.replace("/", "\\") out_win = items_path.replace("/", "\\") proc = subprocess.run( [exe_win, "-pvf", pvf_win, "-out", out_win, "-compact"], capture_output=True, text=True, timeout=300, ) return proc.returncode == 0 and os.path.exists(items_path) except Exception: return False def _load_item_names(): """加载 dnf_items.json（{item_id: name} 映射），失败返回空 dict。 PVF 比名称表新（或名称表缺失）时，自动调用 pvfexport.exe 重新导出。 """ global _ITEM_NAMES, _ITEM_NAMES_LOADED if _ITEM_NAMES_LOADED: return _ITEM_NAMES _ITEM_NAMES_LOADED = True _ITEM_NAMES = {} items_path = _items_json_path() pvf_path = _find_pvf_path() if not items_path or not pvf_path: # 回退到打包资源目录（旧行为，兼容无 GM配置 目录的情况） candidate = os.path.join(_resource_dir(), "dnf_items.json") if os.path.exists(candidate): try: with open(candidate, "r", encoding="utf-8") as f: _ITEM_NAMES = json.load(f) except Exception: _ITEM_NAMES = {} return _ITEM_NAMES # 需要重导出：名称表缺失，或 PVF 修改时间晚于名称表 need_export = not os.path.exists(items_path) if not need_export: try: need_export = os.path.getmtime(pvf_path) > os.path.getmtime(items_path) except OSError: need_export = True if need_export: with _EXPORT_LOCK: # 二次确认（避免并发重复触发） need_export = not os.path.exists(items_path) if not need_export: try: need_export = os.path.getmtime(pvf_path) > os.path.getmtime(items_path) except OSError: need_export = True if need_export: _run_pvfexport(items_path, pvf_path) if os.path.exists(items_path): try: with open(items_path, "r", encoding="utf-8") as f: data = json.load(f) if isinstance(data, dict): _ITEM_NAMES = data except Exception: _ITEM_NAMES = {} return _ITEM_NAMES def item_name(item_id): """返回 item_id 对应的物品名称，未知则返回空字符串。""" names = _load_item_names() return names.get(str(item_id), "") class DNFDatabase(object): def __init__(self, path): self.path = path self.con = None self._lock = threading.RLock() def connect(self): if self.con is None: self.con = sqlite3.connect(self.path, check_same_thread=False) self.con.row_factory = sqlite3.Row return self.con def close(self): with self._lock: if self.con is not None: self.con.close() self.con = None def _exec(self, fn): """run fn(connection) under the thread lock.""" with self._lock: return fn(self.connect()) # ---- low level ------------------------------------------------- def list_tables(self): def fn(con): cur = con.cursor() cur.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name") return [r["name"] for r in cur.fetchall()] return self._exec(fn) def list_records(self): def fn(con): cur = con.cursor() cur.execute("SELECT record_kind, record_key, length(payload) AS size " "FROM dnf_repository_records ORDER BY record_kind, record_key") return [dict(r) for r in cur.fetchall()] return self._exec(fn) def get_record(self, kind, key): def fn(con): cur = con.cursor() cur.execute("SELECT payload, checksum, updated_at FROM dnf_repository_records " "WHERE record_kind=? AND record_key=?", (kind, key)) r = cur.fetchone() if r is None: return None return {"payload": r["payload"], "checksum": r["checksum"], "updated_at": r["updated_at"]} return self._exec(fn) def get_record_json(self, kind, key): r = self.get_record(kind, key) if r is None: return None try: return json.loads(r["payload"].decode("utf-8")) except (ValueError, AttributeError): return None def put_record_json(self, kind, key, obj): """overwrite a record's JSON payload, recomputing the sha256 checksum.""" blob = json.dumps(obj, ensure_ascii=False).encode("utf-8") checksum = hashlib.sha256(blob).hexdigest() def fn(con): cur = con.cursor() cur.execute("UPDATE dnf_repository_records SET payload=?, checksum=? " "WHERE record_kind=? AND record_key=?", (sqlite3.Binary(blob), checksum, kind, key)) if cur.rowcount == 0: cur.execute("INSERT INTO dnf_repository_records " "(record_kind, record_key, payload, checksum, updated_at) " "VALUES (?, ?, ?, ?, ?)", (kind, key, sqlite3.Binary(blob), checksum, "")) con.commit() self._exec(fn) # ---- domain helpers ------------------------------------------- def list_accounts(self): def fn(con): cur = con.cursor() cur.execute("SELECT account_id, username, state, created_at, last_login_at " "FROM dnf_login_accounts ORDER BY username") return [dict(r) for r in cur.fetchall()] return self._exec(fn) def list_characters(self): """character records (record_kind='character') with parsed summary.""" def fn(con): out = [] cur = con.cursor() cur.execute("SELECT record_key, payload FROM dnf_repository_records " "WHERE record_kind='character' ORDER BY record_key") for r in cur.fetchall(): try: ch = json.loads(r["payload"].decode("utf-8")) except (ValueError, AttributeError): continue out.append({ "character_id": ch.get("character_id", r["record_key"]), "account_id": ch.get("account_id", ""), "name": ch.get("name", ""), "job": ch.get("job", "0"), "level": ch.get("level", 0), "updated_at": ch.get("updated_at", ""), }) return out return self._exec(fn) def get_character(self, char_id): return self.get_record_json("character", str(char_id)) def get_character_stats(self, char_id): ch = self.get_character(str(char_id)) return (ch or {}).get("stats", {}) if ch else None def set_character_field(self, char_id, field, value): """set a top-level field (name/job/level) on a character record.""" ch = self.get_character(str(char_id)) if ch is None: raise KeyError("角色 %s 不存在" % char_id) ch[field] = value self.put_record_json("character", str(char_id), ch) def set_character_stat(self, char_id, stat_name, value): """set a value inside character.stats.""" ch = self.get_character(str(char_id)) if ch is None: raise KeyError("角色 %s 不存在" % char_id) ch.setdefault("stats", {})[stat_name] = value self.put_record_json("character", str(char_id), ch) def grant_item(self, char_id, item_id, count=1): """add an item to a character's inventory (SQLite mode). slot keys are "{container}:{slot}" (e.g. "0:23"). We append to the next free slot of container 0 (the main bag).""" inv = self.get_record_json("inventory", str(char_id)) or {} slots = inv.setdefault("slots", {}) used = set() for k in slots: if ":" in k: cont, _, slot = k.partition(":") if cont == "0" and slot.isdigit(): used.add(int(slot)) slot = 1 while slot in used: slot += 1 slots["0:%d" % slot] = {"item_id": item_id, "count": count, "expire_at": "0001-01-01T00:00:00Z"} inv["updated_at"] = self._now_iso() self.put_record_json("inventory", str(char_id), inv) return {"slot": "0:%d" % slot} def inventory_list_slots(self, char_id): """列出角色背包所有行（SQLite 直读），供「背包管理」展示。 返回 [{"collection":"slots","entry_key":key,"item_id":..,"count":..,"bind":..}, ...]。 key 格式 "{container}:{slot}"（如 "0:23"）。""" inv = self.get_record_json("inventory", str(char_id)) or {} slots = inv.get("slots", {}) out = [] for key, v in sorted(slots.items()): if not isinstance(v, dict): continue iid = v.get("item_id", 0) out.append({ "collection": "slots", "entry_key": key, "item_id": iid, "count": int(v.get("count", 0) or 0), "bind": bool(v.get("bind", False)), "item_name": item_name(iid), }) return out def inventory_delete_slot(self, char_id, entry_key): """删除背包里指定的一行（SQLite 直读），key 格式 "{container}:{slot}"。""" inv = self.get_record_json("inventory", str(char_id)) if inv is None: raise KeyError("角色 %s 的背包记录不存在" % char_id) slots = inv.get("slots", {}) if entry_key not in slots: raise KeyError("背包行 %s 不存在" % entry_key) removed = slots.pop(entry_key) inv["slots"] = slots inv["updated_at"] = self._now_iso() self.put_record_json("inventory", str(char_id), inv) return { "entry_key": entry_key, "item_id": removed.get("item_id", 0) if isinstance(removed, dict) else 0, "removed": removed.get("count", 1) if isinstance(removed, dict) else 1, } def set_fatigue(self, char_id, fatigue, fatigue_limit=None): """设置疲劳值（fatigue）与疲劳上限（fatigue_limit）。 - fatigue : 当前疲劳值（剩余疲劳） - fatigue_limit : 疲劳上限（每天 6 点重置回该值，默认 156 / 黑钻 188） 所谓「无限疲劳」即把两者都设成一个很大的值，同时把已用疲劳 （fatigue_display_update）清零，保证客户端显示正确。 """ ch = self.get_character(str(char_id)) if ch is None: raise KeyError("角色 %s 不存在" % char_id) stats = ch.setdefault("stats", {}) fatigue = int(fatigue) if fatigue < 0: fatigue = 0 stats["fatigue"] = fatigue if fatigue_limit is not None: limit = int(fatigue_limit) if limit < 0: limit = 0 stats["fatigue_limit"] = limit # 已用疲劳 = 上限 - 当前，客户端据此显示；拉满时清零 limit = stats.get("fatigue_limit", 0) used = max(0, limit - fatigue) stats["fatigue_display_update"] = used ch["updated_at"] = self._now_iso() self.put_record_json("character", str(char_id), ch) return {"fatigue": stats["fatigue"], "fatigue_limit": stats.get("fatigue_limit", 0), "used": used} def get_fatigue(self, char_id): """读取角色的疲劳相关信息。""" ch = self.get_character(str(char_id)) if ch is None: return None stats = (ch or {}).get("stats", {}) fatigue = stats.get("fatigue") if fatigue is None: # 兼容旧字段名 for key in ("fp", "疲劳"): if key in stats: fatigue = stats[key] break return { "fatigue": fatigue, "fatigue_limit": stats.get("fatigue_limit"), "used": stats.get("fatigue_display_update"), } @staticmethod def _now_iso(): import datetime return datetime.datetime.now().isoformat() # ---- delete / wipe (dangerous, irreversible) ------------------ def delete_character(self, char_id): """Delete a character and every record that belongs to it. Record kinds whose key is the raw character id (character, equipment, inventory, mailbox, quest, skill, dungeon_permission) plus the `character:{id}:...` settings keys are removed. Account-scoped data (account, account_inventory, account-v1:* dungeon permission) is kept. Returns the number of rows deleted. """ cid = str(char_id) def fn(con): cur = con.cursor() # character-scoped records keyed directly by the id kinds = ("character", "equipment", "inventory", "mailbox", "quest", "skill", "dungeon_permission") qmarks = ",".join("?" for _ in kinds) cur.execute( "DELETE FROM dnf_repository_records " "WHERE (record_kind IN (%s) AND record_key=?) " " OR record_key LIKE ?" % qmarks, tuple(kinds) + (cid, "character:" + cid + ":%")) deleted = cur.rowcount con.commit() return deleted return self._exec(fn) def delete_account(self, account_id): """Delete an account and all records referencing it. Removes the dnf_login_accounts row and every record whose key is the account id, the `account:{id}:...` settings keys, or an `account-v1:` dungeon-permission key belonging to that account (the latter requires reading the account-v1 key to resolve its owner, so here we delete by exact key match plus the account-scoped prefix matches we know). """ aid = str(account_id) def fn(con): cur = con.cursor() # login account row cur.execute("DELETE FROM dnf_login_accounts WHERE account_id=?", (aid,)) n1 = cur.rowcount # repository records: exact key, or account-prefixed settings cur.execute( "DELETE FROM dnf_repository_records " "WHERE record_key=? OR record_key LIKE ?", (aid, "account:" + aid + ":%")) n2 = cur.rowcount con.commit() return {"account_rows": n1, "record_rows": n2} return self._exec(fn) def list_character_record_kinds(self, char_id): """Return the record_kind/record_key rows belonging to a character, for the confirmation UI to show exactly what will be deleted.""" cid = str(char_id) def fn(con): cur = con.cursor() cur.execute( "SELECT record_kind, record_key FROM dnf_repository_records " "WHERE record_key=? OR record_key LIKE ? " "ORDER BY record_kind", (cid, "character:" + cid + ":%")) return [dict(r) for r in cur.fetchall()] return self._exec(fn) def list_account_record_kinds(self, account_id): """Return records belonging to an account (for confirmation UI).""" aid = str(account_id) def fn(con): cur = con.cursor() cur.execute( "SELECT record_kind, record_key FROM dnf_repository_records " "WHERE record_key=? OR record_key LIKE ? " "ORDER BY record_kind", (aid, "account:" + aid + ":%")) return [dict(r) for r in cur.fetchall()] return self._exec(fn) # ================================================================== # 拍卖行（上下架工具） # # 服务端把挂单全部存放在托管角色 system:auction:escrow 的 inventory # 记录里，slots 的 key 为 auction:listing:，价格/卖家/时间等字段 # 编码在 ItemStack.extra 里（见 90server go-server 的 auction/owner.go）。 # 这里按同一套数据结构直接读写 SQLite，行为与游戏内一致。 # ================================================================== AUCTION_ESCROW_CHAR = "system:auction:escrow" AUCTION_LISTING_PREFIX = "auction:listing:" AUCTION_DELIVERY_PREFIX = "auction:delivery:" AUCTION_REFUND_PREFIX = "auction:refund:" # 上架押金 / 手续费（与服务端常量一致） AUCTION_DEPOSIT = 10000 # 上架押金（取消上架不退） AUCTION_SALE_FEE_PERCENT = 5 # 成交手续费（%） # GM 挂单 ID 使用高位段，避免与服务端进程内计数器（重启后从 1 递增）碰撞 AUCTION_GM_ID_BASE = 9000000000000000000 def _auction_escrow(self): """读取（或初始化）拍卖行托管角色的 inventory 记录。""" return self.get_record_json("inventory", self.AUCTION_ESCROW_CHAR) or {} def _auction_save_escrow(self, escrow): escrow.setdefault("character_id", self.AUCTION_ESCROW_CHAR) escrow.setdefault("slots", {}) escrow["updated_at"] = self._now_iso() self.put_record_json("inventory", self.AUCTION_ESCROW_CHAR, escrow) @staticmethod def _auction_next_id(slots): """返回一个未被占用、且落在高位段的挂单 ID。 服务端的挂单 ID 是进程内计数器（重启后从 1 重新开始，碰撞时重试 8 次）， 因此 GM 工具造的挂单若占用小 ID，会让重启后的服务端连续撞号。这里直接 使用一个极大的固定段，既不会与进程内计数器冲突，也能通过单调递增避免 已产生过投递/退款的 ID 被复用。""" best = DNFDatabase.AUCTION_GM_ID_BASE - 1 for key in slots: if not key.startswith(DNFDatabase.AUCTION_LISTING_PREFIX): continue try: best = max(best, int(key[len(DNFDatabase.AUCTION_LISTING_PREFIX):])) except (TypeError, ValueError): continue return best + 1 def auction_list_listings(self): """返回当前所有挂单（含已过期的），解析成前端友好的结构。""" escrow = self._auction_escrow() slots = escrow.get("slots", {}) out = [] for key, stack in slots.items(): if not key.startswith(self.AUCTION_LISTING_PREFIX): continue if not isinstance(stack, dict): continue extra = stack.get("extra", {}) or {} listing_id = key[len(self.AUCTION_LISTING_PREFIX):] try: unit_price = int(extra.get("unit_price", "0") or 0) start_price = int(extra.get("start_price", "-1") or -1) buyout_price = int(extra.get("buyout_price", "-1") or -1) deposit = int(extra.get("deposit", "0") or 0) fee_percent = int(extra.get("fee_percent", "0") or 0) bid_total = int(extra.get("bid_total", "0") or 0) created_at = int(extra.get("created_at", "0") or 0) expires_at = int(extra.get("expires_at", "0") or 0) source_slot = extra.get("source_slot", "") list_type = extra.get("list_type", "0") except (ValueError, TypeError): continue count = int(stack.get("count", 1) or 1) now = int(time.time()) out.append({ "listing_id": listing_id, "item_id": stack.get("item_id", 0), "count": count, "seller_id": extra.get("seller_id", ""), "seller_name": extra.get("seller_name", ""), "list_type": list_type, "source_slot": source_slot, "unit_price": unit_price, "start_price": start_price, "buyout_price": buyout_price, "deposit": deposit, "fee_percent": fee_percent, "bid_total": bid_total, "bidder_id": extra.get("bidder_id", ""), "created_at": created_at, "expires_at": expires_at, "expired": expires_at > 0 and expires_at < now, }) # 新→旧排序（GM 高位段 id 天然更大，排前面） out.sort(key=lambda x: x["listing_id"], reverse=True) return out def auction_escrow_gold(self): """返回托管角色的金币（用于核对）。""" ch = self.get_character(self.AUCTION_ESCROW_CHAR) if not ch: return 0 return int((ch.get("stats") or {}).get("gold", 0) or 0) def auction_list(self, seller_id, item_id, count, unit_price, start_price=None, buyout_price=None, duration_hours=24): """上架一件物品。 行为对齐服务端 registerWithin： - 从卖家背包移除物品，写入 escrow 挂单槽 - 押金从卖家金币扣除，转入 escrow 角色金币 - 挂单 Extra 记录全部价格/时间/卖家字段 :param start_price: 每件起拍价（-1 表示不设竞拍，一口价模式） :param buyout_price: 每件一口价（-1 表示不设一口价） """ seller_id = str(seller_id) item_id = int(item_id) count = int(count) unit_price = int(unit_price) if unit_price <= 0: raise ValueError("单价必须大于 0") if count <= 0: raise ValueError("数量必须大于 0") # 一口价：未显式给定时默认 = 单价；起拍价：未给定时默认 -1（无竞拍） if buyout_price is None: buyout_price = unit_price else: buyout_price = int(buyout_price) if start_price is None: start_price = -1 else: start_price = int(start_price) # 校验：至少一口价或起拍价其一有效 if buyout_price <= 0 and start_price <= 0: raise ValueError("一口价和起拍价不能同时为空") if start_price > 0 and start_price > unit_price: raise ValueError("起拍价不能高于单价") # 卖家角色 seller = self.get_character(seller_id) if seller is None: raise KeyError("卖家角色 %s 不存在" % seller_id) seller_name = seller.get("name", "") seller_account = seller.get("account_id", "") # 卖家背包 inv = self.get_record_json("inventory", seller_id) or {} slots = inv.setdefault("slots", {}) # 找到包含该物品的背包槽（主背包 0:*） source_key = None source_stack = None for k, v in slots.items(): if not isinstance(v, dict): continue if str(v.get("item_id")) == str(item_id) and int(v.get("count", 0) or 0) >= count: source_key = k source_stack = v break if source_key is None: raise ValueError("卖家背包中该物品数量不足") # 托管角色与 escrow escrow = self._auction_escrow() escrow_slots = escrow.setdefault("slots", {}) listing_id = self._auction_next_id(escrow_slots) listing_key = self.AUCTION_LISTING_PREFIX + str(listing_id) now = int(time.time()) expires = now + int(duration_hours) * 3600 # 挂单 ItemStack：复制原物品（含 extra），再写入挂单字段 listed = dict(source_stack) listed["count"] = count extra = dict(source_stack.get("extra", {}) or {}) extra["seller_id"] = seller_id extra["seller_name"] = seller_name or "" extra["list_type"] = "0" extra["source_slot"] = source_key.split(":", 1)[-1] if ":" in source_key else source_key extra["unit_price"] = str(unit_price) extra["start_price"] = str(start_price) extra["buyout_price"] = str(buyout_price) extra["deposit"] = str(self.AUCTION_DEPOSIT) extra["fee_percent"] = str(self.AUCTION_SALE_FEE_PERCENT) extra["created_at"] = str(now) extra["expires_at"] = str(expires) listed["extra"] = extra # 卖家背包：数量扣减或整槽删除 remaining = int(source_stack.get("count", 0) or 0) - count if remaining <= 0: del slots[source_key] else: source_stack["count"] = remaining slots[source_key] = source_stack inv["updated_at"] = self._now_iso() # 押金：卖家 -> escrow stats = seller.setdefault("stats", {}) gold = int(stats.get("gold", 0) or 0) if gold < self.AUCTION_DEPOSIT: raise ValueError("卖家金币不足，无法缴纳上架押金 %d" % self.AUCTION_DEPOSIT) stats["gold"] = gold - self.AUCTION_DEPOSIT seller["updated_at"] = self._now_iso() escrow_char = self.get_character(self.AUCTION_ESCROW_CHAR) if escrow_char is None: escrow_char = { "character_id": self.AUCTION_ESCROW_CHAR, "account_id": "system:auction", "name": "拍卖行托管", "job": "0", "level": 0, "stats": {"gold": 0}, "created_at": self._now_iso(), "updated_at": self._now_iso(), } escrow_stats = escrow_char.setdefault("stats", {}) escrow_stats["gold"] = int(escrow_stats.get("gold", 0) or 0) + self.AUCTION_DEPOSIT escrow_char["updated_at"] = self._now_iso() # 写入挂单槽 escrow_slots[listing_key] = listed escrow["updated_at"] = self._now_iso() # 持久化（顺序：背包 -> 角色 -> escrow 角色 -> escrow 背包） self.put_record_json("inventory", seller_id, inv) self.put_record_json("character", seller_id, seller) self.put_record_json("character", self.AUCTION_ESCROW_CHAR, escrow_char) self._auction_save_escrow(escrow) return { "listing_id": listing_id, "item_id": item_id, "count": count, "unit_price": unit_price, "start_price": start_price, "buyout_price": buyout_price, "deposit": self.AUCTION_DEPOSIT, "expires_at": expires, "seller_gold_after": stats["gold"], } def auction_cancel(self, listing_id): """下架（取消）一件挂单。 行为对齐服务端 Cancel： - 从 escrow 移除挂单槽 - 物品退回卖家背包（游戏内经邮件，这里直接写入背包，GM 工具便于核对） - 押金不退（取消上架押金没收，服务端注释明确此为 gold sink） - 若有竞价，退款给竞价者（这里按相同 refund 槽逻辑写入） """ listing_id = str(listing_id) escrow = self._auction_escrow() slots = escrow.setdefault("slots", {}) listing_key = self.AUCTION_LISTING_PREFIX + listing_id stack = slots.get(listing_key) if stack is None: raise KeyError("挂单 %s 不存在" % listing_id) extra = stack.get("extra", {}) or {} seller_id = extra.get("seller_id", "") item_id = stack.get("item_id", 0) count = int(stack.get("count", 1) or 1) # 移除挂单 del slots[listing_key] # 物品退回卖家背包（找空闲槽位） if seller_id: inv = self.get_record_json("inventory", seller_id) or {} inv_slots = inv.setdefault("slots", {}) returned = dict(stack) returned.pop("extra", None) # 去掉挂单字段，还原为普通物品 # 尝试放回原槽位，否则找空闲槽 target_key = None source_slot = extra.get("source_slot", "") preferred = "0:" + source_slot if source_slot else None if preferred and preferred not in inv_slots: target_key = preferred else: used = set() for k in inv_slots: if ":" in k: c, _, s = k.partition(":") if c == "0" and s.isdigit(): used.add(int(s)) slot = 1 while slot in used: slot += 1 target_key = "0:%d" % slot inv_slots[target_key] = returned inv["updated_at"] = self._now_iso() self.put_record_json("inventory", seller_id, inv) # 竞价退款：bid_total > 0 时退回竞价者（写 refund 槽，与 releaseListingBid 一致） bid_total = int(extra.get("bid_total", "0") or 0) bidder_id = extra.get("bidder_id", "") refund = None if bid_total > 0 and bidder_id: refund_key = self.AUCTION_REFUND_PREFIX + listing_id + ":" + bidder_id slots[refund_key] = { "item_id": item_id, "count": bid_total, "extra": { "seller_id": bidder_id, "listing_id": listing_id, "sold_count": "0", "refund": "1", }, } # 退款金币写回竞价者角色 bidder = self.get_character(bidder_id) if bidder: bstats = bidder.setdefault("stats", {}) bstats["gold"] = int(bstats.get("gold", 0) or 0) + bid_total bidder["updated_at"] = self._now_iso() self.put_record_json("character", bidder_id, bidder) refund = {"bidder_id": bidder_id, "gold": bid_total} escrow["updated_at"] = self._now_iso() self._auction_save_escrow(escrow) return { "listing_id": listing_id, "item_id": item_id, "count": count, "seller_id": seller_id, "returned_to": seller_id, "deposit_forfeited": self.AUCTION_DEPOSIT, "refund": refund, }