158 lines
5.7 KiB
Python
158 lines
5.7 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""vps_mirror_lib — corpus/dmsk NAS→VPS 镜像链纯逻辑核（spec §20.11.2）。
|
||
|
||
journal 指纹（分区聚合 bytes+max mtime）/diff/水位线计算。bash 端
|
||
push_vps_mirror_daily.sh 消费本模块 CLI 输出执行 scp；pytest 直测纯函数。
|
||
仅 stdlib（NAS 宿主 python3.8 兼容）。
|
||
|
||
语义：分区粒度增量、只增不删（远端不镜像本地删除；分区重写靠指纹变化
|
||
触发整分区重推吸收）、幂等可重跑、journal 删除=全量重建。
|
||
|
||
CLI（输出协议 TAB 分隔，bash 消费）：
|
||
plan [--dry-run] [--journal-dir DIR]
|
||
扫描+diff，打印 PUSH 行与 PLAN 摘要；写 pending（除非 --dry-run）
|
||
与水位线临时文件 watermark_<family>.json
|
||
commit [--journal-dir DIR]
|
||
pending 原子晋升为 journal（只在 bash 全部推送成功后调用，
|
||
保证 journal 记录的正是推送时扫到的指纹，mid-push 变更次日必重推）
|
||
协议行：
|
||
PUSH<TAB>family<TAB>remote_parent_posix<TAB>local_partition_abs
|
||
PLAN<TAB>push=N<TAB>bytes=N
|
||
"""
|
||
import argparse
|
||
import datetime
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
FAMILIES = {
|
||
"corpus": {
|
||
"domains": ["news_meta", "news_fulltext", "ann_meta", "ann_pdf",
|
||
"flash_meta"], # §20.12.3 快讯域(09-22): 新域=指纹全新, 首跑自动全量推一次
|
||
"local_root": "/volume1/stock/corpus",
|
||
"remote_root": "C:/sanguo_vnpy_v2/data/corpus",
|
||
},
|
||
"dmsk": {
|
||
"domains": ["balance", "cashflow", "express", "forecast", "income"],
|
||
"local_root": "/volume1/stock/fundamentals/dmsk",
|
||
"remote_root": "C:/sanguo_vnpy_v2/data/fundamentals/dmsk",
|
||
},
|
||
}
|
||
JOURNAL_DIR_DEFAULT = "/volume1/stock/vps-mirror"
|
||
WATERMARK_NAME = "_mirror_watermark.json"
|
||
|
||
|
||
def fingerprint(partition_dir):
|
||
# type: (str) -> list
|
||
"""分区指纹 = [聚合 bytes, max mtime_ns]。任一变化即判该分区需重推。"""
|
||
total = 0
|
||
max_ns = 0
|
||
for dirpath, _dirnames, filenames in os.walk(partition_dir):
|
||
for fn in filenames:
|
||
st = os.stat(os.path.join(dirpath, fn))
|
||
total += st.st_size
|
||
if st.st_mtime_ns > max_ns:
|
||
max_ns = st.st_mtime_ns
|
||
return [total, max_ns]
|
||
|
||
|
||
def scan_family(root, domains):
|
||
# type: (str, list) -> dict
|
||
"""{domain/分区名: 指纹}。分区=域目录下一级子目录（dt= 或 ann_pdf 年目录）。"""
|
||
inv = {}
|
||
for domain in domains:
|
||
droot = os.path.join(root, domain)
|
||
if not os.path.isdir(droot):
|
||
continue
|
||
for name in sorted(os.listdir(droot)):
|
||
p = os.path.join(droot, name)
|
||
if os.path.isdir(p):
|
||
inv["%s/%s" % (domain, name)] = fingerprint(p)
|
||
return inv
|
||
|
||
|
||
def diff(journal, inventory):
|
||
# type: (dict, dict) -> tuple
|
||
"""返回 (待推送键列表, 新 journal 状态)。只增不删：journal 有而本地无的键
|
||
不产生任何动作（远端保留），也不从新状态里显式剔除以外的处理。"""
|
||
to_push = [k for k in sorted(inventory) if journal.get(k) != inventory[k]]
|
||
return to_push, dict(inventory)
|
||
|
||
|
||
def watermark_payload(inventory):
|
||
# type: (dict) -> dict
|
||
"""单族水位线：各 domain 的最大分区值（dt= 前缀剥掉；年目录原样）。"""
|
||
latest = {}
|
||
for key in inventory:
|
||
domain, part = key.split("/", 1)
|
||
val = part[3:] if part.startswith("dt=") else part
|
||
if domain not in latest or val > latest[domain]:
|
||
latest[domain] = val
|
||
return {
|
||
"generated_at": datetime.datetime.now().isoformat(timespec="seconds"),
|
||
"domains": latest,
|
||
}
|
||
|
||
|
||
def commit(journal_path, pending_path):
|
||
# type: (str, str) -> None
|
||
"""pending 原子晋升为 journal；无 pending = no-op。"""
|
||
if os.path.exists(pending_path):
|
||
os.replace(pending_path, journal_path)
|
||
|
||
|
||
def _load_journal(journal_path):
|
||
if os.path.exists(journal_path):
|
||
with open(journal_path) as f:
|
||
return json.load(f)
|
||
return {}
|
||
|
||
|
||
def main(argv=None):
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("mode", choices=["plan", "commit"])
|
||
ap.add_argument("--dry-run", action="store_true",
|
||
help="plan 只打印不写 pending")
|
||
ap.add_argument("--journal-dir", default=JOURNAL_DIR_DEFAULT)
|
||
args = ap.parse_args(argv)
|
||
journal_path = os.path.join(args.journal_dir, "journal.json")
|
||
pending_path = os.path.join(args.journal_dir, "journal.pending.json")
|
||
|
||
if args.mode == "commit":
|
||
commit(journal_path, pending_path)
|
||
return 0
|
||
|
||
os.makedirs(args.journal_dir, exist_ok=True)
|
||
old_journal = _load_journal(journal_path)
|
||
new_state = {}
|
||
push_count = 0
|
||
push_bytes = 0
|
||
lines = []
|
||
for fam in sorted(FAMILIES):
|
||
spec = FAMILIES[fam]
|
||
inv = scan_family(spec["local_root"], spec["domains"])
|
||
to_push, merged = diff(old_journal.get(fam, {}), inv)
|
||
new_state[fam] = merged
|
||
for key in to_push:
|
||
domain, part = key.split("/", 1)
|
||
local_abs = os.path.join(spec["local_root"], domain, part)
|
||
remote_parent = "%s/%s" % (spec["remote_root"], domain)
|
||
lines.append("PUSH\t%s\t%s\t%s" % (fam, remote_parent, local_abs))
|
||
push_count += 1
|
||
push_bytes += inv[key][0]
|
||
wm_path = os.path.join(args.journal_dir, "watermark_%s.json" % fam)
|
||
with open(wm_path, "w") as f:
|
||
json.dump(watermark_payload(inv), f, ensure_ascii=False)
|
||
if not args.dry_run:
|
||
with open(pending_path, "w") as f:
|
||
json.dump(new_state, f, ensure_ascii=False)
|
||
for line in lines:
|
||
print(line)
|
||
print("PLAN\tpush=%d\tbytes=%d" % (push_count, push_bytes))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|