server.py

# -*- coding: utf-8 -*-
"""
BarVision 불량판정 로직트리 뷰어 - 로컬 웹서버

- C:\\glim\\Parameter\\SPEC\\*.ini  (모델 파라미터) 리스트업 / 읽기 / 저장
- D:\\DATA\\LOT_DATA\\**\\*DEFECT_Detail*.csv 탐색 / 파싱
- 정적 웹 UI 서빙

표준 라이브러리만 사용 (별도 설치 불필요). Python 3.8+
"""

import os
import io
import sys
import csv
import json
import shutil
import datetime
import threading
import webbrowser
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs, unquote

# ----------------------------------------------------------------------------
# 설정
# ----------------------------------------------------------------------------
FROZEN = getattr(sys, "frozen", False)

# BASE_DIR : 사용자가 만지는 파일(설명 JSON)이 놓이는 곳 = exe 가 있는 폴더
# BUNDLE_DIR : 웹 리소스가 들어 있는 곳 = onefile 임시 해제 폴더
if FROZEN:
	BASE_DIR   = os.path.dirname(os.path.abspath(sys.executable))
	BUNDLE_DIR = getattr(sys, "_MEIPASS", BASE_DIR)
else:
	BASE_DIR   = os.path.dirname(os.path.abspath(__file__))
	BUNDLE_DIR = BASE_DIR

WEB_DIR    = os.path.join(BUNDLE_DIR, "web")

SPEC_DIR   = r"C:\glim\Parameter\SPEC"          # 모델 파라미터 INI (Constants.h PATH_PARAM)
BACKUP_DIR = os.path.join(SPEC_DIR, "_backup")  # 저장 전 원본 백업 위치

CSV_ROOTS  = [r"D:\DATA\LOT_DATA", r"D:\DATA"]  # detail CSV 탐색 루트
DESC_PATH  = os.path.join(BASE_DIR, "defect_desc.json")   # 불량 설명 저장 파일
FEAT_PATH  = os.path.join(BASE_DIR, "feature_desc.json")  # feature 설명 저장 파일
CSV_PATTERN_PARTS = ("DEFECT_Detail", "Defect_Detail")

INI_ENCODING = "cp949"   # 멀티바이트 빌드 기준
PORT = 8731

# 모델 INI 에서 읽어오는 판정 파라미터 전체.
# ParamData.cpp:169~192 의 SerGet 순서 + Constants.h 의 인덱스 상수와 1:1로 맞춘다.
#   key      : INI 키 이름
#   internal : 소스상의 인덱스 상수명
#   tree     : Get_Defect_Type 분류 트리가 직접 참조하는가
TREE_PARAMS = [
	{"key": "dParam1",  "internal": "D_IDX_TH_W",               "name": "후보 임계 (백)",     "type": "float", "tree": False},
	{"key": "dParam2",  "internal": "D_IDX_TH_B",               "name": "후보 임계 (흑)",     "type": "float", "tree": False},
	{"key": "dParam3",  "internal": "D_IDX_RATIO_W",            "name": "백 종횡비",          "type": "float", "tree": True},
	{"key": "dParam4",  "internal": "D_IDX_HEIGHT_Y_W",         "name": "백 Y높이",           "type": "float", "tree": False},
	{"key": "dParam5",  "internal": "D_IDX_RATIO_B",            "name": "흑 종횡비",          "type": "float", "tree": True},
	{"key": "dParam6",  "internal": "D_IDX_HEIGHT_Y_B",         "name": "흑 Y높이",           "type": "float", "tree": False},
	{"key": "dParam7",  "internal": "D_IDX_VAL_PINHOLE",        "name": "핀홀 밝기차 TH",          "type": "float", "tree": True},
	{"key": "dParam8",  "internal": "D_IDX_VAL_EXTRUDING",      "name": "돌출 밝기차 TH",          "type": "float", "tree": True},
	{"key": "dParam9",  "internal": "D_IDX_BIG_WIDTH_LINE",     "name": "거대 라인 폭",       "type": "float", "tree": False},
	{"key": "dParam10", "internal": "D_IDX_LINE_V",             "name": "라인 밝기차 TH",          "type": "float", "tree": True},
	{"key": "dParam11", "internal": "D_IDX_COMPACT",            "name": "진원도",             "type": "float", "tree": True},
	{"key": "dParam12", "internal": "D_IDX_WIDTH_LINE_B",       "name": "흑 라인 폭",         "type": "float", "tree": False},
	{"key": "dParam13", "internal": "D_IDX_ANGLE",              "name": "기울기 기준",        "type": "float", "tree": False},
	{"key": "dParam14", "internal": "D_IDX_TH_B_ISLAND",        "name": "무지부 흑 후보 임계", "type": "float", "tree": False},
	{"key": "dParam15", "internal": "D_IDX_PREPROCESS_TARGET_VAL", "name": "전처리 목표 밝기 (절대)", "type": "float", "tree": False},
	{"key": "dParam16", "internal": "D_IDX_PREPROCESS_WEIGHT_W", "name": "전처리 가중치 (백)", "type": "float", "tree": False},
	{"key": "dParam17", "internal": "D_IDX_PREPROCESS_WEIGHT_B", "name": "전처리 가중치 (흑)", "type": "float", "tree": False},
	{"key": "dParam18", "internal": "D_IDX_BINARY_TH_W",        "name": "이진화 임계 (백)",   "type": "float", "tree": False},
	{"key": "dParam19", "internal": "D_IDX_BINARY_TH_B",        "name": "이진화 임계 (흑)",   "type": "float", "tree": False},
	{"key": "dParam20", "internal": "D_IDX_AREA",               "name": "최소 면적",          "type": "float", "tree": True},
	{"key": "dParam21", "internal": "D_IDX_AREA_DARK_PERCENT",  "name": "영역 내 비율",       "type": "float", "tree": True},
	{"key": "dParam22", "internal": "D_IDX_EDGE_TH",            "name": "엣지 임계",          "type": "float", "tree": False},
	{"key": "dParam23", "internal": "D_IDX_EDGE_OFFSET",        "name": "엣지 오프셋",        "type": "float", "tree": False},
	{"key": "dParam24", "internal": "D_IDX_TAPE_TH",            "name": "테이프 임계",        "type": "float", "tree": False},
	{"key": "m_nMinDarkDefectValue", "internal": "m_nMinDarkDefectValue", "name": "흑 밝기차 최소 TH", "type": "int", "tree": True},
]

# 소스에 상수로 박혀 있는 값 (INI 에 없음)
HARDCODED_PARAMS = [
	{"key": "__angle_low",  "name": "스크래치 각도 하한 (<=)", "value": 10.0, "note": "ClientDlg.cpp:9519 하드코딩"},
	{"key": "__angle_high", "name": "스크래치 각도 상한 (>=)", "value": 80.0, "note": "ClientDlg.cpp:9519 하드코딩"},
]


# ----------------------------------------------------------------------------
# INI 처리 : 라인 단위로 다뤄 원본 포맷/주석/순서를 보존한다
# ----------------------------------------------------------------------------
def list_models():
	result = []
	if not os.path.isdir(SPEC_DIR):
		return result
	for fname in sorted(os.listdir(SPEC_DIR)):
		full = os.path.join(SPEC_DIR, fname)
		if not os.path.isfile(full):
			continue
		if not fname.lower().endswith(".ini"):
			continue
		st = os.stat(full)
		result.append({
			"name": os.path.splitext(fname)[0],
			"file": fname,
			"path": full,
			"size": st.st_size,
			"mtime": datetime.datetime.fromtimestamp(st.st_mtime).strftime("%Y-%m-%d %H:%M:%S"),
		})
	return result


def read_ini_lines(path):
	with io.open(path, "r", encoding=INI_ENCODING, errors="replace") as f:
		return f.read().splitlines()


def parse_ini_values(path):
	"""키=값 전체를 dict 로. (대괄호 인덱스가 붙은 키도 그대로 문자열 키로 취급)"""
	values = {}
	for line in read_ini_lines(path):
		s = line.strip()
		if not s or s.startswith(";") or s.startswith("#") or s.startswith("["):
			continue
		if "=" not in s:
			continue
		k, v = s.split("=", 1)
		values[k.strip()] = v.strip()
	return values


def parse_ini_text(text):
	"""파일이 아니라 본문 문자열에서 키=값을 뽑는다 (브라우저로 올린 INI 용)."""
	values = {}
	for line in text.splitlines():
		s = line.strip()
		if not s or s.startswith(";") or s.startswith("#") or s.startswith("["):
			continue
		if "=" not in s:
			continue
		k, v = s.split("=", 1)
		values[k.strip()] = v.strip()
	return values


def build_params(all_values):
	"""키=값 dict 를 화면이 쓰는 파라미터 목록으로 변환."""
	params = []
	for p in TREE_PARAMS:
		raw = all_values.get(p["key"])
		item = dict(p)
		item["raw"] = raw
		item["found"] = raw is not None
		try:
			item["value"] = float(raw) if raw not in (None, "") else None
		except ValueError:
			item["value"] = None
		params.append(item)
	return params


def get_model_params(model_name):
	path = os.path.join(SPEC_DIR, model_name + ".ini")
	if not os.path.isfile(path):
		raise FileNotFoundError(path)

	all_values = parse_ini_values(path)
	params = []
	for p in TREE_PARAMS:
		raw = all_values.get(p["key"])
		item = dict(p)
		item["raw"] = raw
		item["found"] = raw is not None
		try:
			item["value"] = float(raw) if raw not in (None, "") else None
		except ValueError:
			item["value"] = None
		params.append(item)

	return {
		"model": model_name,
		"path": path,
		"params": params,
		"hardcoded": HARDCODED_PARAMS,
		"total_keys": len(all_values),
	}


def save_model_params(model_name, updates):
	"""INI 의 해당 키 값만 교체. 원본은 _backup 폴더에 타임스탬프로 보존."""
	path = os.path.join(SPEC_DIR, model_name + ".ini")
	if not os.path.isfile(path):
		raise FileNotFoundError(path)

	editable = {p["key"] for p in TREE_PARAMS}
	targets = {}
	for k, v in updates.items():
		if k not in editable:
			continue           # 트리에서 쓰는 키 외에는 절대 건드리지 않는다
		targets[k] = str(v)

	if not targets:
		return {"changed": [], "backup": None}

	# 백업
	if not os.path.isdir(BACKUP_DIR):
		os.makedirs(BACKUP_DIR)
	stamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
	backup_path = os.path.join(BACKUP_DIR, "%s_%s.ini" % (model_name, stamp))
	shutil.copy2(path, backup_path)

	lines = read_ini_lines(path)
	changed = []
	remaining = dict(targets)

	for i, line in enumerate(lines):
		s = line.strip()
		if not s or "=" not in s or s.startswith(";") or s.startswith("#") or s.startswith("["):
			continue
		k = s.split("=", 1)[0].strip()
		if k in remaining:
			old = s.split("=", 1)[1].strip()
			new = remaining.pop(k)
			if old != new:
				lines[i] = "%s=%s" % (k, new)
				changed.append({"key": k, "old": old, "new": new})

	# INI 에 없던 키는 끝에 추가
	for k, v in remaining.items():
		lines.append("%s=%s" % (k, v))
		changed.append({"key": k, "old": None, "new": v})

	with io.open(path, "w", encoding=INI_ENCODING, errors="replace", newline="\r\n") as f:
		f.write("\n".join(lines) + "\n")

	return {"changed": changed, "backup": backup_path}


def load_json_doc(path, packed_name, key):
	"""사용자가 저장한 파일이 우선, 없으면 배포본에 동봉된 초안을 읽는다."""
	src = path
	if not os.path.isfile(src):
		packed = os.path.join(BUNDLE_DIR, packed_name)
		src = packed if os.path.isfile(packed) else None
	if not src:
		return {key: {}, "path": path}
	with io.open(src, "r", encoding="utf-8") as f:
		return {key: json.load(f), "path": path}


# ----------------------------------------------------------------------------
# CSV 처리
# ----------------------------------------------------------------------------
def find_csv_files(limit=300):
	found = []
	seen = set()
	for root in CSV_ROOTS:
		if not os.path.isdir(root):
			continue
		for dirpath, dirnames, filenames in os.walk(root):
			for fn in filenames:
				if not fn.lower().endswith(".csv"):
					continue
				if not any(part.lower() in fn.lower() for part in CSV_PATTERN_PARTS):
					continue
				full = os.path.join(dirpath, fn)
				if full in seen:
					continue
				seen.add(full)
				try:
					st = os.stat(full)
				except OSError:
					continue
				found.append({
					"path": full,
					"name": fn,
					"size": st.st_size,
					"mtime_ts": st.st_mtime,
					"mtime": datetime.datetime.fromtimestamp(st.st_mtime).strftime("%Y-%m-%d %H:%M"),
				})
				if len(found) >= limit:
					break
			if len(found) >= limit:
				break
	found.sort(key=lambda x: x["mtime_ts"], reverse=True)
	for f in found:
		f.pop("mtime_ts", None)
	return found


def load_csv(path, max_rows=5000):
	if not os.path.isfile(path):
		raise FileNotFoundError(path)

	with io.open(path, "r", encoding=INI_ENCODING, errors="replace", newline="") as f:
		reader = csv.reader(f)
		rows = []
		header = None
		for i, rec in enumerate(reader):
			if not rec:
				continue
			if header is None:
				header = [c.strip() for c in rec]
				continue
			rows.append(rec)
			if len(rows) >= max_rows:
				break

	if header is None:
		return {"path": path, "header": [], "rows": [], "truncated": False}

	# 데이터 필드 수가 헤더보다 많은 경우가 있다(로거 trailing comma / 헤더 누락).
	# 헤더 이름 기준으로만 매핑하고, 남는 뒤쪽 필드는 버린다.
	dicts = []
	for rec in rows:
		d = {}
		for idx, name in enumerate(header):
			d[name] = rec[idx].strip() if idx < len(rec) else ""
		dicts.append(d)

	return {
		"path": path,
		"header": header,
		"rows": dicts,
		"row_count": len(dicts),
		"truncated": len(rows) >= max_rows,
	}


# ----------------------------------------------------------------------------
# HTTP 핸들러
# ----------------------------------------------------------------------------
class Handler(BaseHTTPRequestHandler):
	server_version = "BarLogicTree/1.0"

	def log_message(self, fmt, *args):
		pass  # 콘솔 조용히

	# ---------- 공통 응답 ----------
	def _send_json(self, obj, code=200):
		body = json.dumps(obj, ensure_ascii=False).encode("utf-8")
		self.send_response(code)
		self.send_header("Content-Type", "application/json; charset=utf-8")
		self.send_header("Content-Length", str(len(body)))
		self.send_header("Cache-Control", "no-store")
		self.end_headers()
		self.wfile.write(body)

	def _send_file(self, path):
		if not os.path.isfile(path):
			self.send_error(404, "Not Found")
			return
		ext = os.path.splitext(path)[1].lower()
		ctype = {
			".html": "text/html; charset=utf-8",
			".js":   "application/javascript; charset=utf-8",
			".css":  "text/css; charset=utf-8",
			".json": "application/json; charset=utf-8",
		}.get(ext, "application/octet-stream")
		with open(path, "rb") as f:
			body = f.read()
		self.send_response(200)
		self.send_header("Content-Type", ctype)
		self.send_header("Content-Length", str(len(body)))
		self.send_header("Cache-Control", "no-store")
		self.end_headers()
		self.wfile.write(body)

	# ---------- GET ----------
	def do_GET(self):
		parsed = urlparse(self.path)
		route = parsed.path
		qs = parse_qs(parsed.query)

		try:
			if route == "/" or route == "/index.html":
				self._send_file(os.path.join(WEB_DIR, "index.html"))
				return

			if route.startswith("/web/"):
				rel = route[len("/web/"):]
				safe = os.path.normpath(os.path.join(WEB_DIR, rel))
				if not safe.startswith(WEB_DIR):
					self.send_error(403, "Forbidden")
					return
				self._send_file(safe)
				return

			if route == "/api/models":
				self._send_json({"spec_dir": SPEC_DIR, "models": list_models()})
				return

			if route == "/api/params":
				model = qs.get("model", [""])[0]
				if not model:
					self._send_json({"error": "model 파라미터가 필요합니다."}, 400)
					return
				self._send_json(get_model_params(model))
				return

			if route == "/api/csvfiles":
				self._send_json({"roots": CSV_ROOTS, "files": find_csv_files()})
				return

			if route == "/api/desc":
				self._send_json(load_json_doc(DESC_PATH, "defect_desc.json", "desc"))
				return

			if route == "/api/featdesc":
				self._send_json(load_json_doc(FEAT_PATH, "feature_desc.json", "desc"))
				return

			if route == "/api/csv":
				path = unquote(qs.get("path", [""])[0])
				if not path:
					self._send_json({"error": "path 파라미터가 필요합니다."}, 400)
					return
				self._send_json(load_csv(path))
				return

			self.send_error(404, "Not Found")

		except FileNotFoundError as e:
			self._send_json({"error": "파일을 찾을 수 없습니다: %s" % e}, 404)
		except Exception as e:
			self._send_json({"error": "%s: %s" % (type(e).__name__, e)}, 500)

	# ---------- POST ----------
	def do_POST(self):
		parsed = urlparse(self.path)
		route = parsed.path
		length = int(self.headers.get("Content-Length") or 0)
		raw = self.rfile.read(length) if length else b"{}"

		try:
			payload = json.loads(raw.decode("utf-8"))
		except Exception:
			self._send_json({"error": "잘못된 JSON 본문입니다."}, 400)
			return

		try:
			if route == "/api/params":
				model = payload.get("model")
				values = payload.get("values") or {}
				if not model:
					self._send_json({"error": "model 이 필요합니다."}, 400)
					return
				result = save_model_params(model, values)
				result["model"] = model
				self._send_json(result)
				return

			if route == "/api/desc" or route == "/api/featdesc":
				desc = payload.get("desc")
				if not isinstance(desc, dict):
					self._send_json({"error": "desc 는 객체여야 합니다."}, 400)
					return
				target = DESC_PATH if route == "/api/desc" else FEAT_PATH
				with io.open(target, "w", encoding="utf-8") as f:
					f.write(json.dumps(desc, ensure_ascii=False, indent="\t"))
				self._send_json({"saved": len(desc), "path": target})
				return

			if route == "/api/initext":
				# 브라우저에서 직접 올린 INI 본문 (SPEC 폴더 밖의 파일 대응)
				text = payload.get("text") or ""
				all_values = parse_ini_text(text)
				self._send_json({
					"model": payload.get("name") or "(열어온 파일)",
					"path": payload.get("name") or "(열어온 파일)",
					"params": build_params(all_values),
					"hardcoded": HARDCODED_PARAMS,
					"total_keys": len(all_values),
					"readonly": True,
				})
				return

			if route == "/api/csvtext":
				# 브라우저에서 직접 올린 CSV 본문 파싱 (원격 PC 파일 대응)
				text = payload.get("text") or ""
				reader = csv.reader(io.StringIO(text, newline=""))
				header = None
				dicts = []
				for rec in reader:
					if not rec:
						continue
					if header is None:
						header = [c.strip() for c in rec]
						continue
					d = {}
					for idx, name in enumerate(header):
						d[name] = rec[idx].strip() if idx < len(rec) else ""
					dicts.append(d)
				self._send_json({
					"path": payload.get("name") or "(업로드)",
					"header": header or [],
					"rows": dicts,
					"row_count": len(dicts),
					"truncated": False,
				})
				return

			self.send_error(404, "Not Found")

		except FileNotFoundError as e:
			self._send_json({"error": "파일을 찾을 수 없습니다: %s" % e}, 404)
		except Exception as e:
			self._send_json({"error": "%s: %s" % (type(e).__name__, e)}, 500)


def main():
	httpd = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
	url = "http://127.0.0.1:%d/" % PORT
	print("=" * 62)
	print(" BarVision 불량판정 로직트리 뷰어")
	print("=" * 62)
	print(" 주소      : %s" % url)
	print(" 모델 INI  : %s" % SPEC_DIR)
	print(" CSV 탐색  : %s" % ", ".join(CSV_ROOTS))
	print(" 종료      : 이 창에서 Ctrl+C")
	print("=" * 62)
	threading.Timer(1.0, lambda: webbrowser.open(url)).start()
	try:
		httpd.serve_forever()
	except KeyboardInterrupt:
		print("\n종료합니다.")
		httpd.shutdown()


if __name__ == "__main__":
	main()