88 lines
2.6 KiB
Python
88 lines
2.6 KiB
Python
import hashlib
|
|
from pathlib import Path
|
|
from typing import Optional, Union
|
|
|
|
from .utils import register_log_file, register_reference, touch_log_file
|
|
|
|
LOG_DIR = Path("./instance/log")
|
|
|
|
|
|
def _ensure_log_dir() -> None:
|
|
LOG_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def _canonicalize_literal(value: str) -> str:
|
|
raw = value.strip()
|
|
if not raw:
|
|
raise ValueError("Literal value cannot be empty.")
|
|
|
|
sign = ""
|
|
digits = raw
|
|
if raw[0] in "+-":
|
|
sign = "-" if raw[0] == "-" else ""
|
|
digits = raw[1:]
|
|
|
|
if not digits or not digits.isdigit():
|
|
raise ValueError(f"Literal must be an integer string, got: {value!r}")
|
|
|
|
digits = digits.lstrip("0") or "0"
|
|
if digits == "0":
|
|
sign = ""
|
|
return f"{sign}{digits}"
|
|
|
|
|
|
def _is_in_log_dir(path: Path) -> bool:
|
|
try:
|
|
path.resolve().relative_to(LOG_DIR.resolve())
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
class StreamNumber:
|
|
def __init__(
|
|
self,
|
|
file_path: Optional[Union[str, Path]] = None,
|
|
*,
|
|
literal: Optional[str] = None,
|
|
):
|
|
if (file_path is None) == (literal is None):
|
|
raise ValueError("Provide exactly one of file_path or literal.")
|
|
|
|
if literal is not None:
|
|
normalized = _canonicalize_literal(literal)
|
|
_ensure_log_dir()
|
|
literal_hash = hashlib.sha1(normalized.encode()).hexdigest()[:10]
|
|
self.path = LOG_DIR / f"literal_{literal_hash}.txt"
|
|
self.path.write_text(normalized, encoding="utf-8")
|
|
else:
|
|
self.path = Path(file_path)
|
|
if not self.path.exists():
|
|
raise FileNotFoundError(self.path)
|
|
|
|
if _is_in_log_dir(self.path):
|
|
register_log_file(self.path)
|
|
register_reference(self.path)
|
|
|
|
self.hash = hashlib.sha1(str(self.path).encode()).hexdigest()[:10]
|
|
|
|
def __repr__(self):
|
|
return f"<StreamNumber {self.path.name}>"
|
|
|
|
def stream(self, chunk_size=4096):
|
|
"""Yield chunks of digits as strings."""
|
|
if _is_in_log_dir(self.path):
|
|
touch_log_file(self.path)
|
|
with open(self.path, "r", encoding="utf-8") as f:
|
|
while chunk := f.read(chunk_size):
|
|
yield chunk.strip().replace(",", ".")
|
|
|
|
def write_stage(self, stage, data: str):
|
|
"""Write intermediate stage result."""
|
|
_ensure_log_dir()
|
|
stage_file = LOG_DIR / f"{self.hash}_stage_{stage}.bin"
|
|
with open(stage_file, "wb") as f:
|
|
f.write(data.encode())
|
|
register_log_file(stage_file)
|
|
return stage_file
|