269 lines
9.2 KiB
Python
269 lines
9.2 KiB
Python
import pyodbc
|
|
|
|
# 선호 순서 — 앞쪽일수록 우선. 설치 환경마다 다를 수 있어 자동 탐지 후 선택.
|
|
_PREFERRED_DRIVERS = (
|
|
"ODBC Driver 18 for SQL Server",
|
|
"ODBC Driver 17 for SQL Server",
|
|
"ODBC Driver 13 for SQL Server",
|
|
"SQL Server Native Client 11.0",
|
|
"SQL Server",
|
|
)
|
|
|
|
|
|
def _pick_driver() -> "str | None":
|
|
"""이 PC에 설치된 ODBC 드라이버 중 사용할 SQL Server 드라이버를 선택.
|
|
선호 순서대로 먼저 찾고, 없으면 이름에 'SQL Server'가 포함된 아무 드라이버라도 사용."""
|
|
available = pyodbc.drivers()
|
|
for name in _PREFERRED_DRIVERS:
|
|
if name in available:
|
|
return name
|
|
for name in available:
|
|
if "SQL Server" in name:
|
|
return name
|
|
return None
|
|
|
|
|
|
class SQLClient:
|
|
def __init__(self):
|
|
self.conn = None
|
|
self.cursor = None
|
|
|
|
def connect(self, server: str, database: str,
|
|
username: str, password: str) -> bool:
|
|
driver = _pick_driver()
|
|
if driver is None:
|
|
print("[DB] 연결 실패: 설치된 SQL Server ODBC 드라이버가 없습니다. "
|
|
"'ODBC Driver 18 for SQL Server'를 설치하세요.")
|
|
self.conn = None
|
|
return False
|
|
try:
|
|
conn_str = (
|
|
f"DRIVER={{{driver}}};"
|
|
f"SERVER={server};"
|
|
f"DATABASE={database};"
|
|
f"UID={username};"
|
|
f"PWD={password};"
|
|
f"TrustServerCertificate=yes;"
|
|
f"Encrypt=optional;"
|
|
)
|
|
self.conn = pyodbc.connect(conn_str, timeout=10)
|
|
self.cursor = self.conn.cursor()
|
|
print(f"[DB] 연결 성공: {server}/{database} (드라이버: {driver})")
|
|
return True
|
|
except Exception as e:
|
|
print(f"[DB] 연결 실패: {e} (드라이버: {driver})")
|
|
self.conn = None
|
|
return False
|
|
|
|
def disconnect(self):
|
|
if self.cursor:
|
|
self.cursor.close()
|
|
if self.conn:
|
|
self.conn.close()
|
|
self.conn = None
|
|
self.cursor = None
|
|
print("[DB] 연결 해제")
|
|
|
|
def is_connected(self) -> bool:
|
|
return self.conn is not None
|
|
|
|
@staticmethod
|
|
def _norm_id(value) -> str:
|
|
return str(value).strip() if value is not None else ""
|
|
|
|
def get_wk_results(self) -> list:
|
|
"""
|
|
vi_AI_WK_Result 뷰 조회.
|
|
반환: [{
|
|
"article_id": ..., "machine_id": ..., "machine": ...,
|
|
"work_start_date": ..., "work_start_time": ...,
|
|
}, ...]
|
|
"""
|
|
if not self.is_connected():
|
|
return []
|
|
try:
|
|
self.cursor.execute("""
|
|
SELECT ArticleID, MachineID, Machine,
|
|
WorkStartDate, WorkStartTime
|
|
FROM vi_AI_WK_Result
|
|
WHERE ArticleID IS NOT NULL
|
|
ORDER BY ArticleID
|
|
""")
|
|
rows = self.cursor.fetchall()
|
|
return [self._row_to_wk_result(row) for row in rows]
|
|
except Exception as e:
|
|
print(f"[DB] WK_Result 조회 실패: {e}")
|
|
return []
|
|
|
|
def get_wk_result_article_ids(self) -> list:
|
|
"""vi_AI_WK_Result에 있는 ArticleID 목록 (중복 제거, 순서 유지)."""
|
|
seen = set()
|
|
ids = []
|
|
for row in self.get_wk_results():
|
|
norm = self._norm_id(row["article_id"])
|
|
if norm and norm not in seen:
|
|
seen.add(norm)
|
|
ids.append(row["article_id"])
|
|
return ids
|
|
|
|
def get_wk_result_map(self) -> dict:
|
|
"""ArticleID(정규화) → WK_Result 행. 동일 ID가 여러 행이면 마지막 행 사용."""
|
|
result = {}
|
|
for row in self.get_wk_results():
|
|
norm = self._norm_id(row["article_id"])
|
|
if norm:
|
|
result[norm] = row
|
|
return result
|
|
|
|
@staticmethod
|
|
def _row_to_wk_result(row) -> dict:
|
|
return {
|
|
"article_id": row[0],
|
|
"machine_id": row[1],
|
|
"machine": row[2],
|
|
"work_start_date": row[3],
|
|
"work_start_time": row[4],
|
|
}
|
|
|
|
@staticmethod
|
|
def format_db_value(value) -> str:
|
|
if value is None:
|
|
return "—"
|
|
if hasattr(value, "strftime"):
|
|
if hasattr(value, "hour"):
|
|
return value.strftime("%H:%M:%S")
|
|
return value.strftime("%Y-%m-%d")
|
|
text = str(value).strip()
|
|
return text if text else "—"
|
|
|
|
def get_reflector_list_ordered(self, article_ids: list) -> list:
|
|
"""article_ids 순서를 유지한 제품 목록 (PatMax 슬롯 순서용)."""
|
|
if not article_ids:
|
|
return []
|
|
by_norm = {
|
|
self._norm_id(item["article_id"]): item
|
|
for item in self.get_reflector_list(article_ids=article_ids)
|
|
}
|
|
ordered = []
|
|
for article_id in article_ids:
|
|
item = by_norm.get(self._norm_id(article_id))
|
|
if item is not None:
|
|
ordered.append(item)
|
|
return ordered
|
|
|
|
def split_articles_by_wk(self, mes_selected_ids: "list | None" = None) -> tuple:
|
|
"""
|
|
vi_AI_mt_Article 목록을 WK_Result 작업 대상 / 기타로 분류.
|
|
반환: (active_list, inactive_list)
|
|
"""
|
|
if mes_selected_ids is not None:
|
|
if len(mes_selected_ids) == 0:
|
|
return [], []
|
|
all_items = self.get_reflector_list_ordered(mes_selected_ids)
|
|
else:
|
|
all_items = self.get_reflector_list()
|
|
|
|
if not all_items:
|
|
return [], []
|
|
|
|
wk_norm = set(self.get_wk_result_map().keys())
|
|
active = [
|
|
item for item in all_items
|
|
if self._norm_id(item["article_id"]) in wk_norm
|
|
]
|
|
inactive = [
|
|
item for item in all_items
|
|
if self._norm_id(item["article_id"]) not in wk_norm
|
|
]
|
|
return active, inactive
|
|
|
|
def get_inspectable_articles(self, mes_selected_ids: "list | None" = None) -> list:
|
|
"""
|
|
vi_AI_mt_Article ∩ vi_AI_WK_Result.
|
|
mes_selected_ids 지정 시 관리자 MES 선택과도 교집합.
|
|
"""
|
|
wk_ids = self.get_wk_result_article_ids()
|
|
if not wk_ids:
|
|
return []
|
|
|
|
wk_by_norm = {self._norm_id(article_id): article_id for article_id in wk_ids}
|
|
|
|
if mes_selected_ids is not None:
|
|
if len(mes_selected_ids) == 0:
|
|
return []
|
|
filter_ids = [
|
|
wk_by_norm[self._norm_id(article_id)]
|
|
for article_id in mes_selected_ids
|
|
if self._norm_id(article_id) in wk_by_norm
|
|
]
|
|
else:
|
|
filter_ids = wk_ids
|
|
|
|
if not filter_ids:
|
|
return []
|
|
return self.get_reflector_list(article_ids=filter_ids)
|
|
|
|
def get_all_articles(self) -> list:
|
|
"""
|
|
vi_AI_mt_Article 뷰 전체 조회 (관리자 MES 제품 선택용).
|
|
반환: [{"article_id": ..., "article": ..., "buyer_article_no": ...}, ...]
|
|
"""
|
|
if not self.is_connected():
|
|
return []
|
|
try:
|
|
self.cursor.execute("""
|
|
SELECT ArticleID, Article, BuyerArticleNo
|
|
FROM vi_AI_mt_Article
|
|
ORDER BY ArticleID
|
|
""")
|
|
rows = self.cursor.fetchall()
|
|
return [self._row_to_article(row) for row in rows]
|
|
except Exception as e:
|
|
print(f"[DB] 전체 제품 조회 실패: {e}")
|
|
return []
|
|
|
|
def get_reflector_list(self, article_ids: "list | None" = None) -> list:
|
|
"""
|
|
vi_AI_mt_Article 뷰에서 제품 목록 조회.
|
|
article_ids 지정 시 해당 ID만, 미지정 시 REF 포함 제품 전체.
|
|
반환: [{"article_id": ..., "article": ..., "buyer_article_no": ...}, ...]
|
|
"""
|
|
if not self.is_connected():
|
|
return []
|
|
try:
|
|
if article_ids:
|
|
placeholders = ",".join("?" * len(article_ids))
|
|
self.cursor.execute(f"""
|
|
SELECT ArticleID, Article, BuyerArticleNo
|
|
FROM vi_AI_mt_Article
|
|
WHERE ArticleID IN ({placeholders})
|
|
ORDER BY ArticleID
|
|
""", article_ids)
|
|
else:
|
|
self.cursor.execute("""
|
|
SELECT ArticleID, Article, BuyerArticleNo
|
|
FROM vi_AI_mt_Article
|
|
WHERE Article LIKE '%REF%'
|
|
ORDER BY ArticleID
|
|
""")
|
|
rows = self.cursor.fetchall()
|
|
return [self._row_to_article(row) for row in rows]
|
|
except Exception as e:
|
|
print(f"[DB] 조회 실패: {e}")
|
|
return []
|
|
|
|
@staticmethod
|
|
def _row_to_article(row) -> dict:
|
|
return {
|
|
"article_id": row[0],
|
|
"article": row[1],
|
|
"buyer_article_no": row[2],
|
|
}
|
|
|
|
def save_inspection_result(self, article_id: str,
|
|
result: str, score: float) -> bool:
|
|
"""검사 결과 저장 — 테이블 확정 후 구현."""
|
|
# TODO: 결과 저장 테이블 확정 후 쿼리 구현
|
|
print(f"[DB] 검사 결과 저장: {article_id} {result} {score}")
|
|
return True
|