728x90
from langchain_core.tools import tool
from langchain.tools import tool
from langchain_openai import ChatOpenAI
import os
import base64
import numpy as np
import pandas as pd
from urllib.parse import urlencode, quote
from dotenv import load_dotenv
load_dotenv()
base_url = os.getenv('base_url')
api_key = os.getenv('api_key')
model = os.getenv('model')
from .Graph_Agent import draw_multi_values_doughnut_graph_api
import requests
import matplotlib.pyplot as plt
import io
from pytz import timezone
import smtplib
from core.graph.state import State
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.image import MIMEImage
from email.mime.base import MIMEBase
from email import encoders
from email.header import Header
import markdown
from core.graph.state import State
from core.tools.func import *
#from core.tools.query import extract_factory_list
import yaml
with open("config/prompt.yaml", "r", encoding="utf-8") as f:
prompts = yaml.safe_load(f)
import json
# config.json 불러오기
with open("config/config.json", "r", encoding="utf-8") as f:
config = json.load(f)
# NG, CHECK 등 불량 결과 리스트 가져오기
defect_results = config["defect_result"]
# main에 사용한 logger 불러오기
import logging
logger = logging.getLogger("main")
import re
def remove_img_tags(html: str) -> str:
"""HTML 문자열에서 <img> 태그를 제거합니다."""
return re.sub(r'<img[^>]*>', '', html)
# anomaly data 가져옴
def get_anomaly_data_from_DB(): # 지금은 한 개 임의로 가져옴
return f"""
SELECT L.*, R.line_code FROM tb_ims_anomalies L
JOIN vw_ims_basic_info R
ON L.line_desc = R.line_desc AND L.equipment_id = R.equipment_id
WHERE L.product = "VS"
LIMIT 1
"""
class Mailing_Agent:
def __init__(self,anomaly_info):
self.name = 'Mailing_Agent'
self.mail_creator = ChatOpenAI(
openai_api_base=base_url,
openai_api_key=api_key,
model=model,
# temperature=0,
# top_p=1
)
self.translater = ChatOpenAI(
openai_api_base=base_url,
openai_api_key=api_key,
model="translate-gpt-4.1", #gpt 4.1 형식 고정 / 지시 엄수
# model="stablelm2:12b",
# temperature=0,
# top_p=1
)
self.anomaly_info = anomaly_info
self.description = prompts["Mailing_Agent"]["description"]
# ✅ 클래스 함수 객체의 docstring 수정
Mailing_Agent.run.__doc__ = self.description
#self.run.__doc__ = self.description
def __str__(self):
return f"[{self.name}] {self.description}"
def anomaly_mailSending(self, emailBody, anomaly_info):
_DEBUG_MODE = True # DEBUG
self.mail_sending_count = 0
# Now
today_now = datetime.now(timezone('Asia/Seoul')).strftime('%Y-%m-%d %H:%M:%S')
#today_hour = datetime.now(timezone('Asia/Seoul')).strftime('%Y-%m-%d %H:00:00')
# for row in rows:
mail_sent_count = int(anomaly_info['is_mail_sent'])
if mail_sent_count >= 0:
# if mail_sent_count < 1 :
self.mail_sending_count += 1
if "@lge.com" in anomaly_info['mailing_group_name'] :
recipient = f"""{anomaly_info['mailing_group_name']}"""
else :
recipient = f"""{anomaly_info['mailing_group_name']}@lge.com"""
#recipient = "jaewon.kim@lge.com"
# recipient_list = ["seungjong.yoo@lge.com", "kd.noh@lge.com"]
recipient_list = ["seungjong.yoo@lge.com", "kd.noh@lge.com",
# "DL-insp-team@lge.com",
# "DL-vs1-pt@lge.com","DL-vs2-pt@lge.com",
"chihwan.song@lge.com",
"kyungseok.oh@lge.com","wh81.park@lge.com","jethro.song@lge.com",
"daniel.lim@lge.com",
]
recipient = ",".join(recipient_list)
##############
# 발신자 부분 #
##############
sender = "TEST_ZENIS_SERVER@lge.com"
# if "@lge.com" in anomaly_info['issue_owner'] :
# sender = f"""{anomaly_info['issue_owner']}"""
# else :
# sender = f"""{anomaly_info['issue_owner']}@lge.com"""
eCc = ""
eBcc = ""
header = f"""[ZENIS][{self.factory_name}] AI found abnormal inspection equipment status."""
# if doughnut_graph_byte:
# doughnut_graph_str = base64.b64encode(doughnut_graph_byte).decode("utf-8")
logger.info(f"======================================================================")
logger.info(f"[mailing agent] emailTo= {recipient}")
logger.info(f"[mailing agent] emailSender= {sender}")
logger.info(f"[mailing agent] emailSubject= {header}")
logger.info(f"[mailing agent] id= {anomaly_info['id']}")
logger.info(f"======================================================================")
#self.logger.info("emailBody=",emailBody)
#################################################################
# 첨부 화일이 있을때 아래 코드 필요
#################################################################
#temp_dir = os.path.join(os.path.dirname(__file__), 'tmp')
#os.makedirs(temp_dir, exist_ok=True) # 디렉토리가 없으면 생성
#file_paths = []
#for file in files:
# image_url = file.filename
# logger.info("IMAGE:", image_url)
# if file:
# file_path = os.path.join(temp_dir, image_url)
# file.save(file_path)
# file_paths.append(file_path)
################################################################
eMailSERVER = "lgekrhqmh01.lge.com"
smtp = smtplib.SMTP(eMailSERVER, 25)
msg = MIMEMultipart('alternative')
msg['Subject'] = header
msg['From'] = sender
msg['To'] = recipient
msg['Cc'] = eCc
msg.attach(MIMEText(emailBody, 'html', 'utf-8'))
#################################################################
# 첨부 화일이 있을때 아래 코드 필요
#################################################################
#for file_path in file_paths:
# with open(file_path, 'rb') as attachment:
# if file_path.lower().endswith(('png', 'jpg', 'jpeg', 'gif')):
# img = MIMEImage(attachment.read())
# img.add_header('Content-Disposition', f'attachment; filename="{Header(os.path.basename(file_path), "utf-8").encode()}"')
# msg.attach(img)
# else:
# part = MIMEBase('application', 'octet-stream')
# part.set_payload(attachment.read())
# encoders.encode_base64(part)
# part.add_header('Content-Disposition', f'attachment; filename="{Header(os.path.basename(file_path), "utf-8").encode()}"')
# msg.attach(part)
##############################################################
try:
to_addrs = recipient.split(",") + eCc.split(",") + eBcc.split(",")
to_addrs = [addr.strip() for addr in to_addrs if addr.strip()]
smtp.sendmail(sender, to_addrs, msg.as_string())
smtp.quit()
## Optionally, delete the file after sending the email
#for file_path in file_paths:
# os.remove(file_path)
except Exception as e:
logger.info(f"Email sent Failed= {e}")
# update is_mail_sent field
mail_sent_count += 1
#logger.info("mail_sent_count=",mail_sent_count)
#logger.info("id=",anomaly_info['id'])
#####################
# DB Update #
#####################
# SQL = f"""UPDATE tb_ims_anomalies T SET T.is_mail_sent={mail_sent_count} WHERE T.id ={row['id']}"""
# curs.execute(SQL)
# mydb.commit() #commit를 해야 실제 DB가 update됨
logger.info(f"[mailing agent] {self.mail_sending_count}개 mail을 보냈습니다.\n")
return True
def format_ai_analysis_html(self, ai_text: str) -> str:
#html_text = markdown.markdown(ai_text)
# 마크다운 전체를 HTML로 변환 (표 포함)
html_text = markdown.markdown(ai_text, extensions=['tables'])
return f"<div font-size:14px;'>{html_text}</div>"
def run(self, state: State,factory_name):
"""
지정된 규칙과 그동안의 정보들을 이용하여 메일을 보냅니다
"""
anomaly_info = self.anomaly_info
self.factory_name = factory_name
actions = state["planned_actions"]
logger.info(f"[메일 요청] {factory_name} 공정 분석 결과를 메일로 발송합니다.")
trend_graph_html = ""
reason_graph_html = ""
time_stamp_processed_df, search_df, filtered_anomaly_period_origin_df = state['time_stamp_processed_df'], state['search_df'] , state['filtered_anomaly_period_origin_df']
analysis_margin_df = state['analysis_margin_df']
filtered_anomaly_with_margin_df = state['search_result']['filtered_anomaly_with_margin_df']
daily_origin_df = state['daily_origin_df']
analysis_daily_df = state['analysis_daily_df']
# filtered_origin_df_NG = filtered_origin_df[filtered_origin_df['insp_judge_code'] == 'NG'].copy()
filtered_origin_df_NG = filtered_anomaly_period_origin_df[
filtered_anomaly_period_origin_df["insp_judge_code"].isin(defect_results)
].copy()
# filtered_with_margin_df_NG = filtered_with_margin_df[filtered_with_margin_df['insp_judge_code'] == 'NG'].copy()
#np.where(조건, 참일 때 값, 거짓일 때 값)
# >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
# CHECK 일 때는 REASON을 CHECK 로 채우는 Logic
# filtered_origin_df_NG['reason'] = np.where(
# filtered_origin_df_NG['insp_judge_code'].isin(['OK', 'NG']),
# filtered_origin_df_NG['reason'],
# filtered_origin_df_NG['insp_judge_code']
# )
# <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<
filtered_with_margin_df_NG = filtered_anomaly_with_margin_df[
filtered_anomaly_with_margin_df['insp_judge_code'].isin(defect_results)
].copy()
# >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
# CHECK 일 때는 REASON을 CHECK 로 채우는 Logic
# filtered_with_margin_df_NG['reason'] = np.where(
# filtered_with_margin_df_NG['insp_judge_code'].isin(['OK', 'NG']),
# filtered_with_margin_df_NG['reason'],
# filtered_with_margin_df_NG['insp_judge_code']
# )
# <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<
daily_origin_df_NG = daily_origin_df[
daily_origin_df["insp_judge_code"].isin(defect_results)
].copy()
num_daily_origin_df = len(daily_origin_df)
num_daily_origin_df_NG = len(daily_origin_df_NG)
num_filtered_origin_df = len(filtered_anomaly_period_origin_df)
num_analysis_margin_df = len(filtered_anomaly_with_margin_df)
num_filtered_origin_df_NG = len(filtered_origin_df_NG)
num_analysis_margin_df_NG = len(filtered_with_margin_df_NG)
margin_NG_ratio = (num_analysis_margin_df_NG / num_analysis_margin_df) * 100
daily_NG_ratio = (num_daily_origin_df_NG/num_daily_origin_df) * 100
# 불량 항목과 Model Suffix별 개수 계산
reason_counts = filtered_origin_df_NG['reason'].value_counts()
suffix_counts = filtered_origin_df_NG['product_specification_id'].value_counts()
reason_list = reason_counts.index.tolist()
margin_reason_counts = filtered_with_margin_df_NG['reason'].value_counts()
daily_reason_counts = daily_origin_df_NG['reason'].value_counts()
# 문자열로 변환 (예: "CAMC (1개), BT_READ (2개)")
reason_summary = ", ".join([f"{r} ({c}개)" for r, c in reason_counts.items()])
suffix_summary = ", ".join([f"{s} ({c}개)" for s, c in suffix_counts.items()])
# filtered_origin_df = filtered_origin_df.dropna(subset=['local_date', 'local_time','facility_code'])
graph_img_lst = state["graph_dict"].keys()
trend_graph_str_base64 = state["graph_dict"].get("draw_multi_values_graph_api")
reason_doughnut_graph_base64 = state["graph_dict"].get("draw_multi_values_doughnut_graph_api")
# trend_graph_str_base64, reason_doughnut_graph_base64 = state["graph_dict"]["draw_multi_values_graph_api"], state["graph_dict"]["draw_multi_values_doughnut_graph_api"]
save = "./output" # 저장할 폴더 경로
# 폴더가 없으면 생성
os.makedirs(save, exist_ok=True)
graph_img_lst = state["graph_dict"].keys()
for graph_img_key in graph_img_lst:
img_path = os.path.join(save, f"{graph_img_key}.png")
with open(img_path, "wb") as f:
f.write(base64.b64decode(state["graph_dict"][graph_img_key]))
# logger.info(filtered_origin_df)
actions = state["planned_actions"]
# HTML 표 생성
df_html = filtered_origin_df_NG[[
'time_stamp', 'insp_judge_code', 'lot_id', 'product_specification_id', 'reason'
]].rename(columns={
'time_stamp': '시간',
'insp_judge_code': 'OK/NG/CHECK',
'lot_id': 'wip_id',
'product_specification_id': 'Model-Suffix',
'reason': '불량 항목'
})
# html_table = df_html.to_html(index=False, escape=False, border=0) #border : 외각 표시
rows = ""
# DataFrame을 직접 HTML로 렌더링
for _, r in df_html.iterrows():
rows += f"""
<tr>
<td style="padding:8px 20px;text-align:center;color:#333;">{r['시간']}</td>
<td style="padding:8px 20px;text-align:center;color:#333;">{r['OK/NG/CHECK']}</td>
<td style="padding:8px 20px;text-align:center;color:#333;">{r['wip_id']}</td>
<td style="padding:8px 20px;text-align:center;color:#333;">{r['Model-Suffix']}</td>
<td style="padding:8px 20px;text-align:center;color:#333;">{r['불량 항목']}</td>
</tr>
"""
styled_html = f"""
<div class="table-container" style="background-color:#ffffff;border-radius:10px;
box-shadow:0 4px 12px rgba(0,0,0,0.08);overflow:hidden;width:900px;margin:30px 0;">
<table style="border-collapse:collapse;width:100%;font-family:'Noto Sans KR','Roboto',sans-serif;
font-size:15px;">
<caption style="caption-side:top;font-size:1.3em;font-weight:700;color:#A50034;
padding:14px;text-align:left;">불량 내용 Detail</caption>
<thead style="background-color:#A50034;color:#ffffff;">
<tr>
<th style="padding:8px 20px;text-align:center;">시간</th>
<th style="padding:8px 20px;text-align:center;">OK/NG/CHECK</th>
<th style="padding:8px 20px;text-align:center;">WIP_ID</th>
<th style="padding:8px 20px;text-align:center;">Model-Suffix</th>
<th style="padding:8px 20px;text-align:center;">불량 항목</th>
</tr>
</thead>
<tbody>
{rows}
</tbody>
</table>
</div>
"""
# AI 분석 생성
analysis_prompt = f"""
다음은 공정 이상 탐지 결과입니다.
- 사용자 요청: {state["user_prompt"]}
- 이상치 정보: {self.anomaly_info}
- 해당 기간 전체 생산량 : {num_filtered_origin_df}
절대로 전체 공정 NG 데이터에 있는 정보 안에서만 대답해야 됩니다.
이상치 정보에 대한 분석을 반드시 포함하고 기간, 통계값 분석 등을 포함할 수록 좋습니다.
또한 전체 공정 NG 데이터와 이상치 정보를 함께 이용한 분석도 반드시 포함해야 합니다
각 그래프에서 나타나는 주요 패턴, 이상 징후, 시점 등을 포함해
종합적인 통계 분석 보고서를 작성해 주세요.
데이터프레임 설명
- is_checked : 생산량 (is_checked라 작성하지 말고 생산 수량이라 작성하세요)
- is_defect : 불량 수량 (is_defect라 작성하지 말고 불량 수량이라 작성하세요)
아래 지침에 따라 보고서를 작성하세요.
- 아래 지침과 별개의 내용은 기재하지 마세요
- 데이터에 기반한 분석만 하고 예측 및 추측은 절대 금지합니다
[지침]
- 불량 항목은 filtered_origin_df_NG의 reason 컬럼 값만 그대로 사용.
- reason 값이 없으면 "불명"으로 표시하고, 추측하지 말 것.
1. **그래프별 분석**
- 생산량 추이, 불량률 추이, 시간대별 NG 분포 등 3\~4개 항목을 설명.
- 각 그래프에 대해 패턴, 이상 징후, 원인 추정, 향후 조치 제안을 간단히 작성.
2. **해당 시점 분석**
- 해당 기간 전체 공정 NG 데이터 : {filtered_origin_df_NG}
- filtered_origin_df_NG 에 대한 분석을 작성.
- OK 항목은 기재하지 말고 NG항목에 대한 분석을 자세히 기재
- 분석내용에는 시간, insp_judge_code, 불량 여부, wip_id, product_specification_id, reason 등을 포함해라
- 용어 관련
- insp_judge_code 는 'insp_judge_code'가 아닌 'OK/NG/CHECK' 라 표현
- product_specification_id는 'product_specification_id' 가 아닌 'Model-Suffix' 로 표현해라
- reason은 'reason'이 아닌 '불량 항목' 으로 표현해라
- filtered_origin_df_NG 'filtered_origin_df_NG'가 아닌 '공정 DB NG데이터'라 표현
- reason(불량 항목) 정보는 반드시 "filtered_origin_df_NG"를 기반으로만 적을 것, 마음대로 변경하지 않아야 함
3. **종합 결론**
- 생산량·가동률·품질 측면에서의 전반적 상태를 요약.
- 반복되는 문제나 계절성 패턴을 명시.
4 ** AI 답변 생성 과정 **
- 핵심 정보를 추출하는 과정을 설명하세요
[출력 형식]
- 표는 Markdown 표 형식으로 작성.
- 각 항목은 간결하지만 구체적으로 기술.
- 불필요한 금융·거래 데이터 언급은 제외하고 제조업 KPI만 반영.
"""
# ================= 그래프 생성 =================
plt.figure(figsize=(8, 4))
# 🔸 변경: 선 색상을 LG Red(#A50034)로 변경
plt.plot(
analysis_daily_df['time_stamp'],
analysis_daily_df['defect_rate'],
marker='o',
linestyle='-',
color='steelblue',
#color='#A50034', # LG Red
markerfacecolor='steelblue', # 점: steelblue
markeredgecolor='steelblue'
# markerfacecolor='#009E73', # 점: 청록색
# markeredgecolor='#009E73' # 점: 청록색
# markerfacecolor='#A50034', # LG Red
# markeredgecolor='#A50034' # LG Red
)
# 🔸 변경: 제목과 축 스타일을 LG 디자인에 맞게 조정
plt.title('Defect Rate Trend (with Margin)', fontsize=14, color='#A50034', fontweight='bold')
plt.xlabel('Time', fontsize=12, color='#333')
plt.ylabel('Defect Rate', fontsize=12, color='#333')
# 🔸 변경: 격자선 색상과 투명도 조정
plt.grid(True, color='#ddd', linestyle='--', linewidth=0.8, alpha=0.7)
# 🔸 변경: 배경색을 흰색으로 고정
plt.gcf().set_facecolor('white')
# 🔸 변경: anomaly 구간 시각화 추가
start_time = pd.to_datetime(self.anomaly_info['start_period'])
end_time = pd.to_datetime(self.anomaly_info['end_period'])
# anomaly 구간을 반투명한 붉은 영역으로 표시
plt.axvspan(start_time, end_time, color='#A50034', alpha=0.15, label='Anomaly Period')
# anomaly 시작/종료 시점에 수직선 표시
plt.axvline(start_time, color='#A50034', linestyle='--', linewidth=1.5)
plt.axvline(end_time, color='#A50034', linestyle='--', linewidth=1.5)
# 범례 추가
plt.legend(loc='upper right', fontsize=10)
plt.tight_layout()
# 그래프를 base64로 인코딩
buf = io.BytesIO()
plt.savefig(buf, format='png')
buf.seek(0)
analysis_margin_graph_base64 = base64.b64encode(buf.read()).decode('utf-8')
plt.close()
# 가시화를 위한 EMWA 생성
analysis_daily_df_cp = analysis_daily_df.copy()
# # Exponentially Weighted Moving Average 지수가중이동평균
analysis_daily_df_cp['ewma'] = analysis_daily_df_cp['defect_rate'].ewm(halflife=120, adjust=False).mean()
# ================= 그래프 생성 =================
plt.figure(figsize=(12, 4))
# 🔸 변경: 선 색상을 LG Red(#A50034)로 변경
plt.plot(
analysis_daily_df_cp['time_stamp'],
analysis_daily_df_cp['ewma'],
marker='o',
linestyle='-',
color='steelblue',
#color='#A50034', # LG Red
markerfacecolor='steelblue', # 점: steelblue
markeredgecolor='steelblue'
# markerfacecolor='#009E73', # 점: 청록색
# markeredgecolor='#009E73' # 점: 청록색
# markerfacecolor='#A50034', # LG Red
# markeredgecolor='#A50034' # LG Red
)
# 🔸 변경: 제목과 축 스타일을 LG 디자인에 맞게 조정
plt.title('EWMA Trend (during 24 hours)', fontsize=14, color='#A50034', fontweight='bold')
plt.xlabel('Time', fontsize=12, color='#333')
plt.ylabel('EWMA', fontsize=12, color='#333')
# 🔸 변경: 격자선 색상과 투명도 조정
plt.grid(True, color='#ddd', linestyle='--', linewidth=0.8, alpha=0.7)
# 🔸 변경: 배경색을 흰색으로 고정
plt.gcf().set_facecolor('white')
# 🔸 변경: anomaly 구간 시각화 추가
start_time = pd.to_datetime(self.anomaly_info['start_period'])
end_time = pd.to_datetime(self.anomaly_info['end_period'])
# anomaly 구간을 반투명한 붉은 영역으로 표시
plt.axvspan(start_time, end_time, color='#A50034', alpha=0.15, label='Anomaly Period')
# anomaly 시작/종료 시점에 수직선 표시
plt.axvline(start_time, color='#A50034', linestyle='--', linewidth=1.5)
plt.axvline(end_time, color='#A50034', linestyle='--', linewidth=1.5)
# 범례 추가
plt.legend(loc='upper right', fontsize=10)
plt.tight_layout()
# 그래프를 base64로 인코딩
buf = io.BytesIO()
plt.savefig(buf, format='png')
buf.seek(0)
analysis_ewma_graph_base64 = base64.b64encode(buf.read()).decode('utf-8')
plt.close()
# 🔸 도넛 그래프 생성 (Graph_Agent 함수 사용)
# ✅ numpy.int64 → int 변환
values = [int(v) for v in daily_reason_counts.values]
labels = list(daily_reason_counts.index)
chart_config = {
"type": "doughnut",
"data": {
"labels": labels,
"datasets": [
{"data": values}
],
},
"options": {
"plugins": {
"legend": {
"position": "right", # 오른쪽
"align": "end", # 아래쪽 정렬
"labels": {
"font": {"size": 25} # 범례 폰트 크기
}
},
"doughnutlabel": {
"labels": [
{"text": str(sum(values)), "font": {"size": 30, "weight": "bold"}}, # 중앙 합계 크기
{"text": "total", "font": {"size": 25}} # 중앙 total 크기
],
},
"datalabels": { # 각 데이터 조각 위 숫자 표시
"font": {"size": 25},
"formatter": "Math.round" # 값 표시 형식 (선택)
},
},
"font": {"size": 25}, # 전체 기본 폰트 크기
},
}
logger.info("[Processing] Quickchart API를 호출합니다")
resp = requests.get(
"https://quickchart.io/chart",
params={
"c": json.dumps(chart_config),
"format": "png",
"bkg": "white"
}
)
if resp.status_code != 200:
logger.error(f"QuickChart API 호출 실패: {resp.status_code}, {resp.text}")
else:
doughnut_graph_base64 = base64.b64encode(resp.content).decode("utf-8")
# 🔸 두 그래프를 나란히 배치
graphs_margin_html = f"""
<div style="display: flex; justify-content: space-between; align-items: flex-start; gap: 20px;">
<div style="flex: 1; text-align: center;">
<img src="data:image/png;base64,{analysis_margin_graph_base64}"
alt="Margin Defect Rate Graph" style="max-width: 100%; height: auto;" />
</div>
<div style="flex: 1; text-align: center;">
<img src="data:image/png;base64,{doughnut_graph_base64}"
alt="Defect Reason Doughnut Graph" style="max-width: 100%; height: auto;" />
</div>
</div>
"""
additional_html = f"""
<div style="flex: 1; text-align: center;">
<img src="data:image/png;base64,{analysis_ewma_graph_base64}"
alt="24 hours EWMA Trend Graph" style="max-width: 100%; height: auto;" />
</div>
"""
if trend_graph_str_base64 is not None:
trend_graph_html = f"""
<div style="flex: 1; text-align: left;">
<h2>이상감지 시점 그래프:</h2>
<img src="data:image/png;base64,{trend_graph_str_base64}" alt="Trend Graph" style="max-width: 100%; height: auto;" />
</div>
"""
if reason_doughnut_graph_base64 is not None:
reason_graph_html = f"""
<div style="flex: 1; text-align: left;">
<h2>불량 항목 :</h2>
<img src="data:image/png;base64,{reason_doughnut_graph_base64}" alt="Reason Graph" style="max-width: 100%; height: auto;" />
</div>
"""
graph_section_html = f"""
<div style="display: flex; justify-content: space-between; align-items: flex-start; gap: 20px; margin-top: 10px;">
<div style="flex: 1; text-align: left;">
<h3>이상감지 시점 그래프:</h3>
<img src="data:image/png;base64,{trend_graph_str_base64}" alt="Trend Graph" style="max-width: 100%; height: auto;" />
</div>
<div style="flex: 1; text-align: left;">
<h3 style="margin-bottom: 8px;">불량 항목</h3>
<img src="data:image/png;base64,{reason_doughnut_graph_base64}" alt="Reason Graph" style="max-width: 100%; height: auto;" />
</div>
</div>
"""
# 엑셀 불러오기
df = pd.read_excel("manual/saved_reason_solution_method_manual_sample.xlsx")
# 필요한 컬럼만 추출 (GMES, 원인, 점검/대응 방안, OWNER)
try:
df = df[["GMES", "원인", "점검/대응 방안", "OWNER"]]
except IndexError:
logger.error(f"Please Check the manual Column name : 'GMES', '원인', '점검/대응 방안', 'OWNER'] ")
solution_html = '''
<table role="presentation" cellspacing="0" cellpadding="0" border="0" width="900" style="width:900px;
border-collapse:collapse;font-family:Arial,'Noto Sans KR',sans-serif;font-size:14px;color:#333;background-color:#ffffff;">
<tr>
<td style="padding:14px;font-size:20px;font-weight:bold;color:#A50034;">
불량 항목별 원인 및 조치
</td>
</tr>
<tr>
<td>
<table role="presentation" cellspacing="0" cellpadding="0" border="1" width="900" style="width:900px;border-collapse:collapse;border:1px solid #ddd;">
<thead>
<tr style="background-color:#A50034;color:#ffffff;">
<th style="padding:8px;text-align:center;width:100px;">불량 항목</th>
<th style="padding:8px;text-align:center;width:90px;">Model</th>
<th style="padding:8px;text-align:center;width:220px;">이슈 원인</th>
<th style="padding:8px;text-align:center;width:240px;">조치 방안</th>
<th style="padding:8px;text-align:center;width:70px;">OWNER</th>
<th style="padding:8px;text-align:center;width:70px;">Link</th>
</tr>
</thead>
<tbody>
'''
streamlit_url = "http://10.197.51.108:9501/"
for i, reason in enumerate(reason_list):
bg_color = "#f3f5f8" if i % 2 else "#ffffff"
match = df[df["GMES"] == reason]
if not match.empty:
cause = str(match["원인"].values[0])
action = str(match["점검/대응 방안"].values[0])
owner = str(match["OWNER"].values[0])
else:
cause, action, owner = "[내용 없음]", "[내용 없음]", "[내용 없음]"
model = "model"
params = {
"subsidiary": self.anomaly_info["subsidiary"],
"line_name": self.anomaly_info["line_desc"],
"process_name": self.anomaly_info["equipment_desc"],
"process_id": self.anomaly_info["equipment_id"],
"model": model,
"mes_code": reason,
"ng_defect_cause": cause,
"trouble_shooting": action,
}
link = f"{streamlit_url}?{urlencode(params, quote_via=quote)}"
solution_html += f'''
<tr style="background-color:{bg_color};">
<td style="padding:8px;text-align:center;font-weight:bold;">{reason}</td>
<td style="padding:8px;text-align:center;">{model}</td>
<td style="padding:8px;text-align:center;">{cause}</td>
<td style="padding:8px;text-align:left;">{action}</td>
<td style="padding:8px;text-align:center;">{owner}</td>
<td style="padding:8px;text-align:center;">
<a href="{link}" target="_blank" style="color:#0563C1;text-decoration:underline;">View</a>
</td>
</tr>
'''
solution_html += '''
</tbody>
</table>
</td>
</tr>
</table>
'''
email_html = f""
# 표 대신 그래프 삽입
emailBody_header = f"""
<h1>공정 이상 탐지 분석 보고서 <strong>{self.anomaly_info['equipment_desc']}</strong></h1>
<div style="height:8px;background-color:#A50034;"></div>
하기 메일은 검사 자율 관제에서 시범으로 운영하는 메일입니다. 데이터는 해당 공정의 실제 실시간 데이터를 사용하고 있습니다.<br>
해당 메일은 발신 전용 메일로 기타 문의사항이나 건의사항은 seungjong.yoo@lge.com 으로 회신 주시기 바랍니다 <br>
"""
emailBody_report = f"""
<div class="table-container" style="background-color:#ffffff;border-radius:10px;
box-shadow:0 4px 12px rgba(0,0,0,0.08);overflow:hidden;width:900px;margin:20px 0;">
<table style="border-collapse:collapse;width:100%;font-family:'Noto Sans KR','Roboto',sans-serif;
font-size:15px;border:1px solid #ddd;table-layout:auto;word-break:keep-all;white-space:nowrap;">
<thead style="background-color:#A50034;color:#ffffff;">
<tr>
<th style="padding:8px 18px;text-align:center;">항목</th>
<th style="padding:8px 18px;text-align:center;">내용</th>
</tr>
</thead>
<tbody>
<tr style="background-color:#ffffff;">
<td style="padding:5px 18px;text-align:center;color:#333;">법인(subsidiary)</td>
<td style="padding:5px 18px;text-align:center;color:#333;"><b>{self.anomaly_info['subsidiary']}</b></td>
</tr>
<tr style="background-color:#f3f5f8;">
<td style="padding:5px 18px;text-align:center;color:#333;">제품(product)</td>
<td style="padding:5px 18px;text-align:center;color:#333;"><b>{self.anomaly_info['product']}</b></td>
</tr>
<tr style="background-color:#ffffff;">
<td style="padding:5px 18px;text-align:center;color:#333;">라인(line)</td>
<td style="padding:5px 18px;text-align:center;color:#333;"><b>{self.anomaly_info['line_desc']}</b></td>
</tr>
<tr style="background-color:#f3f5f8;">
<td style="padding:5px 18px;text-align:center;color:#333;">장비(equipment)</td>
<td style="padding:5px 18px;text-align:center;color:#333;"><b>{self.anomaly_info['equipment_desc']}</b></td>
</tr>
<tr style="background-color:#ffffff;">
<td style="padding:5px 18px;text-align:center;color:#333;">이상 감지 시작시점</td>
<td style="padding:5px 18px;text-align:center;color:#333;"><b>{self.anomaly_info['start_period']}</b></td>
</tr>
<tr style="background-color:#f3f5f8;">
<td style="padding:5px 18px;text-align:center;color:#333;">이상 감지 종료시점</td>
<td style="padding:5px 18px;text-align:center;color:#333;"><b>{self.anomaly_info['end_period']}</b></td>
</tr>
<tr style="background-color:#ffffff;">
<td style="padding:5px 18px;text-align:center;color:#333;">불량항목</td>
<td style="padding:5px 18px;text-align:center;color:#333;"><b>{reason_summary}</b></td>
</tr>
<tr style="background-color:#f3f5f8;">
<td style="padding:5px 18px;text-align:center;color:#333;">불량 Model Suffix</td>
<td style="padding:5px 18px;text-align:center;color:#333;"><b>{suffix_summary}</b></td>
</tr>
</tbody>
</table>
</div>
<a href="http://zenis.lge.com:9090/anomaly_analysis?id={self.anomaly_info['id']}">
추가적인 정보를 확인하거나 해당 이슈가 해결이 된 경우 여기를 눌러 anomaly 상태를 해제해주세요.
</a><br>
로그인 창이 나오면 로그인 후 링크를 다시 눌러 주세요
<br><br>
<hr><h1><strong>이상 감지 시점 불량률 Trend</strong></h1>
이상 감지 시점 <strong style="color: red;">당시의</strong> (분 당) 불량률<sup><a href="#" >[1]</a></sup> Trend와 불량항목 순위를 보여 줍니다 </br>
<span style="font-size:13px;color:#777;margin-left:4px;">
*[1] 분 당 불량률: 일정 시간(1분) 동안 생산된 제품 중 불량품이 차지하는 비율을 나타내는 지표
</span>
{graph_section_html}
<div class="table-container" style="background-color:#ffffff;border-radius:10px;box-shadow:0 4px 12px rgba(0,0,0,0.08);overflow:hidden;width:900px;margin:30px 0;">
<table style="border-collapse:collapse;width:100%;font-family:'Noto Sans KR','Roboto',sans-serif;font-size:15px;">
<caption style="caption-side:top;font-size:1.3em;font-weight:700;color:#A50034;padding:14px;text-align:left;">이상 탐지 정보</caption>
<thead style="background-color:#A50034;color:#ffffff;">
<tr>
<th style="padding:8px 20px;text-align:center;">항목</th>
<th style="padding:8px 20px;text-align:center;">내용</th>
</tr>
</thead>
<tbody>
<tr><td style="font-weight:600;color:#333;text-align:center;">시작 시점</td><td style="text-align:center;">{self.anomaly_info['start_period']}</td></tr>
<tr style="background-color:#f3f5f8;"><td style="font-weight:600;color:#333;text-align:center;">종료 시점</td><td style="text-align:center;">{self.anomaly_info['end_period']}</td></tr>
<tr><td style="font-weight:600;color:#333;text-align:center;">총 생산량</td><td style="text-align:center;">{num_filtered_origin_df}개</td></tr>
<tr style="background-color:#f3f5f8;"><td style="font-weight:600;color:#333;text-align:center;">불량 수량(NG)</td><td style="text-align:center;">{num_filtered_origin_df_NG}개</td></tr>
<tr><td style="font-weight:600;color:#333;text-align:center;">불량률</td><td style="text-align:center;">{round((num_filtered_origin_df_NG / num_filtered_origin_df) * 100, 2) if num_filtered_origin_df > 0 else 0.0}%</td></tr>
<tr style="background-color:#f3f5f8;"><td style="font-weight:600;color:#333;text-align:center;">이상치 정도</td><td style="text-align:center;">{self.anomaly_info['severity']}</td></tr>
</tbody>
</table>
</div>
{styled_html}
<br>
<hr><h1 id="solution"><strong>이상 원인 및 조치 방법</strong></h1>
개발 진행 중
{solution_html}
<br><br>
<hr><h2>[참고] 이상 감지 시점 전 (24시간) 불량률 Trend</strong></h2>
이상 감지 시점 <strong style="color: red;">전후 24시간</strong>의 분당 불량률 Trend와 불량항목 순위를 보여줍니다
{graphs_margin_html}
<div class="table-container" style="background-color:#ffffff;border-radius:10px;box-shadow:0 4px 12px rgba(0,0,0,0.08);overflow:hidden;width:900px;margin:30px 0;">
<table style="border-collapse:collapse;width:100%;font-family:'Noto Sans KR','Roboto',sans-serif;font-size:15px;">
<caption style="caption-side:top;font-size:1.3em;font-weight:700;color:#A50034;padding:14px;text-align:left;">이상 감지 전 24시간 생산 요약</caption>
<thead style="background-color:#A50034;color:#ffffff;">
<tr>
<th style="padding:8px 20px;text-align:center;">불량 항목</th>
<th style="padding:8px 20px;text-align:center;">개수 (개)</th>
</tr>
</thead>
<tbody>
<tr><td style="font-weight:600;color:#333;text-align:center;">전체 생산 수</td><td style="text-align:center;">{num_daily_origin_df}개</td></tr>
<tr style="background-color:#f3f5f8;"><td style="font-weight:600;color:#333;text-align:center;">OK 개수</td><td style="text-align:center;">{(num_analysis_margin_df-num_analysis_margin_df_NG)}개</td></tr>
<tr><td style="font-weight:600;color:#333;text-align:center;">NG 개수</td><td style="text-align:center;">{num_daily_origin_df_NG}개</td></tr>
<tr style="background-color:#f3f5f8;"><td style="font-weight:600;color:#333;text-align:center;">전체 불량률(1시간)</td><td style="text-align:center;">{daily_NG_ratio:.2f}%</td></tr>
</tbody>
</table>
</div>
<br>
<hr><h2>[참고] 이상 감지 시점 전 (24시간) EWMA Trend</strong></h2>
ewma는 트랜드가 반영된 누적 불량률으로써 해당 기준으로 이상치를 판단합니다
{additional_html}
"""
email_html += emailBody_report
clean_html = remove_img_tags(emailBody_report) + solution_html
report_prompt = f"""
다음은 공정 이상 탐지 분석결과입니다.
- 사용자 요청: {clean_html}에 대한 분석
절대로 clean_html에 있는 정보 안에서만 대답해야 됩니다.
아래 지침에 따라 보고서를 작성하세요.
- 아래 지침과 별개의 내용은 기재하지 마세요
[지침]
- 한국어로 답변을 해야 합니다
- 반드시 1줄에서 2줄 사이에 Text 로만 나타내야 합니다.
- 데이터에 기반한 분석만 하고 예측 및 추측은 절대 금지합니다.
- 법인, 제품, 라인, 장비 정보를 반드시 포함하여서 어떤 설비가 이상있는지를 알려주고 이상감지 시작지점 ~ 이상감지 종료 시점에 어떤 model suffix에 어떤 불량항목이 몇개 있는지 알려줘야 합니다
"""
logger.info(f"[Processing] 메일 내용을 생성 중입니다")
ai_analysis_text = self.mail_creator.invoke([{"role": "user", "content": report_prompt}])
# ai_analysis_text = self.mail_creator.invoke([{"role": "user", "content": analysis_prompt}])
# AIMessage → 문자열 변환
if hasattr(ai_analysis_text, "content"):
ai_analysis_text = ai_analysis_text.content
else:
ai_analysis_text = str(ai_analysis_text)
ai_analysis_html = self.format_ai_analysis_html(ai_analysis_text)
ai_response_html = f"""
<h3>분석 요약 (AI 생성)</h3>
<p>(아래 내용은 공정 데이터를 기반으로 AI가 요약한 텍스트 입니다)<br>
{ai_analysis_html}</p>
"""
logger.info(f"[Processing] 번역을 시작합니다")
translate_result = self.mail_creator.invoke([{"role": "user", "content": f"translate into English below : {ai_analysis_text}"}])
# AIMessage 객체일 경우 content 속성 사용
if hasattr(translate_result, "content"):
translate_text = translate_result.content
else:
translate_text = str(translate_result)
translate_html = self.format_ai_analysis_html(translate_text)
ai_response_html += translate_html
email_html = emailBody_header + ai_response_html+ '<br><strong>원인 및 조치 방법</strong> : <sup><a href="#solution">[바로가기]</a></sup><hr>' + email_html
# ✅ 메일 전송 직전 감싸기 (메일 본문을 1024px로 고정하고 가운데 정렬)
# margin:0 auto; : 가운데 margin:0 왼쪽
email_html = f"""
<div style="width:1024px; margin:0; background-color:#f9f9f9; padding:20px;
font-family:'Noto Sans KR','Roboto',sans-serif; box-sizing:border-box;">
<div style="background-color:#ffffff; border-radius:10px; padding:30px;
box-shadow:0 4px 12px rgba(0,0,0,0.1);">
{email_html}
</div>
</div>
"""
# ✅ output 폴더에 저장
output_dir = "output"
os.makedirs(output_dir, exist_ok=True)
output_path = os.path.join(output_dir, "email_result.html")
with open(output_path, "w", encoding="utf-8") as f:
print(f"Complete to save at : {output_path}")
f.write(email_html)
self.anomaly_mailSending(email_html,self.anomaly_info)
return stateimport streamlit as st
import yaml
import streamlit_authenticator as stauth
from bcrypt import hashpw, gensalt
from urllib.parse import unquote
st.set_page_config(page_title="Trouble Shooting Editor", layout="wide")
# ----------------------------
# 1) 설정 로드
# ----------------------------
with open("config/user_info.yaml", encoding="utf-8") as file:
config = yaml.load(file, Loader=stauth.SafeLoader)
allowed_emails = config["preauthorized"]["emails"]
authenticator = stauth.Authenticate(
config["credentials"],
config["cookie"]["name"],
config["cookie"]["key"],
config["cookie"]["expiry_days"],
)
# ----------------------------
# 2) URL 파라미터 안전하게 읽기
# ----------------------------
def get_query_param(name, default=None):
value = st.query_params.get(name, default)
if value is None:
return default
if isinstance(value, list):
return value[0] if value else default
return value
# URL 파라미터 읽기
incoming_params = {
"subsidiary": get_query_param("subsidiary"),
"line_name": get_query_param("line_name"),
"process_name": get_query_param("process_name"),
"process_id": get_query_param("process_id"),
"model": get_query_param("model"),
"mes_code": get_query_param("mes_code"),
"ng_defect_cause": get_query_param("ng_defect_cause"),
"trouble_shooting": get_query_param("trouble_shooting"),
}
# 필요하면 session_state에 저장
if "mail_params" not in st.session_state:
st.session_state.mail_params = incoming_params
else:
# URL로 새 값이 들어온 경우 갱신
for k, v in incoming_params.items():
if v is not None:
st.session_state.mail_params[k] = v
params = st.session_state.mail_params
# ----------------------------
# 3) 로그인 폼
# ----------------------------
if "authentication_status" not in st.session_state or st.session_state["authentication_status"] in (None, False):
try:
authenticator.login(
location="main",
fields={
"Form name": "Trouble_Shooting Editor",
"Username": "USER_ID",
"Password": "USER_PW",
"Login": "Login",
"Captcha": "Captcha",
},
captcha=False,
key="auth_login_form",
)
auth_status = st.session_state.get("authentication_status", None)
if auth_status:
st.rerun()
except Exception as e:
st.error(f"로그인 오류: {e}")
# ----------------------------
# 4) 회원가입 폼
# ----------------------------
st.sidebar.subheader("회원가입")
st.sidebar.info("아래 아이디로만 가입이 가능합니다")
new_username = st.sidebar.selectbox("새 사용자 ID", allowed_emails)
new_name = st.sidebar.text_input("이름")
new_password = st.sidebar.text_input("비밀번호", type="password")
new_password_confirm = st.sidebar.text_input("비밀번호 확인", type="password")
if st.sidebar.button("회원가입", key="register_button"):
if new_username in config["credentials"]["usernames"]:
st.sidebar.error("이미 가입된 사용자 ID입니다.")
elif new_username not in allowed_emails:
st.sidebar.error("회원가입이 허용되지 않은 이메일입니다.")
elif new_password != new_password_confirm:
st.sidebar.error("비밀번호가 일치하지 않습니다.")
elif not new_name.strip():
st.sidebar.error("이름을 입력해주세요.")
elif not new_password.strip():
st.sidebar.error("비밀번호를 입력해주세요.")
else:
hashed_password = hashpw(new_password.encode(), gensalt()).decode()
config["credentials"]["usernames"][new_username] = {
"name": new_name,
"password": hashed_password,
}
with open("config/user_info.yaml", "w", encoding="utf-8") as file:
yaml.dump(config, file, allow_unicode=True)
st.sidebar.success("회원가입이 완료되었습니다. 로그인 해주세요.")
# ----------------------------
# 5) 로그인 성공 후 화면
# ----------------------------
if st.session_state.get("authentication_status"):
name = st.session_state.get("name", "")
st.title(f"Welcome *{name}*")
st.subheader("전달받은 Trouble Shooting 정보")
col1, col2 = st.columns(2)
with col1:
st.text_input("Subsidiary", value=params.get("subsidiary") or "", disabled=True)
st.text_input("Line Name", value=params.get("line_name") or "", disabled=True)
st.text_input("Process Name", value=params.get("process_name") or "", disabled=True)
st.text_input("Process ID", value=params.get("process_id") or "", disabled=True)
with col2:
st.text_input("Model", value=params.get("model") or "", disabled=True)
st.text_input("Mes Code", value=params.get("mes_code") or "", disabled=True)
st.text_input("NG Defect Cause", value=params.get("ng_defect_cause") or "", disabled=True)
st.text_area("Trouble Shooting", value=params.get("trouble_shooting") or "", height=200, disabled=True)
authenticator.logout(button_name="Logout", location="main", key="auth_logout_btn")
elif st.session_state.get("authentication_status") is False:
st.error("Username/password is incorrect")
else:
st.warning("Please enter your username and password")
user_id_lst = list(config["credentials"]["usernames"].keys())
st.sidebar.header("ID를 찾아보세요")
st.sidebar.selectbox("ID를 입력하세요", user_id_lst)import streamlit as st
import yaml
import streamlit_authenticator as stauth
from bcrypt import hashpw, gensalt
# --- 1) 설정 로드 ---
with open('config/user_info.yaml') as file:
config = yaml.load(file, Loader=stauth.SafeLoader)
# --- 2) 이메일 리스트 로드 ---
allowed_emails = config['preauthorized']['emails']
# --- 3) Authenticator 인스턴스 ---
authenticator = stauth.Authenticate(
config['credentials'],
config['cookie']['name'],
config['cookie']['key'],
config['cookie']['expiry_days'],
)
# --- 4) URL 파라미터 읽기 ---
query_params = st.query_params
subsidiary = query_params.get('subsidiary', [None])[0]
line_name = query_params.get('line_name', [None])[0]
process_name = query_params.get('process_name', [None])[0]
process_id = query_params.get('process_id', [None])[0]
model = query_params.get('model', [None])[0]
mes_code = query_params.get('mes_code', [None])[0]
ng_defect_cause = query_params.get('ng_defect_cause', [None])[0]
trouble_shooting = query_params.get('trouble_shooting', [None])[0]
# col1, col2 = st.columns([1,1])
# --- 4) 로그인 폼: '아직 인증 안 됐을 때'에만 렌더링 ---
if 'authentication_status' not in st.session_state or st.session_state['authentication_status'] in (None, False):
try:
# 구버전(v0.2.x)은 반환값이 없고 세션 상태만 갱신합니다.
# 신버전(v0.3.x)은 반환값을 주지만, 여기선 굳이 받지 않습니다(버전 호환 목적).
authenticator.login(
location="main",
fields={
'Form name': 'Trouble_Shooting Editor',
'Username': 'USER_ID',
'Password': 'USER_PW',
'Login': 'Login',
'Captcha': 'Captcha'
},
captcha=False,
key="auth_login_form"
)
# 세션 상태에서 인증 결과 확인
auth_status = st.session_state.get('authentication_status', None)
# 로그인 성공 시 즉시 리런 → 다음 렌더에서 폼이 없어짐
if auth_status:
st.rerun() # Streamlit 1.27+ / 구버전은 st.experimental_rerun()
except Exception as e:
st.error(e)
# 회원가입 폼
st.sidebar.subheader("회원가입")
st.sidebar.info("아래 아이디로만 가입이 가능합니다")
new_username = st.sidebar.selectbox("새 사용자 ID",allowed_emails)
new_name = st.sidebar.text_input("이름")
new_password = st.sidebar.text_input("비밀번호", type="password")
new_password_confirm = st.sidebar.text_input("비밀번호 확인", type="password")
if st.sidebar.button("회원가입", key="register_button"):
if new_username in config['credentials']['usernames']:
st.sidebar.error("이미 가입된 사용자 ID입니다.")
elif new_username not in allowed_emails:
st.sidebar.error("회원가입이 허용되지 않은 이메일입니다.")
elif new_password != new_password_confirm:
st.sidebar.error("비밀번호가 일치하지 않습니다.")
else:
hashed_password = hashpw(new_password.encode(), gensalt()).decode()
config['credentials']['usernames'][new_username] = {
'name': new_name,
'password': hashed_password
}
with open('config/user_info.yaml', 'w') as file:
yaml.dump(config, file)
st.sidebar.success("회원가입이 완료되었습니다. 로그인 해주세요.")
# --- 5) 로그인 상태별 UI ---
if st.session_state.get('authentication_status'):
# 구/신버전 모두에서 세션 상태에 name/username이 들어옵니다.
name = st.session_state.get("name", "")
st.title(f'Welcome *{name}*')
# --- 7) URL 파라미터로 전달된 정보 표시 ---
st.write(f"Subsidiary: {subsidiary}")
st.write(f"Line Name: {line_name}")
st.write(f"Process Name: {process_name}")
st.write(f"Process ID: {process_id}")
st.write(f"Model: {model}")
st.write(f"Mes Code: {mes_code}")
st.write(f"NG Defect Cause: {ng_defect_cause}")
st.write(f"Trouble Shooting: {trouble_shooting}")
# 로그아웃 버튼 (위치/키 지정 권장)
authenticator.logout(button_name="Logout", location="main", key="auth_logout_btn") #location="sidebar"
elif st.session_state.get('authentication_status') is False:
st.error('Username/password is incorrect')
else:
st.warning('Please enter your username and password')
user_ID_lst = list(config['credentials']['usernames'].keys())
st.sidebar.header("ID를 찾아보세요")
st.sidebar.selectbox("ID를 입력하세요", user_ID_lst)from urllib.parse import urlencode, quote
solution_html = '''
<table role="presentation" cellspacing="0" cellpadding="0" border="0" width="900" style="width:900px;border-collapse:collapse;font-family:Arial,'Noto Sans KR',sans-serif;font-size:14px;color:#333;background-color:#ffffff;">
<tr>
<td style="padding:14px;font-size:20px;font-weight:bold;color:#A50034;">
불량 항목별 원인 및 조치
</td>
</tr>
<tr>
<td>
<table role="presentation" cellspacing="0" cellpadding="0" border="1" width="900" style="width:900px;border-collapse:collapse;border:1px solid #ddd;">
<thead>
<tr style="background-color:#A50034;color:#ffffff;">
<th style="padding:8px;text-align:center;width:100px;">불량 항목</th>
<th style="padding:8px;text-align:center;width:90px;">Model</th>
<th style="padding:8px;text-align:center;width:220px;">이슈 원인</th>
<th style="padding:8px;text-align:center;width:240px;">조치 방안</th>
<th style="padding:8px;text-align:center;width:70px;">OWNER</th>
<th style="padding:8px;text-align:center;width:70px;">Link</th>
</tr>
</thead>
<tbody>
'''
streamlit_url = "http://10.197.51.108:9501/"
for i, reason in enumerate(reason_list):
bg_color = "#f3f5f8" if i % 2 else "#ffffff"
match = df[df["GMES"] == reason]
if not match.empty:
cause = str(match["원인"].values[0])
action = str(match["점검/대응 방안"].values[0])
owner = str(match["OWNER"].values[0])
else:
cause, action, owner = "[내용 없음]", "[내용 없음]", "[내용 없음]"
model = "model"
params = {
"subsidiary": self.anomaly_info["subsidiary"],
"line_name": self.anomaly_info["line_desc"],
"process_name": self.anomaly_info["equipment_desc"],
"process_id": self.anomaly_info["equipment_id"],
"model": model,
"mes_code": reason,
"ng_defect_cause": cause,
"trouble_shooting": action,
}
link = f"{streamlit_url}?{urlencode(params, quote_via=quote)}"
solution_html += f'''
<tr style="background-color:{bg_color};">
<td style="padding:8px;text-align:center;font-weight:bold;">{reason}</td>
<td style="padding:8px;text-align:center;">{model}</td>
<td style="padding:8px;text-align:center;">{cause}</td>
<td style="padding:8px;text-align:left;">{action}</td>
<td style="padding:8px;text-align:center;">{owner}</td>
<td style="padding:8px;text-align:center;">
<a href="{link}" target="_blank" style="color:#0563C1;text-decoration:underline;">View</a>
</td>
</tr>
'''
solution_html += '''
</tbody>
</table>
</td>
</tr>
</table>
''' solution_html = '''
<div class="table-container" style="background-color:#ffffff;border-radius:10px;box-shadow:0 4px 12px rgba(0,0,0,0.08);overflow:hidden;width:900px;margin:30px 0;">
<table style="border-collapse:collapse;width:100%;font-family:'Noto Sans KR','Roboto',sans-serif;font-size:15px;table-layout:fixed;">
<caption style="caption-side:top;font-size:1.3em;font-weight:700;color:#A50034;padding:14px;text-align:left;">불량 항목별 원인 및 조치</caption>
<thead style="background-color:#A50034;color:#ffffff;">
<tr>
<th style="padding:8px 20px;text-align:center;width:120px;">불량 항목</th>
<th style="padding:8px 20px;text-align:center;width:120px;">Model</th>
<th style="padding:8px 20px;text-align:center;width:250px;">이슈 원인</th>
<th style="padding:8px 20px;text-align:center;width:300px;">조치 방안</th>
<th style="padding:8px 20px;text-align:center;width:80px;">OWNER</th>
<th style="padding:8px 20px;text-align:center;width:80px;">Link</th>
</tr>
</thead>
<tbody>
'''
streamlit_url = "http://10.197.51.108:9501/"
for i, reason in enumerate(reason_list):
bg_color = "#f3f5f8" if i % 2 else "#ffffff"
# GMES 컬럼에서 reason과 일치하는 행 찾기
match = df[df["GMES"] == reason]
if not match.empty:
cause = match["원인"].values[0]
action = match["점검/대응 방안"].values[0]
owner = match["OWNER"].values[0]
else:
cause, action, owner = "[내용 없음]", "[내용 없음]", "[내용 없음]"
# # Model 정보 추출
# model_suffix = self.anomaly_info['product_specification_id']
# model = model_suffix.split('-')[0] if '-' in model_suffix else model_suffix
# URL 파라미터 생성
params = (
f"subsidiary={self.anomaly_info['subsidiary']}&"
f"line_name={self.anomaly_info['line_desc']}&"
f"process_name={self.anomaly_info['equipment_desc']}&"
f"process_id={self.anomaly_info['equipment_id']}&"
f"model= {'model'}&"
f"mes_code={reason}&"
f"ng_defect_cause={cause}&"
f"trouble_shooting={action}"
)
link = f"{streamlit_url}?{params}"
solution_html += f'''
<tr style="background-color:{bg_color};">
<td style="text-align:center;font-weight:600;color:#333;">{reason}</td>
<td style="text-align:center;font-weight:600;color:#333;">{"model"}</td>
<td style="text-align:center;">{cause}</td>
<td style="text-align:left;padding-left:10px;">{action}</td>
<td style="text-align:center;">{owner}</td>
<td style="text-align:center;">
<a href="{link}" target="_blank">View Details</a>
</td>
</tr>
'''
# 마무리
solution_html += '''
</tbody>
</table>
</div>
'''import streamlit as st
import yaml
import streamlit_authenticator as stauth
with open('config/user_info.yaml') as file:
config = yaml.load(file, Loader=stauth.SafeLoader)
## yaml 파일 데이터로 객체 생성
authenticator = stauth.Authenticate(
config['credentials'],
config['cookie']['name'],
config['cookie']['key'],
config['cookie']['expiry_days'],
)
## 로그인 위젯 렌더링
## log(in/out)(로그인 위젯 문구, 버튼 위치)
## 버튼 위치 = "main" or "sidebar"
name, authentication_status, username = authenticator.login("Login","main")
# authentication_status : 인증 상태 (실패=>False, 값없음=>None, 성공=>True)
if authentication_status == False:
st.error("Username/password is incorrect")
if authentication_status == None:
st.warning("Please enter your username and password")
if authentication_status:
authenticator.logout("Logout","sidebar")
st.sidebar.title(f"Welcome {name}")
## 로그인 후 기능들 작성 ##
ValueError: Location must be one of 'main' or 'sidebar' or 'unrendered'
2026-03-13 04:02:06.503 Uncaught app execution
Traceback (most recent call last):
File "/usr/local/lib/python3.13/site-packages/streamlit/runtime/scriptrunner/exec_code.py", line 129, in exec_func_with_error_handling
result = func()
File "/usr/local/lib/python3.13/site-packages/streamlit/runtime/scriptrunner/script_runner.py", line 671, in code_to_exec
exec(code, module.__dict__) # noqa: S102
~~~~^^^^^^^^^^^^^^^^^^^^^^^
File "/app/trouble_shooting_UI.py", line 18, in <module>
name, authentication_status, username = authenticator.login("Login","main")
~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.13/site-packages/streamlit_authenticator/views/authentication_view.py", line 329, in login
raise ValueError("Location must be one of 'main' or 'sidebar' or 'unrendered'")
ValueError: Location must be one of 'main' or 'sidebar' or 'unrendered'from langchain_chroma import Chroma
vectorstore = Chroma.from_documents(
documents=clean_splits,
embedding=embedding_model,
collection_name="temp_test_collection"
)
print("count =", vectorstore._collection.count())import os
import shutil
from pathlib import Path
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
# -------------------------------
# 설정
# -------------------------------
persist_directory = "./chroma_store"
collection_name = "reason_manual_v1"
# -------------------------------
# 1. 기존 DB 완전 삭제
# -------------------------------
p = Path(persist_directory)
if p.exists():
print("Removing old Chroma DB...")
shutil.rmtree(p)
# 새 폴더 생성
p.mkdir(parents=True, exist_ok=True)
# 권한 확인
print("dir exists:", p.exists())
print("dir writable:", os.access(p, os.W_OK))
# -------------------------------
# 2. Embedding 모델
# -------------------------------
embedding_model = OpenAIEmbeddings(
model="text-embedding-3-large", # 서버 지원 모델
api_key=api_key,
base_url=base_url,
# OpenAI 호환 서버 안정 옵션
tiktoken_enabled=False,
check_embedding_ctx_length=False,
)
# -------------------------------
# 3. 문서 sanity check
# -------------------------------
print("documents:", len(clean_splits))
print("sample:", clean_splits[0].page_content[:200])
# -------------------------------
# 4. Vector DB 생성 + 저장
# -------------------------------
vectorstore = Chroma.from_documents(
documents=clean_splits,
embedding=embedding_model,
persist_directory=persist_directory,
collection_name=collection_name,
)
print("DB count:", vectorstore._collection.count())
# -------------------------------
# 5. 검색 테스트
# -------------------------------
query = "설비 이상 원인 분석 방법은?"
docs = vectorstore.similarity_search(query, k=3)
print("\nSearch results\n")
for i, d in enumerate(docs, 1):
print(f"--- doc {i} ---")
print(d.page_content[:300])
print(d.metadata)
print()import shutil
from pathlib import Path
from langchain_chroma import Chroma
persist_directory = "./chroma_store_new" # 새 폴더로 바꿔서 테스트
collection_name = "reason_manual_v1"
p = Path(persist_directory)
if p.exists():
shutil.rmtree(p)
p.mkdir(parents=True, exist_ok=True)
vectorstore = Chroma.from_documents(
documents=clean_splits,
embedding=embedding_model,
persist_directory=str(p),
collection_name=collection_name
)
print("count =", vectorstore._collection.count())import shutil
import os
from langchain_chroma import Chroma
persist_directory = "./chroma_store"
# 1) 현재 프로세스에서 vectorstore 객체 있으면 커널 재시작이 제일 안전
# 주피터면 Restart Kernel 권장
# 2) 기존 DB 삭제
if os.path.exists(persist_directory):
shutil.rmtree(persist_directory)
# 3) 다시 생성
vectorstore = Chroma.from_documents(
documents=clean_splits, # 네가 정리한 문서 리스트
embedding=embedding_model,
persist_directory=persist_directory,
collection_name="reason_manual_v1"
)
print(vectorstore._collection.count())import os
import shutil
from dotenv import load_dotenv
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
load_dotenv()
base_url = os.getenv("base_url")
api_key = os.getenv("api_key")
collection_name = "reason_manual_v1"
persist_directory = "./chroma_store"
# 기존 테스트 흔적이 있으면 지우고 새로 시작
if os.path.exists(persist_directory):
shutil.rmtree(persist_directory)
embedding_model = OpenAIEmbeddings(
model="text-embedding-3-large", # 서버가 지원하는 모델명으로 바꿔
api_key=api_key,
base_url=base_url,
# 중요: OpenAI 호환 서버에서 invalid input type 방지
tiktoken_enabled=False,
check_embedding_ctx_length=False,
# 선택
chunk_size=100
)
# all_splits 점검
print("num docs =", len(all_splits))
print("type =", type(all_splits[0]))
print("sample text =", repr(all_splits[0].page_content[:200]))
print("sample metadata =", all_splits[0].metadata)
# 빈 문자열/비문자열 제거
clean_splits = []
for doc in all_splits:
text = doc.page_content
if text is None:
continue
if not isinstance(text, str):
text = str(text)
text = text.strip()
if text == "":
continue
doc.page_content = text
clean_splits.append(doc)
print("clean docs =", len(clean_splits))
vectorstore = Chroma.from_documents(
documents=clean_splits,
embedding=embedding_model,
persist_directory=persist_directory,
collection_name=collection_name
)
print("count =", vectorstore._collection.count())
query = "설비 이상 원인 분석 방법은?"
docs = vectorstore.similarity_search(query, k=3)
for i, d in enumerate(docs, 1):
print(f"\n--- doc {i} ---")
print(d.page_content[:300])
print(d.metadata)
--------------------------------------------------------------------------- BadRequestError Traceback (most recent call last) Cell In[60], line 35 33 if count == 0: 34 print("Empty collection. Adding documents...") ---> 35 vectorstore.add_documents(all_splits) 37 print("after count =", vectorstore._collection.count()) 39 # 저장된 데이터 일부 확인 File ~/.pyenv/versions/Agent/lib/python3.10/site-packages/langchain_core/vectorstores/base.py:258, in VectorStore.add_documents(self, documents, **kwargs) 256 texts = [doc.page_content for doc in documents] 257 metadatas = [doc.metadata for doc in documents] --> 258 return self.add_texts(texts, metadatas, **kwargs) 259 msg = ( 260 f"`add_documents` and `add_texts` has not been implemented " 261 f"for {self.__class__.__name__} " 262 ) 263 raise NotImplementedError(msg) File ~/.pyenv/versions/Agent/lib/python3.10/site-packages/langchain_chroma/vectorstores.py:627, in Chroma.add_texts(self, texts, metadatas, ids, **kwargs) 625 texts = list(texts) 626 if self._embedding_function is not None: --> 627 embeddings = self._embedding_function.embed_documents(texts) 628 if metadatas: 629 # fill metadatas with empty dicts if somebody
...
-> 1047 raise self._make_status_error_from_response(err.response) from None 1049 break 1051 assert response is not None, "could not resolve response (should never happen)" BadRequestError: Error code: 400 - {'error': {'message': 'invalid input type', 'type': 'api_error', 'param': None, 'code': None}}
import os
from dotenv import load_dotenv
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
load_dotenv()
base_url = os.getenv("base_url") # 예: OpenAI 호환 임베딩 서버 주소
api_key = os.getenv("api_key")
collection_name = "reason_manual_v1"
persist_directory = "./chroma_store"
# OpenAIEmbeddings
embedding_model = OpenAIEmbeddings(
model="text-embedding-3-large", # 실제 쓰려는 임베딩 모델명으로 변경
api_key=api_key,
base_url=base_url, # OpenAI 공식이면 생략 가능
)
# Chroma 로드
vectorstore = Chroma(
persist_directory=persist_directory,
embedding_function=embedding_model,
collection_name=collection_name,
)
# 비어 있으면 문서 추가
count = vectorstore._collection.count()
print("before count =", count)
if count == 0:
print("Empty collection. Adding documents...")
vectorstore.add_documents(all_splits)
print("after count =", vectorstore._collection.count())
# 저장된 데이터 일부 확인
result = vectorstore.get(limit=2)
print(type(result))
print(result.keys())
print(result)
# 검색
query = "설비 이상 원인 분석 방법은?"
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
docs = retriever.invoke(query)
for i, d in enumerate(docs, 1):
print(f"\n--- doc {i} ---")
print(d.page_content[:300])
print(d.metadata)
ollama를 써서
import os
from dotenv import load_dotenv
load_dotenv()
base_url = os.getenv('base_url')
api_key = os.getenv('api_key')
collection_name = "reason_manual_v1"
# from langchain_openai import OpenAIEmbeddings
# embedding_model = OpenAIEmbeddings(
# openai_api_base=base_url,
# openai_api_key=api_key,
# model='jeffh/intfloat-multilingual-e5-large-instruct:q8_0',
# )
from langchain_community.embeddings import OllamaEmbeddings
embedding_model = OllamaEmbeddings(
model='jeffh/intfloat-multilingual-e5-large-instruct:q8_0', # Ollama에 미리 pull된 임베딩 모델
base_url=base_url
)
이렇게 바꾸고
from langchain_chroma import Chroma
persist_directory = './chroma_store'
if not os.path.exists(persist_directory):
print("Creating new Chroma store")
vectorstore = Chroma.from_documents(
documents=all_splits,
embedding=embedding_model,
persist_directory=persist_directory,
collection_name=collection_name
)
else:
print("Loading existing Chroma store")
vectorstore = Chroma(
persist_directory = persist_directory,
embedding_function = embedding_model,
collection_name=collection_name
)
vectorstore.get()
이거 출력했더니
from langchain_chroma import Chroma
persist_directory = './chroma_store'
if not os.path.exists(persist_directory):
print("Creating new Chroma store")
vectorstore = Chroma.from_documents(
documents=all_splits,
embedding=embedding_model,
persist_directory=persist_directory,
collection_name=collection_name
)
else:
print("Loading existing Chroma store")
vectorstore = Chroma(
persist_directory = persist_directory,
embedding_function = embedding_model,
collection_name=collection_name
)
vectorstore.get()
이렇게 나와
A futuristic smart factory control room dashboard showing real-time manufacturing process data,
multiple graphs and data streams flowing across large digital screens,
camera slowly zooming in,
high-tech industrial AI monitoring system,
cinematic lighting, ultra realistic, 4K
728x90