fix(misc): CSV column name mismatches in data loading (#32)

* fix: CSV column name mismatches in data loading

- sect.py: Fix headquarter_name/headquarter_desc -> name/desc when reading sect_region.csv
- sect.py: Move sid initialization before technique lookup to fix unbound variable bug
- technique.py: Change sect (name) to sect_id (int) to match technique.csv column
- elixir.py: Remove redundant get_int(row, "id") that reads non-existent column

These fixes ensure:
1. Sect headquarters display correct location names (e.g., "大千光极城" instead of "不夜城")
2. Sect techniques are correctly associated and displayed
3. Technique sect restrictions work properly

* fix: update main.py to use sect_id, add CSV loading tests

- main.py: Change technique.sect to technique.sect_id in API response
- Add tests/test_csv_loading.py to verify CSV column names match code

* test: add API test for /api/meta/game_data endpoint

Verify that techniques in API response use sect_id field (not sect)

* fix: add None check for hq_region in AvatarFactory

Remove redundant code that adds sect headquarters to known_regions.
This logic is already handled by Avatar._init_known_regions() which
uses sect_id matching (more reliable than name-based lookup).

The removed code was causing a flaky test (~1% failure rate) because
resolve_query returns None in test environments with simplified maps.
This commit is contained in:
Zihao Xu
2026-01-17 23:31:15 -08:00
committed by GitHub
parent 0acf72a313
commit 7edae9188b
6 changed files with 178 additions and 28 deletions

View File

@@ -147,7 +147,6 @@ def _load_elixirs() -> tuple[Dict[int, Elixir], Dict[str, List[Elixir]]]:
df = game_configs["elixir"] df = game_configs["elixir"]
for row in df: for row in df:
elixir_id = get_int(row, "id")
name = get_str(row, "name") name = get_str(row, "name")
desc = get_str(row, "desc") desc = get_str(row, "desc")
price = get_int(row, "price") price = get_int(row, "price")

View File

@@ -180,8 +180,8 @@ def _load_sects() -> tuple[dict[int, Sect], dict[str, Sect]]:
sid = get_int(sr, "sect_id", -1) sid = get_int(sr, "sect_id", -1)
if sid == -1: if sid == -1:
continue continue
hq_name = get_str(sr, "headquarter_name") hq_name = get_str(sr, "name")
hq_desc = get_str(sr, "headquarter_desc") hq_desc = get_str(sr, "desc")
hq_by_sect_id[sid] = (hq_name, hq_desc) hq_by_sect_id[sid] = (hq_name, hq_desc)
# 可能不存在 technique 配表或未添加 sect 列,做容错 # 可能不存在 technique 配表或未添加 sect 列,做容错
@@ -192,11 +192,13 @@ def _load_sects() -> tuple[dict[int, Sect], dict[str, Sect]]:
name = get_str(row, "name") name = get_str(row, "name")
image_path = assets_base / f"{name}.png" image_path = assets_base / f"{name}.png"
# 先读取当前宗门 ID供后续使用
sid = get_int(row, "id")
# 收集该宗门下配置的功法名称 # 收集该宗门下配置的功法名称
technique_names: list[str] = [] technique_names: list[str] = []
# 检查 tech_df 是否存在以及是否有数据 # 检查 tech_df 是否存在以及是否有数据
if tech_df: if tech_df:
# 检查是否存在 sect 字段 (检查第一行或当前行)
technique_names = [ technique_names = [
get_str(t, "name") get_str(t, "name")
for t in tech_df for t in tech_df
@@ -229,7 +231,6 @@ def _load_sects() -> tuple[dict[int, Sect], dict[str, Sect]]:
} }
# 从 sect_region.csv 中优先取驻地名称/描述;否则兼容旧列或退回宗门名 # 从 sect_region.csv 中优先取驻地名称/描述;否则兼容旧列或退回宗门名
sid = get_int(row, "id")
csv_hq = hq_by_sect_id.get(sid) csv_hq = hq_by_sect_id.get(sid)
hq_name_from_csv = (csv_hq[0] if csv_hq else "").strip() if csv_hq else "" hq_name_from_csv = (csv_hq[0] if csv_hq else "").strip() if csv_hq else ""
hq_desc_from_csv = (csv_hq[1] if csv_hq else "").strip() if csv_hq else "" hq_desc_from_csv = (csv_hq[1] if csv_hq else "").strip() if csv_hq else ""

View File

@@ -62,8 +62,8 @@ class Technique:
desc: str desc: str
weight: float weight: float
condition: str condition: str
# 归属宗门名称None/空表示无宗门要求(散修可修) # 归属宗门IDNone 表示无宗门要求(散修可修)
sect: Optional[str] = None sect_id: Optional[int] = None
# 影响角色或系统的效果 # 影响角色或系统的效果
effects: dict[str, object] = field(default_factory=dict) effects: dict[str, object] = field(default_factory=dict)
effect_desc: str = "" effect_desc: str = ""
@@ -125,9 +125,8 @@ def loads() -> tuple[dict[int, Technique], dict[str, Technique]]:
condition = get_str(row, "condition") condition = get_str(row, "condition")
weight = get_float(row, "weight", 1.0) weight = get_float(row, "weight", 1.0)
sect = get_str(row, "sect") sect_id_raw = get_int(row, "sect_id", -1)
if not sect: sect_id = sect_id_raw if sect_id_raw > 0 else None
sect = None
effects = load_effect_from_str(get_str(row, "effects")) effects = load_effect_from_str(get_str(row, "effects"))
from src.classes.effect import format_effects_to_text from src.classes.effect import format_effects_to_text
@@ -141,7 +140,7 @@ def loads() -> tuple[dict[int, Technique], dict[str, Technique]]:
desc=get_str(row, "desc"), desc=get_str(row, "desc"),
weight=weight, weight=weight,
condition=condition, condition=condition,
sect=sect, sect_id=sect_id,
effects=effects, effects=effects,
effect_desc=effect_desc, effect_desc=effect_desc,
) )
@@ -232,24 +231,22 @@ def get_random_upper_technique_for_avatar(avatar) -> Technique | None:
def get_technique_by_sect(sect) -> Technique: def get_technique_by_sect(sect) -> Technique:
""" """
简化版:仅按宗门筛选并按权重抽样,不考虑灵根与 condition。 简化版:仅按宗门筛选并按权重抽样,不考虑灵根与 condition。
- 散修sect 为 None/空只从无宗门要求sect 为空)的功法中抽样; - 散修sect 为 None只从无宗门要求sect_id 为 None)的功法中抽样;
- 有宗门:从无宗门 + 该宗门的功法中抽样; - 有宗门:从"无宗门 + 该宗门"的功法中抽样;
若集合为空,则退回全量功法。 若集合为空,则退回全量功法。
""" """
import random import random
sect_name: Optional[str] = None target_sect_id: Optional[int] = None
if sect is not None: if sect is not None:
sect_name = getattr(sect, "name", sect) target_sect_id = getattr(sect, "id", None)
if isinstance(sect_name, str):
sect_name = sect_name.strip() or None
allowed_sects: set[Optional[str]] = {None, ""} allowed_sect_ids: set[Optional[int]] = {None}
if sect_name is not None: if target_sect_id is not None:
allowed_sects.add(sect_name) allowed_sect_ids.add(target_sect_id)
def _in_allowed_sect(t: Technique) -> bool: def _in_allowed_sect(t: Technique) -> bool:
return (t.sect in allowed_sects) or (t.sect is None) or (t.sect == "") return t.sect_id in allowed_sect_ids
candidates: List[Technique] = [t for t in techniques_by_id.values() if _in_allowed_sect(t)] candidates: List[Technique] = [t for t in techniques_by_id.values() if _in_allowed_sect(t)]
if not candidates: if not candidates:

View File

@@ -1191,7 +1191,7 @@ def get_game_data():
"name": t.name, "name": t.name,
"grade": t.grade.value, "grade": t.grade.value,
"attribute": t.attribute.value, "attribute": t.attribute.value,
"sect": t.sect "sect_id": t.sect_id
} }
for t in techniques_by_id.values() for t in techniques_by_id.values()
] ]

View File

@@ -459,12 +459,6 @@ class AvatarFactory:
# 自己.relations[师傅] = MASTER (自己认为师傅是师傅) # 自己.relations[师傅] = MASTER (自己认为师傅是师傅)
plan.master_avatar.set_relation(avatar, Relation.APPRENTICE) plan.master_avatar.set_relation(avatar, Relation.APPRENTICE)
# 宗门弟子天生知道宗门总部位置
if avatar.sect is not None:
res = resolve_query(avatar.sect.headquarter.name, world, expected_types=[Region])
hq_region = res.obj
avatar.known_regions.add(hq_region.id)
if avatar.technique is not None: if avatar.technique is not None:
mapped = attribute_to_root(avatar.technique.attribute) mapped = attribute_to_root(avatar.technique.attribute)
if mapped is not None: if mapped is not None:

159
tests/test_csv_loading.py Normal file
View File

@@ -0,0 +1,159 @@
"""
测试 CSV 数据加载的正确性。
验证代码中使用的列名与 CSV 文件中的实际列名匹配。
"""
import pytest
from src.classes.sect import sects_by_id, sects_by_name, Sect
from src.classes.technique import techniques_by_id, techniques_by_name, Technique
class TestSectLoading:
"""测试宗门数据加载"""
def test_sect_headquarter_name_loaded(self):
"""测试宗门驻地名称正确加载(来自 sect_region.csv 的 name 列)"""
# 不夜城 (sect_id=12) 的驻地应该是 "大千光极城"
sect = sects_by_id.get(12)
assert sect is not None, "宗门 ID=12 应该存在"
assert sect.name == "不夜城", "宗门名称应该是 '不夜城'"
assert sect.headquarter.name == "大千光极城", (
f"驻地名称应该是 '大千光极城',而不是 '{sect.headquarter.name}'"
)
def test_sect_headquarter_desc_loaded(self):
"""测试宗门驻地描述正确加载(来自 sect_region.csv 的 desc 列)"""
sect = sects_by_id.get(12)
assert sect is not None
# 验证描述不为空且包含关键词
assert sect.headquarter.desc, "驻地描述不应为空"
assert "极光" in sect.headquarter.desc, "驻地描述应该包含 '极光'"
def test_all_sects_have_headquarters(self):
"""测试所有宗门都有驻地信息"""
for sect_id, sect in sects_by_id.items():
assert sect.headquarter is not None, f"宗门 {sect.name} (ID={sect_id}) 应该有驻地"
assert sect.headquarter.name, f"宗门 {sect.name} 的驻地名称不应为空"
def test_sect_techniques_loaded(self):
"""测试宗门功法列表正确加载"""
# 明心剑宗 (sect_id=1) 应该有功法
sect = sects_by_id.get(1)
assert sect is not None, "宗门 ID=1 应该存在"
assert len(sect.technique_names) > 0, (
f"宗门 '{sect.name}' 应该有独门功法,但 technique_names 为空"
)
def test_sect_without_techniques(self):
"""测试没有配置功法的宗门(不夜城 sect_id=12"""
sect = sects_by_id.get(12)
assert sect is not None
# 不夜城在 technique.csv 中没有配置功法,所以应该是空列表
assert sect.technique_names == [], (
f"宗门 '{sect.name}' 不应该有独门功法"
)
class TestTechniqueLoading:
"""测试功法数据加载"""
def test_technique_sect_id_loaded(self):
"""测试功法的 sect_id 正确加载(来自 technique.csv 的 sect_id 列)"""
# 草字剑诀 (id=30) 属于明心剑宗 (sect_id=1)
technique = techniques_by_id.get(30)
assert technique is not None, "功法 ID=30 应该存在"
assert technique.name == "草字剑诀", f"功法名称应该是 '草字剑诀',而不是 '{technique.name}'"
assert technique.sect_id == 1, (
f"功法 '草字剑诀' 的 sect_id 应该是 1而不是 {technique.sect_id}"
)
def test_technique_without_sect(self):
"""测试散修功法(没有宗门限制)的 sect_id 为 None"""
# 金刚不坏体 (id=1) 是散修功法
technique = techniques_by_id.get(1)
assert technique is not None, "功法 ID=1 应该存在"
assert technique.sect_id is None, (
f"散修功法 '{technique.name}' 的 sect_id 应该是 None而不是 {technique.sect_id}"
)
def test_sect_techniques_match(self):
"""测试宗门功法和功法的宗门ID相互匹配"""
for sect_id, sect in sects_by_id.items():
for tech_name in sect.technique_names:
technique = techniques_by_name.get(tech_name)
assert technique is not None, f"功法 '{tech_name}' 应该存在"
assert technique.sect_id == sect_id, (
f"功法 '{tech_name}' 的 sect_id ({technique.sect_id}) "
f"应该匹配宗门 '{sect.name}' 的 ID ({sect_id})"
)
class TestElixirLoading:
"""测试丹药数据加载"""
def test_elixir_loaded_with_item_id(self):
"""测试丹药使用 item_id 列正确加载"""
from src.classes.elixir import elixirs_by_id
# 验证丹药已加载且 ID 不为 0如果用错误的列名会得到默认值 0
assert len(elixirs_by_id) > 0, "应该加载到丹药数据"
for elixir_id, elixir in elixirs_by_id.items():
assert elixir_id > 0, f"丹药 '{elixir.name}' 的 ID 应该大于 0"
assert elixir.id == elixir_id, f"丹药 ID 不匹配: {elixir.id} != {elixir_id}"
class TestGameDataAPI:
"""测试 /api/meta/game_data API 返回正确的数据结构"""
@pytest.fixture
def client(self):
"""创建测试客户端"""
from fastapi.testclient import TestClient
from src.server.main import app
return TestClient(app)
def test_game_data_techniques_have_sect_id(self, client):
"""测试 /api/meta/game_data 返回的功法包含 sect_id 字段(而非 sect"""
response = client.get("/api/meta/game_data")
assert response.status_code == 200
data = response.json()
assert "techniques" in data, "响应应该包含 techniques 字段"
techniques = data["techniques"]
assert len(techniques) > 0, "应该有功法数据"
for tech in techniques:
# 确保使用 sect_id 而非 sect
assert "sect_id" in tech, (
f"功法 '{tech.get('name', 'unknown')}' 应该有 sect_id 字段"
)
assert "sect" not in tech, (
f"功法 '{tech.get('name', 'unknown')}' 不应该有 sect 字段(应使用 sect_id"
)
# 验证 sect_id 的值类型正确
sect_id = tech["sect_id"]
assert sect_id is None or isinstance(sect_id, int), (
f"功法 '{tech.get('name')}' 的 sect_id 应该是 None 或 int而不是 {type(sect_id)}"
)
def test_game_data_sects_structure(self, client):
"""测试 /api/meta/game_data 返回的宗门数据结构正确"""
response = client.get("/api/meta/game_data")
assert response.status_code == 200
data = response.json()
assert "sects" in data, "响应应该包含 sects 字段"
sects = data["sects"]
assert len(sects) > 0, "应该有宗门数据"
for sect in sects:
assert "id" in sect, "宗门应该有 id 字段"
assert "name" in sect, "宗门应该有 name 字段"
assert sect["id"] > 0, f"宗门 '{sect.get('name')}' 的 ID 应该大于 0"
if __name__ == "__main__":
pytest.main([__file__, "-v"])