From 62bbdd11915821e9b260fa48bf3b0ce431f7ea24 Mon Sep 17 00:00:00 2001 From: xbfighting Date: Tue, 7 Jul 2026 22:50:27 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20infoharbor=5Fex.code=20=E5=AD=98?= =?UTF-8?q?=E5=9C=A8=E4=BD=86=E4=B8=8D=E5=8F=AF=E8=AF=BB=E6=97=B6=E5=9B=9E?= =?UTF-8?q?=E9=80=80=E5=8D=A0=E4=BD=8D=E7=AC=A6=EF=BC=8C=E4=B8=8D=E4=B8=AD?= =?UTF-8?q?=E6=96=AD=E5=90=8C=E6=AD=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SMB 挂载下运行中的通达信会锁定 hq_cache 文件(EPERM), _load_real_names 只兜了文件缺失,read_text 抛 OSError 直接 中断 daily/minutes 同步。现与缺失场景同样降级处理。 Co-Authored-By: Claude Fable 5 --- tdx2db/reader.py | 13 ++++++++++--- tests/test_reader.py | 23 +++++++++++++++++++++++ 2 files changed, 33 insertions(+), 3 deletions(-) diff --git a/tdx2db/reader.py b/tdx2db/reader.py index ecfdc1e..56e597f 100644 --- a/tdx2db/reader.py +++ b/tdx2db/reader.py @@ -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] diff --git a/tests/test_reader.py b/tests/test_reader.py index bfa76ae..9f55448 100644 --- a/tests/test_reader.py +++ b/tests/test_reader.py @@ -9,6 +9,8 @@ import shutil from pathlib import Path +import os + import pytest from tdx2db.reader import TdxDataReader @@ -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'