Aislo/resources/master_data/scripts/material_merge.py

"""재료 자재품목 항목 통합(요구 819 · 827 · 828 · 830 ~ 832 · PLAN 830-2) — build_재료_자재품목 이 부름.
시험 = test_830_2_material_merge.

묶음 = 이름에서 지역 · 관급/사급 꼬리표 · 제조사 꼬리(「-거양」) · 조합 이름 · 상하차 글을 뺀
  이름 · 규격 · 단위 · 상하차가 같은 줄.
  규격 = 앞말 「조합공통품목,」 · 규격 속 지역 괄호 뗌
  단위 = M=m · m²=㎡ · Ton=톤 · ㎏=kg 표기 하나로 · 비교는 공백을 뺀 글
열 한 벌 = COLUMNS — 상하차 = 단위 바로 뒤 · 가격 열 = 조달청(관급) · 물가지 넷 · 일반가격(사급).
상하차 칸(요구 828 · 831) = 상차 · 하차 · -(기본 · 모름) — 이름 · 규격의 인도 조건 글을 떼어 이 칸으로
  「도착도」 · 「하차도」 = 하차 · 「채취장상차(도)」 · 「하치장 · 생산공장 상차도」 · 「(상차도)」 = 상차
남는 줄 = 사급(관급 아닌) 줄 가운데 전국 › 지역 글 없음(전국으로 봄) › 서울 › 서울 권역 글 › 경기
  › 그 밖 지역(요구 830 · 831 — 조달청과 같은 차례) · 같은 차례 여럿(경기 · 그 밖)이면 값이 중간인 줄
  (짝수면 아래 중간 · 다시 지어도 같은 줄)
  · 남길 줄의 값 칸이 전부 비고 다른 줄에 값이 있으면 다음 차례 줄(브)
  · 관급 줄만 있는 묶음 = 가장 작은 키 한 줄(브) · 나머지 줄은 지움(이력 안 남김)
덧정보가 하나도 없는 묶음(원래부터 겹친 줄 — 식생 · 종자 등)은 합치지 않음.
조달청 칸 값은 모두 비움(EST 「가격정보」 — 요구 827 · 831) · 일반가격 = 시스템 관리자가 넣는 칸(빌더가 이어 받음).
이름에서 뗀 말은 비고 「통합: …」 로.
"""

import re

MARKET = ("물가자료", "유통물가", "물가정보", "거래가격")
LOAD = "상하차"  # 표지 칸(값 칸 아님)
LOADS = ("-", "상차", "하차")  # 첫째 = 기본(모름)
GOVT = "조달청"  # 관급 가격 칸 — 가장 낮은 값 고르기에 안 넣음(요구 826 · 830)
OWN = "일반가격"  # 시스템 관리자 손 입력 칸(사급)
COLUMNS = (  # 열 한 벌 · 차례(요구 830 · 831)
    "키",
    "구분",
    "상세구분",
    "이름",
    "규격",
    "단위",
    LOAD,
    GOVT,
    *MARKET,
    OWN,
    "출처",
    "비고",
    "면수",
)
GYEONGGI = {"경기", "수원", "용인", "포천", "가평", "여주", "화성"}
CITY = set(
    """서울 부산 대구 인천 광주 대전 울산 세종 제주 청주 전주 포항 창원 춘천 충주 강릉
    천안 원주 여수 구미 순천 수원 목포 금산 영천 논산 삼척 홍성 익산 청도 양양 남원
    공주 군산 김천 가평 용인 포천 경주 아산 제천 여주 보령 부여 당진 청원 음성 서산
    영동 진천 예산 장성 고흥 고창 광양 고령 성주 영주 안동 칠곡 상주 문경 합천 화성
    경산 전국""".split()
)
SIDO = "서울|부산|대구|인천|광주|대전|울산|세종|제주|경기|강원|충북|충남|전북|전남|경북|경남|충청|전라|경상"
_CITY = "|".join(sorted(CITY, key=len, reverse=True))
UNIT = {"M": "m", "m²": "㎡", "Ton": "톤", "ton": "톤", "㎏": "kg"}
SUPPLY = (  # (꼬리표 글, 공급) — 이름 끝 · 괄호 꼴
    (r"\(관급\)", "관급"),
    (r"[-/ ]관급$", "관급"),
    (r"(?<=\))관급$", "관급"),
    (r",관급(?=\))", "관급"),
    (r"\(사급\)", "사급"),
    (r"-사급$", "사급"),
    (r",사급$", "사급"),
)
NAME_LOAD = re.compile(r",\s*(도착도|하차도|채취장상차도|채취장상차)$")  # 모 래, 서울, 도착도
SPEC_LOAD = re.compile(
    r"\s*(?:\((상차도)\)|((?:하치장|생산공장)\s*상차도))$"
)  # HD=10mm 하치장 상차도
COOP = re.compile(r"^조합공통품목(?:\(([^)]*)\))?,?\s*")


def load_of(text: str) -> str:
    """인도 조건 글 → 상하차 값 — 도착 · 하차 = 하차 · 상차 = 상차."""
    return "하차" if text.startswith(("도착", "하차")) else "상차"


def place(row: dict, load: str) -> dict:
    """COLUMNS 차례의 줄 — 첫 칸(열쇠 · 키)은 그대로 · 상하차 = load · 조달청 = 빔 · 옛 칸(관급)은 뺌."""
    first = next(iter(row))
    out = {first: row[first]}
    for c in COLUMNS[1:]:
        out[c] = load if c == LOAD else None if c == GOVT else row.get(c)
    return out


def ws(s: str | None) -> str:
    return re.sub(r"\s+", " ", s or "").strip()


def nows(s: str) -> str:
    return re.sub(r"\s+", "", s)


def _region(name: str, spec: str):
    """(지역 글, 지역을 뗀 이름, 지역을 뗀 규격) — 지역 꼴 일곱(조사 818)."""
    m = re.match(r"^(.*?)\(([^()]+)\)-사급$", name)  # 레미콘(서울)-사급
    if m and m[2] in CITY:
        return m[2], m[1] + "-사급", spec
    m = re.match(r"^(.*?)\((SD\d+),([^()]+)\)-관급$", name)  # 이형철근(SD400,제주)-관급
    if m and m[3] in CITY:
        return m[3], f"{m[1]}({m[2]})-관급", spec
    m = re.match(r"^(.*?\(관급\))-(.+)$", name)  # 레미콘(관급)-강원도(태백시)
    if m:
        tail = m[2]
        head = re.sub(r"[(\s,].*", "", tail)
        if re.match(rf"^({SIDO})", head) or head in CITY or "광역시" in head or head.endswith("도"):
            return tail, m[1], spec
        return None, name, spec
    m = re.match(r"^(.*?),\s*([가-힣]+)\s*\(관급\)$", name)
    if m and m[2] in CITY:
        return m[2], m[1] + "(관급)", spec
    if "(관급)" in name or name.endswith("-관급") or "-사급" in name:
        return None, name, spec
    parts = [p.strip() for p in name.split(",")]  # 합판(내수), 서울 · 모 래, 여수, 도착도
    hit = [i for i, p in enumerate(parts) if i > 0 and p in CITY]
    if hit:
        return parts[hit[0]], ", ".join(p for j, p in enumerate(parts) if j != hit[0]), spec
    m = re.search(
        rf"\((?:([가-힣]+)·)?({_CITY}|충남)(지역)?(,관급)?\)", name
    )  # H 형강(전국) · 모래(건조상태·전국)
    if m:
        keep = f"({m[1]})" if m[1] else ""
        rest = "(관급)" if m[4] else ""
        return m[2], name[: m.start()] + keep + rest + name[m.end() :], spec
    m = re.search(rf"\s?\(({_CITY})(지역)?\)", spec)  # 콘크리트근가 0.7M(제주지역)
    if m:
        return m[1], name, spec[: m.start()] + spec[m.end() :]
    if spec in CITY:  # 품질시험 현장출장비 · 포항
        return spec, name, ""
    return None, name, spec


def parse(row: dict) -> dict:
    """줄 → 덧정보를 뗀 이름 · 규격 · 상하차 · 묶음 열쇠 · 뗀 말."""
    name, spec = row.get("이름") or "", row.get("규격") or ""
    words = []
    region, base, spec2 = _region(name, spec)
    m = re.search(r"-거양$", base)
    if m:
        base = base[: m.start()]
        words.append("제조사 거양")
    supply = ""
    for pat, kind in SUPPLY:
        if re.search(pat, base):
            supply = supply or kind
            base = re.sub(pat, "", base)
    if supply:
        words.append(supply)
    load = row.get(LOAD) or LOADS[0]
    m = NAME_LOAD.search(base)
    if m:
        load = load_of(m[1])
        words.append(m[1])
        base = base[: m.start()]
    base = re.sub(r"[\s,]+$", "", base)
    m = COOP.match(spec2)
    if m:
        spec2 = spec2[m.end() :]
        words.append(f"조합 {m[1]}" if m[1] else "조합공통품목")
    m = SPEC_LOAD.search(spec2)
    if m:
        load = load_of(m[1] or m[2])
        words.append(m[1] or m[2])
        spec2 = spec2[: m.start()]
    spec2 = spec2.strip()
    unit = ws(row.get("단위"))
    tagged = bool(region or words)
    return {
        "이름": base if tagged else name,
        "규격": spec2 if tagged else spec,
        LOAD: load,
        "지역": region or "",
        "공급": supply,
        "뗀 말": words,
        "덧정보": tagged,
        "열쇠": (nows(base), nows(spec2), UNIT.get(unit, unit), load),
    }


def price(row: dict):
    v = [row[s] for s in MARKET if row.get(s) is not None]
    return min(v) if v else None


RANKS = ("전국", "지역 없음", "서울", "서울 권역", "경기", "그 밖")


def _rank(region: str) -> int:
    if region == "전국":
        return 0
    if not region:
        return 1
    if region == "서울":
        return 2
    if region.startswith("서울"):
        return 3
    if region in GYEONGGI or region.startswith("경기"):
        return 4
    return 5


def _middle(pool: list) -> list:
    """값이 중간인 줄부터 가까운 차례(짝수면 아래 중간 · 값 빈 줄은 끝) — 다시 지어도 같은 차례."""
    pool = sorted(pool, key=lambda p: (price(p[0]) is None, price(p[0]) or 0))
    mid = max((sum(1 for p in pool if price(p[0]) is not None) - 1) // 2, 0)
    return sorted(pool, key=lambda p: (abs(pool.index(p) - mid), pool.index(p)))


def _order(pool: list) -> list:
    """남길 차례 — 전국 › 지역 없음 › 서울 › 서울 권역 › 경기 › 그 밖 · 경기 · 그 밖은 값 중간부터 ·
    그 위 차례는 앞 줄(작은 키)부터."""
    out = []
    for rank in range(len(RANKS)):
        same = [p for p in pool if _rank(p[1]["지역"]) == rank]
        out += _middle(same) if rank >= 4 else same
    return out


def merge(rows: list[dict]) -> tuple[list[dict], list[dict]]:
    """통합한 줄(조달청 칸 비움 · COLUMNS 차례) · 기록(덧정보를 뗀 묶음마다 남은 줄 · 지운 줄 · 기준 지역 · 「브」 규칙).
    줄 차례 = 키 차례(빌더가 원문 차례로 짓고 대장 키도 그 차례) — 같은 순위 · 관급만 묶음은 앞 줄을 남김."""
    groups: dict = {}
    for r in rows:
        p = parse(r)
        groups.setdefault(p["열쇠"], []).append((r, p))
    new, log = {}, []
    for g in groups.values():
        if not any(p["덧정보"] for _, p in g):
            continue
        sup = [x for x in g if x[1]["공급"] != "관급"]
        rules = []
        if sup:
            order = _order(sup)
            best = next((x for x in order if price(x[0]) is not None), order[0])
            region = best[1]["지역"]
            if _rank(region) == 1 and len(g) > 1:
                rules.append("지역 없음 = 전국")
            if _rank(region) >= 4:
                rules.append(RANKS[_rank(region)])
                if sum(1 for x in sup if _rank(x[1]["지역"]) == _rank(region)) > 1:
                    rules.append("값 중간 지역")
            if best is not order[0]:
                rules.append("값 빈 줄 건너뜀")
        else:
            best = g[0]
            region = ""
            if len(g) > 1:
                rules.append("관급만 = 작은 키")
        words = list(dict.fromkeys(w for _, p in g for w in p["뗀 말"]))
        regions = sorted({p["지역"] for _, p in g if p["지역"]})
        bits = [f"기준 지역 {region or '없음'}" if regions else ""] if sup else ["관급 줄만"]
        if len(g) > 1:
            bits.append(f"{len(g)}줄 합침")
        if len(regions) > 1 or (regions and region != regions[0]):
            bits.append(f"지역 {len(regions)}곳")
        if words:
            bits.append("뗀 말 " + ", ".join(words))
        r, info = best
        note = "통합: " + " · ".join(b for b in bits if b)
        new[id(r)] = place(
            {
                **r,
                "이름": info["이름"],
                "규격": info["규격"],
                "비고": " · ".join(x for x in (r.get("비고"), note) if x),
            },
            info[LOAD],
        )
        gone = [o for o, _ in g if o is not r]
        for o in gone:
            new[id(o)] = None
        log.append(
            {"남은": new[id(r)], "원래": r, "지운": gone, "기준 지역": region, "규칙": rules}
        )
    out = []
    for r in rows:
        r = new.get(id(r), place(r, r.get(LOAD) or LOADS[0]))
        if r is not None:
            out.append(r)
    return out, log