eomsangdonandClaude Opus 5.5
e9e949112c
- 새 파일 — Words(식 풀기 · 장비 갈래 · 조건 낱말) · Finger(블록 글 · STmate 자원 → 지문 · 기계 줄 · 직접 줄 호표) · Shares(절 몫 벡터) · Ours(마스터 판 열쇠 · 우리 지문 캐시 tmp/m02_hopyo · 엔진 셈 캐시 · 자재 값 찾기 건너뜀) · Option(옵션 공간 · 거꾸로 찾기 · 견줌) · Fit(손잡이 한 곳 · 순위 · 뒤 셈 · 맞춤셈 캐시) · Combine(조합 · 섞인 줄 몫 먼저) · Verdict(블록 셈 · 맞음 / 다른 점 / 후보) - Match — b_order(이름 · 근거 차례 · 같음 근거 검사 · 옛 판 글 · 조항 오타) · clause(윗마디 0.9 · 나 창 Basis.clause 있으면 그것) · 이름 후보 로직은 접지 않음 - 정답 표 fixtures/m02_hopyo_golden.json(사람 136 · 새 파일 51) · 골든 도우미 - 북부청 92 줄 실측 — 1순위 87 · 다섯 안 91 · 맞음 41 틀림 0 · 다른 점 28 틀림 0 · 조합 맞음 57/66 · 우리 지문 셈 37 초 · 캐시 읽기 0.05 초 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Ji4nuyLVHcxrXWHTdb5nRD
412 lines
17 KiB
Python
412 lines
17 KiB
Python
"""M02 일위대가 맞춤 — 순위 · 판정 · 블록 셈 · 뒤 셈 · 맞춤셈 캐시(PLAN 338-1 · 설계 2-4 · 2-6 ·
|
|
2-10).
|
|
|
|
순위 = 로직마다 가장 닮은 변형의 지문 점수(인력 · 장비 · 자재 · 율 · 단위 · 조건 · 조항 · 이름 ·
|
|
금액 없음)
|
|
→ 지문 위 POOL ∪ 이름 · 근거 차례(`Match.b_order`) 위 다섯 → 점수 + BETA × 차례 덤 → 위 다섯 ·
|
|
같음(이름 같음 · 씨앗) 1위 고정 — 지문 「딱 맞음」이 다른 로직이면 지문 · 섞인 줄은 이름 · 근거
|
|
차례로
|
|
몫 풀이(`Combine.solve`).
|
|
판정(1위) = 맞음(옵션 수준 ≤ 1 · 엑셀 몫을 다 덮음 · 숫자 하나뿐 줄 아님) · 다른 점(옵션 수준 2 또는
|
|
직종 묶음 같음 + 인용 조항 같음) · 후보 — 글은 숫자 사실만(엑셀 · 우리 품).
|
|
뒤 셈 = 올리기 · 보기가 `ensure` → 일감 줄 하나 · 블록 차례로 셈 · 20 블록마다 캐시 덮음 ·
|
|
캐시 = `tmp/m02_hopyo/맞춤_<HP16>_<판16>.json`. 서버 단독 · 원본 읽기만.
|
|
손잡이(아래 상수)는 336 · K5 실험 값 그대로 — 한 파일로 다시 맞추지 않음.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import queue
|
|
import threading
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from common_util.common_util_json import atomic_write_json
|
|
from M02_MasterTemplete import M02_Hopyo_Basis as hopyo_basis
|
|
from M02_MasterTemplete import M02_Hopyo_Finger as finger
|
|
from M02_MasterTemplete import M02_Hopyo_Match as hopyo_match
|
|
from M02_MasterTemplete import M02_Hopyo_Ours as ours
|
|
from M02_MasterTemplete import M02_Hopyo_Store as hopyo_store
|
|
from M02_MasterTemplete.M02_Hopyo_Words import close, cond_score, cosine, jacc, nfkc
|
|
|
|
# ── 손잡이(한 곳) ──────────────────────────────────────────────────────
|
|
W = {
|
|
"인력": 3.0,
|
|
"장비": 2.0,
|
|
"자재": 1.0,
|
|
"율": 0.3,
|
|
"단위": 0.5,
|
|
"조건": 0.6,
|
|
"조항": 1.5,
|
|
"이름": 0.6,
|
|
}
|
|
BETA = 1.0
|
|
RB = (1, 0.6, 0.45, 0.35, 0.25)
|
|
POOL = 40
|
|
TOP_OPT = 5
|
|
CAP = 400
|
|
TOL = (0.005, "R", 0.03)
|
|
SURE = (0.03, 0.3)
|
|
MIXED_CUT = 0.35
|
|
COMBO = (0.85, 0.75, 0.6)
|
|
COVER = 0.6
|
|
SHARES = (0.1, 0.2, 0.3, 0.5, 0.7, 0.8, 0.9)
|
|
EVERY = 20 # 블록 몇 개마다 캐시 덮음
|
|
COUNT = {"개소", "개", "본", "ea", "조", "매", "대", "기", "공"}
|
|
HOUR = {"hr", "시간", "h"}
|
|
NO_NUMBER = "숫자 판정 없음"
|
|
_STATE: dict[str, Any] = {"판": None}
|
|
|
|
|
|
# ── 우리 지문 정리(점수용) ─────────────────────────────────────────────
|
|
def mat_words(name: Any) -> set[str]:
|
|
t = nfkc(name).lower().replace(".", "").replace(" ", "")
|
|
out = set(hopyo_match.words(name)) | ({"pvc"} if "pvc" in t else set())
|
|
return {w for w in out if len(w) >= 2}
|
|
|
|
|
|
def _prep(o: dict[str, Any], grams: dict[str, set]) -> dict[str, Any]:
|
|
"""기계경비 로직은 조종원 뺌 · 모두 0 인 변형 뺌 · 섞기 전 변형 덧 · 고를 수 있는 장비."""
|
|
vs = []
|
|
for v in o["변형"]:
|
|
L = {} if o["기계경비"] else {t: q for t, q in v["L"].items() if q > 0}
|
|
X = [(set(a), set(b), q) for a, b, q in v["X"] if q > 0]
|
|
M = {k: q for k, q in v["M"].items() if q > 0}
|
|
if not (L or X or M) and (v["L"] or v["X"] or v["M"]):
|
|
continue
|
|
vs.append({"in": v["in"], "L": L, "X": X, "M": M})
|
|
ri, rm = (v.get("mix") or {}).get("인력비율"), (v.get("mix") or {}).get("기계비율")
|
|
if (ri and 0 < ri < 100) or (rm and 0 < rm < 100): # 섞기 전 품(인력 · 기계 100%)
|
|
Lu = {t: q * 100 / ri for t, q in L.items()} if ri and 0 < ri < 100 else L
|
|
Xu = [(a, b, q * 100 / rm) for a, b, q in X] if rm and 0 < rm < 100 else X
|
|
vs.append({"in": {**v["in"], "_섞기전": "100%"}, "L": Lu, "X": Xu, "M": M})
|
|
if not vs: # 엔진이 못 돈 로직 — 직종 이름만(품 모름)
|
|
X0 = [(set(a), set(b), None) for a, b in o["장비"] if a]
|
|
vs = [{"in": None, "L": {t: None for t in o["직종"]}, "X": X0, "M": {}}]
|
|
opt = []
|
|
if o["기계경비"] and (o["장비고름"] or o["장비"]):
|
|
ts = set().union(*[set(a) for a, _ in o["장비고름"] + o["장비"]])
|
|
sp = set().union(*[set(b) for _, b in o["장비고름"]]) if o["장비고름"] else set()
|
|
opt = [(ts, sp)]
|
|
mats = set().union(*[mat_words(n) for n in o["자재"]], set())
|
|
for v in vs:
|
|
for n in v["M"]:
|
|
mats |= mat_words(n)
|
|
return {
|
|
"로직": o["로직"],
|
|
"단위": o["단위"],
|
|
"vs": vs,
|
|
"opt": opt,
|
|
"matw": mats,
|
|
"pctv": {round(p, 3) for _, p in o["덧줄율"]},
|
|
"nameg": grams.get(o["로직"], set()),
|
|
"nc": {k: set(v) for k, v in o["이름조건"].items()},
|
|
"oc": {k: set(v) for k, v in o["고름조건"].items()},
|
|
}
|
|
|
|
|
|
def prepared() -> list[dict[str, Any]] | None:
|
|
"""점수용 우리 지문 — 우리 지문 준비 전 None."""
|
|
table = ours.table()
|
|
if table is None:
|
|
return None
|
|
if _STATE["판"] != id(table):
|
|
grams = {x["로직"]: x["grams"] for x in hopyo_match.targets()["로직"]}
|
|
_STATE.update(판=id(table), 표=[_prep(o, grams) for o in table])
|
|
return _STATE["표"]
|
|
|
|
|
|
# ── 엑셀 지문 정리 · 점수 ──────────────────────────────────────────────
|
|
def ex_prep(efp: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
**efp,
|
|
"E": {t: o["품"] for t, o in efp["인력"].items() if o["품"]},
|
|
"Eb": {t: o["밑품"] for t, o in efp["인력"].items() if o["밑품"]},
|
|
"matl": [w for w in (mat_words(m["이름"]) for m in efp["자재"]) if w],
|
|
"pctv": {round(v, 3) for _, v in efp["율"]},
|
|
"nameg": hopyo_match._grams(hopyo_match.tidy(efp["이름"])),
|
|
"theirs": [list(x) for x in efp["출처"]],
|
|
"조건s": {k: set(v) for k, v in efp["조건"].items()},
|
|
"이름조건s": {k: set(v) for k, v in efp["이름조건"].items()},
|
|
}
|
|
|
|
|
|
def unit_fit(a: Any, b: Any) -> bool:
|
|
"""단위 맞음 + 별칭(개소 = 개 = 본 · hr = 시간) — 품 크기 견줌에만."""
|
|
if hopyo_match.fits(a, b):
|
|
return True
|
|
na = "".join(str(a or "").split()).replace("/", "").lower()
|
|
for nb in ("".join(x.split()).replace("/", "").lower() for x in str(b or "").split("·")):
|
|
if na == nb or (na in COUNT and nb in COUNT) or (na in HOUR and nb in HOUR):
|
|
return True
|
|
return False
|
|
|
|
|
|
def lab_sim(E: dict, Eb: dict, ov: dict, unit_ok: bool) -> tuple[float, float | None]:
|
|
if not ov:
|
|
return 0.0, None
|
|
keys = set(E) | set(ov)
|
|
w = lambda t: 0.5 if t == "보통인부" else 1.0 # noqa: E731
|
|
st = sum(w(t) for t in set(E) & set(ov)) / sum(w(t) for t in keys)
|
|
if any(q is None for q in ov.values()): # 품 모름 — 직종 겹침만
|
|
return 0.6 * st, None
|
|
shape = max(cosine(E, ov), cosine(Eb, ov))
|
|
if not unit_ok:
|
|
return 0.35 * st + 0.25 * shape, None
|
|
mag = sum(max(close(E.get(t), ov.get(t)), close(Eb.get(t), ov.get(t))) for t in keys) / len(
|
|
keys
|
|
)
|
|
return 0.35 * st + 0.25 * shape + 0.40 * mag, mag
|
|
|
|
|
|
def mach_sim(EM: list, OX: list, opt: list, unit_ok: bool) -> float:
|
|
if not OX and not opt:
|
|
return 0.0
|
|
got = []
|
|
for m in EM:
|
|
mt, ms = set(m["갈래"]), set(m["규격"])
|
|
best = 0.0
|
|
for ot, osp, oh in OX:
|
|
j = jacc(mt, ot)
|
|
if not j:
|
|
continue
|
|
sp = (1.0 if ms & osp else 0.0) if ms and osp else 0.5
|
|
hc = max(close(m.get("시간"), oh), close(m.get("밑시간"), oh)) if unit_ok and oh else 0
|
|
best = max(best, 0.6 * j + 0.2 * sp + 0.2 * hc)
|
|
for ot, osp in opt:
|
|
if mt and mt <= ot:
|
|
sp = (1.0 if ms & osp else 0.0) if ms and osp else 0.5
|
|
best = max(best, 0.6 + 0.2 * sp + 0.2 * (1.0 if unit_ok else 0.0))
|
|
got.append(best)
|
|
et = set().union(*[set(m["갈래"]) for m in EM])
|
|
ot_all = set().union(*[t for t, _, _ in OX]) if OX else set()
|
|
extra = len(ot_all - et) / max(1, len(ot_all | et)) if ot_all else 0.0
|
|
return sum(got) / len(got) * (1 - 0.5 * extra)
|
|
|
|
|
|
def score_pair(e: dict, o: dict, table: dict | None = None) -> tuple[float, dict, dict | None]:
|
|
"""(점수, 채널, 고른 변형) — 로직마다 가장 닮은 변형 · `table` = `Basis.ours()`(한 번 읽어
|
|
넘김)."""
|
|
unit_ok = unit_fit(e["단위"], o["단위"])
|
|
ch = {"단위": 1.0 if unit_ok else 0.0}
|
|
g1, g2 = e["nameg"], o["nameg"]
|
|
ch["이름"] = 2 * len(g1 & g2) / (len(g1) + len(g2)) if g1 and g2 else 0.0
|
|
our = (table if table is not None else hopyo_basis.ours()).get(o["로직"])
|
|
c = hopyo_match.clause(e["theirs"], our)[0] if (our and e["theirs"]) else 0.0
|
|
ch["조항"] = c * (0.5 if e["옛판"] else 1.0)
|
|
hit, _ = cond_score(e["조건s"], o["nc"], o["oc"])
|
|
_, clash = cond_score(e["이름조건s"], o["nc"], {})
|
|
ch["조건"] = hit - clash
|
|
if e["matl"]:
|
|
near = lambda ws: any(a in b or b in a for a in ws for b in o["matw"]) # noqa: E731
|
|
ch["자재"] = sum(1 for ws in e["matl"] if near(ws)) / len(e["matl"])
|
|
ch["율"] = 1.0 if (e["pctv"] and o["pctv"] and e["pctv"] & o["pctv"]) else 0.0
|
|
base = sum(W.get(k, 0) * v for k, v in ch.items())
|
|
best, bestv, bestch = None, None, {}
|
|
for v in o["vs"]:
|
|
s, vch = base, {}
|
|
if e["E"]:
|
|
ls, mag = lab_sim(e["E"], e["Eb"], v["L"], unit_ok)
|
|
vch["인력"], vch["품"] = ls, mag
|
|
s += W["인력"] * ls
|
|
elif v["L"]:
|
|
s -= 0.6
|
|
if e["장비"]:
|
|
vch["장비"] = mach_sim(e["장비"], v["X"], o["opt"], unit_ok)
|
|
s += W["장비"] * vch["장비"]
|
|
elif v["X"] or o["opt"]:
|
|
s -= 0.4
|
|
if best is None or s > best:
|
|
best, bestv, bestch = s, v, vch
|
|
return best, {**ch, **bestch}, bestv
|
|
|
|
|
|
def rank(e: dict, n: int = POOL) -> list[tuple]:
|
|
"""[(점수, 로직, 채널, 변형)] — 위 n."""
|
|
table = hopyo_basis.ours()
|
|
out = [(*_swap(score_pair(e, o, table), o["로직"]),) for o in prepared() or []]
|
|
out.sort(key=lambda x: -x[0])
|
|
return out[:n]
|
|
|
|
|
|
def _swap(got: tuple, code: str) -> tuple:
|
|
return got[0], code, got[1], got[2]
|
|
|
|
|
|
def sure(e: dict, pool: list[tuple]) -> bool:
|
|
"""딱 맞음 — 1위 직종 · 품(3% 안)이 다 맞음(보통인부 하나뿐 줄 아님) 또는 장비 갈래 · 시간 맞음
|
|
·
|
|
2위와 점수 차 SURE[1] 넘음."""
|
|
if not pool:
|
|
return False
|
|
s1, _, ch, _ = pool[0]
|
|
gap = s1 - pool[1][0] if len(pool) > 1 else 9
|
|
if e["E"]:
|
|
rich = len(e["E"]) >= 2 or set(e["E"]) != {"보통인부"} or bool(e["장비"])
|
|
ok = rich and (ch.get("품") or 0) >= 1 - SURE[0] and ch.get("인력", 0) >= 1 - SURE[0]
|
|
else:
|
|
ok = bool(e["장비"]) and ch.get("장비", 0) >= 0.95
|
|
return ok and gap > SURE[1]
|
|
|
|
|
|
def fuse(e: dict, pool: list[tuple], border: list[str]) -> list[str]:
|
|
"""지문 점수 + BETA × 이름 · 근거 차례 덤 → 위 다섯."""
|
|
bonus = {k: RB[i] for i, k in enumerate(border[:5])}
|
|
cand = {k: s for s, k, _, _ in pool}
|
|
by = {o["로직"]: o for o in prepared() or []}
|
|
table = hopyo_basis.ours()
|
|
for k in bonus:
|
|
if k not in cand and k in by:
|
|
cand[k] = score_pair(e, by[k], table)[0]
|
|
return sorted(cand, key=lambda k: -(cand[k] + BETA * bonus.get(k, 0)))[:TOP_OPT]
|
|
|
|
|
|
def single(e: dict) -> bool:
|
|
"""숫자 하나뿐 줄 — 직종 하나 또는 장비 시간 하나(기계 1 시간 줄 뺌) · 같은 숫자 로직이 많아
|
|
우연 맞음."""
|
|
E = [k for k, x in e["인력"].items() if x.get("품")]
|
|
X = [m for m in e["장비"] if m.get("시간")]
|
|
return len(E) + len(X) == 1 and not (X and abs(X[0]["시간"] - 1.0) < 1e-9)
|
|
|
|
|
|
def has_numbers(efp: dict) -> bool:
|
|
return bool(efp["인력"] or any(m.get("시간") for m in efp["장비"]) or efp["자재"])
|
|
|
|
|
|
# ── 블록 셈 ────────────────────────────────────────────────────────────
|
|
def compute(block: dict | None, line: dict, got: dict | None = None) -> dict[str, Any]:
|
|
"""블록 + 줄 하나 → 맞춤셈 한 칸(설계 3-2 꼴)."""
|
|
from M02_MasterTemplete import M02_Hopyo_Verdict as verdict # 판정이 이 파일을 부름
|
|
|
|
return verdict.block(block, line, got or hopyo_match.targets())
|
|
|
|
|
|
# ── 뒤 셈 · 캐시 ───────────────────────────────────────────────────────
|
|
_JOBS: queue.Queue = queue.Queue()
|
|
_PENDING: set[str] = set()
|
|
_WORKER: dict[str, Any] = {"스레드": None}
|
|
_LOCK = threading.Lock()
|
|
|
|
|
|
def _hp_path(folder: Path, key: str) -> Path:
|
|
return hopyo_store._path(folder, key)
|
|
|
|
|
|
def cache_path(folder: Path, key: str) -> Path:
|
|
hp = hashlib.sha256(_hp_path(folder, key).read_bytes()).hexdigest()[:16]
|
|
return ours.folder() / f"맞춤_{hp}_{ours.version()}.json"
|
|
|
|
|
|
def units(doc: dict[str, Any]) -> list[tuple[str, dict | None, dict]]:
|
|
"""셈 단위 [(열쇠, 블록, 줄)] — 블록 + 줄 열쇠 한 번 · 블록 없는 기계 줄 · 직접 줄 호표."""
|
|
blocks = {int(b["번"]): b for b in doc.get("단산") or [] if b.get("번") is not None}
|
|
out, seen = [], set()
|
|
for h in doc.get("호표") or []:
|
|
for line in h.get("줄") or []:
|
|
ref = line.get("참조") or {}
|
|
kind = finger.kind_of(line)
|
|
if ref.get("종류") == "단산":
|
|
key = unit_key(line)
|
|
if key not in seen:
|
|
seen.add(key)
|
|
out.append((key, blocks.get(int(ref.get("번") or 0)), line))
|
|
elif kind == "기계":
|
|
key = unit_key(line)
|
|
if key not in seen:
|
|
seen.add(key)
|
|
out.append((key, None, line))
|
|
if (fp := finger.direct(h)) is not None and f"호표:{h['번']}" not in seen:
|
|
seen.add(f"호표:{h['번']}")
|
|
out.append(
|
|
(f"호표:{h['번']}", {"_지문": fp}, {k: h.get(k) for k in ("이름", "규격", "단위")})
|
|
)
|
|
return out
|
|
|
|
|
|
def unit_key(line: dict[str, Any]) -> str:
|
|
ref = line.get("참조") or {}
|
|
if ref.get("종류") == "단산":
|
|
return f"단산:{ref.get('번')}|{hopyo_match.line_key(line)}"
|
|
return f"줄:{hopyo_match.line_key(line)}"
|
|
|
|
|
|
def _read(path: Path) -> dict[str, Any] | None:
|
|
try:
|
|
doc = json.loads(path.read_bytes())
|
|
except (OSError, ValueError):
|
|
return None
|
|
return doc if isinstance(doc, dict) else None
|
|
|
|
|
|
def result(folder: Path, key: str) -> dict[str, Any] | None:
|
|
"""맞춤셈 캐시(이 HP 파일 · 지금 판) — 없으면 None."""
|
|
try:
|
|
return _read(cache_path(folder, key))
|
|
except OSError:
|
|
return None
|
|
|
|
|
|
def status(folder: Path, key: str) -> dict[str, Any]:
|
|
"""{끝, 한, 모두, 까닭?}."""
|
|
got = result(folder, key)
|
|
if got and got.get("끝"):
|
|
return {"끝": True, "한": got["셈"][0], "모두": got["셈"][1]}
|
|
why = None if ours.ready() else "우리 로직 준비 중"
|
|
done, total = got["셈"] if got else (0, None)
|
|
return {"끝": False, "한": done, "모두": total, **({"까닭": why} if why else {})}
|
|
|
|
|
|
def run_job(folder: Path, key: str) -> dict[str, Any]:
|
|
"""HP 파일 하나를 끝까지 셈(뒤 스레드 · 시험이 바로 부름) → 캐시."""
|
|
if ours.table() is None:
|
|
thread = ours.warm(build_missing=True)
|
|
if thread is not None:
|
|
thread.join()
|
|
path = cache_path(folder, key)
|
|
doc = hopyo_store.read(folder, key)
|
|
todo = units(doc)
|
|
got = _read(path) or {}
|
|
done = got.get("블록") if got.get("판") == ours.version() else None
|
|
out = {"판": ours.version(), "지문판": ours.FP_VERSION, "셈": [0, len(todo)], "끝": False}
|
|
out["블록"] = dict(done or {})
|
|
targets = hopyo_match.targets()
|
|
for at, (unit, block, line) in enumerate(todo, 1):
|
|
if unit not in out["블록"]:
|
|
out["블록"][unit] = compute(block, line, targets)
|
|
out["셈"] = [at, len(todo)]
|
|
if at % EVERY == 0:
|
|
atomic_write_json(path, out)
|
|
out["끝"] = True
|
|
atomic_write_json(path, out)
|
|
return out
|
|
|
|
|
|
def _work() -> None:
|
|
while True:
|
|
folder, key = _JOBS.get()
|
|
try:
|
|
run_job(folder, key)
|
|
except Exception: # noqa: BLE001 — 한 파일이 깨져도 일감 줄은 계속(보기에서 「셈 중」 그대로)
|
|
pass
|
|
finally:
|
|
with _LOCK:
|
|
_PENDING.discard(f"{folder}|{key}")
|
|
|
|
|
|
def ensure(folder: Path, key: str) -> dict[str, Any]:
|
|
"""캐시 있으면 끝 · 없으면 일감 줄에 넣음(같은 파일은 한 번) → {끝, 한, 모두, 까닭?}."""
|
|
now = status(folder, key)
|
|
if now["끝"]:
|
|
return now
|
|
with _LOCK:
|
|
tag = f"{folder}|{key}"
|
|
if tag not in _PENDING:
|
|
_PENDING.add(tag)
|
|
_JOBS.put((Path(folder), key))
|
|
if _WORKER["스레드"] is None or not _WORKER["스레드"].is_alive():
|
|
_WORKER["스레드"] = threading.Thread(target=_work, name="m02-hopyo-fit", daemon=True)
|
|
_WORKER["스레드"].start()
|
|
return now
|