Aislo/tools/stmate-test/dbf_rw.py

# -*- coding: utf-8 -*-
"""STMATE DBF 무손실 읽기·고쳐 쓰기 — 손대지 않은 칸은 원래 바이트 그대로.

근거: STC_컨테이너_구조.md §3·§4(필드명 기반 매핑 · 미지원 칸 원문 보존) · STC_필드사전_핵심.md
규칙:
  - 헤더(필드 설명자·끝 0x0D 뒤 꼬리 포함)·레코드·파일 끝(0x1A) 원문 보존
  - 칸 값 비교는 뜻으로 — 숫자는 자리수 맞춰, 글자는 앞뒤 공백 무시, 빈 숫자 = 0
  - 같은 값이면 원래 바이트 유지(앞 공백·빈칸 표기 보존) · 다른 값만 그 칸 폭에 맞춰 씀
  - 글자는 cp949 엄격 인코딩(못 쓰는 글자는 오류 — '?' 로 바꾸지 않음) · 두 바이트 글자 중간에서 자르지 않음
  - 숫자는 실무 STC 7건 전 칸과 같은 꼴(소수 dec 자리 고정 · 오른쪽 맞춤)
  - 고친 표만 헤더 날짜(1~3바이트)를 오늘로 — dBase 마지막 갱신일
  - 암호 표(버전 0x06 · BDQTY)는 dbf_crypt 로 자리별 변환 — 표에 없는 자리는 쓰지 않음
"""

from __future__ import annotations

import datetime
import struct
from dataclasses import dataclass

ENC = "cp949"


@dataclass(frozen=True)
class Field:
    name: str
    type: str
    len: int
    dec: int
    off: int


def _num_text(v: float, dec: int) -> str:
    return f"{v:.{dec}f}" if dec > 0 else str(int(round(v)))


def _to_float(v: object) -> float | None:
    if v is None:
        return None
    s = str(v).strip().replace(",", "")
    if s in ("", "None", "null"):
        return None
    return float(s)


def cut_cp949(b: bytes, n: int) -> bytes:
    """n 바이트 안으로 자르되 두 바이트 글자 중간을 피함."""
    if len(b) <= n:
        return b
    i = 0
    while i < n:
        step = 2 if b[i] >= 0x81 else 1
        if i + step > n:
            break
        i += step
    return b[:i]


class DbfTable:
    """DBF 한 개 — 레코드는 bytearray 로 들고 칸 단위로만 고침."""

    def __init__(self, b: bytes, codec=None):
        if len(b) < 32:
            raise ValueError("DBF 헤더 길이 부족")
        self.version = b[0]
        nrec, hlen, rlen = struct.unpack_from("<IHH", b, 4)
        self.head = bytearray(b[:hlen])
        self.rlen = rlen
        self.fields: dict[str, Field] = {}
        self.order: list[str] = []
        off, o = 1, 32
        while o < hlen and b[o] != 0x0D:
            name = b[o : o + 11].split(b"\x00", 1)[0].decode("ascii", "replace").strip()
            f = Field(name, chr(b[o + 11]), b[o + 16], b[o + 17], off)
            self.fields[name] = f
            self.order.append(name)
            off += f.len
            o += 32
        body_end = hlen + nrec * rlen
        self.recs = [bytearray(b[hlen + i * rlen : hlen + (i + 1) * rlen]) for i in range(nrec)]
        self.recs = [r for r in self.recs if len(r) == rlen]
        self.tail = bytes(b[body_end:]) if len(b) > body_end else b"\x1a"
        self.codec = codec
        self.dirty = False

    # ── 읽기 ──
    def __len__(self) -> int:
        return len(self.recs)

    def raw(self, i: int, name: str) -> bytes:
        f = self.fields[name]
        val = bytes(self.recs[i][f.off : f.off + f.len])
        return self.codec.decode(val, f.off) if self.codec else val

    def get(self, i: int, name: str) -> str:
        return self.raw(i, name).decode(ENC, "replace").strip()

    def deleted(self, i: int) -> bool:
        return self.recs[i][0:1] == b"*"

    def row(self, i: int) -> dict:
        return {n: self.get(i, n) for n in self.order}

    # ── 쓰기 ──
    def set_raw(self, i: int, name: str, val: bytes) -> bool:
        f = self.fields[name]
        val = bytes(val[: f.len]).ljust(f.len, b" ")
        if self.codec:
            val = self.codec.encode(val, f.off)
        if bytes(self.recs[i][f.off : f.off + f.len]) == val:
            return False
        self.recs[i][f.off : f.off + f.len] = val
        self.dirty = True
        return True

    def encode_value(self, name: str, value: object) -> bytes:
        f = self.fields[name]
        if f.type in ("N", "F"):
            v = _to_float(value)
            if v is None:
                return b" " * f.len
            txt = _num_text(v, f.dec)
            if len(txt) > f.len:
                raise ValueError(f"{name} 칸 폭 {f.len} 넘는 숫자: {txt}")
            return txt.encode("ascii").rjust(f.len)
        if f.type == "L":
            s = str(value).strip().upper()[:1] if value is not None else ""
            s = {"1": "T", "0": "F", "Y": "T", "N": "F"}.get(s, s)
            return (s or " ").encode("ascii").ljust(f.len)
        s = "" if value is None else str(value)
        try:
            enc = s.encode(ENC)
        except UnicodeEncodeError as e:
            raise ValueError(f"{name} 칸에 cp949 로 쓸 수 없는 글자: {s!r}") from e
        return cut_cp949(enc, f.len).ljust(f.len, b" ")

    def same_value(self, i: int, name: str, value: object) -> bool:
        f = self.fields[name]
        old = self.raw(i, name)
        if f.type in ("N", "F"):
            ov = _to_float(old.decode("ascii", "replace"))
            nv = _to_float(value)
            if ov is None or nv is None:
                return (ov or 0.0) == (nv or 0.0)
            return _num_text(ov, f.dec) == _num_text(nv, f.dec)
        if f.type == "L":
            return self.encode_value(name, value).strip() == old.strip()
        new = "" if value is None else str(value)
        return old.decode(ENC, "replace").strip() == new.strip()

    def set(self, i: int, name: str, value: object) -> bool:
        """값이 뜻으로 다를 때만 씀 — 바뀌었으면 True."""
        f = self.fields[name]
        if f.type == "M" or (f.type not in ("C", "N", "F", "L", "D")):
            raise ValueError(f"{name}({f.type}) 칸은 set_raw 로만 씀")
        if self.same_value(i, name, value):
            return False
        return self.set_raw(i, name, self.encode_value(name, value))

    def blank_record(self) -> bytearray:
        rec = bytearray(b" " * self.rlen)
        if self.codec:
            for f in self.fields.values():
                rec[f.off : f.off + f.len] = self.codec.encode(b" " * f.len, f.off)
        return rec

    def insert(self, pos: int, values: dict | None = None, template: int | None = None) -> int:
        """pos 자리에 레코드를 끼움(template 레코드 복사 또는 빈칸) — 끼운 자리 반환."""
        rec = bytearray(self.recs[template]) if template is not None else self.blank_record()
        rec[0:1] = b" "
        pos = max(0, min(pos, len(self.recs)))
        self.recs.insert(pos, rec)
        self.dirty = True
        for k, v in (values or {}).items():
            if k in self.fields and self.fields[k].type != "M":
                self.set_raw(pos, k, self.encode_value(k, v))
        return pos

    def to_bytes(self, today: datetime.date | None = None) -> bytes:
        head = bytearray(self.head)
        struct.pack_into("<I", head, 4, len(self.recs))
        if self.dirty:
            d = today or datetime.date.today()
            head[1], head[2], head[3] = d.year - 1900, d.month, d.day
        return bytes(head) + b"".join(bytes(r) for r in self.recs) + self.tail


def find_rows(tbl: DbfTable, name: str) -> dict[str, int]:
    """name 칸 값 → 레코드 자리(삭제 레코드 제외 · 같은 값이면 첫 자리)."""
    out: dict[str, int] = {}
    for i in range(len(tbl)):
        if not tbl.deleted(i):
            out.setdefault(tbl.get(i, name), i)
    return out