feat(data): corpus/dmsk 消费接口(spec §20.11.3 实施,设计已入档)——partition_reader 通用读取核(核内零语义:不做 as-of/PIT+零静默去重;失败分级:缺数据空 df 不 raise/跨分区列集与类型漂移 PartitionSchemaError fail loud;双机根解析 SANGUO_CORPUS_ROOT/SANGUO_FUND_DMSK_ROOT env 覆盖+水位线读取)+corpus_reader(get_news_meta/get_ann_meta symbols 下推 stock_code/sec_code/get_news_fulltext(ids) art_code 批量/ann_pdf_dir;schema 契约 09-17 NAS 实测落 docstring:稳定 id=art_code/announcement_id 与 PDF 同名)+fund_dmsk_reader(get_domain 五域+noticed_by 纯过滤 NOTICE_DATE 缺列 fail loud;契约:五域 NOTICE_DATE 实测全在/SECUCODE 带后缀 SECURITY_CODE 裸 6 位/单位元);TDD 23 例,全套 414 绿 [vps] [no-doc]
This commit is contained in:
@@ -0,0 +1,53 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""corpus 语料薄封装（spec §20.11.3）。列名随源直存零转换（§15.2 双层读法）。
|
||||
|
||||
schema 契约（2026-09-17 NAS 实测落档，§20.11.3 数据契约③）：
|
||||
- news_meta: art_code(稳定 id)/stock_code(裸 6 位)/show_time/title/summary/
|
||||
media_name/url/first_seen_date
|
||||
- news_fulltext: art_code(=news_meta join 键)/url/show_time/content_text/fetch_date
|
||||
- ann_meta: announcement_id(稳定 id，与 ann_pdf/<year>/<id>.pdf 同名)/
|
||||
sec_code(裸 6 位)/sec_name/title/short_title/ann_time/ann_type(|| 分隔多码)/
|
||||
ann_type_name/column_id/adjunct_url/adjunct_size/first_seen_date
|
||||
- ann_pdf: <corpus_root>/ann_pdf/<year>/<announcement_id>.pdf
|
||||
|
||||
稳定 id（数据契约②配方）：art_code/announcement_id 为源站稳定 id；爬虫层
|
||||
跨天重复行会在数据里原样出现——核内零去重，消费方按 id 自行幂等。
|
||||
"""
|
||||
import os
|
||||
|
||||
import sanguo_data.partition_reader as pr
|
||||
|
||||
|
||||
def get_news_meta(start=None, end=None, symbols=None):
|
||||
"""新闻元数据（dt 分区直读）。symbols=裸 6 位代码列表，下推过滤 stock_code。"""
|
||||
df = pr.read_partitions(pr.corpus_root(), "news_meta", start, end)
|
||||
if symbols and not df.empty:
|
||||
df = df[df["stock_code"].isin(set(symbols))]
|
||||
return df
|
||||
|
||||
|
||||
def get_ann_meta(start=None, end=None, symbols=None):
|
||||
"""公告元数据。symbols 下推过滤 sec_code；ann_type/标题已透传（因子线
|
||||
选择性抽取 PDF 用，不盲抽）。"""
|
||||
df = pr.read_partitions(pr.corpus_root(), "ann_meta", start, end)
|
||||
if symbols and not df.empty:
|
||||
df = df[df["sec_code"].isin(set(symbols))]
|
||||
return df
|
||||
|
||||
|
||||
def get_news_fulltext(ids):
|
||||
"""按 art_code 批量取全文（因子回访高信号条目/LLM 分块天然按 id）。"""
|
||||
df = pr.read_partitions(pr.corpus_root(), "news_fulltext")
|
||||
if ids and not df.empty:
|
||||
df = df[df["art_code"].isin(set(ids))]
|
||||
return df
|
||||
|
||||
|
||||
def ann_pdf_dir(year):
|
||||
"""公告 PDF 年目录（文件名=announcement_id.pdf，与 ann_meta 同名 join）。"""
|
||||
return os.path.join(pr.corpus_root(), "ann_pdf", str(year))
|
||||
|
||||
|
||||
def get_watermark():
|
||||
"""镜像水位线（None=尚无成功推送记录）。"""
|
||||
return pr.read_watermark(pr.corpus_root())
|
||||
@@ -0,0 +1,39 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""dmsk 财务五域薄封装（spec §20.11.3）。
|
||||
|
||||
schema 契约（2026-09-17 NAS 实测落档，§20.11.3 数据契约）：
|
||||
- 五域每行 NOTICE_DATE 必在（实测全命中；缺失=写路径事故，消费方按整期
|
||||
弃用=宁缺毋假红线）
|
||||
- 代码双列：SECUCODE 带后缀（301569.SZ）/ SECURITY_CODE 裸 6 位
|
||||
- 金额单位=元（EM datacenter 原生，非万元）
|
||||
- REPORT_DATE=分区键同值（dt=<REPORT_DATE>）
|
||||
|
||||
noticed_by=纯 filter 非 PIT merge（文档明示）：只做 NOTICE_DATE<=noticed_by
|
||||
的行级过滤（日期前缀比较）；「static 基线为主+dmsk 新鲜度补丁」的 PIT 合并
|
||||
归因子侧 fundamental_pit（§19.3 下游读法），不在数据线。
|
||||
"""
|
||||
import sanguo_data.partition_reader as pr
|
||||
|
||||
DOMAINS = ("balance", "cashflow", "express", "forecast", "income")
|
||||
|
||||
|
||||
def get_domain(domain, start=None, end=None, noticed_by=None):
|
||||
"""读五域之一（dt=REPORT_DATE 分区直读，零转换零去重）。
|
||||
|
||||
start/end 过滤的是 REPORT_DATE 分区值；noticed_by 过滤 NOTICE_DATE
|
||||
（接受 'YYYY-MM-DD' 或全时间戳，按日期前缀比较）。
|
||||
"""
|
||||
if domain not in DOMAINS:
|
||||
raise ValueError("domain 必须是 %s 之一, 收到 %r" % (DOMAINS, domain))
|
||||
df = pr.read_partitions(pr.fund_dmsk_root(), domain, start, end)
|
||||
if noticed_by is not None and not df.empty:
|
||||
if "NOTICE_DATE" not in df.columns:
|
||||
raise KeyError(
|
||||
"域 %s 缺 NOTICE_DATE 列(数据契约①违约, 整期弃用勿消费)" % domain)
|
||||
df = df[df["NOTICE_DATE"].astype(str).str[:10] <= noticed_by[:10]]
|
||||
return df
|
||||
|
||||
|
||||
def get_watermark():
|
||||
"""镜像水位线（None=尚无成功推送记录）。"""
|
||||
return pr.read_watermark(pr.fund_dmsk_root())
|
||||
@@ -0,0 +1,105 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""通用 dt= 分区 parquet 读取核（spec §20.11.3，corpus/dmsk 消费接口基座）。
|
||||
|
||||
核内零语义（因子线 09-17 契约）：
|
||||
- 不做 as-of/PIT 合并——因子侧 fundamental_pit 是唯一正确性模型，两处实现会漂
|
||||
且 PIT 错是静默错；dmsk 按 REPORT_DATE 分区，「as of D」本就无法靠分区剪枝。
|
||||
- 零静默去重——幂等去重归写路径；镜像重放/分区重写出重复行原样透出，
|
||||
取哪行=消费方语义决策。
|
||||
|
||||
失败分级（策略线 09-17 契约）：
|
||||
- 缺数据（根/域/分区不存在）→ 空 DataFrame，绝不 raise
|
||||
（「数据失败≠交易信号」，消费方沿用上次快照）；
|
||||
- 数据在但形状错（跨分区列集/类型漂移）→ PartitionSchemaError fail loud
|
||||
（静默 coerce 会埋前视/单位错，研究侧静默错最贵）。
|
||||
"""
|
||||
import glob
|
||||
import json
|
||||
import os
|
||||
|
||||
import pandas as pd
|
||||
|
||||
_IS_WIN = os.name == "nt"
|
||||
DEFAULT_CORPUS_ROOT_WIN = r"C:\sanguo_vnpy_v2\data\corpus"
|
||||
DEFAULT_CORPUS_ROOT_POSIX = "/volume1/stock/corpus"
|
||||
DEFAULT_DMSK_ROOT_WIN = r"C:\sanguo_vnpy_v2\data\fundamentals\dmsk"
|
||||
DEFAULT_DMSK_ROOT_POSIX = "/volume1/stock/fundamentals/dmsk"
|
||||
WATERMARK_NAME = "_mirror_watermark.json"
|
||||
_DT_PREFIX = "dt="
|
||||
|
||||
|
||||
class PartitionSchemaError(RuntimeError):
|
||||
"""跨分区 schema 漂移（列集/类型不一致）。"""
|
||||
|
||||
|
||||
def corpus_root():
|
||||
"""corpus 族根（§20.11.3：双机各读各的，env 覆盖同 SANGUO_DATA_ROOT 模式）。"""
|
||||
return os.environ.get("SANGUO_CORPUS_ROOT") or (
|
||||
DEFAULT_CORPUS_ROOT_WIN if _IS_WIN else DEFAULT_CORPUS_ROOT_POSIX)
|
||||
|
||||
|
||||
def fund_dmsk_root():
|
||||
"""dmsk 族根。"""
|
||||
return os.environ.get("SANGUO_FUND_DMSK_ROOT") or (
|
||||
DEFAULT_DMSK_ROOT_WIN if _IS_WIN else DEFAULT_DMSK_ROOT_POSIX)
|
||||
|
||||
|
||||
def _partitions(domain_root):
|
||||
if not os.path.isdir(domain_root):
|
||||
return []
|
||||
out = []
|
||||
for name in sorted(os.listdir(domain_root)):
|
||||
p = os.path.join(domain_root, name)
|
||||
if os.path.isdir(p) and name.startswith(_DT_PREFIX):
|
||||
out.append((name[len(_DT_PREFIX):], p))
|
||||
return out
|
||||
|
||||
|
||||
def list_dates(root, domain):
|
||||
"""可用分区日期列表（dt= 前缀剥掉，升序）。缺根/缺域=空列表。"""
|
||||
return [d for d, _ in _partitions(os.path.join(root, domain))]
|
||||
|
||||
|
||||
def _schema_sig(df):
|
||||
return tuple((str(c), str(df[c].dtype)) for c in df.columns)
|
||||
|
||||
|
||||
def read_partitions(root, domain, start=None, end=None):
|
||||
"""读域的 dt= 分区并拼接（含分区内全部 part-*.parquet）。
|
||||
|
||||
start/end 为 ISO 日期字符串，含端点、字符串比较。零去重零转换；
|
||||
跨文件 schema 漂移 raise PartitionSchemaError；无数据返空 DataFrame。
|
||||
"""
|
||||
parts = _partitions(os.path.join(root, domain))
|
||||
if start is not None:
|
||||
parts = [(d, p) for d, p in parts if d >= start]
|
||||
if end is not None:
|
||||
parts = [(d, p) for d, p in parts if d <= end]
|
||||
frames = []
|
||||
sig = None
|
||||
for _, pdir in parts:
|
||||
for pf in sorted(glob.glob(os.path.join(pdir, "*.parquet"))):
|
||||
df = pd.read_parquet(pf)
|
||||
cur_sig = _schema_sig(df)
|
||||
if sig is None:
|
||||
sig = cur_sig
|
||||
elif cur_sig != sig:
|
||||
raise PartitionSchemaError(
|
||||
"schema 漂移: %s 签名 %s != 首文件 %s (域=%s)" % (
|
||||
pf, cur_sig, sig, domain))
|
||||
frames.append(df)
|
||||
if not frames:
|
||||
return pd.DataFrame()
|
||||
return pd.concat(frames, ignore_index=True)
|
||||
|
||||
|
||||
def read_watermark(root):
|
||||
"""镜像水位线（§20.11.2 推送链落在族根的 _mirror_watermark.json）。
|
||||
|
||||
返回 {generated_at, domains:{domain: 最大分区值}} 或 None（无文件）。
|
||||
"""
|
||||
p = os.path.join(root, WATERMARK_NAME)
|
||||
if not os.path.exists(p):
|
||||
return None
|
||||
with open(p) as f:
|
||||
return json.load(f)
|
||||
@@ -0,0 +1,60 @@
|
||||
# corpus_reader 薄封装测试（spec §20.11.3：symbols 下推/ids 批取/容错分级）
|
||||
# 2026-09-17 TDD RED -> GREEN
|
||||
import os
|
||||
|
||||
import pandas as pd
|
||||
import pytest
|
||||
|
||||
import sanguo_data.corpus_reader as cr
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tree(tmp_path, monkeypatch):
|
||||
root = tmp_path / "corpus"
|
||||
monkeypatch.setenv("SANGUO_CORPUS_ROOT", str(root))
|
||||
(root / "news_meta" / "dt=2026-09-17").mkdir(parents=True)
|
||||
(root / "news_fulltext" / "dt=2026-09-17").mkdir(parents=True)
|
||||
(root / "ann_meta" / "dt=2026-09-17").mkdir(parents=True)
|
||||
(root / "ann_pdf" / "2026").mkdir(parents=True)
|
||||
pd.DataFrame({
|
||||
"art_code": ["A1", "A2"], "stock_code": ["600519", "000001"],
|
||||
"title": ["t1", "t2"],
|
||||
}).to_parquet(root / "news_meta" / "dt=2026-09-17" / "part-0.parquet", index=False)
|
||||
pd.DataFrame({
|
||||
"art_code": ["A1", "A2"], "content_text": ["c1", "c2"],
|
||||
}).to_parquet(root / "news_fulltext" / "dt=2026-09-17" / "part-0.parquet", index=False)
|
||||
pd.DataFrame({
|
||||
"announcement_id": ["1225567781"], "sec_code": ["605488"],
|
||||
"title": ["理财公告"], "ann_type": ["01010503||011705"],
|
||||
}).to_parquet(root / "ann_meta" / "dt=2026-09-17" / "part-0.parquet", index=False)
|
||||
return root
|
||||
|
||||
|
||||
class TestNews:
|
||||
def test_get_news_meta_symbols_pushdown(self, tree):
|
||||
df = cr.get_news_meta(symbols=["600519"])
|
||||
assert list(df["art_code"]) == ["A1"]
|
||||
|
||||
def test_get_news_meta_no_symbols_all(self, tree):
|
||||
assert len(cr.get_news_meta()) == 2
|
||||
|
||||
def test_get_news_fulltext_by_ids(self, tree):
|
||||
df = cr.get_news_fulltext(["A2", "A1"])
|
||||
assert set(df["art_code"]) == {"A1", "A2"}
|
||||
|
||||
def test_empty_root_returns_empty_no_raise(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("SANGUO_CORPUS_ROOT", str(tmp_path / "empty"))
|
||||
assert cr.get_news_meta().empty
|
||||
assert cr.get_news_fulltext(["A1"]).empty
|
||||
|
||||
|
||||
class TestAnn:
|
||||
def test_get_ann_meta_symbols(self, tree):
|
||||
df = cr.get_ann_meta(symbols=["605488"])
|
||||
assert list(df["announcement_id"]) == ["1225567781"]
|
||||
|
||||
def test_get_ann_meta_miss_returns_empty(self, tree):
|
||||
assert cr.get_ann_meta(symbols=["999999"]).empty
|
||||
|
||||
def test_ann_pdf_dir(self, tree):
|
||||
assert cr.ann_pdf_dir(2026) == str(tree / "ann_pdf" / "2026")
|
||||
@@ -0,0 +1,60 @@
|
||||
# fund_dmsk_reader 薄封装测试（spec §20.11.3：noticed_by 纯过滤+NOTICE_DATE 契约 fail loud）
|
||||
# 2026-09-17 TDD RED -> GREEN
|
||||
import pandas as pd
|
||||
import pytest
|
||||
|
||||
import sanguo_data.fund_dmsk_reader as dr
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tree(tmp_path, monkeypatch):
|
||||
root = tmp_path / "dmsk"
|
||||
monkeypatch.setenv("SANGUO_FUND_DMSK_ROOT", str(root))
|
||||
(root / "balance" / "dt=2026-06-30").mkdir(parents=True)
|
||||
(root / "income" / "dt=2026-06-30").mkdir(parents=True)
|
||||
pd.DataFrame({
|
||||
"SECUCODE": ["301569.SZ", "000002.SZ"],
|
||||
"SECURITY_CODE": ["301569", "000002"],
|
||||
"NOTICE_DATE": ["2026-09-17 00:00:00", "2026-08-01 00:00:00"],
|
||||
"REPORT_DATE": ["2026-06-30", "2026-06-30"],
|
||||
"TOTAL_ASSETS": [2429015351.79, 1.0],
|
||||
}).to_parquet(root / "balance" / "dt=2026-06-30" / "part-0.parquet", index=False)
|
||||
pd.DataFrame({
|
||||
"SECUCODE": ["301569.SZ"],
|
||||
"SECURITY_CODE": ["301569"],
|
||||
"NOTICE_DATE": ["2026-09-16 00:00:00"],
|
||||
"REPORT_DATE": ["2026-06-30"],
|
||||
}).to_parquet(root / "income" / "dt=2026-06-30" / "part-0.parquet", index=False)
|
||||
return root
|
||||
|
||||
|
||||
class TestGetDomain:
|
||||
def test_returns_rows_with_all_domains_listed(self, tree):
|
||||
assert dr.DOMAINS == ("balance", "cashflow", "express", "forecast", "income")
|
||||
df = dr.get_domain("balance")
|
||||
assert len(df) == 2
|
||||
|
||||
def test_bad_domain_raises(self, tree):
|
||||
with pytest.raises(ValueError):
|
||||
dr.get_domain("nope")
|
||||
|
||||
def test_noticed_by_pure_filter(self, tree):
|
||||
df = dr.get_domain("balance", noticed_by="2026-09-01")
|
||||
assert list(df["SECURITY_CODE"]) == ["000002"]
|
||||
|
||||
def test_noticed_by_accepts_full_timestamp(self, tree):
|
||||
df = dr.get_domain("balance", noticed_by="2026-09-17 00:00:00")
|
||||
assert len(df) == 2
|
||||
|
||||
def test_noticed_by_missing_column_fails_loud(self, tmp_path, monkeypatch):
|
||||
root = tmp_path / "d2"
|
||||
monkeypatch.setenv("SANGUO_FUND_DMSK_ROOT", str(root))
|
||||
(root / "balance" / "dt=2026-06-30").mkdir(parents=True)
|
||||
pd.DataFrame({"SECURITY_CODE": ["000001"]}).to_parquet(
|
||||
root / "balance" / "dt=2026-06-30" / "p.parquet", index=False)
|
||||
with pytest.raises(KeyError):
|
||||
dr.get_domain("balance", noticed_by="2026-09-17")
|
||||
|
||||
def test_missing_root_empty_no_raise(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("SANGUO_FUND_DMSK_ROOT", str(tmp_path / "absent"))
|
||||
assert dr.get_domain("balance").empty
|
||||
@@ -0,0 +1,107 @@
|
||||
# partition_reader 通用分区读取核测试（spec §20.11.3：核内零语义+失败分级）
|
||||
# 2026-09-17 TDD RED -> GREEN
|
||||
import json
|
||||
import os
|
||||
|
||||
import pandas as pd
|
||||
import pytest
|
||||
|
||||
import sanguo_data.partition_reader as pr
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tree(tmp_path):
|
||||
root = tmp_path / "corpus"
|
||||
(root / "news_meta" / "dt=2026-09-16").mkdir(parents=True)
|
||||
(root / "news_meta" / "dt=2026-09-17").mkdir(parents=True)
|
||||
(root / "news_meta" / "not-a-partition").mkdir(parents=True)
|
||||
pd.DataFrame({"art_code": ["a1", "a2"], "stock_code": ["600519", "000001"]}) \
|
||||
.to_parquet(root / "news_meta" / "dt=2026-09-16" / "part-0.parquet", index=False)
|
||||
pd.DataFrame({"art_code": ["a3"], "stock_code": ["600519"]}) \
|
||||
.to_parquet(root / "news_meta" / "dt=2026-09-17" / "part-0.parquet", index=False)
|
||||
pd.DataFrame({"art_code": ["a4"], "stock_code": ["300750"]}) \
|
||||
.to_parquet(root / "news_meta" / "dt=2026-09-17" / "part-1.parquet", index=False)
|
||||
return str(root)
|
||||
|
||||
|
||||
class TestRoots:
|
||||
def test_env_override(self, monkeypatch, tmp_path):
|
||||
monkeypatch.setenv("SANGUO_CORPUS_ROOT", str(tmp_path))
|
||||
assert pr.corpus_root() == str(tmp_path)
|
||||
|
||||
def test_default_posix(self, monkeypatch):
|
||||
monkeypatch.delenv("SANGUO_CORPUS_ROOT", raising=False)
|
||||
monkeypatch.setattr(pr, "_IS_WIN", False)
|
||||
assert pr.corpus_root() == "/volume1/stock/corpus"
|
||||
|
||||
def test_default_win(self, monkeypatch):
|
||||
monkeypatch.delenv("SANGUO_CORPUS_ROOT", raising=False)
|
||||
monkeypatch.setattr(pr, "_IS_WIN", True)
|
||||
assert pr.corpus_root() == r"C:\sanguo_vnpy_v2\data\corpus"
|
||||
|
||||
|
||||
class TestListDates:
|
||||
def test_sorted_dt_values_ignores_non_partition(self, tree):
|
||||
assert pr.list_dates(tree, "news_meta") == ["2026-09-16", "2026-09-17"]
|
||||
|
||||
def test_missing_domain_empty(self, tree):
|
||||
assert pr.list_dates(tree, "nope") == []
|
||||
|
||||
|
||||
class TestReadPartitions:
|
||||
def test_concat_all_partitions_and_part_files(self, tree):
|
||||
df = pr.read_partitions(tree, "news_meta")
|
||||
assert list(df["art_code"]) == ["a1", "a2", "a3", "a4"]
|
||||
|
||||
def test_start_end_filter_inclusive(self, tree):
|
||||
df = pr.read_partitions(tree, "news_meta", start="2026-09-17", end="2026-09-17")
|
||||
assert list(df["art_code"]) == ["a3", "a4"]
|
||||
df = pr.read_partitions(tree, "news_meta", start="2026-09-16", end="2026-09-16")
|
||||
assert list(df["art_code"]) == ["a1", "a2"]
|
||||
|
||||
def test_duplicate_rows_survive_no_silent_dedup(self, tmp_path):
|
||||
# 核内零去重（因子线契约）：同键行跨分区重复必须原样透出
|
||||
root = tmp_path / "d"
|
||||
for dt in ("dt=2026-09-16", "dt=2026-09-17"):
|
||||
(root / "x" / dt).mkdir(parents=True)
|
||||
pd.DataFrame({"SECURITY_CODE": ["000001"], "REPORT_DATE": ["2026-06-30"]}) \
|
||||
.to_parquet(root / "x" / dt / "part-0.parquet", index=False)
|
||||
df = pr.read_partitions(str(root), "x")
|
||||
assert len(df) == 2
|
||||
|
||||
def test_missing_root_empty_df(self, tmp_path):
|
||||
df = pr.read_partitions(str(tmp_path / "absent"), "news_meta")
|
||||
assert isinstance(df, pd.DataFrame) and df.empty
|
||||
|
||||
def test_empty_domain_dir_empty_df(self, tmp_path):
|
||||
(tmp_path / "corpus" / "news_meta").mkdir(parents=True)
|
||||
df = pr.read_partitions(str(tmp_path / "corpus"), "news_meta")
|
||||
assert df.empty
|
||||
|
||||
def test_schema_column_drift_fails_loud(self, tmp_path):
|
||||
root = tmp_path / "d"
|
||||
(root / "x" / "dt=2026-09-16").mkdir(parents=True)
|
||||
(root / "x" / "dt=2026-09-17").mkdir(parents=True)
|
||||
pd.DataFrame({"a": [1], "b": [2]}).to_parquet(root / "x" / "dt=2026-09-16" / "p.parquet", index=False)
|
||||
pd.DataFrame({"a": [1], "c": [3]}).to_parquet(root / "x" / "dt=2026-09-17" / "p.parquet", index=False)
|
||||
with pytest.raises(pr.PartitionSchemaError):
|
||||
pr.read_partitions(str(root), "x")
|
||||
|
||||
def test_schema_dtype_drift_fails_loud(self, tmp_path):
|
||||
root = tmp_path / "d"
|
||||
(root / "x" / "dt=2026-09-16").mkdir(parents=True)
|
||||
(root / "x" / "dt=2026-09-17").mkdir(parents=True)
|
||||
pd.DataFrame({"a": [1]}).to_parquet(root / "x" / "dt=2026-09-16" / "p.parquet", index=False)
|
||||
pd.DataFrame({"a": [1.5]}).to_parquet(root / "x" / "dt=2026-09-17" / "p.parquet", index=False)
|
||||
with pytest.raises(pr.PartitionSchemaError):
|
||||
pr.read_partitions(str(root), "x")
|
||||
|
||||
|
||||
class TestWatermark:
|
||||
def test_read_watermark_dict(self, tree):
|
||||
with open(os.path.join(tree, pr.WATERMARK_NAME), "w") as f:
|
||||
json.dump({"generated_at": "2026-09-17T16:00:00", "domains": {"news_meta": "2026-09-17"}}, f)
|
||||
assert pr.read_watermark(tree)["domains"]["news_meta"] == "2026-09-17"
|
||||
|
||||
def test_read_watermark_absent_none(self, tree):
|
||||
assert pr.read_watermark(tree) is None
|
||||
Reference in New Issue
Block a user