Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 10 additions & 3 deletions tdx2db/reader.py
Original file line number Diff line number Diff line change
Expand Up @@ -62,11 +62,18 @@ def _load_real_names(self) -> dict:
dict: 6位纯数字 code -> 股票名
"""
f = self.tdx_path / 'T0002' / 'hq_cache' / 'infoharbor_ex.code'
if not f.exists():
logger.warning(f"缺少 {f},stock_info.name 回退为占位符(深A/上A + code)")
try:
if not f.exists():
logger.warning(f"缺少 {f},stock_info.name 回退为占位符(深A/上A + code)")
return {}
text = f.read_text(encoding='gbk', errors='replace')
except OSError as e:
# 文件存在但不可读(SMB 挂载下被运行中的通达信锁定等场景,
# 表现为 EPERM)——名称是可选增强,不应中断 daily/minutes 同步
logger.warning(f"读取 {f} 失败({e}),stock_info.name 回退为占位符")
return {}
names = {}
for line in f.read_text(encoding='gbk', errors='replace').splitlines():
for line in text.splitlines():
p = line.strip().split('|')
if len(p) >= 2 and p[0] and p[1]:
names[p[0]] = p[1]
Expand Down
23 changes: 23 additions & 0 deletions tests/test_reader.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@
import shutil
from pathlib import Path

import os

import pytest

from tdx2db.reader import TdxDataReader
Expand Down Expand Up @@ -155,3 +157,24 @@ def test_fivemin_records(self, tdx_reader):
assert first['volume'] == 200000 # lc5 无系数换算
assert first['code'] == '000001'
assert first['market'] == 0


class TestStockListNamesUnreadable:
@pytest.mark.skipif(os.name == 'nt', reason='chmod 权限位在 Windows 无效')
def test_fallback_placeholder_when_file_unreadable(self, tmp_path):
"""名称文件存在但不可读(SMB 锁定 EPERM 等)回退占位符,不中断同步"""
import shutil as _shutil
d = tmp_path / 'vipdoc' / 'sz' / 'lday'
d.mkdir(parents=True)
_shutil.copy(FIXTURES / 'sz000001.day', d / 'sz000001.day')
hq = tmp_path / 'T0002' / 'hq_cache'
hq.mkdir(parents=True)
f = hq / 'infoharbor_ex.code'
f.write_bytes('000001|平安银行|x\n'.encode('gbk'))
f.chmod(0o000)
try:
df = TdxDataReader(tdx_path=str(tmp_path)).get_stock_list()
finally:
f.chmod(0o644)
names = dict(zip(df.code, df.name))
assert names['sz000001'] == '深Asz000001'
Loading