import sys import re import html as _html import os from datetime import datetime from urllib.parse import urlparse, unquote import psycopg2 from psycopg2.extras import DictCursor from rapidfuzz import fuzz from PySide6.QtCore import Qt, QAbstractTableModel, QModelIndex, QObject, Signal, QThread, QUrl, QByteArray from PySide6.QtWidgets import ( QApplication, QWidget, QVBoxLayout, QHBoxLayout, QLineEdit, QPushButton, QTableView, QSplitter, QLabel, QStyledItemDelegate, QTextEdit, QTextBrowser, QCompleter, QDialog, QMessageBox, QProgressDialog, QAbstractItemView, QScrollArea, QGridLayout, QTabWidget, QPlainTextEdit ) from PySide6.QtGui import QTextDocument, QPixmap, QBrush, QColor, QImage, QPainter import requests from io import BytesIO from functools import lru_cache import importlib try: # optional dependency _pil = importlib.import_module("PIL.Image") _pil_ops = importlib.import_module("PIL.ImageOps") _pil_filter = importlib.import_module("PIL.ImageFilter") _imagehash = importlib.import_module("imagehash") # 모듈을 그대로 사용 (Image.open / Image.Resampling / Image.LANCZOS 등) Image = _pil ImageOps = _pil_ops ImageFilter = _pil_filter imagehash = _imagehash except Exception: Image = None ImageOps = None ImageFilter = None imagehash = None @lru_cache(maxsize=2048) def _fetch_http_bytes(url: str, *, timeout: int = 10) -> bytes | None: """ http/https 리소스 다운로드 헬퍼. - 일부 CDN(예: alicdn)은 기본 요청을 차단(420/403)하므로 UA/Referer를 강제한다. - 너무 큰 파일(10MB 초과)은 메모리/시간 문제로 스킵. """ u = str(url or "").strip() if not u: return None if u.startswith("//"): u = "https:" + u if not (u.startswith("http://") or u.startswith("https://")): return None try: # referer는 host 기반으로 맞추는 게 차단 회피에 유리 try: m = re.match(r"^(https?://[^/]+)", u, flags=re.IGNORECASE) referer = (m.group(1) + "/") if m else "https://www.alibaba.com/" except Exception: referer = "https://www.alibaba.com/" headers = { "User-Agent": "Mozilla/5.0", "Referer": referer, "Accept": "image/avif,image/webp,image/apng,image/*,*/*;q=0.8", } r = requests.get(u, timeout=int(timeout), headers=headers) r.raise_for_status() data = bytes(r.content) if len(data) > 10 * 1024 * 1024: return None return data except Exception: return None def _norm_iname_db_exact(s: str | None) -> str: """ DB에서 정확일치 비교에 쓰는 정규화. - 현재 SQL의: lower(regexp_replace(trim(m.iname), '\s+', ' ', 'g')) 와 동일한 규칙 """ s = _html.unescape(str(s or "")).strip().lower() s = re.sub(r"\s+", " ", s) return s def _sort_price(value) -> int: """ 정렬용 price 키. - DB NULL/비정상 값이 섞여도 sort가 터지지 않도록 항상 int를 반환 - None/파싱실패는 매우 큰 값으로 보내서 뒤로 정렬 """ try: if value is None: return 10**18 return int(value) except Exception: return 10**18 class InameSearchWorker(QObject): finished = Signal(list) # group_rows error = Signal(str) def __init__(self, iname: str, min_ratio: int): super().__init__() self.iname = iname self.min_ratio = min_ratio self._conn = None def cancel(self): try: if self._conn is not None: self._conn.cancel() except Exception: pass def run(self): """ 상품명 검색을 별도 스레드에서 수행 (UI 프리징 방지 + progress 팝업 유지). search_by_iname()와 동일한 검색 전략을 '별도 커넥션'으로 실행한다. """ try: conn = psycopg2.connect(**DB_INFO) self._conn = conn cur = conn.cursor(cursor_factory=DictCursor) iname = self.iname min_ratio = int(self.min_ratio or 70) norm = _normalize_iname(iname) # 유사% 100 이상이면: "포함"이 아니라 "정확히 동일한 상품명"만 조회 if min_ratio >= 100: exact_key = _norm_iname_db_exact(iname) cur.execute(""" SELECT g.group_id, m.icode as winner_icode, m.price as winner_price, m.iname AS winner_iname, m.img_url AS winner_img_url, m.content AS winner_content, g.group_key, m.img_url AS match_img_url FROM mlinkdw.shopprod_group2 g JOIN mlinkdw.shopprod_group_map2 m ON m.group_id = g.group_id WHERE lower(regexp_replace(trim(m.iname), '\s+', ' ', 'g')) = %s LIMIT 20000 """, (exact_key,)) group_best = {} for r in cur.fetchall(): dist = _hashab_value(r.get("winner_img_url"), r.get("match_img_url")) # 흐름 표시: 검색 매칭된 이미지(= m.img_url)의 phash/dhash로 그룹 매칭 시뮬레이션 ph_dh = _image_hash_pair_from_url(str(r.get("match_img_url") or "")) if ph_dh: ph_hex, dh_hex = str(ph_dh[0]), str(ph_dh[1]) else: ph_hex, dh_hex = None, None flow = _simulate_match_flow(conn, ph_hex, dh_hex, threshold=12) gid = r["group_id"] group_best[gid] = dict( group_id=gid, winner=r["winner_icode"], iname=(r.get("winner_iname") or ""), price=r["winner_price"], sim=100, img_url=r.get("winner_img_url"), group_key=r["group_key"], hashAB=dist, content=r.get("winner_content"), match_flow=flow ) rows = sorted(group_best.values(), key=lambda x: (_sort_price(x.get("price")), -x.get("sim", 0))) group_rows = [ [r["group_id"], r["winner"], r["iname"], r["price"], r["sim"], r["img_url"], r["group_key"], r["hashAB"], r.get("content"), r["match_flow"]] for r in rows ] self.finished.emit(group_rows) conn.close() return tokens = [t for t in norm.split(" ") if t] tokens = [t for t in tokens if len(t) >= 2][:5] if not tokens: self.finished.emit([]) conn.close() return where_parts = [] params = [] for i, t in enumerate(tokens): if i == 0 and len(t) >= 3: where_parts.append("m.iname ILIKE %s") params.append(f"{t}%") else: where_parts.append("m.iname ILIKE %s") params.append(f"%{t}%") where = " AND ".join(where_parts) # pg_trgm 가능 여부(워커 커넥션 기준) pg_trgm_available = False try: cur2 = conn.cursor() cur2.execute("SELECT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'pg_trgm')") pg_trgm_available = bool(cur2.fetchone()[0]) except Exception: pg_trgm_available = False if pg_trgm_available: try: sim_threshold = max(0.0, min(1.0, min_ratio / 100.0)) cur.execute(f""" SELECT DISTINCT ON (t.group_id) t.group_id, t.winner_icode, t.winner_price, t.winner_iname AS iname, t.winner_img_url AS img_url, t.winner_content AS content, t.group_key AS group_key, t.sim FROM ( SELECT g.group_id, g.winner_icode, g.winner_price, (SELECT iname FROM mlinkdw.shopprod_group_map2 WHERE group_id = g.group_id AND icode = g.winner_icode) AS winner_iname, (SELECT img_url FROM mlinkdw.shopprod_group_map2 WHERE group_id = g.group_id AND icode = g.winner_icode) AS winner_img_url, (SELECT content FROM mlinkdw.shopprod_group_map2 WHERE group_id = g.group_id AND icode = g.winner_icode) AS winner_content, g.group_key, mlinkdw.similarity(lower(m.iname), lower(%s)) AS sim FROM mlinkdw.shopprod_group2 g JOIN mlinkdw.shopprod_group_map2 m ON m.group_id = g.group_id WHERE {where} ) t WHERE t.sim >= %s ORDER BY t.group_id, t.sim DESC LIMIT 5000 """, tuple([iname] + params + [sim_threshold])) group_best = {} for r in cur.fetchall(): sim_pct = int(round(float(r["sim"]) * 100)) gid = r["group_id"] group_best[gid] = dict( group_id=gid, winner=r["winner_icode"], iname=r["iname"], price=r["winner_price"], sim=sim_pct, img_url=r["img_url"], content=r.get("content"), group_key=r["group_key"] ) rows = sorted(group_best.values(), key=lambda x: (_sort_price(x.get("price")), -x.get("sim", 0))) group_rows = [ [r["group_id"], r["winner"], r["iname"], r["price"], r["sim"], r["img_url"], r["group_key"], None, r.get("content"), None] for r in rows ] self.finished.emit(group_rows) conn.close() return except Exception: try: conn.rollback() except Exception: pass # fallback(rapidfuzz) cur.execute(f""" SELECT g.group_id, g.winner_icode, g.winner_price, (SELECT iname FROM mlinkdw.shopprod_group_map2 WHERE group_id = g.group_id AND icode = g.winner_icode) AS winner_iname, (SELECT img_url FROM mlinkdw.shopprod_group_map2 WHERE group_id = g.group_id AND icode = g.winner_icode) AS winner_img_url, (SELECT content FROM mlinkdw.shopprod_group_map2 WHERE group_id = g.group_id AND icode = g.winner_icode) AS winner_content, m.iname AS match_iname, m.img_url AS match_img_url, g.group_key FROM mlinkdw.shopprod_group2 g JOIN mlinkdw.shopprod_group_map2 m ON m.group_id = g.group_id WHERE {where} LIMIT 20000 """, tuple(params)) group_best = {} for r in cur.fetchall(): sim = calc_iname_similarity_advanced(iname, r["match_iname"]) if sim < min_ratio: continue gid = r["group_id"] if gid not in group_best or sim > group_best[gid]["sim"]: group_best[gid] = dict( group_id=gid, winner=r["winner_icode"], iname=r.get("winner_iname"), price=r["winner_price"], sim=sim, img_url=r.get("winner_img_url"), content=r.get("winner_content"), group_key=r["group_key"] ) rows = sorted(group_best.values(), key=lambda x: (_sort_price(x.get("price")), -x.get("sim", 0))) group_rows = [ [r["group_id"], r["winner"], r["iname"], r["price"], r["sim"], r["img_url"], r["group_key"], None, r.get("content"), None] for r in rows ] self.finished.emit(group_rows) conn.close() except Exception as e: self.error.emit(str(e)) class ContentPrefetchWorker(QObject): """ content(HTML)에 포함된 이미지 리소스를 미리 다운로드해서 UI 프리징을 줄이기 위한 워커. - 진행률(progress) 시그널 제공 """ progress = Signal(int, int, str) # done, total, current_url finished = Signal(dict) # {url: bytes} error = Signal(str) def __init__(self, html_text: str, *, timeout: int = 10, max_cache_bytes: int = 10 * 1024 * 1024): super().__init__() self.html_text = str(html_text or "") self.timeout = int(timeout) self.max_cache_bytes = int(max_cache_bytes) self._cancel = False def cancel(self): self._cancel = True def run(self): try: # img src 수집(따옴표/무따옴표 모두) # 예: , , srcs = re.findall(r"""]+src\s*=\s*(?:(["'])(.*?)\1|([^'"\s>]+))""", self.html_text, flags=re.IGNORECASE) urls = [] for q, a, b in srcs: u = a or b if not u: continue u = u.strip() if u.startswith("//"): u = "https:" + u if u.startswith("http://") or u.startswith("https://"): urls.append(u) # 중복 제거(순서 유지) seen = set() uniq = [] for u in urls: if u in seen: continue seen.add(u) uniq.append(u) urls = uniq total = len(urls) cache: dict[str, bytes] = {} done = 0 for u in urls: if self._cancel: break self.progress.emit(done, total, u) try: data = _fetch_http_bytes(u, timeout=int(self.timeout)) or b"" if not data: raise RuntimeError("download failed") if len(data) <= self.max_cache_bytes: cache[u] = data except Exception: # 개별 실패는 무시하고 진행 pass done += 1 self.progress.emit(done, total, u) self.finished.emit(cache) except Exception as e: self.error.emit(str(e)) # ========================= # DB CONFIG # ========================= from db_config import DB_INFO_PSYCOPG2 as DB_INFO # ========================= # UTIL FUNCTIONS # ========================= def calc_similarity(search_name: str, target_name: str) -> int: if not search_name or not target_name: return 0 return fuzz.token_set_ratio(search_name, target_name) def _normalize_iname(text: str) -> str: """ 상품명 비교용 정규화: - 소문자화 - 특수문자 제거/공백 정리 - 한글/영문/숫자/공백 위주로 남김 """ if not text: return "" t = str(text).lower() # 한글/영문/숫자/공백만 남기고 나머지는 공백으로 치환 t = re.sub(r"[^0-9a-z가-힣\s]+", " ", t) t = re.sub(r"\s+", " ", t).strip() return t def calc_iname_similarity_advanced(search_name: str, target_name: str) -> int: """ 상품명 검색(유사도 %) 고도화용 점수. - WRatio: 부분일치/정렬/토큰 등을 종합적으로 반영해 검색 품질이 좋음 - token_set_ratio: 단어 순서가 바뀌거나 일부 단어가 추가된 케이스에 강함 """ s = _normalize_iname(search_name) t = _normalize_iname(target_name) if not s or not t: return 0 return int(max(fuzz.WRatio(s, t), fuzz.token_set_ratio(s, t))) def calc_price_score(winner_price: int, cand_price: int) -> int: if not winner_price or not cand_price: return 0 if cand_price <= winner_price: return 100 diff_ratio = (cand_price - winner_price) / winner_price return max(0, int(100 - diff_ratio * 100)) def calc_image_score(hash_dist: int, max_dist=64) -> int: return max(0, int(100 - (hash_dist / max_dist) * 100)) def _simulate_match_flow(conn, phash_hex: str | None, dhash_hex: str | None, *, threshold: int = 12, candidate_limit: int = 5000) -> str: """ UI 표시용: 거리 기반 그룹 매칭 흐름을 문자열로 리턴. 흐름: EXACT(group_key) → BAND 후보조회 → 해밍거리(best) → threshold 이내면 기존 / 아니면 신규 """ if not phash_hex or not dhash_hex: return "NO_HASH" if not imagehash: return "NO_LIB" # 1) EXACT try: cur = conn.cursor(cursor_factory=DictCursor) try: cur.execute( """ SELECT group_id FROM mlinkdw.shopprod_group2 WHERE phash_hex = %s AND dhash_hex = %s LIMIT 1 """, (phash_hex, dhash_hex), ) except Exception: # 컬럼이 없거나 마이그레이션 전이면 group_key로 fallback conn.rollback() cur.execute( """ SELECT group_id FROM mlinkdw.shopprod_group2 WHERE group_key = %s LIMIT 1 """, (f"{phash_hex}_{dhash_hex}",), ) row = cur.fetchone() if row: gid = row[0] if not isinstance(row, dict) else row.get("group_id") return f"EXACT:{gid}" except Exception: try: conn.rollback() except Exception: pass # 2) BAND 후보 조회 # phash_hex 길이 16(64bit) => 32bit band 2개, 64(256bit) => 8개 def _bands_from_phash_hex(ph: str): ph = (ph or "").strip() if len(ph) not in (16, 64): return [] n = int(ph, 16) num_bands = 2 if len(ph) == 16 else 8 bands = [] for i in range(num_bands): shift = (num_bands - 1 - i) * 32 bands.append((n >> shift) & 0xFFFFFFFF) return bands bands = _bands_from_phash_hex(phash_hex) if not bands: return "NO_BAND" try: cur = conn.cursor(cursor_factory=DictCursor) cur.execute( """ SELECT group_id, phash_hex FROM mlinkdw.shopprod_group2 WHERE phash_bands && %s::int4[] LIMIT %s """, (bands, candidate_limit), ) rows = cur.fetchall() except Exception: try: conn.rollback() except Exception: pass return "BAND_QUERY_FAIL" if not rows: return "NEW" # 3) 후보들 거리 계산 try: target_hash = imagehash.hex_to_hash(phash_hex) except Exception: return "HASH_PARSE_FAIL" best_gid = None best_dist = None for r in rows: try: gid = r["group_id"] if isinstance(r, dict) else r[0] cand_hex = r["phash_hex"] if isinstance(r, dict) else r[1] if not cand_hex: continue cand_hash = imagehash.hex_to_hash(cand_hex) dist = target_hash - cand_hash # int except Exception: continue if best_dist is None or dist < best_dist: best_dist = dist best_gid = gid if best_gid is None or best_dist is None: return "BAND_NO_VALID" if best_dist <= threshold: return f"BAND:gid={best_gid},dist={best_dist},thr={threshold}" return f"NEW:best_gid={best_gid},dist={best_dist},thr={threshold}" @lru_cache(maxsize=512) def _image_hash_pair_from_url(img_url: str): """ URL -> (verify/정규화) -> (phash, dhash) Search_DomeDuplicate_4st3Mro.py 로직과 유사하게 구성. """ if not img_url or not (Image and imagehash): return None try: data = _fetch_http_bytes(str(img_url), timeout=10) if not data: return None bio = BytesIO(data) img = Image.open(bio) try: img.verify() except Exception: # verify 실패해도 열리는 경우가 있어 계속 진행 pass bio.seek(0) img = Image.open(bio) img = ImageOps.exif_transpose(img) # palette(P) + transparency 경고/품질 이슈 회피 try: if getattr(img, "mode", None) == "P" and "transparency" in getattr(img, "info", {}): img = img.convert("RGBA") except Exception: pass img = img.convert("RGB") # 비교 안정화를 위한 약간의 전처리 img = img.filter(ImageFilter.GaussianBlur(radius=0.6)) resampling = getattr(Image, "Resampling", None) method = resampling.LANCZOS if resampling else getattr(Image, "LANCZOS", 1) img = ImageOps.fit(img, (256, 256), method=method) # 기존 Search_DomeDuplicate_* 와 동일하게 기본값(hash_size=8) 사용 → 16 hex chars ph = imagehash.phash(img) dh = imagehash.dhash(img) return ph, dh except Exception: return None def _image_phash_from_url(img_url: str): pair = _image_hash_pair_from_url(img_url) if not pair: return None return pair[0] def _phash_distance(url_a: str, url_b: str) -> int | None: """ 두 이미지 URL의 phash 해밍거리. """ if not (Image and imagehash): return None ha = _image_phash_from_url(url_a) hb = _image_phash_from_url(url_b) if ha is None or hb is None: return None # imagehash.ImageHash 간의 뺄셈은 해밍거리(int)를 반환한다. return ha - hb def _hash_pair_string(img_url: str | None) -> str: """ UI 표시용: 'phash_dhash' (예: e386..._b0c6...) 형태로 반환. """ if not img_url: return "NO_URL" if not (Image and imagehash): return "NO_LIB" pair = _image_hash_pair_from_url(str(img_url)) if not pair: return "HASH_FAIL" ph, dh = pair return f"{ph}_{dh}" def _hashab_value(winner_img_url: str | None, img_url: str | None) -> str: """ hashAB 컬럼 표시용: winner와 대상 이미지의 hash string을 함께 표시. """ a = _hash_pair_string(winner_img_url) b = _hash_pair_string(img_url) return f"{a} | {b}" def _html_to_qimage(html: str, *, width: int = 900, max_height: int = 1600) -> QImage | None: """ winner_content(HTML)을 Qt QTextDocument로 렌더링하여 QImage로 변환. - CSS/JS는 제한적(브라우저와 동일하지 않음) - '같은 그룹 판단'을 위한 보조 신호로 사용 """ try: raw = str(html or "") # DB/JSON 등을 거치며 \" 형태로 이스케이프된 HTML이 들어오는 케이스 대응 # - 예: raw = raw.replace('\\"', '"').replace("\\'", "'") raw = _html.unescape(raw) doc = QTextDocument() # 1) HTML 안의 원격 이미지 리소스를 미리 주입(브라우징/렌더링 안되는 케이스 보완) # - QTextDocument는 기본적으로 http/https를 자동 fetch 하지 않을 수 있음 # - 여기서는 src URL 그대로를 key로 resource 등록 srcs = re.findall(r"""]+src\s*=\s*(?:(["'])(.*?)\1|([^'"\s>]+))""", raw, flags=re.IGNORECASE) urls = [] for q, a, b in srcs: u = (a or b or "").strip() if not u: continue if u.startswith("//"): u = "https:" + u if u.startswith("http://") or u.startswith("https://"): urls.append(u) # 중복 제거(순서 유지) + 너무 많은 이미지 제한 seen = set() uniq = [] for u in urls: if u in seen: continue seen.add(u) uniq.append(u) if len(uniq) >= 20: break for u in uniq: data = _fetch_http_bytes(u, timeout=10) if not data: continue qimg = QImage() try: qimg.loadFromData(data) except Exception: qimg = QImage() if qimg.isNull(): continue try: doc.addResource(QTextDocument.ResourceType.ImageResource, QUrl(u), qimg) except Exception: pass # 2) HTML 설정 후 렌더링 doc.setHtml(raw) doc.setTextWidth(max(200, int(width))) h = int(doc.size().height()) if h <= 0: h = 1 h = min(int(max_height), h) img = QImage(int(width), int(h), QImage.Format_ARGB32) img.fill(Qt.white) p = QPainter(img) try: doc.drawContents(p) finally: p.end() return img except Exception: return None def _qimage_to_pil(img: QImage): """ QImage -> PIL.Image (선택적 의존성). """ if Image is None: return None try: qimg = img.convertToFormat(QImage.Format_RGBA8888) ptr = qimg.bits() try: ptr.setsize(qimg.sizeInBytes()) except Exception: pass data = bytes(ptr) pil_img = Image.frombytes("RGBA", (qimg.width(), qimg.height()), data) pil_img = pil_img.convert("RGB") return pil_img except Exception: return None @lru_cache(maxsize=512) def _content_hash_pair_from_html(html: str): """ HTML -> (phash, dhash) (imagehash.ImageHash 객체) """ if imagehash is None or Image is None: return None qimg = _html_to_qimage(html) if qimg is None: return None pil_img = _qimage_to_pil(qimg) if pil_img is None: return None try: ph = imagehash.phash(pil_img) dh = imagehash.dhash(pil_img) return (ph, dh) except Exception: return None def _content_hash_distance(html_a: str | None, html_b: str | None) -> tuple[int | None, int | None]: """ winner_content(HTML) 간 '렌더링 이미지' 기반 해시 거리(phash/dhash). """ pa = _content_hash_pair_from_html(str(html_a or "")) pb = _content_hash_pair_from_html(str(html_b or "")) if not pa or not pb: return (None, None) try: ph_dist = int(pa[0] - pb[0]) dh_dist = int(pa[1] - pb[1]) return (ph_dist, dh_dist) except Exception: return (None, None) def _extract_img_urls_from_html(html_text: str | None, *, limit: int = 20) -> list[str]: """ content(HTML)에서 img src URL 추출. - 공백/쉼표/따옴표 유무 모두 대응 """ raw = str(html_text or "") # DB/JSON 등을 거치며 \" 형태로 이스케이프된 HTML 대응 raw = raw.replace('\\"', '"').replace("\\'", "'") raw = _html.unescape(raw) # 태그를 먼저 뽑고, 속성(src/data-src/srcset 등)을 안전하게 추출한다. attr_names = ("src", "data-src", "data-original", "data-lazy-src", "data-zoom-image", "data-url") urls: list[str] = [] for m in re.finditer(r"]*>", raw, flags=re.IGNORECASE): tag = m.group(0) or "" for an in attr_names: mm = re.search( rf"""\b{re.escape(an)}\s*=\s*(?:(["'])(.*?)\1|([^'"\s>]+))""", tag, flags=re.IGNORECASE | re.DOTALL, ) if mm: u = (mm.group(2) or mm.group(3) or "").strip() if u: urls.append(u) break mm2 = re.search( r"""\bsrcset\s*=\s*(?:(["'])(.*?)\1|([^'"\s>]+))""", tag, flags=re.IGNORECASE | re.DOTALL, ) if mm2: ss = (mm2.group(2) or mm2.group(3) or "").strip() if ss: first = ss.split(",")[0].strip() first = first.split(" ")[0].strip() if first: urls.append(first) # 깨진 HTML 방어: # - 예: for u2 in re.findall(r"""https?://[^\s<>"']+""", tag, flags=re.IGNORECASE): u2 = str(u2).strip() if u2: urls.append(u2) if not urls: for u in re.findall( r"""https?://[^\s<>"']+\.(?:jpg|jpeg|png|gif|webp|bmp|svg)(?:\?[^\s<>"']*)?""", raw, flags=re.IGNORECASE, ): urls.append(str(u).strip()) # 중복 제거(순서 유지) seen = set() out = [] for u in urls: if u in seen: continue seen.add(u) out.append(u) if len(out) >= int(limit): break return out def _normalize_img_url_token(u: str) -> str | None: """ content_token용 이미지 URL 정규화. - query/fragment 제거(추적 파라미터 등으로 토큰 분산 방지) - scheme/host는 소문자 - path는 unquote 후 소문자(매칭 우선) """ try: # Guard against Postgres btree index entry size limit (~8KB). # Very long URL tokens can crash inserts into shopprod_group_content_token. MAX_LEN = 512 raw = str(u).strip() if not raw: return None # scheme-less URL: //cdn.example.com/x.jpg if raw.startswith("//"): raw = "https:" + raw if raw.startswith("http:////"): raw = "http://" + raw[len("http:////") :] if raw.startswith("https:////"): raw = "https://" + raw[len("https:////") :] if raw.startswith("http:/") and (not raw.startswith("http://")): raw = "http://" + raw[len("http:/") :] if raw.startswith("https:/") and (not raw.startswith("https://")): raw = "https://" + raw[len("https:/") :] if raw.startswith("/"): path = unquote(raw).strip().lower() tok = f"rel:{path}" return tok if len(tok) <= MAX_LEN else None if raw.startswith("./") or raw.startswith("../"): path = unquote(raw).strip().lower() tok = f"rel:{path}" return tok if len(tok) <= MAX_LEN else None if (not re.match(r"^[a-zA-Z][a-zA-Z0-9+.-]*:", raw)) and ("/" in raw): path = unquote(raw).strip().lower() tok = f"rel:{path}" return tok if len(tok) <= MAX_LEN else None p = urlparse(raw) if p.scheme not in ("http", "https"): return None netloc = (p.netloc or "").strip().lower() if not netloc: return None path = unquote(p.path or "").strip() if not path: return None path = path.lower() # http/https는 동일 리소스로 취급(토큰 분산 방지): scheme을 https로 통일 tok = f"https://{netloc}{path}" return tok if len(tok) <= MAX_LEN else None except Exception: return None def _extract_img_filenames_from_html( html_text: str | None, *, max_imgs: int = 20, max_filenames: int = 8, min_fname_len: int = 6, ) -> list[str]: """ (호환용 함수명 유지) content HTML에서 의 "정규화된 URL 토큰"을 추출. - 후보군 폭발을 막기 위해 '안내/공지/배송' 계열은 제외(휴리스틱) """ # 너무 흔한(배송/공지/안내) 이미지 파일명은 토큰으로 쓰면 후보군이 과도해져서 제외한다. stop_filenames = { "배송공지.jpg", "delivery.jpg", "notice.jpg", "noti.jpg", "top_open_dome.jpg", "down_open_dome.jpg", "dome_bottom.jpg", "vender_top.jpg", "top.jpg", "title.jpg", "info-title.jpg", "info.jpg", "information.jpg", "product_info_bottom.jpg", "info_issue.jpg", "도매의신.jpg", "dome.jpg", } stop_stems = { "notice", # 너무 범용적인 파일명 stem은 제외 "img", "image", "photo", "pic", "logo", "banner", } stop_prefixes = { "top", "bottom", "title", "noti", "intro", "dome", "도매의신", "무료배송", "개인정보", "교환반품", "안내서", "공지", "도매신", } stop_substrings = { "배송공지", "delivery", "notice", "top_open_dome", "down_open_dome", "dome_bottom", "vender_top", "info-title", "information", "product_info_bottom", "info_issue", "도매의신", } urls = _extract_img_urls_from_html(html_text, limit=int(max_imgs)) if not urls: return [] out: list[str] = [] seen: set[str] = set() for u in urls: try: p = urlparse(str(u)) fn = os.path.basename(p.path or "") fn = unquote(fn or "").strip().lower() except Exception: continue if not fn: continue stem = fn.rsplit(".", 1)[0].strip() # stoplist(정확/포함) if fn in stop_filenames: continue # stop_filenames 변형(문자 포함)도 제외: 예) steadylady_notice.jpg, NOTICE_end.jpg ... if any(sf in fn for sf in stop_filenames): continue if stem in stop_stems: continue if any(stem.startswith(x) for x in stop_prefixes): continue if any(x in stem for x in stop_substrings): continue tok = _normalize_img_url_token(str(u)) if not tok: continue if tok in seen: continue seen.add(tok) out.append(tok) if len(out) >= int(max_filenames): break return out def _content_imgset_match_stats( html_a: str | None, html_b: str | None, *, max_imgs: int = 12, ph_thr: int = 12, dh_thr: int = 14, ) -> dict: """ content 내 이미지 세트 간 유사도(매칭) 기반 판단. - 각 HTML에서 최대 max_imgs개 이미지 해시(ph/dh)를 계산 - B의 각 이미지가 A의 어떤 이미지와도 (ph<=thr and dh<=thr)면 match로 카운트 """ if imagehash is None or Image is None: return {"err": "NO_LIB"} urls_a = _extract_img_urls_from_html(html_a, limit=max_imgs) urls_b = _extract_img_urls_from_html(html_b, limit=max_imgs) if not urls_a or not urls_b: return {"err": "NO_IMG", "na": len(urls_a), "nb": len(urls_b)} pairs_a = [] for u in urls_a: p = _image_hash_pair_from_url(u) if p: pairs_a.append(p) pairs_b = [] for u in urls_b: p = _image_hash_pair_from_url(u) if p: pairs_b.append(p) if not pairs_a or not pairs_b: return {"err": "HASH_FAIL", "na": len(pairs_a), "nb": len(pairs_b)} match = 0 min_ph = None min_dh = None for pb in pairs_b: best_ok = False for pa in pairs_a: try: ph = int(pa[0] - pb[0]) dh = int(pa[1] - pb[1]) except Exception: continue if min_ph is None or ph < min_ph: min_ph = ph if min_dh is None or dh < min_dh: min_dh = dh if ph <= ph_thr and dh <= dh_thr: best_ok = True break if best_ok: match += 1 denom = min(len(pairs_a), len(pairs_b)) ratio = (match / float(denom)) if denom > 0 else 0.0 return { "err": None, "na": len(pairs_a), "nb": len(pairs_b), "match": match, "ratio": ratio, "min_ph": min_ph, "min_dh": min_dh, "ph_thr": ph_thr, "dh_thr": dh_thr, } def calc_merge_score(sim, price_score, img_score): return int(sim * 0.5 + price_score * 0.3 + img_score * 0.2) def _bands_from_phash_hex_signed(phash_hex: str) -> list[int]: """ phash(hex) -> int4[] 저장용 band 배열(signed 32-bit). - 16 chars (64bit) => 32bit x 2 - 64 chars (256bit) => 32bit x 8 """ ph = (phash_hex or "").strip() if len(ph) not in (16, 64): return [] try: n = int(ph, 16) except Exception: return [] num_bands = 2 if len(ph) == 16 else 8 bands: list[int] = [] for i in range(num_bands): shift = (num_bands - 1 - i) * 32 band = (n >> shift) & 0xFFFFFFFF # unsigned # PostgreSQL int4[](signed) 범위로 매핑 if band >= 0x80000000: band -= 0x100000000 bands.append(int(band)) return bands # ========================= # HIGHLIGHT DELEGATE # ========================= class HighlightDelegate(QStyledItemDelegate): def __init__(self, parent=None): super().__init__(parent) self.tokens = [] self.sim_col = None def set_search_text(self, text: str): self.tokens = [re.escape(t) for t in text.split() if t.strip()] def set_similarity_column(self, col_idx: int): self.sim_col = col_idx def _color_by_similarity(self, sim: int): if sim >= 90: return "#FFF176" if sim >= 80: return "#FFF59D" if sim >= 70: return "#FFFDE7" return None def paint(self, painter, option, index): text = str(index.data(Qt.DisplayRole)) if not self.tokens or self.sim_col is None: super().paint(painter, option, index) return sim = index.sibling(index.row(), self.sim_col).data() if not isinstance(sim, int): super().paint(painter, option, index) return bg = self._color_by_similarity(sim) if not bg: super().paint(painter, option, index) return html = text for token in self.tokens: html = re.sub( f"({token})", rf'\1', html, flags=re.IGNORECASE ) doc = QTextDocument() doc.setHtml(html) painter.save() painter.translate(option.rect.topLeft()) doc.setTextWidth(option.rect.width()) doc.drawContents(painter) painter.restore() # ========================= # SIMPLE TABLE MODEL # ========================= class SimpleTableModel(QAbstractTableModel): def __init__(self, headers): super().__init__() self.headers = headers self.rows = [] # 하이라이트(상품코드) - 단일/다중 모두 지원 self.highlight_icode = None self.highlight_icodes: set[str] | None = None def rowCount(self, parent=QModelIndex()): return len(self.rows) def columnCount(self, parent=QModelIndex()): return len(self.headers) def data(self, index, role): value = self.rows[index.row()][index.column()] # 기본 표시 if role in (Qt.DisplayRole, Qt.ItemDataRole.DisplayRole): return value # ✅ 상품코드 위치 강조 (상품코드 검색 전용) if role in (Qt.BackgroundRole, Qt.ItemDataRole.BackgroundRole): if index.column() == 2: # ICode 컬럼 (RowNo 추가로 +1) v = str(value or "") v_key = v.upper() # 다중 하이라이트 우선 if self.highlight_icodes: if v_key in self.highlight_icodes: return QBrush(QColor("yellow")) # 단일 하이라이트 fallback if self.highlight_icode and v_key == str(self.highlight_icode).upper(): return QBrush(QColor("yellow")) def headerData(self, section, orientation, role): if role == Qt.DisplayRole and orientation == Qt.Horizontal: return self.headers[section] def load(self, rows): self.beginResetModel() self.rows = rows self.endResetModel() # ========================= # IMAGE COMPARE DIALOG # ========================= class CompareThumbnailDialog(QDialog): def __init__(self, w_name, w_img, c_name, c_img, search_text, parent=None): super().__init__(parent) self.setWindowTitle("이미지 비교 (Winner vs Candidate)") self.resize(900, 450) tokens = search_text.split() # 해시 거리 기반 "동일/상이" 판정 표시 self._judge_label = QLabel("") self._judge_label.setTextFormat(Qt.PlainText) self._judge_label.setStyleSheet("color:#333;") dist_text = self._hash_judgement_text(w_img, c_img) self._judge_label.setText(dist_text) left = self._box("WINNER", w_name, w_img, tokens) right = self._box("CANDIDATE", c_name, c_img, tokens) outer = QVBoxLayout(self) outer.addWidget(self._judge_label) row = QHBoxLayout() row.addWidget(left) row.addWidget(right) outer.addLayout(row) def _box(self, title, name, img_url, tokens): box = QVBoxLayout() w = QWidget() w.setLayout(box) lbl = QLabel() lbl.setTextFormat(Qt.RichText) html = name for t in tokens: html = re.sub( f"({re.escape(t)})", r'\1', html, flags=re.IGNORECASE ) lbl.setText(html) img = QLabel("이미지 로딩 중...") img.setAlignment(Qt.AlignCenter) if img_url: try: r = requests.get(img_url, timeout=5) pix = QPixmap() pix.loadFromData(BytesIO(r.content).read()) img.setPixmap(pix.scaled(380, 380, Qt.KeepAspectRatio)) except Exception: img.setText("이미지 로딩 실패") box.addWidget(QLabel(title)) box.addWidget(lbl) box.addWidget(img) return w def _hash_judgement_text(self, url_a: str, url_b: str) -> str: """ UI에 표시할 해시 거리/판정 문자열. - phash/dhash 모두 거리 계산 """ if not (Image and imagehash): return "판정: (해시 라이브러리 없음: pillow/imagehash 설치 필요)" pa = _image_hash_pair_from_url(str(url_a or "")) pb = _image_hash_pair_from_url(str(url_b or "")) if not pa or not pb: return "판정: (해시 계산 실패)" ph_a, dh_a = pa ph_b, dh_b = pb ph_dist = ph_a - ph_b dh_dist = dh_a - dh_b # 기본 임계치(64bit 기준). 필요시 조정 가능 ph_thr = 12 dh_thr = 14 same = (ph_dist <= ph_thr) and (dh_dist <= dh_thr) verdict = "동일(추천)" if same else "상이" return ( f"판정: {verdict} | " f"phash_dist={ph_dist} (<= {ph_thr}), " f"dhash_dist={dh_dist} (<= {dh_thr})" ) # ========================= # IMAGE DIALOG # ========================= class ImageDialog(QDialog): def __init__(self, img_url: str, title: str | None = None, parent=None): super().__init__(parent) self.setWindowTitle(title or "IMAGE") self.resize(900, 700) layout = QVBoxLayout(self) self.url_label = QLabel(img_url or "") self.url_label.setTextInteractionFlags(Qt.TextSelectableByMouse) self.img_label = QLabel("이미지 로딩 중...") self.img_label.setAlignment(Qt.AlignCenter) layout.addWidget(self.url_label) layout.addWidget(self.img_label) self._load_image(img_url) def _load_image(self, img_url: str): if not img_url: self.img_label.setText("img_url이 비어있습니다.") return try: r = requests.get(img_url, timeout=10) r.raise_for_status() pix = QPixmap() pix.loadFromData(BytesIO(r.content).read()) if pix.isNull(): self.img_label.setText("이미지 로딩 실패 (지원되지 않는 포맷)") return self.img_label.setPixmap( pix.scaled(860, 620, Qt.KeepAspectRatio, Qt.SmoothTransformation) ) except Exception as e: self.img_label.setText(f"이미지 로딩 실패\n{e}") class SideBySideImageDialog(QDialog): """ 두 이미지를 나란히 비교해서 보여주는 다이얼로그(비모달 show 권장) - 왼쪽: group_model 선택 이미지 - 오른쪽: item 선택 이미지 """ def __init__( self, left_title: str, left_url: str, right_title: str, right_url: str, *, title: str = "이미지 비교", parent=None, ): super().__init__(parent) self.setWindowTitle(title) self.resize(1200, 750) outer = QHBoxLayout(self) outer.addWidget(self._panel(left_title, left_url)) outer.addWidget(self._panel(right_title, right_url)) def _panel(self, title: str, img_url: str) -> QWidget: w = QWidget() box = QVBoxLayout(w) lbl_title = QLabel(title or "") lbl_title.setStyleSheet("font-weight:600;") lbl_url = QLabel(img_url or "") lbl_url.setTextInteractionFlags(Qt.TextSelectableByMouse) lbl_url.setWordWrap(True) img = QLabel("이미지 로딩 중...") img.setAlignment(Qt.AlignCenter) box.addWidget(lbl_title) box.addWidget(lbl_url) scroll = QScrollArea() scroll.setWidgetResizable(True) scroll.setFrameShape(QScrollArea.NoFrame) inner = QWidget() inner_l = QVBoxLayout(inner) inner_l.addWidget(img) inner_l.addStretch(1) scroll.setWidget(inner) box.addWidget(scroll) self._load_pixmap(img, img_url) return w def _load_pixmap(self, label: QLabel, img_url: str): if not img_url: label.setText("img_url이 비어있습니다.") return try: r = requests.get(img_url, timeout=10) r.raise_for_status() pix = QPixmap() pix.loadFromData(BytesIO(r.content).read()) if pix.isNull(): label.setText("이미지 로딩 실패 (지원되지 않는 포맷)") return label.setPixmap(pix.scaled(560, 680, Qt.KeepAspectRatio, Qt.SmoothTransformation)) except Exception as e: label.setText(f"이미지 로딩 실패\n{e}") class SideBySideTextDialog(QDialog): """ 두 텍스트(예: content HTML)를 나란히 비교해서 보여주는 일반창(비모달 show 권장) - 왼쪽: group 선택 content - 오른쪽: item 선택 content """ def __init__( self, left_title: str, left_text: str, right_title: str, right_text: str, *, title: str = "CONTENT 비교", parent=None, ): super().__init__(parent) self.setWindowTitle(title) self.resize(1200, 750) outer = QHBoxLayout(self) outer.addWidget(self._panel(left_title, left_text)) outer.addWidget(self._panel(right_title, right_text)) def _panel(self, title: str, text: str) -> QWidget: w = QWidget() box = QVBoxLayout(w) lbl_title = QLabel(title or "") lbl_title.setStyleSheet("font-weight:600;") box.addWidget(lbl_title) edit = QPlainTextEdit() edit.setReadOnly(True) try: edit.setLineWrapMode(QPlainTextEdit.WidgetWidth) except Exception: pass edit.setPlainText(str(text or "")) box.addWidget(edit) return w class SideBySideHtmlContentDialog(QDialog): """ 두 content(HTML)을 좌/우 패널로 나란히 비교해서 보여주는 다이얼로그. 각 패널은 HtmlContentDialog와 동일하게: - 미리보기(QTextBrowser 기반 간단 렌더링) - 원본(HTML) (QPlainTextEdit) """ def __init__( self, left_title: str, left_html: str, right_title: str, right_html: str, *, title: str = "CONTENT 비교", parent=None, left_prefetched: dict | None = None, right_prefetched: dict | None = None, ): super().__init__(parent) self.setWindowTitle(title) self.resize(1200, 750) outer = QHBoxLayout(self) outer.addWidget(self._panel(left_title, left_html, prefetched=left_prefetched)) outer.addWidget(self._panel(right_title, right_html, prefetched=right_prefetched)) def _panel(self, title: str, html: str, *, prefetched: dict | None = None) -> QWidget: w = QWidget() box = QVBoxLayout(w) lbl_title = QLabel(title or "") lbl_title.setStyleSheet("font-weight:600;") box.addWidget(lbl_title) tabs = QTabWidget() box.addWidget(tabs) class _RemoteResourceBrowser(QTextBrowser): """ HtmlContentDialog과 동일 목적: QTextBrowser가 http/https 이미지 리소스를 못 가져오는 케이스가 있어, loadResource를 오버라이드해서 직접 다운로드 후 렌더링. """ _cache: dict[str, bytes] = {} def loadResource(self, type: int, name: QUrl): # noqa: A002 (Qt API) try: if name.isEmpty(): return super().loadResource(type, name) url = name.toString() if url.startswith("//"): url = "https:" + url if url.startswith("http://") or url.startswith("https://"): if url in self._cache: data = self._cache[url] if type == QTextDocument.ResourceType.ImageResource: pix = QPixmap() pix.loadFromData(data) return pix return QByteArray(data) data = _fetch_http_bytes(url, timeout=10) if not data: return super().loadResource(type, name) # 과도한 메모리 사용 방지: 너무 큰 파일은 캐시하지 않음(10MB) if len(data) <= 10 * 1024 * 1024: self._cache[url] = data if type == QTextDocument.ResourceType.ImageResource: pix = QPixmap() pix.loadFromData(data) return pix return QByteArray(data) except Exception: return super().loadResource(type, name) return super().loadResource(type, name) raw_html = str(html or "") # DB/JSON escape (\" 등) + HTML entity 정리 후 렌더링 render_html = raw_html.replace('\\"', '"').replace("\\'", "'") render_html = _html.unescape(render_html) browser = _RemoteResourceBrowser() try: if isinstance(prefetched, dict) and prefetched: browser._cache.update(prefetched) # type: ignore[attr-defined] except Exception: pass browser.setOpenExternalLinks(True) browser.setHtml(render_html) tabs.addTab(browser, "미리보기") raw_view = QPlainTextEdit() raw_view.setReadOnly(True) raw_view.setPlainText(raw_html) tabs.addTab(raw_view, "원본(HTML)") return w class HtmlContentDialog(QDialog): """ shopprod_group_map.content(HTML)을 렌더링해서 보여주는 팝업. - QtWebEngine 없이도 QTextBrowser로 기본 HTML 렌더링 가능 """ def __init__(self, html: str, title: str | None = None, parent=None, *, prefetched: dict | None = None): super().__init__(parent) self.setWindowTitle(title or "CONTENT") self.resize(900, 700) outer = QVBoxLayout(self) info = QLabel("content(HTML) 미리보기") outer.addWidget(info) tabs = QTabWidget() outer.addWidget(tabs) class _RemoteResourceBrowser(QTextBrowser): """ QTextBrowser가 http/https 이미지 리소스를 못 가져오는 케이스가 있어, loadResource를 오버라이드해서 직접 다운로드 후 렌더링. """ _cache: dict[str, bytes] = {} def loadResource(self, type: int, name: QUrl): # noqa: A002 (Qt API) try: if name.isEmpty(): return super().loadResource(type, name) url = name.toString() if url.startswith("//"): url = "https:" + url if url.startswith("http://") or url.startswith("https://"): if url in self._cache: data = self._cache[url] if type == QTextDocument.ResourceType.ImageResource: pix = QPixmap() pix.loadFromData(data) return pix return QByteArray(data) data = _fetch_http_bytes(url, timeout=10) if not data: return super().loadResource(type, name) # 과도한 메모리 사용 방지: 너무 큰 파일은 캐시하지 않음(10MB) if len(data) <= 10 * 1024 * 1024: self._cache[url] = data if type == QTextDocument.ResourceType.ImageResource: pix = QPixmap() pix.loadFromData(data) return pix return QByteArray(data) except Exception: # 실패 시 기본 처리 return super().loadResource(type, name) return super().loadResource(type, name) # 일부 DB content가 escape 되어 들어오는 경우가 있어 unescape 후 렌더링 raw_html = str(html or "") render_html = raw_html.replace('\\"', '"').replace("\\'", "'") render_html = _html.unescape(render_html) self.browser = _RemoteResourceBrowser() try: if isinstance(prefetched, dict) and prefetched: # URL -> bytes 캐시 주입 (렌더링 시 네트워크 접근 최소화) self.browser._cache.update(prefetched) # type: ignore[attr-defined] except Exception: pass self.browser.setOpenExternalLinks(True) self.browser.setHtml(render_html) tabs.addTab(self.browser, "미리보기") self.raw_view = QPlainTextEdit() self.raw_view.setReadOnly(True) self.raw_view.setPlainText(raw_html) tabs.addTab(self.raw_view, "원본(HTML)") class MultiImageCompareDialog(QDialog): def __init__(self, items: list[dict], title: str = "IMAGE 비교", parent=None): """ items: [{"name": str, "img_url": str, "group_key": str}, ...] """ super().__init__(parent) self.setWindowTitle(title) self.resize(1200, 800) outer = QVBoxLayout(self) info = QLabel(f"선택된 이미지: {len(items)}개") outer.addWidget(info) scroll = QScrollArea() scroll.setWidgetResizable(True) outer.addWidget(scroll) container = QWidget() grid = QGridLayout(container) grid.setHorizontalSpacing(16) grid.setVerticalSpacing(16) scroll.setWidget(container) cols = 3 for i, it in enumerate(items): name = str(it.get("name") or "") img_url = str(it.get("img_url") or "") group_key = str(it.get("group_key") or "") cell = QWidget() cell_l = QVBoxLayout(cell) lbl_name = QLabel(name) lbl_name.setWordWrap(True) lbl_key = QLabel(f"group_key: {group_key}" if group_key else "group_key: -") lbl_key.setTextInteractionFlags(Qt.TextSelectableByMouse) lbl_img = QLabel("로딩 중...") lbl_img.setAlignment(Qt.AlignCenter) cell_l.addWidget(lbl_name) cell_l.addWidget(lbl_key) cell_l.addWidget(lbl_img) # 이미지 로딩(간단히 동기; 필요 시 스레드로 분리 가능) if img_url: try: r = requests.get(img_url, timeout=10) r.raise_for_status() pix = QPixmap() pix.loadFromData(BytesIO(r.content).read()) if not pix.isNull(): lbl_img.setPixmap( pix.scaled(320, 320, Qt.KeepAspectRatio, Qt.SmoothTransformation) ) else: lbl_img.setText("로드 실패") except Exception as e: lbl_img.setText(f"로드 실패\n{e}") else: lbl_img.setText("img_url 없음") r0 = i // cols c0 = i % cols grid.addWidget(cell, r0, c0) # ========================= # MAIN WINDOW # ========================= class MainWindow(QWidget): def __init__(self): super().__init__() self.search_icode_highlight = None self.setWindowTitle("MLink MERGE 판단 UI (Full Integrated)") self.conn = psycopg2.connect(**DB_INFO) self._pg_trgm_available = None # lazy check self._modeless_windows = [] # modeless 창 레퍼런스 유지(가비지 컬렉션 방지) # Search history self.search_history = [] # Search UI self.group_id_edit = QLineEdit() self.icode_edit = QLineEdit() self.iname_edit = QLineEdit() self.ratio_edit = QLineEdit("30") self.search_btn = QPushButton("검색") self.compare_img_btn = QPushButton("이미지 비교") self.compare_content_img_btn = QPushButton("CONTENT(이미지) 판단") self.merge_group_btn = QPushButton("그룹 병합") # 다중 입력 지원: "123,456 789" 형태 허용 (validator 사용 안함) self.group_id_edit.setPlaceholderText("예: 123456 또는 123,456 789") self.completer = QCompleter(self.search_history) self.completer.setCaseSensitivity(Qt.CaseInsensitive) self.iname_edit.setCompleter(self.completer) top = QHBoxLayout() top.addWidget(QLabel("group_id")) top.addWidget(self.group_id_edit) top.addWidget(QLabel("상품코드")) top.addWidget(self.icode_edit) top.addWidget(QLabel("상품명")) top.addWidget(self.iname_edit) top.addWidget(QLabel("유사%")) top.addWidget(self.ratio_edit) top.addWidget(self.search_btn) top.addWidget(self.compare_img_btn) top.addWidget(self.compare_content_img_btn) top.addWidget(self.merge_group_btn) # Models self.group_model = SimpleTableModel( ["GroupID", "Winner", "Iname", "Price", "Similarity", "img_url", "group_key", "hashAB", "content", "match_flow"] ) self.sub_group_model = SimpleTableModel( ["SubGroupID", "Winner", "Iname", "Similarity", "winner_price", "winner_vender_code"] ) self.item_model = SimpleTableModel( ["#", "Vendor", "ICode", "Name", "Price", "img_url", "content", "Similarity", "PriceScore", "ImageScore", "MERGE_SCORE"] ) # Views self.group_view = QTableView() self.group_view.setModel(self.group_model) self.group_view.setSelectionMode(QAbstractItemView.ExtendedSelection) self.group_view.setSelectionBehavior(QAbstractItemView.SelectRows) self.sub_group_view = QTableView() self.sub_group_view.setModel(self.sub_group_model) self.item_view = QTableView() self.item_view.setModel(self.item_model) self.item_view.setSelectionMode(QAbstractItemView.ExtendedSelection) self.item_view.setSelectionBehavior(QAbstractItemView.SelectRows) # Highlight delegate self.highlight_delegate = HighlightDelegate() # item_model columns: [#, Vendor, ICode, Name, Price, img_url, content, Similarity, PriceScore, ImageScore, MERGE_SCORE] self.highlight_delegate.set_similarity_column(7) self.item_view.setItemDelegateForColumn(3, self.highlight_delegate) right = QSplitter(Qt.Vertical) right.addWidget(self.sub_group_view) right.addWidget(self.item_view) main = QSplitter(Qt.Horizontal) main.addWidget(self.group_view) main.addWidget(right) # Action buttons self.move_to_winner_btn = QPushButton("WINNER 그룹으로 이동") btns = QHBoxLayout() btns.addWidget(self.move_to_winner_btn) layout = QVBoxLayout(self) layout.addLayout(top) layout.addWidget(main) layout.addLayout(btns) # Signals self.search_btn.clicked.connect(self.on_search) self.group_view.clicked.connect(self.on_group_clicked) self.group_view.doubleClicked.connect(self.on_group_double_clicked) self.sub_group_view.clicked.connect(self.on_sub_group_clicked) self.item_view.doubleClicked.connect(self.on_item_double_clicked) self.move_to_winner_btn.clicked.connect(self.on_move_item_to_winner_group) self.compare_img_btn.clicked.connect(self.on_compare_images_clicked) self.compare_content_img_btn.clicked.connect(self.on_compare_content_images_clicked) self.merge_group_btn.clicked.connect(self.on_merge_groups_clicked) # Enter로도 검색되게(편의) try: self.group_id_edit.returnPressed.connect(self.on_search) self.icode_edit.returnPressed.connect(self.on_search) self.iname_edit.returnPressed.connect(self.on_search) except Exception: pass def _show_modeless(self, dlg: QDialog): """ QDialog를 모달(exec) 대신 일반창(show)으로 띄움. - 닫히면 자동 삭제 - Python GC로 바로 사라지지 않도록 레퍼런스 유지 """ try: dlg.setModal(False) except Exception: pass dlg.setAttribute(Qt.WA_DeleteOnClose, True) self._modeless_windows.append(dlg) try: dlg.destroyed.connect(lambda *_: self._modeless_windows.remove(dlg)) # type: ignore[attr-defined] except Exception: pass dlg.show() def _adjust_group_view_columns(self): """ Group 테이블의 Iname 컬럼 폭을 픽셀 고정값으로 설정. """ # 컨텐츠 기준으로 1차 폭 계산 self.group_view.resizeColumnsToContents() iname_col = 2 fixed_width_px = 350 self.group_view.setColumnWidth(iname_col, fixed_width_px) # img_url 컬럼 폭 고정 img_url_col = 5 self.group_view.setColumnWidth(img_url_col, 50) # group_key 컬럼 폭 고정 group_key_col = 6 self.group_view.setColumnWidth(group_key_col, 50) # hashAB 컬럼 폭 고정 hashab_col = 7 self.group_view.setColumnWidth(hashab_col, 60) # content 컬럼 폭 고정 content_col = 8 self.group_view.setColumnWidth(content_col, 60) # match_flow 컬럼 폭 고정 match_col = 9 self.group_view.setColumnWidth(match_col, 240) def _adjust_sub_group_view_columns(self): """ SubGroup 테이블의 Iname 컬럼 폭을 픽셀 고정값으로 설정. """ self.sub_group_view.resizeColumnsToContents() iname_col = 2 fixed_width_px = 350 self.sub_group_view.setColumnWidth(iname_col, fixed_width_px) def _adjust_item_view_columns(self): """ Item 테이블의 Name(iname) 컬럼 폭을 픽셀 고정값으로 설정. """ self.item_view.resizeColumnsToContents() name_col = 3 fixed_width_px = 350 self.item_view.setColumnWidth(name_col, fixed_width_px) # img_url 컬럼 폭 고정 img_url_col = 5 self.item_view.setColumnWidth(img_url_col, 60) # content 컬럼 폭 고정 content_col = 6 self.item_view.setColumnWidth(content_col, 60) def _clear_grids(self): """ 검색 버튼 클릭 시 기존 결과(그리드)를 모두 비움. """ # highlight 초기화 self.item_model.highlight_icode = None self.item_model.highlight_icodes = None self.sub_group_model.highlight_icode = None self.sub_group_model.highlight_icodes = None # 데이터 초기화 self.group_model.load([]) self.sub_group_model.load([]) self.item_model.load([]) # 선택/상태 초기화 self.group_view.clearSelection() self.sub_group_view.clearSelection() self.item_view.clearSelection() self.current_group_id = None self.current_winner_icode = None self.current_winner_price = None # 컬럼 폭 유지(고정 폭 재적용) self._adjust_group_view_columns() self._adjust_sub_group_view_columns() self._adjust_item_view_columns() # ========================= # SEARCH # ========================= def on_search(self): # 새 검색 시작 시 기존 결과를 먼저 비움 self._clear_grids() keyword = self.iname_edit.text().strip() self.highlight_delegate.set_search_text(keyword) if keyword and keyword not in self.search_history: self.search_history.insert(0, keyword) self.search_history = self.search_history[:10] self.completer.model().setStringList(self.search_history) gid_txt = self.group_id_edit.text().strip() if gid_txt: gids = self._parse_group_ids(gid_txt) if not gids: QMessageBox.warning(self, "안내", "group_id는 숫자(여러 개는 쉼표/공백 구분)로 입력해주세요.") return self.search_by_group_ids(gids) elif self.icode_edit.text().strip(): icodes = self._parse_icodes(self.icode_edit.text().strip()) if not icodes: QMessageBox.warning(self, "안내", "상품코드는 숫자/영문(여러 개는 쉼표/공백/줄바꿈 구분)으로 입력해주세요.") return if len(icodes) == 1: self.search_by_icode(icodes[0]) else: self.search_by_icodes(icodes) elif keyword: self.search_by_iname_with_progress(keyword) def _parse_group_ids(self, text: str) -> list[int]: """ group_id 다중 입력 파서: - 구분자: 공백/쉼표/줄바꿈/탭 - 숫자만 허용, 중복 제거(입력 순서 유지) """ s = str(text or "").strip() if not s: return [] parts = re.split(r"[,\s]+", s) out: list[int] = [] seen = set() for p in parts: p = (p or "").strip() if not p: continue if not re.fullmatch(r"\d+", p): continue try: gid = int(p) except Exception: continue if gid in seen: continue seen.add(gid) out.append(gid) return out def _parse_icodes(self, text: str) -> list[str]: """ 상품코드(icode) 다중 입력 파서: - 구분자: 공백/쉼표/줄바꿈/탭 - 허용: 숫자/영문(대/소문자). (예: 59294839, W009EE8) - 중복 제거(입력 순서 유지) """ s = str(text or "").strip() if not s: return [] parts = re.split(r"[,\s]+", s) out: list[str] = [] seen = set() for p in parts: p = (p or "").strip() if not p: continue # 기본적으로 도메/OWN icode는 영문+숫자 조합이므로 이 범위만 허용 if not re.fullmatch(r"[0-9A-Za-z]+", p): continue key = p.upper() if key in seen: continue seen.add(key) out.append(p) return out # ========================= # SEARCH BY GROUP_ID # ========================= def search_by_group_ids(self, group_ids: list[int]): """ group_id(1개/다중)로 그룹 조회. - 1개면 기존처럼 상세까지 자동 로딩 - 여러 개면 좌측 그룹 리스트를 채우고, 첫 번째는 자동 로딩 """ gids = [int(x) for x in (group_ids or []) if x is not None] if not gids: return cur = self.conn.cursor(cursor_factory=DictCursor) # 입력 순서 유지 위해 WITH ORDINALITY 사용 cur.execute( """ WITH v AS ( SELECT * FROM unnest(%s::bigint[]) WITH ORDINALITY AS t(group_id, ord) ) SELECT g.group_id, g.winner_icode, g.winner_price, g.group_key, gm.iname AS winner_iname, gm.img_url AS winner_img_url, gm.content AS winner_content FROM v JOIN mlinkdw.shopprod_group2 g ON g.group_id = v.group_id LEFT JOIN mlinkdw.shopprod_group_map2 gm ON gm.group_id = g.group_id AND gm.icode = g.winner_icode ORDER BY v.ord """, (gids,), ) rows = cur.fetchall() if not rows: QMessageBox.information(self, "안내", "입력한 group_id를 찾지 못했습니다.") return self.group_model.load([ [r["group_id"], r["winner_icode"], r.get("winner_iname"), r["winner_price"], "—", r.get("winner_img_url"), r.get("group_key"), None, r.get("winner_content"), None] for r in rows ]) self._adjust_group_view_columns() self.sub_group_model.load([]) self._adjust_sub_group_view_columns() self.item_model.load([]) self._adjust_item_view_columns() # 1개면 곧장 상세 로딩, 다중이면 첫 번째만 자동 로딩(선택 변경 시 사용자가 클릭) try: first_gid = int(rows[0]["group_id"]) except Exception: first_gid = None if first_gid is not None: self.current_group_id = first_gid try: self.current_winner_icode = rows[0]["winner_icode"] self.current_winner_price = rows[0]["winner_price"] except Exception: self.current_winner_icode = None self.current_winner_price = None # 첫 row 선택 표시(UX) try: self.group_view.selectRow(0) except Exception: pass # 상세 로딩(기존 동작과 동일) self.load_sub_groups(clear_items=False) self.load_group_items(self.current_group_id) def search_by_iname_with_progress(self, iname: str, *, show_progress: bool | None = None): """ search_by_iname 실행 시 진행율(무한) 팝업 표시. 디버깅 옵션: - env: DOME_UI_NO_INAME_PROGRESS=1 이면 progress 팝업을 띄우지 않는다. - 또는 검색 실행 시 Shift 키를 누르고 있으면 progress 팝업을 띄우지 않는다. (단, progress만 생략하고 검색 자체는 QThread에서 수행하여 UI 프리징은 방지한다.) """ min_ratio = int(self.ratio_edit.text() or 70) if show_progress is None: v = str(os.environ.get("DOME_UI_NO_INAME_PROGRESS") or "").strip().lower() show_progress = not (v in ("1", "true", "yes", "y", "on")) try: if QApplication.keyboardModifiers() & Qt.ShiftModifier: show_progress = False except Exception: pass # 진행 팝업(옵션) self._iname_progress = None if bool(show_progress): self._iname_progress = QProgressDialog("상품명 검색중...", "취소", 0, 0, self) self._iname_progress.setWindowTitle("검색") self._iname_progress.setWindowModality(Qt.ApplicationModal) self._iname_progress.setMinimumDuration(0) self._iname_progress.setAutoClose(False) self._iname_progress.setAutoReset(False) self._iname_progress.show() QApplication.processEvents() # 워커 스레드 self._iname_thread = QThread(self) self._iname_worker = InameSearchWorker(iname=iname, min_ratio=min_ratio) self._iname_worker.moveToThread(self._iname_thread) def _cleanup(): if self._iname_progress is not None: try: self._iname_progress.close() except Exception: pass self._iname_thread.quit() def _on_cancel(): try: self._iname_worker.cancel() except Exception: pass _cleanup() def _on_done(group_rows: list): self.group_model.load(group_rows) self._adjust_group_view_columns() self.sub_group_model.load([]) self._adjust_sub_group_view_columns() self.item_model.load([]) self._adjust_item_view_columns() if not group_rows: QMessageBox.information(self, "안내", "이 상품은 여기 없습니다") _cleanup() def _on_err(msg: str): QMessageBox.critical(self, "오류", msg) _cleanup() if self._iname_progress is not None: self._iname_progress.canceled.connect(_on_cancel) self._iname_thread.started.connect(self._iname_worker.run) self._iname_worker.finished.connect(_on_done) self._iname_worker.error.connect(_on_err) self._iname_worker.finished.connect(self._iname_worker.deleteLater) self._iname_worker.error.connect(self._iname_worker.deleteLater) self._iname_thread.finished.connect(self._iname_thread.deleteLater) self._iname_thread.start() def _db_has_pg_trgm(self) -> bool: """ DB에 pg_trgm extension이 설치되어 있는지 확인(1회 캐시). """ if self._pg_trgm_available is not None: return bool(self._pg_trgm_available) try: cur = self.conn.cursor() cur.execute("SELECT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'pg_trgm')") self._pg_trgm_available = bool(cur.fetchone()[0]) except Exception: self._pg_trgm_available = False return bool(self._pg_trgm_available) # # ========================= # # SEARCH BY INAME # # ========================= # def search_by_iname(self, iname): # if self.conn.closed: # self.conn = psycopg2.connect(**DB_INFO) # min_ratio = int(self.ratio_edit.text() or 70) # cur = self.conn.cursor(cursor_factory=DictCursor) # # 1) 검색어 정규화 + 토큰화 # norm = _normalize_iname(iname) # # 유사% 100 이상이면: "포함"이 아니라 "정확히 동일한 상품명"만 조회 # if min_ratio >= 100: # exact_key = _norm_iname_db_exact(iname) # cur.execute(""" # SELECT # g.group_id, # m.icode as winner_icode, # m.price as winner_price, # m.iname AS winner_iname, # m.img_url AS winner_img_url, # m.content AS winner_content, # g.group_key, # m.img_url AS match_img_url # FROM mlinkdw.shopprod_group2 g # JOIN mlinkdw.shopprod_group_map2 m ON m.group_id = g.group_id # WHERE lower(regexp_replace(trim(m.iname), '\s+', ' ', 'g')) = %s # LIMIT 20000 # """, (exact_key,)) # group_best = {} # for r in cur.fetchall(): # gid = r["group_id"] # dist = _hashab_value(r.get("winner_img_url"), r.get("match_img_url")) # ph_dh = _image_hash_pair_from_url(str(r.get("match_img_url") or "")) # if ph_dh: # ph_hex, dh_hex = str(ph_dh[0]), str(ph_dh[1]) # else: # ph_hex, dh_hex = None, None # flow = _simulate_match_flow(self.conn, ph_hex, dh_hex, threshold=12) # group_best[gid] = dict( # group_id=gid, # winner=r["winner_icode"], # iname=(r.get("winner_iname") or ""), # price=r["winner_price"], # sim=100, # img_url=r.get("winner_img_url"), # group_key=r["group_key"], # hashAB=dist, # content=r.get("winner_content"), # match_flow=flow, # ) # rows = sorted(group_best.values(), key=lambda x: (_sort_price(x.get("price")), -x.get("sim", 0))) # self.group_model.load([ # [r["group_id"], r["winner"], r["iname"], r["price"], r["sim"], r["img_url"], r["group_key"], r["hashAB"], r.get("content"), r["match_flow"]] # for r in rows # ]) # self._adjust_group_view_columns() # self.sub_group_model.load([]) # self._adjust_sub_group_view_columns() # self.item_model.load([]) # self._adjust_item_view_columns() # return # tokens = [t for t in norm.split(" ") if t] # # 2) DB 후보 축소: 그룹 내 전체 상품명 대상으로 검색 (요구사항) # # 너무 짧은 토큰은 제외 # tokens = [t for t in tokens if len(t) >= 2][:5] # if not tokens: # # 너무 짧거나 공백/특수문자만 있는 경우 # self.group_model.load([]) # self._adjust_group_view_columns() # self.sub_group_model.load([]) # self._adjust_sub_group_view_columns() # self.item_model.load([]) # self._adjust_item_view_columns() # return # # 첫 토큰은 prefix 매칭을 우선 적용(인덱스 활용 가능성 ↑), 나머지는 contains # try: # from iname_search import build_ilike_where_clause, db_has_pg_trgm_sync # wc = build_ilike_where_clause(tokens, placeholder_style="psycopg2") # where = wc.sql # params = list(wc.params) # pg_trgm_ok = db_has_pg_trgm_sync(self.conn) # except Exception: # # 기존 방식 fallback # where_parts = [] # params = [] # for i, t in enumerate(tokens): # if i == 0 and len(t) >= 3: # where_parts.append("m.iname ILIKE %s") # params.append(f"{t}%") # else: # where_parts.append("m.iname ILIKE %s") # params.append(f"%{t}%") # where = " AND ".join(where_parts) # pg_trgm_ok = self._db_has_pg_trgm() # # pg_trgm 사용 가능하면: DB에서 유사도(similarity) 계산/필터/그룹별 best 1건 추출 # if pg_trgm_ok: # try: # # similarity()는 0~1.0 → %로 변환 # sim_threshold = max(0.0, min(1.0, min_ratio / 100.0)) # # group_id 별로 similarity 최대 1건만 남김 (DISTINCT ON) # cur.execute(f""" # SELECT DISTINCT ON (t.group_id) # t.group_id, # t.winner_icode, # t.winner_price, # t.iname AS iname, # t.img_url AS img_url, # t.content AS content, # t.group_key AS group_key, # t.sim # FROM ( # SELECT # g.group_id, # g.winner_icode, # g.winner_price, # (SELECT iname FROM mlinkdw.shopprod_group_map2 WHERE group_id = g.group_id and icode = g.winner_icode) as iname, # (SELECT img_url FROM mlinkdw.shopprod_group_map2 WHERE group_id = g.group_id and icode = g.winner_icode) as img_url, # (SELECT content FROM mlinkdw.shopprod_group_map2 WHERE group_id = g.group_id and icode = g.winner_icode) as content, # g.group_key, # mlinkdw.similarity(lower(m.iname), lower(%s)) AS sim # FROM mlinkdw.shopprod_group2 g # JOIN mlinkdw.shopprod_group_map2 m ON m.group_id = g.group_id # WHERE {where} # ) t # WHERE t.sim >= %s # ORDER BY t.group_id, t.sim DESC # LIMIT 5000 # """, tuple([iname] + params + [sim_threshold])) # group_best = {} # for r in cur.fetchall(): # # DictCursor row: r["sim"] is float/Decimal # sim_pct = int(round(float(r["sim"]) * 100)) # gid = r["group_id"] # group_best[gid] = dict( # group_id=gid, # winner=r["winner_icode"], # iname=r["iname"], # price=r["winner_price"], # sim=sim_pct, # img_url=r["img_url"], # content=r.get("content"), # group_key=r["group_key"] # ) # rows = sorted( # group_best.values(), # key=lambda x: (_sort_price(x.get("price")), -x.get("sim", 0)), # ) # self.group_model.load([ # [r["group_id"], r["winner"], r["iname"], r["price"], r["sim"], r["img_url"], r["group_key"], None, r.get("content"), None] # for r in rows # ]) # self._adjust_group_view_columns() # self.sub_group_model.load([]) # self._adjust_sub_group_view_columns() # self.item_model.load([]) # self._adjust_item_view_columns() # return # except Exception: # # 권한/extension 미설치/함수 미존재 등 → 기존 방식으로 fallback # self.conn.rollback() # self._pg_trgm_available = False # # 후보 상한(너무 흔한 단어 검색 시 UI 프리징 방지) - fallback(rapidfuzz) # cur.execute(f""" # SELECT # g.group_id, # g.winner_icode, # g.winner_price, # (SELECT iname FROM mlinkdw.shopprod_group_map2 WHERE group_id = g.group_id AND icode = g.winner_icode) AS winner_iname, # (SELECT img_url FROM mlinkdw.shopprod_group_map2 WHERE group_id = g.group_id AND icode = g.winner_icode) AS winner_img_url, # (SELECT content FROM mlinkdw.shopprod_group_map2 WHERE group_id = g.group_id AND icode = g.winner_icode) AS winner_content, # m.iname AS match_iname, # m.img_url AS match_img_url, # g.group_key # FROM mlinkdw.shopprod_group2 g # JOIN mlinkdw.shopprod_group_map2 m ON m.group_id = g.group_id # WHERE {where} # LIMIT 20000 # """, tuple(params)) # group_best = {} # for r in cur.fetchall(): # sim = calc_iname_similarity_advanced(iname, r["match_iname"]) # if sim < min_ratio: # continue # gid = r["group_id"] # if gid not in group_best or sim > group_best[gid]["sim"]: # group_best[gid] = dict( # group_id=gid, # winner=r["winner_icode"], # iname=r.get("winner_iname"), # price=r["winner_price"], # sim=sim, # img_url=r.get("winner_img_url"), # content=r.get("winner_content"), # group_key=r["group_key"] # ) # rows = sorted( # group_best.values(), # key=lambda x: (_sort_price(x.get("price")), -x.get("sim", 0)), # ) # self.group_model.load([ # [r["group_id"], r["winner"], r["iname"], r["price"], r["sim"], r["img_url"], r["group_key"], None, r.get("content"), None] # for r in rows # ]) # self._adjust_group_view_columns() # self.sub_group_model.load([]) # self._adjust_sub_group_view_columns() # self.item_model.load([]) # self._adjust_item_view_columns() # ========================= # SEARCH BY ICODE # ========================= def search_by_icode(self, icode): self.item_model.highlight_icode = icode self.item_model.highlight_icodes = {str(icode).upper()} self.sub_group_model.highlight_icode = icode self.sub_group_model.highlight_icodes = {str(icode).upper()} cur = self.conn.cursor(cursor_factory=DictCursor) # 1️⃣ 해당 상품이 속한 group 찾기 cur.execute(""" SELECT g.group_id, g.winner_icode, g.winner_price, g.group_key FROM mlinkdw.shopprod_group2 g JOIN mlinkdw.shopprod_group_map2 m ON m.group_id = g.group_id WHERE m.icode = %s LIMIT 1 """, (icode,)) g = cur.fetchone() if not g: return group_id = g["group_id"] # winner 상품명(iname) / img_url cur.execute(""" SELECT iname, img_url, content FROM mlinkdw.shopprod_group_map2 WHERE icode = %s LIMIT 1 """, (g["winner_icode"],)) winner_row = cur.fetchone() winner_iname = winner_row["iname"] if winner_row else None winner_img_url = winner_row["img_url"] if winner_row else None winner_content = winner_row["content"] if winner_row else None winner_group_key = g["group_key"] if g else None # 2️⃣ 좌측 Group 1건 표시 self.group_model.load([ [group_id, g["winner_icode"], winner_iname, g["winner_price"], "—", winner_img_url, winner_group_key, None, winner_content, None] ]) self._adjust_group_view_columns() self.current_group_id = group_id self.current_winner_icode = g["winner_icode"] self.current_winner_price = g["winner_price"] # 3️⃣ SubGroup 전체 로딩 (있으면) cur.execute(""" SELECT sg.sub_group_id, sgm.icode, sgm.iname, sgm.price, sgm.vender_code FROM mlinkdw.shopprod_sub_group2 sg JOIN mlinkdw.shopprod_sub_group_map2 sgm ON sgm.sub_group_id = sg.sub_group_id WHERE sg.group_id = %s """, (group_id,)) sub_rows = [ [r["sub_group_id"], r["icode"], r["iname"], 0, r["price"], r["vender_code"]] for r in cur.fetchall() ] self.sub_group_model.load(sub_rows) self._adjust_sub_group_view_columns() # 4️⃣ group_id 전체 상품 로딩 (하단) cur.execute(""" SELECT vender_code, icode, iname, price, img_url, content FROM mlinkdw.shopprod_group_map2 WHERE group_id = %s ORDER BY price """, (group_id,)) items = [] for r in cur.fetchall(): items.append([ "", # row number (filled after sort) r["vender_code"], r["icode"], r["iname"], r["price"], r.get("img_url"), r.get("content"), 0, # similarity (icode 검색이므로 의미 없음) 0, # price_score 0, # image_score 0, # merge_score ]) # row number 채우기 for i, row in enumerate(items, start=1): row[0] = i self.item_model.load(items) self._adjust_item_view_columns() # 5️⃣ 하이라이트 delegate에 icode 전달 self.highlight_delegate.set_search_text("") # 이름 하이라이트 제거 self.item_model.highlight_icode = icode self.item_model.highlight_icodes = {str(icode).upper()} self.item_view.viewport().update() # ========================= # SEARCH BY ICODE (MULTI) # ========================= def search_by_icodes(self, icodes: list[str]): """ 상품코드(다중)로 그룹 조회. - 입력한 icode 순서 유지 - 여러 icode가 같은 group_id로 매핑되면 group은 1번만 표시 - 첫 번째 group은 자동 로딩(기존 group_id 다중검색 UX와 동일) """ codes = [str(x).strip() for x in (icodes or []) if str(x or "").strip()] if not codes: return # group별로 어떤 input_icode가 매칭되었는지 저장(클릭 시 하이라이트 갱신에 사용) self._group_input_icodes: dict[int, list[str]] = {} cur = self.conn.cursor(cursor_factory=DictCursor) cur.execute( """ WITH v AS ( SELECT * FROM unnest(%s::text[]) WITH ORDINALITY AS t(icode, ord) ), found AS ( SELECT v.ord, v.icode AS input_icode, m.group_id, g.winner_icode, g.winner_price, g.group_key, gm.iname AS winner_iname, gm.img_url AS winner_img_url, gm.content AS winner_content FROM v LEFT JOIN mlinkdw.shopprod_group_map2 m ON m.icode = v.icode LEFT JOIN mlinkdw.shopprod_group2 g ON g.group_id = m.group_id LEFT JOIN mlinkdw.shopprod_group_map2 gm ON gm.group_id = g.group_id AND gm.icode = g.winner_icode ) SELECT * FROM found WHERE group_id IS NOT NULL ORDER BY ord """, (codes,), ) rows = cur.fetchall() if not rows: QMessageBox.information(self, "안내", "입력한 상품코드를 찾지 못했습니다.") return # group_id -> input_icode 리스트 구성(입력 순서 유지) for r in rows: try: gid = r.get("group_id") if gid is None: continue gid_i = int(gid) code = str(r.get("input_icode") or "").strip() if not code: continue key = code.upper() arr = self._group_input_icodes.setdefault(gid_i, []) if key not in {x.upper() for x in arr}: arr.append(code) except Exception: continue # 동일 group_id 중복 제거(입력 순서 유지) seen_gid = set() uniq = [] for r in rows: gid = r.get("group_id") if gid in seen_gid: continue seen_gid.add(gid) uniq.append(r) # 다중 입력인 경우: 첫 화면에서는 전체 입력 icode를 모두 하이라이트(노란색) try: self.item_model.highlight_icode = None self.sub_group_model.highlight_icode = None self.item_model.highlight_icodes = {c.upper() for c in codes} self.sub_group_model.highlight_icodes = {c.upper() for c in codes} except Exception: pass self.group_model.load([ [r["group_id"], r.get("winner_icode"), r.get("winner_iname"), r.get("winner_price"), "—", r.get("winner_img_url"), r.get("group_key"), None, r.get("winner_content"), None] for r in uniq ]) self._adjust_group_view_columns() self.sub_group_model.load([]) self._adjust_sub_group_view_columns() self.item_model.load([]) self._adjust_item_view_columns() # 첫 번째 group 자동 로딩 try: first_gid = int(uniq[0]["group_id"]) except Exception: first_gid = None if first_gid is not None: self.current_group_id = first_gid try: self.current_winner_icode = uniq[0].get("winner_icode") self.current_winner_price = uniq[0].get("winner_price") except Exception: self.current_winner_icode = None self.current_winner_price = None try: self.group_view.selectRow(0) except Exception: pass self.load_sub_groups(clear_items=False) self.load_group_items(self.current_group_id) # ========================= # GROUP CLICK # ========================= def on_group_clicked(self, index): self.current_group_id = self.group_model.rows[index.row()][0] # 다중 icode 검색 결과에서 group row 클릭 시: 해당 group에 매핑된 input_icode(들)만 노란색 표시 try: gid = int(self.current_group_id) codes = getattr(self, "_group_input_icodes", {}).get(gid) or [] if codes: self.item_model.highlight_icode = None self.sub_group_model.highlight_icode = None self.item_model.highlight_icodes = {c.upper() for c in codes} self.sub_group_model.highlight_icodes = {c.upper() for c in codes} else: # 매핑이 없으면 하이라이트 해제(기존 동작 유지) self.item_model.highlight_icodes = None self.sub_group_model.highlight_icodes = None except Exception: pass # Group 클릭 시: SubGroup + Item(그룹 전체) 동시 로드 self.load_sub_groups(clear_items=False) self.load_group_items(self.current_group_id) def on_group_double_clicked(self, index): """ - img_url 컬럼 더블클릭 시 단일 이미지 팝업 - content 컬럼 더블클릭 시 HTML 팝업 """ col = index.column() # img_url if col == 5: try: img_url = self.group_model.rows[index.row()][5] except Exception: img_url = None if not img_url: return title = None try: title = str(self.group_model.rows[index.row()][2] or "IMAGE") except Exception: title = None dlg = ImageDialog(str(img_url), title=title, parent=self) self._show_modeless(dlg) return # content if col == 8: try: html = self.group_model.rows[index.row()][8] except Exception: html = None if not html: QMessageBox.information(self, "안내", "content가 비어있습니다.") return title = None try: title = str(self.group_model.rows[index.row()][2] or "CONTENT") except Exception: title = None # content 이미지 리소스가 많으면 렌더링까지 오래 걸릴 수 있어 선다운로드 + progress 표시 html_text = str(html) # 워커 스레드 준비 self._content_thread = QThread(self) self._content_worker = ContentPrefetchWorker(html_text) self._content_worker.moveToThread(self._content_thread) # 진행 팝업(처음엔 indeterminate, total 알면 determinate) prog = QProgressDialog("content 로딩중...", "취소", 0, 0, self) prog.setWindowTitle("CONTENT") prog.setWindowModality(Qt.ApplicationModal) prog.setMinimumDuration(0) prog.setAutoClose(False) prog.setAutoReset(False) def _cleanup(): try: prog.close() except Exception: pass try: self._content_thread.quit() except Exception: pass def _on_cancel(): try: self._content_worker.cancel() except Exception: pass _cleanup() def _on_progress(done: int, total: int, url: str): try: if total and prog.maximum() != total: prog.setRange(0, total) if total: prog.setValue(done) # 너무 긴 URL은 잘라서 표시 if url: u = url if len(url) <= 120 else (url[:120] + "...") prog.setLabelText(f"content 로딩중... ({done}/{total})\n{u}" if total else f"content 로딩중...\n{u}") except Exception: pass def _on_done(cache: dict): _cleanup() dlg = HtmlContentDialog(html_text, title=title, parent=self, prefetched=cache) self._show_modeless(dlg) def _on_err(msg: str): _cleanup() QMessageBox.warning(self, "오류", f"content 로딩 실패: {msg}\n(원본 HTML은 팝업에서 확인 가능합니다.)") dlg = HtmlContentDialog(html_text, title=title, parent=self, prefetched=None) self._show_modeless(dlg) prog.canceled.connect(_on_cancel) self._content_thread.started.connect(self._content_worker.run) self._content_worker.progress.connect(_on_progress) self._content_worker.finished.connect(_on_done) self._content_worker.error.connect(_on_err) self._content_worker.finished.connect(self._content_worker.deleteLater) self._content_worker.error.connect(self._content_worker.deleteLater) self._content_thread.finished.connect(self._content_thread.deleteLater) prog.show() QApplication.processEvents() self._content_thread.start() return def on_item_double_clicked(self, index): """ item_view 더블클릭: - img_url 컬럼 더블클릭 시: (좌)현재 선택된 group_model img_url vs (우)해당 item img_url 비교 팝업 - content 컬럼 더블클릭 시: (좌)현재 선택된 group_model content vs (우)해당 item content 비교 팝업(일반 텍스트) """ col = index.column() # item_model columns: [#, Vendor, ICode, Name, Price, img_url, content, Similarity, PriceScore, ImageScore, MERGE_SCORE] img_url_col = 5 content_col = 6 if col not in (img_url_col, content_col): return # 왼쪽(group) img_url: 현재 group_view 선택 row 우선, 없으면 첫 row gidx = self.group_view.currentIndex() grow = gidx.row() if gidx.isValid() else 0 try: group_url = self.group_model.rows[grow][5] if self.group_model.rows else None group_name = self.group_model.rows[grow][2] if self.group_model.rows else None group_content = self.group_model.rows[grow][8] if self.group_model.rows else None except Exception: group_url = None group_name = None group_content = None item_name = None try: item_name = self.item_model.rows[index.row()][3] except Exception: item_name = None # content 비교 if col == content_col: try: item_content = self.item_model.rows[index.row()][content_col] except Exception: item_content = None if not item_content: QMessageBox.information(self, "안내", "선택한 item의 content가 비어있습니다.") return # content 이미지 리소스가 많으면 렌더링까지 오래 걸릴 수 있어 선다운로드 + progress 표시 item_html = str(item_content or "") group_html = str(group_content or "") # group content가 없으면 item 단일 팝업(프리페치 포함) if not group_html: prog = QProgressDialog("content 로딩중...", "취소", 0, 0, self) prog.setWindowTitle("CONTENT 로딩") prog.setMinimumDuration(0) prog.setAutoClose(True) prog.setAutoReset(True) prog.setWindowModality(Qt.WindowModal) self._item_content_thread = QThread(self) self._item_content_worker = ContentPrefetchWorker(item_html) self._item_content_worker.moveToThread(self._item_content_thread) def _cleanup_one(): try: prog.close() except Exception: pass try: self._item_content_thread.quit() self._item_content_thread.wait(1500) except Exception: pass def _on_cancel_one(): try: self._item_content_worker.cancel() except Exception: pass _cleanup_one() def _on_progress_one(done: int, total: int, url: str): try: if total and prog.maximum() != total: prog.setRange(0, total) if total: prog.setValue(done) if url: u = url if len(url) <= 120 else (url[:120] + "...") prog.setLabelText(f"content 로딩중... ({done}/{total})\n{u}" if total else f"content 로딩중...\n{u}") except Exception: pass def _on_done_one(cache: dict): _cleanup_one() dlg = HtmlContentDialog(item_html, title=str(item_name or "CONTENT"), parent=self, prefetched=cache) self._show_modeless(dlg) def _on_err_one(msg: str): _cleanup_one() QMessageBox.warning(self, "오류", f"content 로딩 실패: {msg}\n(원본 HTML은 팝업에서 확인 가능합니다.)") dlg = HtmlContentDialog(item_html, title=str(item_name or "CONTENT"), parent=self, prefetched=None) self._show_modeless(dlg) prog.canceled.connect(_on_cancel_one) self._item_content_thread.started.connect(self._item_content_worker.run) self._item_content_worker.progress.connect(_on_progress_one) self._item_content_worker.finished.connect(_on_done_one) self._item_content_worker.error.connect(_on_err_one) self._item_content_worker.finished.connect(self._item_content_worker.deleteLater) self._item_content_worker.error.connect(self._item_content_worker.deleteLater) self._item_content_thread.finished.connect(self._item_content_thread.deleteLater) prog.show() QApplication.processEvents() self._item_content_thread.start() return # group + item 비교창(프리페치 포함): 두 content의 img src를 합쳐서 한번에 선다운로드 combined_html = group_html + "\n" + item_html prog = QProgressDialog("content 로딩중...", "취소", 0, 0, self) prog.setWindowTitle("CONTENT 로딩") prog.setMinimumDuration(0) prog.setAutoClose(True) prog.setAutoReset(True) prog.setWindowModality(Qt.WindowModal) self._item_content_cmp_thread = QThread(self) self._item_content_cmp_worker = ContentPrefetchWorker(combined_html) self._item_content_cmp_worker.moveToThread(self._item_content_cmp_thread) def _cleanup_cmp(): try: prog.close() except Exception: pass try: self._item_content_cmp_thread.quit() self._item_content_cmp_thread.wait(1500) except Exception: pass def _on_cancel_cmp(): try: self._item_content_cmp_worker.cancel() except Exception: pass _cleanup_cmp() def _on_progress_cmp(done: int, total: int, url: str): try: if total and prog.maximum() != total: prog.setRange(0, total) if total: prog.setValue(done) if url: u = url if len(url) <= 120 else (url[:120] + "...") prog.setLabelText(f"content 로딩중... ({done}/{total})\n{u}" if total else f"content 로딩중...\n{u}") except Exception: pass def _on_done_cmp(cache: dict): _cleanup_cmp() dlg = SideBySideHtmlContentDialog( left_title=f"GROUP (선택) : {str(group_name or '')}", left_html=group_html, right_title=f"ITEM : {str(item_name or '')}", right_html=item_html, title="CONTENT 비교 (Group vs Item)", parent=self, left_prefetched=cache, right_prefetched=cache, ) self._show_modeless(dlg) def _on_err_cmp(msg: str): _cleanup_cmp() QMessageBox.warning(self, "오류", f"content 로딩 실패: {msg}\n(원본 HTML은 팝업에서 확인 가능합니다.)") dlg = SideBySideHtmlContentDialog( left_title=f"GROUP (선택) : {str(group_name or '')}", left_html=group_html, right_title=f"ITEM : {str(item_name or '')}", right_html=item_html, title="CONTENT 비교 (Group vs Item)", parent=self, left_prefetched=None, right_prefetched=None, ) self._show_modeless(dlg) prog.canceled.connect(_on_cancel_cmp) self._item_content_cmp_thread.started.connect(self._item_content_cmp_worker.run) self._item_content_cmp_worker.progress.connect(_on_progress_cmp) self._item_content_cmp_worker.finished.connect(_on_done_cmp) self._item_content_cmp_worker.error.connect(_on_err_cmp) self._item_content_cmp_worker.finished.connect(self._item_content_cmp_worker.deleteLater) self._item_content_cmp_worker.error.connect(self._item_content_cmp_worker.deleteLater) self._item_content_cmp_thread.finished.connect(self._item_content_cmp_thread.deleteLater) prog.show() QApplication.processEvents() self._item_content_cmp_thread.start() return # img_url 비교 # 오른쪽(item) img_url try: item_url = self.item_model.rows[index.row()][img_url_col] except Exception: item_url = None if not item_url: return # group img_url이 없으면 단일 이미지로만 표시 if not group_url: dlg = ImageDialog(str(item_url), title=str(item_name or "IMAGE"), parent=self) self._show_modeless(dlg) return dlg = SideBySideImageDialog( left_title=f"GROUP (선택) : {str(group_name or '')}", left_url=str(group_url), right_title=f"ITEM : {str(item_name or '')}", right_url=str(item_url), title="이미지 비교 (Group vs Item)", parent=self, ) self._show_modeless(dlg) return def on_compare_images_clicked(self): """ Group 테이블에서 선택된 여러 row의 img_url을 비교 팝업으로 표시. """ sm = self.group_view.selectionModel() if sm is None: return selected_rows = sm.selectedRows() # row 단위 if len(selected_rows) < 2: QMessageBox.information(self, "안내", "비교할 이미지를 2개 이상 선택해주세요.") return # 선택 row에서 (name, img_url, group_key) 수집 items = [] seen = set() for mi in selected_rows: r = mi.row() try: img_url = self.group_model.rows[r][5] name = self.group_model.rows[r][2] group_key = self.group_model.rows[r][6] if len(self.group_model.rows[r]) > 6 else None except Exception: continue if not img_url: continue key = (str(name), str(img_url), str(group_key)) if key in seen: continue seen.add(key) items.append({"name": name, "img_url": img_url, "group_key": group_key}) if len(items) < 2: QMessageBox.information(self, "안내", "선택된 항목 중 img_url이 2개 이상 필요합니다.") return if len(items) == 2: # 2개면 기존 2분할 비교 다이얼로그 재사용 n0 = f'{items[0]["name"]}\n(group_key: {items[0].get("group_key") or "-"})' n1 = f'{items[1]["name"]}\n(group_key: {items[1].get("group_key") or "-"})' dlg = CompareThumbnailDialog( n0, items[0]["img_url"], n1, items[1]["img_url"], self.iname_edit.text().strip(), parent=self ) dlg.exec() return dlg = MultiImageCompareDialog(items, parent=self) dlg.exec() def on_compare_content_images_clicked(self): """ Group 테이블에서 선택된 여러 row의 winner_content(HTML)를 '이미지로 렌더링'한 뒤 pHash/dHash 거리로 유사도를 판단. - 첫 번째 선택을 기준(target)으로 나머지를 비교 - PASS 기준(보수적): ph<=12 and dh<=14 """ if imagehash is None or Image is None: QMessageBox.warning(self, "안내", "CONTENT(이미지) 판단은 imagehash/PIL 패키지가 필요합니다.") return sm = self.group_view.selectionModel() if sm is None: return selected_rows = sm.selectedRows() if len(selected_rows) < 2: QMessageBox.information(self, "안내", "비교할 그룹을 2개 이상 선택해주세요.") return # group_model columns: [GroupID, Winner, Iname, Price, Similarity, img_url, group_key, hashAB, content, match_flow] def _get_row(mi: QModelIndex): try: return self.group_model.rows[mi.row()] except Exception: return None base = _get_row(selected_rows[0]) if not base: return base_gid = base[0] base_iname = base[2] base_html = base[8] if len(base) > 8 else None ph_thr, dh_thr = 12, 14 min_matches = 2 min_ratio = 0.30 lines = [ f"TARGET group_id={base_gid} iname={str(base_iname or '')}", f"PASS 기준: (이미지 매칭) matches>= {min_matches} 또는 ratio>= {min_ratio:.2f} (개별 매칭 조건: ph<= {ph_thr}, dh<= {dh_thr})", "", ] for mi in selected_rows[1:]: r = _get_row(mi) if not r: continue gid = r[0] iname = r[2] html = r[8] if len(r) > 8 else None st = _content_imgset_match_stats(base_html, html, max_imgs=12, ph_thr=ph_thr, dh_thr=dh_thr) if st.get("err"): lines.append(f"- group_id={gid} iname={str(iname or '')}: (실패: {st.get('err')})") continue match = int(st.get("match") or 0) ratio = float(st.get("ratio") or 0.0) ok = (match >= min_matches) or (ratio >= min_ratio) lines.append( f"- group_id={gid} iname={str(iname or '')}: " f"imgsA={st.get('na')} imgsB={st.get('nb')} match={match} ratio={ratio:.2f} " f"(min_ph={st.get('min_ph')} min_dh={st.get('min_dh')}) => {'PASS' if ok else 'FAIL'}" ) QMessageBox.information(self, "CONTENT(이미지) 판단", "\n".join(lines)) def _get_selected_group_rows(self): sm = self.group_view.selectionModel() if sm is None: return [] rows = [] for mi in sm.selectedRows(): # row 단위 r = mi.row() # view row index try: row = self.group_model.rows[r] except Exception: continue rows.append((r, row)) return rows def on_merge_groups_clicked(self): """ 선택된 N개 그룹을 병합(매핑 이동). - TARGET = 선택된 row 중 화면에서 가장 위(최상단) row의 group_id - SOURCE = 그 외 선택된 모든 group_id (TARGET으로 순차 병합) """ sel = self._get_selected_group_rows() if len(sel) < 2: QMessageBox.information(self, "안내", "병합할 그룹을 2개 이상 선택해주세요.") return # group_model columns: [GroupID, Winner, Iname, Price, Similarity, img_url, group_key, hashAB, content, match_flow] try: # 최상단 row를 TARGET으로 sel_sorted = sorted(sel, key=lambda x: x[0]) target_row = sel_sorted[0][1] source_rows = [r for _, r in sel_sorted[1:]] target_gid = int(target_row[0]) target_img = target_row[5] except Exception: QMessageBox.critical(self, "오류", "선택된 행에서 group_id/img_url을 읽지 못했습니다.") return # 동일 판정(비교 다이얼로그와 동일 임계치) if not (Image and imagehash): QMessageBox.critical(self, "오류", "해시 라이브러리(pillow/imagehash)가 없어 병합 판정을 할 수 없습니다.") return pa = _image_hash_pair_from_url(str(target_img or "")) if not pa: QMessageBox.critical(self, "오류", "TARGET 이미지 해시 계산 실패로 병합할 수 없습니다.") return ph_a, dh_a = pa ph_thr = 12 dh_thr = 14 # SOURCE별 거리 요약 source_infos = [] for sr in source_rows: try: sgid = int(sr[0]) simg = sr[5] except Exception: continue if sgid == target_gid: continue pb = _image_hash_pair_from_url(str(simg or "")) if not pb: source_infos.append((sgid, None, None, False, "HASH_FAIL")) continue ph_b, dh_b = pb ph_dist = ph_a - ph_b dh_dist = dh_a - dh_b same = (ph_dist <= ph_thr) and (dh_dist <= dh_thr) source_infos.append((sgid, ph_dist, dh_dist, same, "OK")) if not source_infos: QMessageBox.information(self, "안내", "TARGET 외 병합 가능한 SOURCE가 없습니다.") return lines = [] for sgid, ph_dist, dh_dist, same, status in source_infos: if status != "OK": lines.append(f"- SOURCE {sgid}: 해시 계산 실패") else: lines.append( f"- SOURCE {sgid}: {'동일(추천)' if same else '상이'} " f"(ph={ph_dist}<= {ph_thr}, dh={dh_dist}<= {dh_thr})" ) msg = ( f"TARGET(group_id): {target_gid}\n" f"SOURCE(group_id): {', '.join(str(x[0]) for x in source_infos)}\n\n" f"거리 판정 요약:\n" + "\n".join(lines) + "\n\n" f"DB에서 SOURCE 그룹들의 매핑을 TARGET 그룹으로 이동하시겠습니까?" ) if QMessageBox.question(self, "그룹 병합", msg) != QMessageBox.Yes: return # DB 병합: group_map / sub_group 의 group_id를 target으로 이동 (다중 SOURCE) try: cur = self.conn.cursor() cur.execute("BEGIN") source_gids = [int(x[0]) for x in source_infos] # content_token 정리 정책: # - winner_icode(=winner 상품 1개)의 token만 유지하는 구조이므로 # target에서는 winner가 아닌 token은 제거하되, # SOURCE 그룹들의 token 중 TARGET에 없는 token은 TARGET winner 키로 "추가(누적)"한다. try: # 0) target winner 키 조회 (token을 winner 키로만 유지하기 위함) cur.execute( """ SELECT winner_vender_code, winner_icode FROM mlinkdw.shopprod_group2 WHERE group_id = %s """, (int(target_gid),), ) _wr = cur.fetchone() if _wr: target_winner_vender_code = _wr[0] target_winner_icode = _wr[1] else: target_winner_vender_code = None target_winner_icode = None # 1) target에서 winner가 아닌 token 제거 cur.execute( """ DELETE FROM mlinkdw.shopprod_group_content_token t WHERE t.group_id = %s AND EXISTS ( SELECT 1 FROM mlinkdw.shopprod_group2 g WHERE g.group_id = t.group_id AND (t.vender_code IS DISTINCT FROM g.winner_vender_code OR t.icode IS DISTINCT FROM g.winner_icode) ) """, (int(target_gid),), ) # 2) SOURCE token 중 TARGET에 없는 token만 TARGET(winner 키)에 추가 # - 비교 기준은 "token" (동일 token은 vender/icode가 달라도 중복으로 보지 않음) # - 저장은 winner 키로만 저장하여 1) 정리 정책에 의해 삭제되지 않게 한다. if target_winner_vender_code is not None and target_winner_icode is not None: cur.execute( """ INSERT INTO mlinkdw.shopprod_group_content_token (group_id, vender_code, icode, token, dome_code) SELECT %s AS group_id, %s AS vender_code, %s AS icode, s.token, COALESCE(s.dome_code, '') AS dome_code FROM mlinkdw.shopprod_group_content_token s WHERE s.group_id = ANY(%s) AND NOT EXISTS ( SELECT 1 FROM mlinkdw.shopprod_group_content_token t WHERE t.group_id = %s AND t.vender_code = %s AND t.icode = %s AND t.token = s.token ) ON CONFLICT (group_id, vender_code, icode, token) DO NOTHING """, ( int(target_gid), str(target_winner_vender_code), str(target_winner_icode), source_gids, int(target_gid), str(target_winner_vender_code), str(target_winner_icode), ), ) # 3) source token 삭제(고아/오염 방지) cur.execute( "DELETE FROM mlinkdw.shopprod_group_content_token WHERE group_id = ANY(%s)", (source_gids,), ) except Exception: # token 테이블/권한/스키마 미존재 등은 병합 자체를 막지 않음 pass for source_gid, *_rest in source_infos: # group_map 이동 cur.execute( """ UPDATE mlinkdw.shopprod_group_map2 SET group_id = %s WHERE group_id = %s """, (target_gid, source_gid), ) # sub_group 이동 cur.execute( """ UPDATE mlinkdw.shopprod_sub_group2 SET group_id = %s WHERE group_id = %s """, (target_gid, source_gid), ) # source 그룹이 완전히 비었으면 삭제 시도(실패해도 병합 유지) try: # content_token까지 비었을 때만 삭제(고아 데이터/후보 오염 방지) cur.execute( """ DELETE FROM mlinkdw.shopprod_group2 g WHERE g.group_id = %s AND NOT EXISTS (SELECT 1 FROM mlinkdw.shopprod_group_map2 gm WHERE gm.group_id = g.group_id) AND NOT EXISTS (SELECT 1 FROM mlinkdw.shopprod_sub_group2 sg WHERE sg.group_id = g.group_id) AND NOT EXISTS (SELECT 1 FROM mlinkdw.shopprod_group_content_token ct WHERE ct.group_id = g.group_id) """, (int(source_gid),), ) except Exception: # 삭제 실패는 무시 try: self.conn.rollback() except Exception: pass cur = self.conn.cursor() cur.execute("BEGIN") # 이후 병합 계속 cur.execute("COMMIT") except Exception as e: try: self.conn.rollback() except Exception: pass QMessageBox.critical(self, "오류", f"병합 실패: {e}") return QMessageBox.information( self, "완료", f"병합 완료: TARGET {target_gid} <- SOURCES {', '.join(str(x[0]) for x in source_infos)}" ) # 화면 갱신 self.on_search() def on_move_item_to_winner_group(self): """ item_view에서 선택한 상품(icode)을 기준으로: - shopprod_group에 winner_icode=icode 인 그룹이 있으면 그 group_id로 이동 - 없으면 shopprod_group을 생성 후 그 group_id로 이동 """ sm = self.item_view.selectionModel() if sm is None: QMessageBox.information(self, "안내", "하단 Item에서 먼저 항목을 선택해주세요.") return sel_rows = sm.selectedRows() if not sel_rows: # row 단위 선택이 아닐 수 있어 currentIndex fallback idx = self.item_view.currentIndex() if idx.isValid(): sel_rows = [idx] if not sel_rows: QMessageBox.information(self, "안내", "하단 Item에서 먼저 항목을 선택해주세요.") return # 중복 제거(같은 row가 여러 컬럼 선택으로 잡히는 경우 대비) uniq_row_idx = [] seen_r = set() for mi in sel_rows: rr = mi.row() if rr in seen_r: continue seen_r.add(rr) uniq_row_idx.append(rr) # 선택된 항목들을 읽어오기 selected_items = [] for r in sorted(uniq_row_idx): try: vender_code = self.item_model.rows[r][1] icode = self.item_model.rows[r][2] iname = self.item_model.rows[r][3] price = self.item_model.rows[r][4] img_url = self.item_model.rows[r][5] if len(self.item_model.rows[r]) > 5 else None except Exception: continue selected_items.append( dict( vender_code=str(vender_code or ""), icode=str(icode or ""), iname=str(iname or ""), price=price, img_url=img_url, ) ) selected_items = [it for it in selected_items if it["vender_code"] and it["icode"]] if not selected_items: QMessageBox.information(self, "안내", "선택된 항목에서 vender_code/icode를 읽지 못했습니다.") return # 전체 확인(1회) preview = "\n".join([f"- {it['vender_code']} / {it['icode']} / {it['iname']}" for it in selected_items[:10]]) more = f"\n... (+{len(selected_items) - 10} more)" if len(selected_items) > 10 else "" msg = ( f"선택 항목 {len(selected_items)}건을 WINNER 그룹으로 이동 처리합니다.\n\n" f"{preview}{more}\n\n" f"계속 진행하시겠습니까?" ) if QMessageBox.question(self, "WINNER 그룹으로 이동", msg) != QMessageBox.Yes: return ok = 0 fail = 0 details = [] for it in selected_items: try: self._move_one_item_to_winner_group( vender_code=it["vender_code"], icode=it["icode"], iname=it["iname"], price=it["price"], img_url=it.get("img_url"), ) ok += 1 details.append(f"[OK] {it['icode']}") except Exception as e: fail += 1 details.append(f"[FAIL] {it['icode']}: {e}") QMessageBox.information( self, "완료", f"WINNER 그룹 이동 처리 완료\n- OK: {ok}\n- FAIL: {fail}\n\n" + "\n".join(details[:50]) + ("\n... (truncated)" if len(details) > 50 else ""), ) # 결과 확인을 위해 마지막 선택 icode로 재조회 try: last_icode = selected_items[-1]["icode"] self.icode_edit.setText(str(last_icode)) self.search_by_icode(str(last_icode)) except Exception: pass return def _move_one_item_to_winner_group( self, *, vender_code: str, icode: str, iname: str, price, img_url, manage_tx: bool = True, ): """ 단일 item(=vender_code+icode)을 WINNER 그룹으로 이동(없으면 생성). 실패 시 Exception을 발생시켜 상위에서 집계. """ vender_code = str(vender_code or "") icode = str(icode or "") iname = str(iname or "") # 토큰 재생성에 사용할 content/dome_code (필요 시 DB에서 재조회) content_html = None dome_code = None try: price_i = int(price) if price is not None else None except Exception: price_i = None if not vender_code or not icode: raise RuntimeError("vender_code/icode가 비어있습니다.") if self.conn.closed: self.conn = psycopg2.connect(**DB_INFO) cur = self.conn.cursor(cursor_factory=DictCursor) # 현재 이 상품의 group_id cur.execute( """ SELECT group_id FROM mlinkdw.shopprod_group_map2 WHERE vender_code = %s AND icode = %s LIMIT 1 """, (vender_code, icode), ) row = cur.fetchone() current_gid = int(row["group_id"]) if row and row.get("group_id") is not None else None # winner 그룹 찾기 cur.execute( """ SELECT group_id FROM mlinkdw.shopprod_group2 WHERE winner_icode = %s ORDER BY group_id LIMIT 1 """, (icode,), ) row = cur.fetchone() target_gid = int(row["group_id"]) if row and row.get("group_id") is not None else None created_new = False # 없으면 신규 그룹 생성 if target_gid is None: # 신규 그룹 생성/매칭 로직은 dome_group_match.find_or_create_group_id_sync()로 위임 if not img_url: raise RuntimeError("img_url이 없어 신규 그룹 생성 불가") if not (Image and imagehash): raise RuntimeError("해시 라이브러리(pillow/imagehash) 없음") pair = _image_hash_pair_from_url(str(img_url)) if not pair: raise RuntimeError("img_url 해시 계산 실패") ph, dh = pair # 신규 그룹 생성 시 winner token 반영을 위해 content/dome_code를 전달 try: cur.execute( """ SELECT content, dome_code, vender_grade, reg_date FROM mlinkdw.shopprod_group_map2 WHERE vender_code = %s AND icode = %s LIMIT 1 """, (vender_code, icode), ) rr = cur.fetchone() if rr: try: content_html = rr["content"] dome_code = rr["dome_code"] vender_grade = rr.get("vender_grade") reg_date = rr.get("reg_date") except Exception: content_html = rr[0] dome_code = rr[1] if len(rr) > 1 else None vender_grade = rr[2] if len(rr) > 2 else None reg_date = rr[3] if len(rr) > 3 else None else: content_html = None dome_code = None vender_grade = None reg_date = None except Exception: content_html = None dome_code = None vender_grade = None reg_date = None import dome_group_match as _dgm target_gid = int( _dgm.find_or_create_group_id_sync( self.conn, {}, ph, dh, threshold=8, dhash_threshold=10, iname=iname, iname_threshold=30, fallback_iname_tokens=True, fallback_content_html=content_html, fallback_iname_dome_code=(str(dome_code) if dome_code is not None else None), fallback_content_min_matches=1, fallback_content_min_ratio=0.8, current_vender_code=vender_code, current_icode=icode, current_price=price_i, current_vender_grade=vender_grade, current_reg_date=reg_date, current_dome_code=(str(dome_code) if dome_code is not None else None), current_content_html=(str(content_html) if content_html is not None else None), manage_tx=manage_tx, current_img_url = img_url, color_check = False, ) ) created_new = False # find_or_create_group_id는 "기존 매칭/신규 생성" 모두 가능하여 여기서는 구분하지 않음 if target_gid is None: raise RuntimeError("대상 group_id 결정 실패") if current_gid is not None and int(current_gid) == int(target_gid): return # 이미 속해있으면 성공으로 간주 # group_id 변경 + 기존 그룹 비었으면 삭제 try: cur = self.conn.cursor(cursor_factory=DictCursor) if manage_tx: cur.execute("BEGIN") cur.execute( """ UPDATE mlinkdw.shopprod_group_map2 SET group_id = %s WHERE vender_code = %s AND icode = %s """, (target_gid, vender_code, icode), ) if cur.rowcount == 0: raise RuntimeError("shopprod_group_map 대상 row를 찾지 못했습니다.") # winner 재선정 + token 정리/재생성(+ moved token 누적)은 공통 로직(dome_group_match)로 위임 # - target_gid: 상품이 편입된 그룹(편입 직후 winner 재선정 필요) # - current_gid: 상품이 빠져나간 기존 그룹(기존 winner였던 상품이 빠졌다면 winner 재선정 필요) try: import dome_group_match as _dgm # 1) TARGET winner/token refresh _dgm.refresh_group_winner_and_tokens_if_needed_sync( self.conn, int(target_gid), cur=cur, force_token_cleanup=True, ) # moved item의 content token을 TARGET winner 키로 누적(없는 token만) try: cur.execute( """ SELECT content, dome_code FROM mlinkdw.shopprod_group_map2 WHERE vender_code=%s AND icode=%s LIMIT 1 """, (vender_code, icode), ) rr_mv = cur.fetchone() mv_html = rr_mv.get("content") if rr_mv else None mv_dc = rr_mv.get("dome_code") if rr_mv else None except Exception: mv_html = None mv_dc = None _dgm.merge_moved_item_tokens_into_target_winner_sync( self.conn, target_gid=int(target_gid), moved_content_html=(str(mv_html) if mv_html is not None else None), moved_dome_code=(str(mv_dc) if mv_dc is not None else None), cur=cur, ) # 2) OLD 그룹 winner/token refresh (다른 그룹에서 이동해온 경우에만) try: if current_gid and int(current_gid) != int(target_gid): _dgm.refresh_group_winner_and_tokens_if_needed_sync( self.conn, int(current_gid), cur=cur, force_token_cleanup=True, ) except Exception: pass except Exception: pass if current_gid and int(current_gid) != int(target_gid): try: cur.execute( """ SELECT (SELECT COUNT(*) FROM mlinkdw.shopprod_group_map2 WHERE group_id = %s) AS map_cnt, (SELECT COUNT(*) FROM mlinkdw.shopprod_sub_group2 WHERE group_id = %s) AS sub_cnt """, (current_gid, current_gid), ) r3 = cur.fetchone() map_cnt = r3[0] if r3 else 0 sub_cnt = r3[1] if r3 else 0 if map_cnt == 0 and sub_cnt == 0: # group_map/sub_group가 비었으면 content_token도 정리(고아 토큰 방지) try: cur.execute( "DELETE FROM mlinkdw.shopprod_group_content_token WHERE group_id = %s", (int(current_gid),), ) except Exception: pass cur.execute("DELETE FROM mlinkdw.shopprod_group2 WHERE group_id = %s", (current_gid,)) except Exception: # 삭제 실패는 무시 pass if manage_tx: cur.execute("COMMIT") except Exception as e: try: if manage_tx: self.conn.rollback() except Exception: pass raise RuntimeError(f"group_id 변경 실패: {e}") return def load_sub_groups(self, clear_items: bool = True): cur = self.conn.cursor(cursor_factory=DictCursor) cur.execute(""" SELECT sg.sub_group_id, sm.icode, sm.iname, sm.price, sm.vender_code FROM mlinkdw.shopprod_sub_group2 sg JOIN mlinkdw.shopprod_sub_group_map2 sm ON sm.sub_group_id = sg.sub_group_id WHERE sg.group_id = %s """, (self.current_group_id,)) rows = [ [r[0], r[1], r[2], 0, r[3], r[4]] for r in cur.fetchall() ] self.sub_group_model.load(rows) self._adjust_sub_group_view_columns() if clear_items: self.item_model.load([]) self._adjust_item_view_columns() def load_group_items(self, group_id): """ group_id 기준으로 하단 Item 그리드를 즉시 로드(서브그룹 클릭 없이). """ search_name = self.iname_edit.text().strip() cur = self.conn.cursor(cursor_factory=DictCursor) cur.execute(""" SELECT vender_code, icode, iname, price, img_url, content FROM mlinkdw.shopprod_group_map2 WHERE group_id = %s ORDER BY price """, (group_id,)) # 현재 선택된 group row에서 winner_price를 가져오되, 없으면 0으로 처리 winner_price = 0 try: if self.group_model.rows: winner_price = self.group_model.rows[0][3] except Exception: winner_price = 0 items = [] fetched = cur.fetchall() too_many = len(fetched) >= 20 # winner 이미지 URL (있으면) - 이미지 점수 산출용 try: winner_img_url = self.group_model.rows[0][5] if (self.group_model.rows and len(self.group_model.rows[0]) > 5) else None except Exception: winner_img_url = None for r in fetched: if too_many: # 후보가 많으면(>=20) 점수 계산은 스킵하고 공백으로 표시(디버깅/성능) sim = "" price_score = "" img_score = "" merge_score = "" else: sim = calc_similarity(search_name, r["iname"]) price_score = calc_price_score(winner_price, r["price"]) img_score = 100 if winner_img_url and r.get("img_url"): dist = _phash_distance(str(winner_img_url), str(r["img_url"])) if dist is not None: img_score = calc_image_score(dist, max_dist=64) merge_score = calc_merge_score(sim, price_score, img_score) items.append([ "", # row number (filled after sort) r["vender_code"], r["icode"], r["iname"], r["price"], r.get("img_url"), r.get("content") , sim, price_score, img_score, merge_score, ]) # group 클릭 시 하단 item은 가격 기준 정렬이 기대되는 경우가 많아 price 기준으로 고정 정렬. # (merge_score는 계산/표시만 하고, 정렬 우선순위에서는 제외) items.sort(key=lambda x: (_sort_price(x[4]), str(x[2] or ""))) for i, row in enumerate(items, start=1): row[0] = i self.item_model.load(items) self._adjust_item_view_columns() # ========================= # SUB GROUP CLICK # ========================= def on_sub_group_clicked(self, index): sub_group_id = self.sub_group_model.rows[index.row()][0] self.load_items(sub_group_id) self.item_model.layoutChanged.emit() # ========================= # LOAD ITEMS + MERGE SCORE # ========================= def load_items(self, sub_group_id): search_name = self.iname_edit.text().strip() cur = self.conn.cursor(cursor_factory=DictCursor) # shopprod_sub_group_map에 content 컬럼이 없는 환경도 있을 수 있어 안전하게 fallback 처리 try: cur.execute(""" SELECT vender_code, icode, iname, price, img_url, content FROM mlinkdw.shopprod_sub_group_map2 WHERE sub_group_id = %s """, (sub_group_id,)) has_content = True except Exception: self.conn.rollback() cur = self.conn.cursor(cursor_factory=DictCursor) cur.execute(""" SELECT vender_code, icode, iname, price, img_url FROM mlinkdw.shopprod_sub_group_map2 WHERE sub_group_id = %s """, (sub_group_id,)) has_content = False # winner 이미지 URL (있으면) - 이미지 점수 산출용 winner_img_url = None try: if self.group_model.rows and len(self.group_model.rows[0]) > 5: winner_img_url = self.group_model.rows[0][5] except Exception: winner_img_url = None items = [] for r in cur.fetchall(): sim = calc_similarity(search_name, r["iname"]) price_score = calc_price_score(self.group_model.rows[0][3], r["price"]) # 이미지 점수: phash 거리 기반(의존성 없으면 100으로 유지) img_score = 100 if winner_img_url and r.get("img_url"): dist = _phash_distance(str(winner_img_url), str(r["img_url"])) if dist is not None: # hash_size=8 => 64bit img_score = calc_image_score(dist, max_dist=64) merge_score = calc_merge_score(sim, price_score, img_score) items.append([ "", # row number (filled after sort) r["vender_code"], r["icode"], r["iname"], r["price"], r.get("img_url"), (r.get("content") if has_content else None), sim, price_score, img_score, merge_score, ]) items.sort(key=lambda x: (-x[10], _sort_price(x[4]))) for i, row in enumerate(items, start=1): row[0] = i self.item_model.load(items) self._adjust_item_view_columns() # ========================= # ENTRY # ========================= if __name__ == "__main__": app = QApplication(sys.argv) win = MainWindow() win.resize(1600, 900) win.show() sys.exit(app.exec())