| 37 | |
37 | |
| 38 | |
38 | |
| 39 | def iter_archive_files(zip_path: Path) -> Iterator[tuple[str, bytes]]: |
39 | def iter_archive_files(zip_path: Path) -> Iterator[tuple[str, bytes]]: |
| 40 | − """Yield (member_name, bytes) for every data file, descending into nested zips.""" |
40 | + """Yield (member_name, bytes) for every data file, descending into nested zips. |
| |
41 | + |
| |
42 | + Some vendor archives (option chains from 2023 on) use Deflate64, which Python's zipfile cannot decompress |
| |
43 | + (`NotImplementedError: That compression method is not supported`). In that case the archive is extracted with |
| |
44 | + the system `7z` into a temporary directory next to the archive and the files are streamed from disk.""" |
| |
45 | + try: |
| |
46 | + yield from _iter_zip_members(zip_path) |
| |
47 | + except NotImplementedError: |
| |
48 | + yield from _iter_via_7z(zip_path) |
| |
49 | + |
| |
50 | + |
| |
51 | +def _iter_zip_members(zip_path: Path) -> Iterator[tuple[str, bytes]]: |
| 41 | with zipfile.ZipFile(zip_path) as zf: |
52 | with zipfile.ZipFile(zip_path) as zf: |
| 42 | for info in zf.infolist(): |
53 | for info in zf.infolist(): |
| 43 | if info.is_dir(): |
54 | if info.is_dir(): |
| 53 | yield name, zf.read(info) |
64 | yield name, zf.read(info) |
| 54 | |
65 | |
| 55 | |
66 | |
| |
67 | +def _find_7z() -> str | None: |
| |
68 | + import shutil |
| |
69 | + for cand in ("7zz", "7z", "7za"): |
| |
70 | + p = shutil.which(cand) or shutil.which(cand, path="/opt/homebrew/bin:/usr/local/bin") |
| |
71 | + if p: |
| |
72 | + return p |
| |
73 | + return None |
| |
74 | + |
| |
75 | + |
| |
76 | +def _iter_via_7z(zip_path: Path) -> Iterator[tuple[str, bytes]]: |
| |
77 | + import shutil |
| |
78 | + import subprocess |
| |
79 | + import tempfile |
| |
80 | + tool = _find_7z() |
| |
81 | + if not tool: |
| |
82 | + raise RuntimeError(f"{zip_path.name}: unsupported zip compression (Deflate64) and no 7z binary found (brew install p7zip)") |
| |
83 | + tmp = Path(tempfile.mkdtemp(prefix="unz-", dir=str(zip_path.parent))) |
| |
84 | + try: |
| |
85 | + r = subprocess.run([tool, "x", "-y", "-bd", "-bso0", "-bsp0", f"-o{tmp}", str(zip_path)], capture_output=True, text=True, timeout=6 * 3600) |
| |
86 | + if r.returncode not in (0, 1): # 1 = warnings |
| |
87 | + raise RuntimeError(f"7z extraction failed for {zip_path.name}: {r.stderr[-300:]}") |
| |
88 | + for p in sorted(tmp.rglob("*")): |
| |
89 | + if p.is_file() and p.suffix.lower() in (".txt", ".csv"): |
| |
90 | + yield str(p.relative_to(tmp)), p.read_bytes() |
| |
91 | + elif p.is_file() and p.suffix.lower() == ".zip": |
| |
92 | + yield from ((f"{p.name}!/{n}", b) for n, b in iter_archive_files(p)) |
| |
93 | + finally: |
| |
94 | + shutil.rmtree(tmp, ignore_errors=True) |
| |
95 | + |
| |
96 | + |
| 56 | def parse_bar_file(data: bytes, symbol: str, asset_class: AssetClass, interval: Interval, adjustment: Adjustment, |
97 | def parse_bar_file(data: bytes, symbol: str, asset_class: AssetClass, interval: Interval, adjustment: Adjustment, |
| 57 | source_id: str, ingested_at: datetime, source_tz: str = "America/New_York", |
98 | source_id: str, ingested_at: datetime, source_tz: str = "America/New_York", |
| 58 | currency: str = "USD", venue: str | None = None, revision_id: int = 1) -> tuple[pa.Table, int]: |
99 | currency: str = "USD", venue: str | None = None, revision_id: int = 1) -> tuple[pa.Table, int]: |