ahnyongjooandClaude Sonnet 5.5
964a841de1
- Code\rebar_pipeline, review_pipeline, pier-program을 정션에서 실제 폴더로 전환 - Code\rhino_mcp 및 CLAUDE_rhino배경.md 추가, .mcp.json은 PYTHONPATH로 새 위치 사용 - pier-program 내부 .git 제거(이력은 backup\pier-program_20261028.bundle에 보관) - worktree 가이드 반영(CLAUDE.local.md, HANDOFF.md) - D:\Rhino-MCP 경로 참조를 새 위치로 치환 Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
910 lines
46 KiB
Python
910 lines
46 KiB
Python
"""T-pier modelling app (semi-automatic MVP): local Flask UI over the engine.
|
|
|
|
Flow per pier: upload drawings -> search texts / review table candidates -> edit and confirm sections ->
|
|
run gates G1..G5 one by one (each needs its sections confirmed and the previous gate approved) ->
|
|
approve / reject each gate -> download the .3dm.
|
|
Local only: binds 127.0.0.1 and has no authentication (do not expose it on a network).
|
|
usage: python web/app.py then open http://127.0.0.1:5000
|
|
"""
|
|
import json
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import threading
|
|
import time
|
|
import traceback
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
from flask import Flask, abort, flash, redirect, render_template, request, send_file, url_for
|
|
|
|
APP = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(APP))
|
|
sys.path.insert(0, str(APP / "web"))
|
|
from engine import anchors, geomread, recommend, reader, render, report_files, review as reviewlib, variability, verify # noqa: E402
|
|
from engine.gates import run_batch_session, run_gate # noqa: E402
|
|
from engine.pipeline import GATES, gate_ids # noqa: E402
|
|
import store # noqa: E402
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = "local-only" # session flash messages only; the app is bound to 127.0.0.1
|
|
RHINO_LOCK = threading.Lock() # one Rhino job at a time
|
|
ACTIVE = {} # job id -> thread of this server process
|
|
BATCH = {} # pier name -> batch thread of this server process
|
|
AUTO_NOTE = "일괄 실행 자동 승인"
|
|
GATE_NAMES = {g: n for g, n, _ in GATES}
|
|
|
|
|
|
def pier_or_404(name):
|
|
try:
|
|
return store.Pier(name)
|
|
except (FileNotFoundError, ValueError):
|
|
abort(404)
|
|
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return render_template("index.html", piers=[store.Pier(n) for n in store.piers()], types=store.PIER_TYPES + ["기타"],
|
|
templates=store.templates(), tpl=store.TPL)
|
|
|
|
|
|
@app.post("/pier/new")
|
|
def pier_new():
|
|
try:
|
|
p = store.create_pier(request.form["name"], request.form["type"], request.form["template"])
|
|
except (ValueError, FileNotFoundError) as e:
|
|
flash(str(e), "fail")
|
|
return redirect(url_for("index"))
|
|
flash(f"교각 '{p.name}'을(를) 만들었습니다. 템플릿 값이 들어 있으니 모든 묶음을 확인·수정한 뒤 확정하세요.", "ok")
|
|
return redirect(url_for("pier", name=p.name))
|
|
|
|
|
|
STAGE = "_추천" # data/_추천/<id>: drawings uploaded before the pier exists (S02)
|
|
|
|
|
|
def stage_dir(sid):
|
|
if not sid.isalnum():
|
|
abort(404)
|
|
return store.DATA / STAGE / sid
|
|
|
|
|
|
@app.post("/recommend")
|
|
def recommend_upload():
|
|
"""S02: upload drawings first, estimate the fit answers from the texts, rank the protected templates."""
|
|
files = [f for f in request.files.getlist("dwg") if f.filename]
|
|
if not files or any(not f.filename.lower().endswith(".dwg") for f in files):
|
|
flash("DWG 파일(.dwg)만 올릴 수 있습니다.", "fail")
|
|
return redirect(url_for("index"))
|
|
sid = uuid.uuid4().hex[:12]
|
|
d = stage_dir(sid)
|
|
(d / "dwg").mkdir(parents=True)
|
|
paths = []
|
|
for f in sorted(files, key=lambda f: f.filename):
|
|
dst = d / "dwg" / store.safe_name(Path(f.filename).name)
|
|
f.save(dst)
|
|
paths.append(dst)
|
|
try:
|
|
audit = reader.convert(paths, d / "dxf")
|
|
except Exception as e:
|
|
flash(f"DWG 변환 실패: {e}", "fail")
|
|
return redirect(url_for("index"))
|
|
labels = [f"D{i}" for i in range(1, len(audit) + 1)]
|
|
for a, lab in zip(audit, labels):
|
|
a["label"] = lab
|
|
store.wjson(d / "audit.json", audit)
|
|
texts = reader.build_index([d / "dxf" / a["dxf"] for a in audit], labels, d / "texts.json")
|
|
return recommend_show(sid, texts)
|
|
|
|
|
|
def recommend_show(sid, texts):
|
|
rows = next((t["rows"] for t in reader.table_candidates(texts) if t["key"] == "elevation"), [])
|
|
est = recommend.estimate(texts, rows)
|
|
tpls = store.templates()
|
|
profiles = {t["name"]: recommend.profile(store.rjson(store.template_dir(t["name"]) / "params.json")) for t in tpls}
|
|
return render_template("recommend.html", profiles=profiles, sid=sid, est=est, questions=recommend.QUESTIONS, rank=recommend.rank(est, profiles),
|
|
tpl=store.TPL, ntexts=len(texts), nfiles=len(store.rjson(stage_dir(sid) / "audit.json", [])))
|
|
|
|
|
|
@app.post("/recommend/<sid>/create")
|
|
def recommend_create(sid):
|
|
"""Create the pier from the chosen template and move the staged drawings into it; the answers are kept in pier.json."""
|
|
d = stage_dir(sid)
|
|
if not (d / "texts.json").exists():
|
|
abort(404)
|
|
answers = {q: request.form.get(f"ans_{q}", "").strip() for q, _ in recommend.QUESTIONS}
|
|
answers.update({k: request.form.get(f"ans_{k}", "").strip() for k in ("rebar_shape", "rebar_method")})
|
|
try:
|
|
p = store.create_pier(request.form["name"], request.form["type"], request.form["template"])
|
|
except (ValueError, FileNotFoundError) as e:
|
|
flash(str(e), "fail")
|
|
return redirect(url_for("index"))
|
|
p.drawings.mkdir(parents=True, exist_ok=True)
|
|
for item in d.iterdir():
|
|
shutil.move(str(item), str(p.drawings / item.name))
|
|
d.rmdir()
|
|
answers.update({k: request.form.get(f"ans_{k}", "").strip() for k in ("skew_deg", "bearing_rot_deg", "footing_rot_deg")})
|
|
tname = request.form["template"]
|
|
bad = (recommend.mismatches(answers, recommend.profile(store.rjson(store.template_dir(tname[len(store.TPL):]) / "params.json")))
|
|
if tname.startswith(store.TPL) else [])
|
|
store.wjson(p.dir / "pier.json", {**p.info, "fit_answers": answers, "fit_mismatch": bad})
|
|
flash(f"교각 '{p.name}'을(를) 만들었습니다. 도면 {len(p.texts())}개 문자 색인을 옮겼습니다. 모든 묶음을 확인·수정한 뒤 확정하세요.", "ok")
|
|
return redirect(url_for("pier", name=p.name))
|
|
|
|
|
|
@app.post("/pier/<name>/delete")
|
|
def pier_delete(name):
|
|
"""Delete = move to data/_휴지통 (recoverable). The typed name must match; refused while a job runs."""
|
|
p = pier_or_404(name)
|
|
if request.form.get("confirm_name", "").strip() != p.name:
|
|
flash(f"삭제 안 됨: 확인란에 교각 이름 '{p.name}'을(를) 정확히 입력하세요.", "fail")
|
|
elif anything_running(p):
|
|
flash(f"삭제 안 됨: '{p.name}'은(는) 게이트를 실행 중입니다. 끝난 뒤 다시 시도하세요.", "fail")
|
|
else:
|
|
try:
|
|
dst = store.delete_pier(p.name)
|
|
flash(f"'{p.name}'을(를) 삭제했습니다(휴지통으로 이동: {dst.relative_to(store.DATA)}). "
|
|
f"되돌리려면 이 폴더를 data 폴더로 옮기고 이름을 원래대로 바꾸세요.", "ok")
|
|
except OSError as e:
|
|
flash(f"삭제 실패: {e} (파일이 열려 있으면 닫고 다시 시도하세요)", "fail")
|
|
return redirect(url_for("index"))
|
|
|
|
|
|
@app.route("/pier/<name>")
|
|
def pier(name):
|
|
p = pier_or_404(name)
|
|
secs = []
|
|
req = required(p)
|
|
p.run_autocheck() # stage B: skipped when nothing changed
|
|
for s in p.sections():
|
|
secs.append(dict(s, **p.section_state(s["file"], s["key"]), meta=p.section_meta(s["file"], s["key"]),
|
|
required=(s["file"], s["key"]) in req, chk=p.section_check(s["file"], s["key"])))
|
|
rank = {"b": 0, "n": 1, "w": 2, "g": 3, "x": 4, "?": 1} # worst app check first, required bundles before the others
|
|
secs.sort(key=lambda s: (0 if s["required"] else 1, rank.get(s["chk"]["st"], 1), s["file"], s["key"]))
|
|
origin_same = [s for s in secs if s["required"] and s["meta"]["origin"]] # stage A 2026-10-16
|
|
chain_bad = [c for c in anchors.chains(p.texts()) if not c["ok"]] # stage C 2026-10-19: R2 check of the drawing's own chains
|
|
cnt = {k: 0 for k in "gwbnx"} # stage B 2026-10-16
|
|
for s in secs:
|
|
if s["required"]:
|
|
cnt[s["chk"]["st"] if s["chk"]["st"] in cnt else "n"] += 1
|
|
human = sum(1 for s in secs if s["required"] and s["confirmed"])
|
|
nmatch = len(matched_targets(p))
|
|
gates = [dict(id=g, name=GATE_NAMES[g], **p.gate_status(g)) for g in p.scope()]
|
|
return render_template("pier.html", p=p, sections=secs, gates=gates, need=unconfirmed(p), origin_same=origin_same,
|
|
has_origin=bool(p.info.get("origin")), grades=variability.GRADES, cnt=cnt, human=human, nreq=len(req),
|
|
nmatch=nmatch, ST=verify.ST_NAME, chain_bad=chain_bad,
|
|
todo=[g["id"] for g in gates if not g["approved"]], running=anything_running(p),
|
|
out_of_scope=[g for g in gate_ids() if g not in p.scope()],
|
|
batch=batch_state(p),
|
|
audit=store.rjson(p.drawings / "audit.json", []), ntexts=len(p.texts()))
|
|
|
|
|
|
# ---------------- drawings
|
|
@app.post("/pier/<name>/upload")
|
|
def upload(name):
|
|
p = pier_or_404(name)
|
|
files = [f for f in request.files.getlist("dwg") if f.filename]
|
|
if not files or any(not f.filename.lower().endswith(".dwg") for f in files):
|
|
flash("DWG 파일(.dwg)만 올릴 수 있습니다.", "fail")
|
|
return redirect(url_for("pier", name=name))
|
|
src = p.drawings / "dwg"
|
|
src.mkdir(parents=True, exist_ok=True)
|
|
paths = []
|
|
for f in sorted(files, key=lambda f: f.filename):
|
|
dst = src / store.safe_name(Path(f.filename).name)
|
|
f.save(dst)
|
|
paths.append(dst)
|
|
try:
|
|
audit = reader.convert(paths, p.drawings / "dxf")
|
|
except Exception as e:
|
|
flash(f"DWG 변환 실패: {e}", "fail")
|
|
return redirect(url_for("pier", name=name))
|
|
labels = [f"D{i}" for i in range(1, len(audit) + 1)]
|
|
for a, lab in zip(audit, labels):
|
|
a["label"] = lab
|
|
store.wjson(p.drawings / "audit.json", audit)
|
|
texts = reader.build_index([p.drawings / "dxf" / a["dxf"] for a in audit], labels, p.drawings / "texts.json")
|
|
flash(f"도면 {len(audit)}장 변환, 문자 {len(texts)}개 색인", "ok")
|
|
return redirect(url_for("pier", name=name))
|
|
|
|
|
|
@app.route("/pier/<name>/search")
|
|
def search(name):
|
|
p = pier_or_404(name)
|
|
q, rx = request.args.get("q", ""), bool(request.args.get("regex"))
|
|
hits, err = [], None
|
|
if q:
|
|
try:
|
|
hits = reader.search(p.texts(), q, regex=rx)
|
|
except Exception as e:
|
|
err = f"검색식 오류: {e}"
|
|
audit = store.rjson(p.drawings / "audit.json", [])
|
|
hd, hdwg, ent = request.args.get("h", "").strip(), request.args.get("dwg", ""), None
|
|
if hd: # handle lookup of a drawing object (OPEN #19 A6)
|
|
a = next((a for a in audit if a["label"] == hdwg), None)
|
|
ent = reader.entity_info(p.drawings / "dxf" / a["dxf"], hd) if a else None
|
|
if ent is None:
|
|
err = f"{hdwg} 도면에 핸들 '{hd}' 객체가 없습니다."
|
|
return render_template("search.html", p=p, q=q, regex=rx, hits=hits, err=err, audit=audit,
|
|
h=hd, hdwg=hdwg, ent=ent, dumps=lambda v: json.dumps(v, ensure_ascii=False))
|
|
|
|
|
|
@app.route("/pier/<name>/candidates")
|
|
def candidates(name):
|
|
p = pier_or_404(name)
|
|
texts = p.texts()
|
|
tables = reader.table_candidates(texts) if texts else []
|
|
rows = reader.proposals(reader.field_candidates(tables), p.load("params"), p.load("expected")) if tables else []
|
|
arows, amiss, ameta = p.anchor_rows()
|
|
rank = {"A": 0, "B": 1, "C": 2}
|
|
arows.sort(key=lambda r: (r["same"], rank[r["grade"]], r["field"])) # differing candidates first, best grade first
|
|
chains = anchors.chains(texts)
|
|
return render_template("candidates.html", p=p, tables=tables, rows=rows, els=reader.el_list(texts),
|
|
dumps=lambda v: json.dumps(v, ensure_ascii=False), arows=arows, amiss=amiss, ameta=ameta,
|
|
chains=chains, chain_bad=[c for c in chains if not c["ok"]],
|
|
tpl_anchors=[t["name"] for t in store.templates() if (store.template_dir(t["name"]) / "anchors.json").exists()])
|
|
|
|
|
|
@app.post("/pier/<name>/anchors/apply")
|
|
def anchors_apply(name):
|
|
"""Apply selected out-of-table candidates. Rules: the same candidate must not go into params and expected together (P7), a grade C candidate
|
|
needs a reason, only differing rows are applied; the change is logged and the changed bundles need a new confirmation."""
|
|
p = pier_or_404(name)
|
|
rows, _m, _meta = p.anchor_rows()
|
|
chosen = set(request.form.getlist("field"))
|
|
sel = [r for r in rows if r["field"] in chosen and not r["same"] and not r["self_source"]]
|
|
why = request.form.get("why", "").strip()
|
|
key = lambda r: (r["text"], r["x"], r["y"], r["value"])
|
|
both = {key(r) for r in sel if r["file"] == "params"} & {key(r) for r in sel if r["file"] == "expected"}
|
|
done = store.rjson(p.dir / "anchor_log.json", []) # applied earlier: (file, text, x, y)
|
|
prior = {(f["file"], f["text"], f["x"], f["y"]) for e in done for f in e["fields"] if "file" in f}
|
|
both |= {key(r) for r in sel if ("expected" if r["file"] == "params" else "params", r["text"], r["x"], r["y"]) in prior}
|
|
if not sel:
|
|
flash("반영할 후보를 고르세요(이미 같은 값은 반영하지 않습니다).", "fail")
|
|
elif both:
|
|
flash("같은 도면 문자에서 나온 후보를 params와 expected에 함께 반영할 수 없습니다. 두 값이 같은 근거에서 나오면 검사가 무의미해집니다. 한쪽만 고르세요.", "fail")
|
|
elif any(r["grade"] == "C" for r in sel) and not why:
|
|
flash("C등급 후보는 반영 사유를 입력해야 합니다.", "fail")
|
|
else:
|
|
docs = {"params": p.load("params"), "expected": p.load("expected")}
|
|
for r in sel:
|
|
store.set_field(docs, r["field"], r["value"])
|
|
p.save("params", docs["params"])
|
|
p.save("expected", docs["expected"])
|
|
log = store.rjson(p.dir / "anchor_log.json", [])
|
|
log.append({"at": store.now(), "fields": [{"field": r["field"], "file": r["file"], "from": r["current"], "to": r["value"], "grade": r["grade"],
|
|
"text": r["text"], "x": r["x"], "y": r["y"], "pos": f"{r['dwg']} ({r['x']:.0f}, {r['y']:.0f})"}
|
|
for r in sel], "why": why})
|
|
store.wjson(p.dir / "anchor_log.json", log)
|
|
flash(f"표 밖 치수 후보 {len(sel)}건을 반영했습니다. 바뀐 묶음은 다시 확정해야 합니다.", "ok")
|
|
return redirect(url_for("candidates", name=name))
|
|
|
|
|
|
@app.post("/pier/<name>/anchors/load")
|
|
def anchors_load(name):
|
|
"""Copy the anchors of a protected template into a pier that has none (the person says which series the drawing belongs to)."""
|
|
p = pier_or_404(name)
|
|
t = request.form.get("template", "")
|
|
try:
|
|
src = store.template_dir(t) / "anchors.json"
|
|
if not src.exists():
|
|
raise FileNotFoundError(t)
|
|
except (FileNotFoundError, ValueError):
|
|
flash("앵커가 있는 템플릿을 고르세요.", "fail")
|
|
return redirect(url_for("candidates", name=name))
|
|
if (p.dir / "anchors.json").exists():
|
|
flash("이 교각에는 이미 앵커가 있습니다.", "fail")
|
|
else:
|
|
shutil.copy2(src, p.dir / "anchors.json")
|
|
flash(f"템플릿 '{t}'의 앵커를 불러왔습니다. 앱 확인이 새 도면 기준으로 다시 실행됩니다.", "ok")
|
|
return redirect(url_for("candidates", name=name))
|
|
|
|
|
|
@app.route("/pier/<name>/view/<dwg>")
|
|
def view_drawing(name, dwg):
|
|
"""PNG of the drawing region around (x, y) so a candidate can be checked by eye."""
|
|
p = pier_or_404(name)
|
|
try:
|
|
x, y = float(request.args["x"]), float(request.args["y"])
|
|
half = min(max(float(request.args.get("half", 3000)), 500.0), 20000.0)
|
|
except (KeyError, ValueError):
|
|
abort(400)
|
|
ent = next((a for a in store.rjson(p.drawings / "audit.json", []) if a.get("label") == dwg), None)
|
|
dxf = p.drawings / "dxf" / ent["dxf"] if ent else None
|
|
if not dxf or not dxf.exists():
|
|
abort(404)
|
|
cache = p.drawings / "view_cache"
|
|
cache.mkdir(exist_ok=True)
|
|
out = cache / f"{dwg}_{x:.0f}_{y:.0f}_{half:.0f}.png"
|
|
if not out.exists():
|
|
render.render(str(dxf), x, y, half, str(out))
|
|
return send_file(out, mimetype="image/png")
|
|
|
|
|
|
@app.post("/pier/<name>/candidates/apply")
|
|
def candidates_apply(name):
|
|
p = pier_or_404(name)
|
|
texts = p.texts()
|
|
rows = reader.proposals(reader.field_candidates(reader.table_candidates(texts)), p.load("params"), p.load("expected"))
|
|
chosen = set(request.form.getlist("row"))
|
|
sel = [r for i, r in enumerate(rows) if str(i) in chosen and not r["same"]]
|
|
P, E = p.load("params"), p.load("expected")
|
|
reader.apply_proposals(sel, P, E)
|
|
p.save("params", P)
|
|
p.save("expected", E)
|
|
flash(f"후보값 {len(sel)}건을 반영했습니다. 바뀐 묶음은 다시 확정해야 합니다.", "ok")
|
|
return redirect(url_for("candidates", name=name))
|
|
|
|
|
|
|
|
# ---------------- 도면에서 읽기 (improvement #34-1, 2026-10-20): engine/geomread.py
|
|
def _geomread(p):
|
|
"""-> (state, P', E', changes, notes, message). state: ok / no_texts / no_dxf / frame / error."""
|
|
texts = p.texts()
|
|
if not texts:
|
|
return "no_texts", None, None, [], [], "도면 문자 색인이 없습니다. 도면을 먼저 올리세요."
|
|
ref_dir = store.template_dir("운곡교P1형")
|
|
ref = store.rjson(ref_dir / "params.json")
|
|
if not ref:
|
|
return "error", None, None, [], [], "기준 템플릿(운곡교P1형)을 찾지 못했습니다."
|
|
cov = geomread.frame_ok(texts, store.rjson(ref_dir / "anchors.json"), anchors)
|
|
if cov is not None and cov < geomread.FRAME_MIN_COVERAGE:
|
|
return "frame", None, None, [], [], f"도면 틀 일치율이 {cov * 100:.0f} %입니다(70 % 미만). 운곡교 P1 도면 틀(P1~P4)이 아니라서 읽지 않습니다."
|
|
try:
|
|
dxf = geomread.dxf_of(p.dir, store.rjson(p.drawings / "audit.json", []))
|
|
P, E, changes, notes = geomread.read(p.load("params"), p.load("expected"), ref, texts, dxf, p.name)
|
|
except geomread.FrameError as e:
|
|
return "frame", None, None, [], [], str(e)
|
|
except Exception as e: # a reader must never break the page
|
|
return "error", None, None, [], [], f"읽는 중 오류: {type(e).__name__}: {e}"
|
|
own = [a.get("source_sha256") for a in store.rjson(p.drawings / "audit.json", []) if a.get("source_sha256")]
|
|
changes = changes + geomread.table_changes(P, E, texts) + geomread.extra_changes(P, E, texts, store.rjson(p.dir / "anchors.json"), own) # tables too: one screen brings every template value in line with the drawing; + anchor dimensions and planned FAIL proposals
|
|
return "ok", P, E, changes, notes, f"도면 틀 일치율 {(cov or 0) * 100:.0f} %" if cov is not None else ""
|
|
|
|
|
|
@app.route("/pier/<name>/geomread")
|
|
def geomread_page(name):
|
|
p = pier_or_404(name)
|
|
state, _P, _E, changes, notes, msg = _geomread(p)
|
|
groups = {}
|
|
for i, c in enumerate(changes):
|
|
groups.setdefault(c["group"], []).append((i, c))
|
|
brief = lambda v: v if isinstance(v, str) else json.dumps(v, ensure_ascii=False) # noqa: E731
|
|
return render_template("geomread.html", p=p, state=state, msg=msg, notes=notes, groups=groups, n=len(changes), brief=brief,
|
|
log=store.rjson(p.dir / "read_log.json", [])[-3:], running=anything_running(p))
|
|
|
|
|
|
@app.post("/pier/<name>/geomread/apply")
|
|
def geomread_apply(name):
|
|
"""Write the selected readings into params / expected. The page is recomputed here (indices refer to the same fresh list); a backup is taken first;
|
|
the changed bundles lose their confirmation by themselves (the hash changes) and the app check runs again."""
|
|
p = pier_or_404(name)
|
|
if anything_running(p):
|
|
flash("게이트 실행 중에는 값을 바꿀 수 없습니다.", "fail")
|
|
return redirect(url_for("geomread_page", name=name))
|
|
state, _P, _E, changes, notes, msg = _geomread(p)
|
|
if state != "ok":
|
|
flash(msg, "fail")
|
|
return redirect(url_for("geomread_page", name=name))
|
|
chosen = sorted({int(i) for i in request.form.getlist("row") if i.isdigit() and int(i) < len(changes)})
|
|
if not chosen:
|
|
flash("반영할 줄을 고르세요.", "fail")
|
|
return redirect(url_for("geomread_page", name=name))
|
|
P0, E0 = p.load("params"), p.load("expected")
|
|
bk = p.dir / "backup" / f"before_geomread_{time.strftime('%Y%m%d_%H%M%S')}"
|
|
bk.mkdir(parents=True)
|
|
for f in ("params.json", "expected.json", "confirm.json"):
|
|
if (p.dir / f).exists():
|
|
shutil.copy2(p.dir / f, bk / f)
|
|
P1, E1 = geomread.apply_selected(P0, E0, changes, chosen)
|
|
if any(changes[i]["path"] == "planned_fail" for i in chosen):
|
|
geomread.stamp_pending(E1, request.form.get("by", "").strip(), store.now()[:10])
|
|
p.save("params", P1)
|
|
p.save("expected", E1)
|
|
# a fixed (F) bundle / fixed field changed by a reading needs a logged reason (check C7): the reading itself is the reason
|
|
by_bundle = {}
|
|
for i in chosen:
|
|
c = changes[i]
|
|
key = c["path"].split(".")[0]
|
|
m = p.section_meta(c["file"], key)
|
|
rel = c["path"].split(".", 1)[1] if "." in c["path"] else ""
|
|
if m.get("grade") == "F" or (rel and rel in (m.get("fixed_fields") or [])):
|
|
by_bundle.setdefault((c["file"], key), []).append(c["path"])
|
|
for (fl, key), paths in by_bundle.items():
|
|
p.log_fixed_edit(fl, key, paths, "도면에서 읽기 반영(" + (request.form.get("by", "").strip() or "-") + "): " + "; ".join(paths[:3]))
|
|
log = store.rjson(p.dir / "read_log.json", [])
|
|
log.append({"at": store.now(), "by": request.form.get("by", "").strip(), "count": len(chosen), "backup": bk.name,
|
|
"fields": [f"{changes[i]['file']}:{changes[i]['path']}" for i in chosen]})
|
|
store.wjson(p.dir / "read_log.json", log)
|
|
p.run_autocheck(force=True)
|
|
flash(f"도면에서 읽은 값 {len(chosen)}건을 반영했습니다(백업 {bk.name}). 바뀐 묶음은 다시 확정해야 합니다.", "ok")
|
|
return redirect(url_for("geomread_page", name=name))
|
|
|
|
# ---------------- sections
|
|
@app.route("/pier/<name>/section/<which>/<key>", methods=["GET", "POST"])
|
|
def section(name, which, key):
|
|
p = pier_or_404(name)
|
|
if which not in ("params", "expected"):
|
|
abort(404)
|
|
d = p.load(which)
|
|
if key not in d:
|
|
abort(404)
|
|
text = json.dumps(d[key], ensure_ascii=False, indent=1)
|
|
meta = p.section_meta(which, key)
|
|
if request.method == "POST":
|
|
text = request.form["value"]
|
|
reason = request.form.get("reason", "").strip()
|
|
try:
|
|
new = store.expand_chains(json.loads(text))
|
|
if meta["grade"] == "V3":
|
|
raise ValueError("생성 변수(V3)는 화면에서 직접 고칠 수 없습니다. 판독 스크립트를 다시 실행해 목록을 만든 뒤 반영하세요")
|
|
fixed = ["(묶음 전체)"] if meta["grade"] == "F" and new != d[key] else variability.changed_fixed(d[key], new, meta["fixed_fields"])
|
|
if fixed and not reason:
|
|
raise ValueError(f"고정(F) 항목({', '.join(fixed)})을 바꾸면 같은 템플릿이 아닐 수 있습니다. 바꾸는 사유를 입력하세요")
|
|
p.set_section(which, key, new)
|
|
if fixed:
|
|
p.log_fixed_edit(which, key, fixed, reason)
|
|
flash("저장했습니다. 확정하려면 '확정'을 누르세요." + (" 고정 항목 변경 사유를 기록했습니다." if fixed else ""), "ok")
|
|
return redirect(url_for("section", name=name, which=which, key=key))
|
|
except (ValueError, json.JSONDecodeError) as e:
|
|
flash(f"저장 안 됨: {e}", "fail")
|
|
return render_template("section.html", p=p, which=which, key=key, text=text, st=p.section_state(which, key), meta=meta,
|
|
grades=variability.GRADES)
|
|
|
|
|
|
@app.post("/pier/<name>/section/<which>/<key>/confirm")
|
|
def section_confirm(name, which, key):
|
|
"""Confirm one bundle. A bundle the app check does not report as 일치 (주의 / 불일치 / 근거 없음) needs a reason (D3, stage B 2026-10-16)."""
|
|
p = pier_or_404(name)
|
|
by = request.form.get("by", "").strip()
|
|
why = request.form.get("why", "").strip()
|
|
p.run_autocheck()
|
|
st = p.section_check(which, key)["st"]
|
|
if not by:
|
|
flash("확정자 이름을 입력하세요.", "fail")
|
|
elif st in ("w", "b", "n") and not why:
|
|
flash(f"{which}.{key}는 앱 확인이 '{verify.ST_NAME[st]}'입니다. 사유를 입력해야 확정할 수 있습니다.", "fail")
|
|
else:
|
|
p.confirm(which, key, by, why, st)
|
|
flash(f"{which}.{key} 확정 ({by})" + (f" — 앱 확인 {verify.ST_NAME.get(st, st)}, 사유 기록" if why else ""), "ok")
|
|
return redirect(request.form.get("next") or url_for("pier", name=name))
|
|
|
|
|
|
# ---------------- app check (stage B 2026-10-16)
|
|
@app.post("/pier/<name>/autocheck")
|
|
def autocheck_run(name):
|
|
p = pier_or_404(name)
|
|
a = p.run_autocheck(force=True)
|
|
flash(f"앱 확인을 실행했습니다 ({len(a['sections'])}개 묶음). 앱은 확정하지 않습니다.", "ok")
|
|
return redirect(request.form.get("next") or url_for("pier", name=name))
|
|
|
|
|
|
@app.route("/pier/<name>/evidence/<which>/<key>")
|
|
def evidence(name, which, key):
|
|
p = pier_or_404(name)
|
|
if which not in ("params", "expected") or key not in p.load(which):
|
|
abort(404)
|
|
p.run_autocheck()
|
|
return render_template("evidence.html", p=p, which=which, key=key, chk=p.section_check(which, key), st=p.section_state(which, key),
|
|
meta=p.section_meta(which, key), grades=variability.GRADES, ST=verify.ST_NAME)
|
|
|
|
|
|
def matched_targets(p):
|
|
"""Required bundles the app check reports as 일치 and nobody has confirmed (current): the D4 bulk-confirm targets."""
|
|
p.run_autocheck()
|
|
out = []
|
|
for f, k in sorted(required(p)):
|
|
c, s = p.section_check(f, k), p.section_state(f, k)
|
|
if c["st"] == "g" and not c["stale"] and not s["confirmed"]:
|
|
out.append((f, k))
|
|
return out
|
|
|
|
|
|
def _targets_sha(p, targets):
|
|
return store.sha([[f, k, p.section_check(f, k)["sha"], p.section_check(f, k)["st"]] for f, k in targets])
|
|
|
|
|
|
@app.post("/pier/<name>/confirm_matched/read")
|
|
def confirm_matched_read(name):
|
|
"""Marker: the evidence table of the current target list was read to its end (the page posts this when the scroll box is at the bottom)."""
|
|
p = pier_or_404(name)
|
|
targets = matched_targets(p)
|
|
store.wjson(p.dir / "confirm_view.json", {"sha": _targets_sha(p, targets), "at": store.now(), "n": len(targets)})
|
|
return {"ok": True, "n": len(targets)}
|
|
|
|
|
|
@app.route("/pier/<name>/confirm_matched", methods=["GET", "POST"])
|
|
def confirm_matched(name):
|
|
"""D4: bulk confirmation of the bundles the app check reports as 일치. The evidence table must have been opened (a marker holds the
|
|
hash of the target list), a name is required; anything the app does not report as 일치 is not touched."""
|
|
p = pier_or_404(name)
|
|
targets = matched_targets(p)
|
|
if request.method == "POST":
|
|
by = request.form.get("by", "").strip()
|
|
mark = store.rjson(p.dir / "confirm_view.json", {})
|
|
if not targets:
|
|
flash("확정할 일치 묶음이 없습니다.", "fail")
|
|
elif mark.get("sha") != _targets_sha(p, targets):
|
|
flash("근거표를 열어 끝까지 읽은 뒤에 확정할 수 있습니다(대상 목록이 바뀌었으면 다시 여세요).", "fail")
|
|
elif not by:
|
|
flash("확정자 이름을 입력하세요.", "fail")
|
|
else:
|
|
for f, k in targets:
|
|
p.confirm(f, k, by, "", "g")
|
|
flash(f"앱 확인 '일치' 묶음 {len(targets)}개를 {by}이(가) 확정했습니다. 확정 기록에는 입력한 이름만 남습니다.", "ok")
|
|
return redirect(url_for("pier", name=name))
|
|
return redirect(url_for("confirm_matched", name=name))
|
|
rest = []
|
|
for f, k in sorted(required(p)):
|
|
if (f, k) in targets:
|
|
continue
|
|
s = p.section_state(f, k)
|
|
if not s["confirmed"]:
|
|
rest.append(dict(file=f, key=k, chk=p.section_check(f, k)))
|
|
return render_template("confirm_matched.html", p=p, targets=[dict(file=f, key=k, chk=p.section_check(f, k)) for f, k in targets],
|
|
rest=rest, ST=verify.ST_NAME)
|
|
|
|
|
|
|
|
# ---------------- 검토 요약 일괄 확정 (2026-10-20): one screen to review, one act to confirm
|
|
def review_items(p):
|
|
"""Required bundles that are not (validly) confirmed, each with the app check, what changed since the pier was created, where the values came
|
|
from (app tools) and its review group (A 일치 / B 주의·근거 없음 / C 불일치 / D 결정 기록·참고)."""
|
|
p.run_autocheck()
|
|
org = store.rjson(p.dir / "origin_values.json", {})
|
|
rl, al = store.rjson(p.dir / "read_log.json", []), store.rjson(p.dir / "anchor_log.json", [])
|
|
docs = {"params": p.load("params"), "expected": p.load("expected")}
|
|
out = []
|
|
for f, k in sorted(required(p)):
|
|
s = p.section_state(f, k)
|
|
if s["confirmed"]:
|
|
continue
|
|
c = p.section_check(f, k)
|
|
ov = (org.get(f) or {}).get(k)
|
|
item = dict(file=f, key=k, id=f"{f}:{k}", st=c["st"], summary=c["summary"], checks=c["checks"], sha=c["sha"] if "sha" in c else "",
|
|
stale=s["stale"], prev=(s["by"], s["at"]) if s["stale"] else None, change=reviewlib.bundle_change(docs[f][k], ov),
|
|
prov=reviewlib.provenance(f, k, rl, al), unverified=reviewlib.not_checked(c["checks"], c["st"]), group=reviewlib.group_of(c["st"]))
|
|
if (f, k) == ("expected", "planned_fail"):
|
|
item["detail"] = [f"{x.get('gate', '')} {x.get('item') or x.get('item_prefix', '')} — {x.get('reason', '')} ({x.get('approved', '')})" for x in docs[f][k]]
|
|
item["flags"] = reviewlib.flags(item)
|
|
out.append(item)
|
|
return out
|
|
|
|
|
|
def _review_sha(items):
|
|
return store.sha([[i["id"], i["st"], i["change"]["text"], i.get("sha", "")] for i in items])
|
|
|
|
|
|
@app.post("/pier/<name>/review/read")
|
|
def review_read(name):
|
|
"""Marker: the review list was shown to its end (the page posts this when its last line is visible). Holds the hash of the list that was shown."""
|
|
p = pier_or_404(name)
|
|
items = review_items(p)
|
|
store.wjson(p.dir / "review_view.json", {"sha": _review_sha(items), "at": store.now(), "n": len(items)})
|
|
return {"ok": True, "n": len(items)}
|
|
|
|
|
|
@app.route("/pier/<name>/review", methods=["GET", "POST"])
|
|
def review(name):
|
|
"""One screen: the app prepares what changed / what it verified / what it could not verify; the person ticks what they compared with the drawing
|
|
(every 주의·근거 없음·결정 기록 bundle needs its own tick, 일치 bundles one tick for the group), names themselves and confirms once. 불일치 bundles
|
|
cannot be confirmed. Each confirmation record gets an informative reason (engine/review.auto_reason) + the person's note."""
|
|
p = pier_or_404(name)
|
|
items = review_items(p)
|
|
if request.method == "POST":
|
|
by, note = request.form.get("by", "").strip(), request.form.get("note", "")
|
|
ticked = set(request.form.getlist("ack"))
|
|
mark = store.rjson(p.dir / "review_view.json", {})
|
|
if not by:
|
|
flash("확정자 이름을 입력하세요.", "fail")
|
|
elif mark.get("sha") != _review_sha(items):
|
|
flash("검토 목록을 끝까지 읽은 뒤에 확정할 수 있습니다(목록이 바뀌었으면 다시 여세요).", "fail")
|
|
else:
|
|
done, skipped = [], []
|
|
for it in items:
|
|
if it["group"] == "C":
|
|
continue
|
|
if it["group"] == "A":
|
|
if not request.form.get("ack_matched"):
|
|
continue
|
|
elif it["id"] not in ticked:
|
|
skipped.append(it["id"])
|
|
continue
|
|
p.confirm(it["file"], it["key"], by, reviewlib.auto_reason(it, note), it["st"])
|
|
done.append(it["id"])
|
|
if not done:
|
|
flash("확정한 묶음이 없습니다. 도면과 대조한 묶음에 체크하세요(일치 묶음은 아래 '읽었습니다'에 체크).", "fail")
|
|
else:
|
|
flash(f"{len(done)}개 묶음을 {by}이(가) 확정했습니다. 사유는 앱이 아는 사실(앱 확인·변경·출처)과 입력한 의견으로 기록했습니다."
|
|
+ (f" 체크하지 않은 {len(skipped)}개는 미확정으로 남았습니다." if skipped else ""), "ok")
|
|
return redirect(url_for("review", name=name) if [i for i in items if i["id"] not in done] else url_for("pier", name=name))
|
|
return redirect(url_for("review", name=name))
|
|
cnt = {g: sum(1 for i in items if i["group"] == g) for g in "ABCD"}
|
|
return render_template("review.html", p=p, items=items, cnt=cnt, ST=verify.ST_NAME, by_group={g: [i for i in items if i["group"] == g] for g in "ABCD"})
|
|
|
|
|
|
def required(p):
|
|
"""Bundles any gate of the pier's scope needs (existing ones)."""
|
|
return {(r["file"], r["key"]) for g in p.scope() for r in p.gate_requirements(g) if r["exists"]}
|
|
|
|
|
|
def unconfirmed(p):
|
|
"""Required sections (all gates) that exist and are not confirmed, without duplicates."""
|
|
out = []
|
|
for g in p.scope():
|
|
for r in p.gate_requirements(g):
|
|
if r["exists"] and not r["confirmed"] and (r["file"], r["key"]) not in out:
|
|
out.append((r["file"], r["key"]))
|
|
return out
|
|
|
|
|
|
# ---------------- gates
|
|
def gate_blockers(p, gate):
|
|
out = [f"{r['file']}.{r['key']} 미확정" for r in p.gate_requirements(gate) if r["exists"] and not r["confirmed"]]
|
|
out += [f"{r['file']}.{r['key']} 없음" for r in p.gate_requirements(gate) if not r["exists"]]
|
|
if p.info.get("type") not in store.PIER_TYPES:
|
|
out.append("T형이 아닌 교각")
|
|
ids = p.scope()
|
|
if gate not in ids:
|
|
return [f"{gate}는 이 교각의 범위 밖"]
|
|
k = ids.index(gate)
|
|
if k > 0 and not p.gate_status(ids[k - 1])["approved"]:
|
|
out.append(f"{ids[k - 1]} 미승인")
|
|
return out
|
|
|
|
|
|
def job_state(job):
|
|
"""A queued/running job counts only while its thread lives in this server process; a job left behind by a
|
|
stopped server (or test process) is shown as 중단됨 and may be run again."""
|
|
if job and job["state"] in ("대기", "실행 중"):
|
|
t = ACTIVE.get(job.get("id"))
|
|
if job.get("pid") != os.getpid() or not (t and t.is_alive()):
|
|
return "중단됨"
|
|
return job and job["state"]
|
|
|
|
|
|
def _gate_job(p, gate, jid):
|
|
jp = p.run_dir / f"{gate}_job.json"
|
|
inputs = p.inputs_sha()
|
|
log = []
|
|
|
|
def put(state, **kw):
|
|
store.wjson(jp, dict({"id": jid, "pid": os.getpid(), "state": state, "at": store.now(), "log": log}, **kw))
|
|
|
|
def note(m):
|
|
log.append(m)
|
|
put("실행 중")
|
|
with RHINO_LOCK:
|
|
try:
|
|
note("Rhino 작업 시작")
|
|
rep = run_gate(gate, p.dir / "params.json", p.dir / "expected.json", p.run_dir, log=note)
|
|
rep["inputs_sha"], rep["job_id"] = inputs, jid
|
|
store.wjson(p.run_dir / f"{gate}_gate.json", rep)
|
|
put("완료")
|
|
except Exception:
|
|
put("오류", error=traceback.format_exc())
|
|
|
|
|
|
def batch_state(p):
|
|
"""State of the batch run of this pier (실행 중 only while its thread lives in this server process)."""
|
|
b = store.rjson(p.run_dir / "batch.json")
|
|
if b and b["state"] == "실행 중":
|
|
t = BATCH.get(p.name)
|
|
if b.get("pid") != os.getpid() or not (t and t.is_alive()):
|
|
b = dict(b, state="중단됨", reason="killed")
|
|
return b
|
|
|
|
|
|
def _batch(p, by, gates_todo):
|
|
bp = p.run_dir / "batch.json"
|
|
log = []
|
|
|
|
started = store.now()
|
|
|
|
def put(state, **kw):
|
|
"""batch.json; started = when this run began, reason (중단 only): blocked / error / fail -- the gates page tells the three apart (#34-5)"""
|
|
store.wjson(bp, dict({"state": state, "by": by, "pid": os.getpid(), "started": started, "at": store.now(), "log": log}, **kw))
|
|
put("실행 중")
|
|
need = list(gates_todo)
|
|
# 1. gates with a current PASS result: approve without running again (in order, until one needs a run)
|
|
while need:
|
|
g = need[0]
|
|
st = p.gate_status(g)
|
|
rep = st["report"] or {}
|
|
if not (rep.get("verdict") == "PASS" and not st["stale"]):
|
|
break
|
|
p.set_approval(g, "승인", by, AUTO_NOTE)
|
|
s = rep["summary"]
|
|
log.append(f"{g} 기존 PASS 결과 {s['passed']}/{s['total']} (계획 FAIL {len(rep['planned_fails'])}) → 승인 (재실행 없음)")
|
|
put("실행 중")
|
|
need.pop(0)
|
|
if not need:
|
|
return put("완료")
|
|
# 2. the rest: one Rhino session (gate checks + negative tests only, no per-gate save / reload, G5 saved once)
|
|
missing = [f"{g}: {b}" for g in need for b in gate_blockers(p, g) if not b.endswith("미승인")]
|
|
if missing:
|
|
log.append("실행 불가: " + ", ".join(missing))
|
|
return put("중단", reason="blocked")
|
|
jid, inputs = uuid.uuid4().hex[:12], p.inputs_sha()
|
|
ACTIVE[jid] = threading.current_thread()
|
|
jlog = []
|
|
|
|
def jobs(state, **kw):
|
|
for g in need:
|
|
store.wjson(p.run_dir / f"{g}_job.json",
|
|
dict({"id": jid, "pid": os.getpid(), "state": state, "at": store.now(), "log": jlog}, **kw))
|
|
|
|
def note(m):
|
|
jlog.append(m)
|
|
log.append(m)
|
|
jobs("실행 중")
|
|
put("실행 중")
|
|
jobs("대기")
|
|
log.append(f"{need[0]}~{need[-1]} 단일 세션 실행 (게이트 검사·음성 시험만, 중간 저장·재로드 없음)")
|
|
put("실행 중")
|
|
with RHINO_LOCK:
|
|
try:
|
|
reps = run_batch_session(need, p.dir / "params.json", p.dir / "expected.json", p.run_dir, log=note)
|
|
except Exception:
|
|
tb = traceback.format_exc()
|
|
jobs("오류", error=tb)
|
|
log.append("오류 — 게이트 화면에서 내용을 확인하세요")
|
|
return put("중단", reason="error", error=(tb.strip().splitlines() or ["?"])[-1])
|
|
for g in need:
|
|
reps[g]["inputs_sha"], reps[g]["job_id"], reps[g]["at"] = inputs, jid, store.now()
|
|
store.wjson(p.run_dir / f"{g}_gate.json", reps[g])
|
|
jobs("완료")
|
|
# 3. approve in order; stop at the first gate that is not PASS
|
|
for g in need:
|
|
rep = reps[g]
|
|
if rep["verdict"] != "PASS":
|
|
log.append(f"{g} {rep['verdict']} — 승인하지 않고 멈춤 (결과를 확인하세요)")
|
|
return put("중단", reason="fail", gate=g)
|
|
p.set_approval(g, "승인", by, AUTO_NOTE)
|
|
s = rep["summary"]
|
|
log.append(f"{g} PASS {s['passed']}/{s['total']} (계획 FAIL {len(rep['planned_fails'])}) → 승인")
|
|
put("실행 중")
|
|
put("완료")
|
|
|
|
|
|
def anything_running(p):
|
|
if (batch_state(p) or {}).get("state") == "실행 중":
|
|
return True
|
|
return any(job_state(p.gate_status(g)["job"]) in ("대기", "실행 중") for g in gate_ids())
|
|
|
|
|
|
@app.post("/pier/<name>/gates/run_all")
|
|
def gates_run_all(name):
|
|
p = pier_or_404(name)
|
|
by = request.form.get("by", "").strip()
|
|
if not by:
|
|
flash("판정자 이름을 입력하세요.", "fail")
|
|
return redirect(url_for("gates", name=name))
|
|
if anything_running(p):
|
|
flash("이미 실행 중인 작업이 있습니다.", "fail")
|
|
return redirect(url_for("gates", name=name))
|
|
todo = [g for g in p.scope() if not p.gate_status(g)["approved"]]
|
|
if not todo:
|
|
flash("모든 게이트가 이미 승인되어 있습니다.", "ok")
|
|
return redirect(url_for("gates", name=name))
|
|
b = gate_blockers(p, todo[0])
|
|
if b:
|
|
flash(f"{todo[0]} 실행 불가: " + ", ".join(b), "fail")
|
|
return redirect(url_for("gates", name=name))
|
|
store.wjson(p.run_dir / "batch.json", {"state": "실행 중", "by": by, "pid": os.getpid(), "at": store.now(),
|
|
"log": ["일괄 실행 대기"]})
|
|
t = threading.Thread(target=_batch, args=(p, by, todo), daemon=True)
|
|
BATCH[p.name] = t
|
|
t.start()
|
|
flash(f"{todo[0]}~{todo[-1]} 연속 실행을 시작했습니다. PASS인 게이트만 '{by}' 이름으로 자동 승인합니다.", "ok")
|
|
return redirect(url_for("gates", name=name))
|
|
|
|
|
|
@app.post("/pier/<name>/gate/<gate>/run")
|
|
def gate_run(name, gate):
|
|
p = pier_or_404(name)
|
|
if gate not in gate_ids():
|
|
abort(404)
|
|
if (batch_state(p) or {}).get("state") == "실행 중":
|
|
flash("일괄 실행 중에는 개별 실행을 할 수 없습니다.", "fail")
|
|
return redirect(url_for("gates", name=name))
|
|
b = gate_blockers(p, gate)
|
|
if b:
|
|
flash(f"{gate} 실행 불가: " + ", ".join(b), "fail")
|
|
return redirect(url_for("gates", name=name))
|
|
if job_state(p.gate_status(gate)["job"]) in ("대기", "실행 중"):
|
|
flash(f"{gate}는 이미 실행 중입니다.", "fail")
|
|
return redirect(url_for("gates", name=name))
|
|
# Mark the job as queued (with its id) before the thread starts, so no request sees the previous run's state.
|
|
jid = uuid.uuid4().hex[:12]
|
|
store.wjson(p.run_dir / f"{gate}_job.json",
|
|
{"id": jid, "pid": os.getpid(), "state": "대기", "at": store.now(), "log": []})
|
|
t = threading.Thread(target=_gate_job, args=(p, gate, jid), daemon=True)
|
|
ACTIVE[jid] = t
|
|
t.start()
|
|
flash(f"{gate} 실행을 시작했습니다. 1~2분 걸립니다.", "ok")
|
|
return redirect(url_for("gates", name=name))
|
|
|
|
|
|
@app.post("/pier/<name>/gate/<gate>/decide")
|
|
def gate_decide(name, gate):
|
|
p = pier_or_404(name)
|
|
st = p.gate_status(gate)
|
|
by, decision, note = request.form.get("by", "").strip(), request.form["decision"], request.form.get("note", "")
|
|
if not st["report"] or st["stale"]:
|
|
flash("판정할 최신 결과가 없습니다. 게이트를 다시 실행하세요.", "fail")
|
|
elif not by:
|
|
flash("판정자 이름을 입력하세요.", "fail")
|
|
elif decision == "승인" and st["report"]["verdict"] != "PASS":
|
|
flash("FAIL 게이트는 승인할 수 없습니다. 입력을 고치거나 계획 FAIL로 등록(expected.planned_fail)한 뒤 다시 실행하세요.", "fail")
|
|
elif decision == "반려" and not note.strip():
|
|
flash("반려 사유를 입력하세요.", "fail")
|
|
else:
|
|
p.set_approval(gate, decision, by, note)
|
|
flash(f"{gate} {decision} ({by})", "ok")
|
|
return redirect(url_for("gates", name=name))
|
|
|
|
|
|
@app.route("/pier/<name>/gates")
|
|
def gates(name):
|
|
p = pier_or_404(name)
|
|
rows = [dict(id=g, name=GATE_NAMES[g], blockers=gate_blockers(p, g), req=p.gate_requirements(g), **p.gate_status(g))
|
|
for g in p.scope()]
|
|
for r in rows:
|
|
r["state"] = job_state(r["job"])
|
|
batch = batch_state(p)
|
|
# #34-5 (2026-10-20): a batch that did not finish must not look like a finished one -- say what stopped it, which results are older than the run,
|
|
# and show the partial results that were computed before it stopped (never used for approval or the report)
|
|
aborted = bool(batch) and batch["state"] in ("중단", "중단됨")
|
|
since = (batch or {}).get("started") or (batch or {}).get("at") or ""
|
|
for r in rows:
|
|
r["older"] = bool(aborted and r["report"] and (r["report_at"] or "") < since)
|
|
r["show_partial"] = bool(r["partial"] and (aborted or r["state"] in ("오류", "중단됨")))
|
|
running = any(r["state"] in ("대기", "실행 중") for r in rows) or (batch or {}).get("state") == "실행 중"
|
|
return render_template("gates.html", p=p, rows=rows, running=running, batch=batch, aborted=aborted,
|
|
todo=[g for g in p.scope() if not p.gate_status(g)["approved"]],
|
|
out_of_scope=[g for g in gate_ids() if g not in p.scope()])
|
|
|
|
|
|
@app.post("/pier/<name>/report")
|
|
def report_make(name):
|
|
p = pier_or_404(name)
|
|
if not any(p.gate_status(g)["report"] for g in p.scope()):
|
|
flash("게이트 결과가 없습니다. 게이트를 먼저 실행하세요.", "fail")
|
|
return redirect(url_for("gates", name=name))
|
|
try:
|
|
stale = {g for g in p.scope() if p.gate_status(g)["stale"]}
|
|
out = report_files.write(p.dir, p.run_dir, request.form.get("by", "").strip(), stale,
|
|
[g for g in gate_ids() if g not in p.scope()])
|
|
except Exception as e:
|
|
flash(f"보고서 생성 실패: {e}", "fail")
|
|
return redirect(url_for("gates", name=name))
|
|
c = out["check"]
|
|
flash(f"보고서 초안 생성: {out['md'].name} · 자체 점검 {'통과' if c['ok'] else '미통과'} "
|
|
f"(장 누락 {len(c['chapters_missing'])}, 재료표 {c['schedule_rows']}/{c['schedule_marks']}행"
|
|
f"{' — 재료표를 읽지 못함' if c.get('schedule_unread') else ''}, 지적 {c['finding_rows']}건)",
|
|
"ok" if c["ok"] else "fail")
|
|
return redirect(url_for("gates", name=name))
|
|
|
|
|
|
@app.route("/pier/<name>/report/<kind>")
|
|
def report_get(name, kind):
|
|
p = pier_or_404(name)
|
|
files = sorted(p.run_dir.glob(f"보고서_*.{kind}")) if kind in ("md", "pdf") else []
|
|
if not files:
|
|
abort(404)
|
|
return send_file(files[-1], as_attachment=True, download_name=files[-1].name)
|
|
|
|
|
|
@app.route("/pier/<name>/download/<gate>")
|
|
def download(name, gate):
|
|
p = pier_or_404(name)
|
|
f = p.run_dir / f"{gate}.3dm"
|
|
rep = p.gate_status(gate)["report"] if gate in gate_ids() else None
|
|
if not rep or not rep.get("model") or not f.exists(): # batch run: only G5 is saved
|
|
abort(404)
|
|
return send_file(f, as_attachment=True, download_name=f"{p.name}_{gate}.3dm")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="127.0.0.1", port=5000, debug=False)
|