로그인 기능
This commit is contained in:
9700X_PC
parent
ebd8f58fc3
commit
193dfcd141
Binary file not shown.
File diff suppressed because one or more lines are too long
Binary file not shown.
|
|
@ -1,7 +0,0 @@
|
|||
[GPT]
|
||||
API_KEY='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'
|
||||
|
||||
[Filters]
|
||||
banned_tags = 오늘출발, 오늘발송
|
||||
banned_words = 금지어1, 금지어2, 금지어3
|
||||
disallowed_words = 비허용단어1, 비허용단어2, 비허용단어3
|
||||
18
main.py
18
main.py
|
|
@ -4,9 +4,9 @@ from PySide6.QtWidgets import QApplication
|
|||
from src.gui import TaobaoScraperApp
|
||||
from src.databaseManager import DatabaseManager
|
||||
from src.loggerModule import Logger
|
||||
from src.user_info_dialog import UserInfoDialog
|
||||
|
||||
import ctypes
|
||||
from ctypes import wintypes
|
||||
|
||||
|
||||
# COM 초기화 (멀티스레드 모드)
|
||||
|
|
@ -23,14 +23,18 @@ if __name__ == "__main__":
|
|||
|
||||
app = QApplication(sys.argv)
|
||||
|
||||
logger = Logger(log_file="Scrapper2.log", logger_name="Scrapper_Logger", level=logging.DEBUG)
|
||||
logger = Logger(log_file="Scrapper2.log", logger_name="Scrapper_Logger", level=logging.INFO)
|
||||
|
||||
db_manager = DatabaseManager(logger) # 데이터베이스 매니저 인스턴스 생성
|
||||
window = TaobaoScraperApp(logger, db_manager)
|
||||
|
||||
window.show()
|
||||
# 로그인 다이얼로그 실행
|
||||
login_dialog = UserInfoDialog()
|
||||
if login_dialog.exec(): # 로그인 성공 시
|
||||
window = TaobaoScraperApp(logger, db_manager)
|
||||
window.show()
|
||||
sys.exit(app.exec()) # 메인 UI 실행
|
||||
else: # 로그인 실패 시
|
||||
sys.exit(0) # 프로그램 종료
|
||||
|
||||
uninitialize_com() # COM 해제
|
||||
|
||||
sys.exit(app.exec())
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -3,9 +3,10 @@ import asyncio
|
|||
class ProcessingThread(QThread):
|
||||
progress = Signal(str) # 진행 상태를 GUI에 전달할 시그널
|
||||
|
||||
def __init__(self, post_processor):
|
||||
def __init__(self, post_processor, keyword_manager):
|
||||
super().__init__()
|
||||
self.post_processor = post_processor
|
||||
self.keyword_manager = keyword_manager
|
||||
# self.folder_path = folder_path
|
||||
|
||||
def run(self):
|
||||
|
|
@ -14,6 +15,8 @@ class ProcessingThread(QThread):
|
|||
"""
|
||||
try:
|
||||
self.progress.emit(f"DB 파일 처리 시작")
|
||||
|
||||
self.post_processor.set_keyword_manager(self.keyword_manager)
|
||||
|
||||
asyncio.run(self.post_processor.post_by_DB())
|
||||
|
||||
|
|
|
|||
Binary file not shown.
|
|
@ -0,0 +1,10 @@
|
|||
[GPT_API]
|
||||
API_KEY=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
|
||||
|
||||
[Kipris_API]
|
||||
API_KEY = X9Tz3JqC/JcCwxnNewA6qdloIN6QFIitVBgS1a2KVDYk1AmddaDTvzr6+t3dyLZV3gh2TPXdNhxsRQwaKP673Q==
|
||||
|
||||
[Filters]
|
||||
banned_tags = 오늘출발, 오늘발송
|
||||
banned_words = 금지어1, 금지어2, 금지어3
|
||||
disallowed_words = 비허용단어1, 비허용단어2, 비허용단어3
|
||||
|
|
@ -5,10 +5,14 @@ import pandas as pd
|
|||
import xlwings as xw
|
||||
import logging
|
||||
from PySide6.QtWidgets import QMessageBox
|
||||
from PySide6.QtCore import QObject, Signal
|
||||
|
||||
|
||||
class ExcelExporter:
|
||||
class ExcelExporter(QObject):
|
||||
progress_signal = Signal(int) # 프로그레스바 업데이트용 시그널
|
||||
|
||||
def __init__(self, logger, db_manager, base_excel_path="src\\baseXLS_Percenty.xlsx"):
|
||||
super().__init__() # QObject의 __init__ 메서드 호출 (문제 해결)
|
||||
self.logger =logger
|
||||
self.db_manager = db_manager # DBManager 인스턴스
|
||||
self.base_excel_path = base_excel_path
|
||||
|
|
@ -41,7 +45,8 @@ class ExcelExporter:
|
|||
self.logger.log(f"xlwings 시작", level=logging.DEBUG)
|
||||
|
||||
try:
|
||||
for i in range(0, len(df), 50):
|
||||
total_rows = len(df)
|
||||
for i in range(0, total_rows, 50):
|
||||
df_subset = df.iloc[i:i+50]
|
||||
self.logger.log(f"{i}번째 출력할 데이터:\n{df_subset}", level=logging.DEBUG) # 데이터 검증 로그 추가
|
||||
|
||||
|
|
@ -76,6 +81,10 @@ class ExcelExporter:
|
|||
|
||||
self.logger.log(f"{index + 1}번째 행 기록 완료", level=logging.DEBUG)
|
||||
|
||||
# 프로그레스바 업데이트
|
||||
progress = int((index + 1) / total_rows * 100)
|
||||
self.progress_signal.emit(progress)
|
||||
|
||||
wb.save(part_file_name) # SaveCopyAs 대신 save 사용
|
||||
wb.close()
|
||||
self.saved_files.append(part_file_name)
|
||||
|
|
|
|||
272
src/gui.py
272
src/gui.py
|
|
@ -1,69 +1,177 @@
|
|||
from PySide6.QtWidgets import QWidget, QVBoxLayout, QPushButton, QLabel, QMessageBox, QFileDialog
|
||||
from PySide6.QtCore import Slot
|
||||
import os
|
||||
|
||||
from PySide6.QtWidgets import QApplication, QWidget, QVBoxLayout, QPushButton, QLabel, QMessageBox, QTextBrowser, QDialog, QProgressBar, QTextEdit, QHBoxLayout, QMenuBar, QMenu
|
||||
from PySide6.QtGui import QAction
|
||||
from PySide6.QtCore import Qt, Slot, Signal, QThread
|
||||
import os, sys
|
||||
import configparser
|
||||
import logging
|
||||
from src.playwright_thread import PlaywrightThread
|
||||
from src.excel_export import ExcelExporter
|
||||
from src.post_processor import PostProcessor
|
||||
from src.xlsProcessingThread import XLSProcessingThread
|
||||
from src.ProcessingThread_for_DB import ProcessingThread
|
||||
from src.keyword.keyword_manager import KeywordManager
|
||||
from src.categoryManager import CategoryManager
|
||||
from src.user_info_dialog import UserInfoDialog
|
||||
|
||||
class TaobaoScraperApp(QWidget):
|
||||
def __init__(self, logger, db_manager):
|
||||
super().__init__()
|
||||
|
||||
self.logger = logger
|
||||
|
||||
self.logger.set_gui_logger(self.append_to_log)
|
||||
|
||||
self.is_logged_in = False # 로그인 상태 플래그
|
||||
|
||||
self.db_manager = db_manager
|
||||
self.setWindowTitle("Taobao Scraper")
|
||||
self.setWindowTitle("내차는 언제타냐 - 역소싱기")
|
||||
self.layout = QVBoxLayout()
|
||||
|
||||
# 메뉴바 생성
|
||||
self.menu_bar = QMenuBar(self)
|
||||
|
||||
# 설정 메뉴
|
||||
self.settings_menu = QMenu("설정", self)
|
||||
self.menu_bar.addMenu(self.settings_menu)
|
||||
|
||||
# 설정 메뉴에 금지어 관리 추가
|
||||
self.manage_keywords_action = QAction("금지어 관리", self)
|
||||
self.manage_keywords_action.triggered.connect(self.manage_keywords)
|
||||
self.settings_menu.addAction(self.manage_keywords_action)
|
||||
|
||||
# 설정 메뉴에 금지 카테고리 관리 추가
|
||||
self.manage_category_action = QAction("금지 카테고리 관리", self)
|
||||
self.manage_category_action.triggered.connect(self.manage_cat)
|
||||
self.settings_menu.addAction(self.manage_category_action)
|
||||
|
||||
# 도움말 메뉴
|
||||
self.help_menu = QMenu("도움말", self)
|
||||
self.menu_bar.addMenu(self.help_menu)
|
||||
|
||||
# 도움말 메뉴에 항목 추가
|
||||
self.help_action = QAction("도움말 보기", self)
|
||||
self.help_action.triggered.connect(self.show_help_dialog)
|
||||
self.help_menu.addAction(self.help_action)
|
||||
|
||||
# 사용자 정보 메뉴
|
||||
self.user_menu = QMenu("사용자 정보", self)
|
||||
self.menu_bar.addMenu(self.user_menu)
|
||||
|
||||
# 사용자 정보 메뉴에 항목 추가
|
||||
self.user_info_action = QAction("사용자 정보 보기", self)
|
||||
self.user_info_action.triggered.connect(self.show_user_info_dialog)
|
||||
self.user_menu.addAction(self.user_info_action)
|
||||
|
||||
# 로그창 추가
|
||||
self.log_text_edit = QTextEdit()
|
||||
self.log_text_edit.setReadOnly(True)
|
||||
self.log_text_edit.setMinimumHeight(200)
|
||||
|
||||
# Logger의 log_signal을 로그창에 연결
|
||||
self.logger.log_signal.connect(self.update_log_window)
|
||||
|
||||
# 프로그레스바 추가
|
||||
self.progress_bar = QProgressBar()
|
||||
self.progress_bar.setAlignment(Qt.AlignCenter)
|
||||
self.progress_bar.setValue(0)
|
||||
|
||||
self.start_button = QPushButton("시작")
|
||||
self.start_button.clicked.connect(self.start_scraping)
|
||||
self.excel_button = QPushButton("엑셀출력")
|
||||
self.excel_button.clicked.connect(self.save_to_excel)
|
||||
self.post_db__button = QPushButton("DB로 후처리")
|
||||
self.post_db__button = QPushButton("후처리")
|
||||
self.post_db__button.clicked.connect(self.post_process_by_DB)
|
||||
self.post_xls_button = QPushButton("수집맨xls로 후처리")
|
||||
self.post_xls_button.clicked.connect(self.post_process_by_xls)
|
||||
self.close_button = QPushButton("닫기")
|
||||
self.close_button.clicked.connect(self.close)
|
||||
|
||||
self.layout.addWidget(QLabel("Taobao Scraper"))
|
||||
self.layout.addWidget(self.start_button)
|
||||
self.layout.addWidget(self.post_db__button)
|
||||
self.layout.addWidget(self.post_xls_button)
|
||||
self.layout.addWidget(self.excel_button)
|
||||
self.layout.addWidget(self.close_button)
|
||||
|
||||
# 레이아웃 구성
|
||||
button_layout = QHBoxLayout()
|
||||
button_layout.addWidget(self.start_button)
|
||||
button_layout.addWidget(self.post_db__button)
|
||||
button_layout.addWidget(self.excel_button)
|
||||
button_layout.addWidget(self.close_button)
|
||||
|
||||
self.layout.addWidget(self.menu_bar)
|
||||
self.layout.addLayout(button_layout)
|
||||
self.layout.addWidget(QLabel("진행 상황"))
|
||||
self.layout.addWidget(self.progress_bar)
|
||||
self.layout.addWidget(QLabel("로그 출력"))
|
||||
self.layout.addWidget(self.log_text_edit)
|
||||
|
||||
self.setLayout(self.layout)
|
||||
|
||||
# base_dir 경로 가져오기
|
||||
base_dir = self.get_base_dir()
|
||||
self.xls_file_path = os.path.join(base_dir, "baseXLS_Percenty.xlsx")
|
||||
config_path = os.path.join(base_dir, "config.ini")
|
||||
self.config = self.load_config(config_path)
|
||||
|
||||
self.playwright_thread = PlaywrightThread(self.logger, self.db_manager)
|
||||
self.playwright_thread.data_collected.connect(self.on_data_collected)
|
||||
self.playwright_thread.progress_signal.connect(self.update_progress_bar)
|
||||
|
||||
self.excel_exporter = ExcelExporter(self.logger, self.db_manager)
|
||||
self.postProcessor = PostProcessor(self.logger, self.db_manager)
|
||||
self.categoryManager = CategoryManager(self.logger, self.xls_file_path)
|
||||
|
||||
self.excel_exporter = ExcelExporter(logger=self.logger, db_manager=self.db_manager)
|
||||
self.excel_exporter.progress_signal.connect(self.update_progress_bar)
|
||||
|
||||
self.postProcessor = PostProcessor(self.logger, self.db_manager, self.config, self.categoryManager)
|
||||
self.keyword_manager = KeywordManager(logger=self.logger, config=self.config, parent=self)
|
||||
|
||||
|
||||
def load_config(self, file_path: str) -> configparser.ConfigParser:
|
||||
"""
|
||||
config.ini 파일을 읽어서 ConfigParser 객체로 반환
|
||||
"""
|
||||
config = configparser.ConfigParser()
|
||||
try:
|
||||
config.read(file_path, encoding="utf-8") # UTF-8 인코딩으로 설정 파일 읽기
|
||||
self.logger.log(f"Config 파일 '{file_path}' 로드 성공", level=logging.INFO)
|
||||
except Exception as e:
|
||||
self.logger.log(f"Config 파일 로드 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
return config
|
||||
|
||||
def get_base_dir(self):
|
||||
"""
|
||||
실행 환경에 따라 base_dir을 설정하는 메서드.
|
||||
cx_Freeze로 패키징된 경우 실행 파일의 경로, 일반 Python 환경일 경우 __file__을 기준으로 설정.
|
||||
"""
|
||||
if getattr(sys, 'frozen', False): # 패키징된 경우
|
||||
base_dir = os.path.dirname(sys.executable)
|
||||
internal_dir = os.path.join(base_dir, '_internal') # _internal 디렉토리 포함
|
||||
if os.path.exists(internal_dir): # _internal 디렉토리가 존재하면 base_dir로 설정
|
||||
return internal_dir
|
||||
|
||||
else: # 일반 Python 실행 환경
|
||||
base_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
return base_dir
|
||||
|
||||
|
||||
@Slot(str)
|
||||
def append_to_log(self, message):
|
||||
self.log_text_edit.appendHtml(message) # log_widget은 QTextEdit 또는 QPlainTextEdit
|
||||
|
||||
@Slot()
|
||||
def start_scraping(self):
|
||||
self.logger.log(f"Playwright 스레드 시작.", level=logging.INFO)
|
||||
self.progress_bar.setValue(1)
|
||||
self.playwright_thread.start()
|
||||
|
||||
@Slot()
|
||||
def post_process_by_DB(self):
|
||||
self.logger.log(f"수집된 DB로 후처리.", level=logging.INFO)
|
||||
self.progress_bar.setValue(1)
|
||||
self.postProcessor.post_by_DB()
|
||||
|
||||
@Slot(str)
|
||||
def post_process_by_DB(self):
|
||||
def post_process_by_DB(self, ):
|
||||
"""
|
||||
폴더가 선택되었을 때 호출되는 슬롯
|
||||
"""
|
||||
self.logger.log(f"DB로 후처리 작업을 시작합니다.", level=logging.INFO)
|
||||
self.progress_bar.setValue(1)
|
||||
|
||||
# 스레드를 생성하여 작업 실행
|
||||
self.db_thread = ProcessingThread(self.postProcessor)
|
||||
self.db_thread = ProcessingThread(self.postProcessor, self.keyword_manager)
|
||||
self.db_thread.progress.connect(self.on_DB_progress)
|
||||
self.db_thread.start()
|
||||
|
||||
|
|
@ -74,46 +182,39 @@ class TaobaoScraperApp(QWidget):
|
|||
"""
|
||||
self.logger.log(message, level=logging.INFO)
|
||||
|
||||
@Slot(str)
|
||||
def manage_keywords(self):
|
||||
self.logger.log("금지어 관리 버튼 클릭됨", level=logging.DEBUG)
|
||||
self.keyword_manager.exec()
|
||||
|
||||
@Slot()
|
||||
def post_process_by_xls(self):
|
||||
self.logger.log(f"수집맨 XLS로 후처리.", level=logging.INFO)
|
||||
def manage_cat(self):
|
||||
"""
|
||||
'스스 카테고리' 시트를 열고 사용자에게 지침 메시지를 표시합니다.
|
||||
"""
|
||||
try:
|
||||
|
||||
default_folder = os.path.join(os.getcwd(), 'XLS')
|
||||
selected_folder = QFileDialog.getExistingDirectory(None, "XLS 폴더 선택", default_folder)
|
||||
# 파일이 존재하는지 확인
|
||||
if not os.path.exists(self.xls_file_path):
|
||||
QMessageBox.warning(self, "파일 없음", f"'{self.xls_file_path}' 파일이 존재하지 않습니다.")
|
||||
self.logger.log(f"'{self.xls_file_path}' 파일이 존재하지 않습니다.", level=logging.ERROR)
|
||||
return
|
||||
|
||||
if not selected_folder:
|
||||
self.logger.warning("폴더 선택이 취소되었습니다.")
|
||||
return
|
||||
# 엑셀 파일 열기 (시스템 기본 프로그램 사용)
|
||||
os.startfile(self.xls_file_path) # Windows 환경에서 엑셀 열기
|
||||
|
||||
self.logger.info(f"선택된 폴더: {selected_folder}")
|
||||
self.postProcessor.post_by_XLS(selected_folder)
|
||||
# 사용자 메시지 표시
|
||||
QMessageBox.information(
|
||||
self,
|
||||
"카테고리 관리",
|
||||
"금지할 카테고리 옆에 False를 입력하면 금지, 비어있거나 True면 허용"
|
||||
)
|
||||
self.logger.log(f"'{self.xls_file_path}' 파일이 실행되었습니다.", level=logging.INFO)
|
||||
|
||||
def post_process_by_xls(self):
|
||||
self.logger.log(f"수집맨 XLS로 후처리.", level=logging.INFO)
|
||||
|
||||
default_folder = os.path.join(os.getcwd(), 'XLS')
|
||||
self.logger.log(f"1", level=logging.INFO)
|
||||
|
||||
# 폴더 선택 다이얼로그 설정 (폴더 트리 모드)
|
||||
dialog = QFileDialog(self, "XLS 폴더 선택")
|
||||
self.logger.log(f"2", level=logging.INFO)
|
||||
|
||||
dialog.setFileMode(QFileDialog.Directory)
|
||||
self.logger.log(f"3", level=logging.INFO)
|
||||
|
||||
dialog.setOption(QFileDialog.ShowDirsOnly, True)
|
||||
self.logger.log(f"4", level=logging.INFO)
|
||||
|
||||
dialog.setDirectory(default_folder)
|
||||
self.logger.log(f"5", level=logging.INFO)
|
||||
|
||||
# 네이티브 다이얼로그 비활성화
|
||||
dialog.setOption(QFileDialog.DontUseNativeDialog, True)
|
||||
|
||||
# 비차단 방식으로 다이얼로그 열기
|
||||
dialog.fileSelected.connect(self.on_folder_selected) # 폴더 선택 시 슬롯 호출
|
||||
dialog.open()
|
||||
except Exception as e:
|
||||
# 오류 처리
|
||||
QMessageBox.critical(self, "오류", f"엑셀 파일 열기 중 오류 발생: {e}")
|
||||
self.logger.log(f"엑셀 파일 열기 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
@Slot(str)
|
||||
def on_folder_selected(self, selected_folder):
|
||||
|
|
@ -136,6 +237,7 @@ class TaobaoScraperApp(QWidget):
|
|||
|
||||
@Slot()
|
||||
def save_to_excel(self):
|
||||
self.progress_bar.setValue(1)
|
||||
success = self.excel_exporter.save_to_excel()
|
||||
if success:
|
||||
QMessageBox.information(self, "엑셀 출력", "엑셀 파일로 저장 완료")
|
||||
|
|
@ -148,3 +250,65 @@ class TaobaoScraperApp(QWidget):
|
|||
QMessageBox.information(self, "수집 완료", message)
|
||||
else:
|
||||
QMessageBox.warning(self, "수집 실패", message)
|
||||
|
||||
@Slot(int)
|
||||
def update_progress_bar(self, value):
|
||||
"""
|
||||
프로그레스바 업데이트
|
||||
"""
|
||||
self.progress_bar.setValue(value)
|
||||
|
||||
@Slot(str)
|
||||
def update_log_window(self, message):
|
||||
"""
|
||||
Logger로부터 전달받은 메시지를 로그창에 추가
|
||||
"""
|
||||
self.log_text_edit.append(message)
|
||||
|
||||
@Slot()
|
||||
def show_help_dialog(self):
|
||||
"""
|
||||
도움말 다이얼로그를 띄웁니다.
|
||||
"""
|
||||
self.logger.log("도움말 메뉴 클릭됨", level=logging.INFO)
|
||||
|
||||
# 다이얼로그 생성
|
||||
dialog = QDialog(self)
|
||||
dialog.setWindowTitle("도움말")
|
||||
dialog.setMinimumSize(400, 300)
|
||||
|
||||
# 다이얼로그 레이아웃
|
||||
dialog_layout = QVBoxLayout()
|
||||
|
||||
# 도움말 내용 추가
|
||||
help_text_browser = QTextBrowser()
|
||||
help_text_browser.setText("""
|
||||
<h3>Taobao Scraper 도움말</h3>
|
||||
<p>본 프로그램은 Taobao 데이터를 스크래핑하고 엑셀 파일로 출력하거나, DB를 후처리하는 기능을 제공합니다.</p>
|
||||
<ul>
|
||||
<li><b>시작:</b> 데이터를 스크래핑 시작</li>
|
||||
<li><b>엑셀 출력:</b> 데이터를 엑셀 파일로 저장</li>
|
||||
<li><b>후처리:</b> DB의 데이터를 후처리</li>
|
||||
<li><b>설정 > 금지어 관리:</b> 금지어 목록을 관리</li>
|
||||
<li><b>설정 > 금지 카테고리 관리:</b> 금지된 카테고리를 설정</li>
|
||||
</ul>
|
||||
<p>추가적인 문의 사항이 있으면 개발자에게 문의하세요.</p>
|
||||
""")
|
||||
|
||||
# 다이얼로그에 텍스트 브라우저 추가
|
||||
dialog_layout.addWidget(help_text_browser)
|
||||
|
||||
# 다이얼로그 레이아웃 설정
|
||||
dialog.setLayout(dialog_layout)
|
||||
|
||||
# 다이얼로그 실행
|
||||
dialog.exec()
|
||||
|
||||
@Slot()
|
||||
def show_user_info_dialog(self):
|
||||
"""
|
||||
사용자 정보 다이얼로그를 표시합니다.
|
||||
"""
|
||||
self.logger.log("사용자 정보 메뉴 클릭됨", level=logging.INFO)
|
||||
dialog = UserInfoDialog(self)
|
||||
dialog.exec()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,874 @@
|
|||
from PySide6.QtWidgets import (
|
||||
QWidget, QDialog, QVBoxLayout, QHBoxLayout, QLineEdit, QPushButton, QListWidget, QMessageBox, QListWidgetItem, QLabel, QScrollArea, QTabWidget
|
||||
)
|
||||
from PySide6.QtGui import QPixmap, QKeyEvent, QColor
|
||||
from PySide6.QtCore import Qt, QEvent
|
||||
|
||||
import os
|
||||
import logging
|
||||
import re
|
||||
import requests
|
||||
from src.keyword.kwDBManager import KeywordDBManager
|
||||
from src.keyword.kiprisAPI import Kipris_API
|
||||
|
||||
|
||||
class KeywordManager(QDialog):
|
||||
def __init__(self, logger, config, parent=None):
|
||||
super().__init__(parent)
|
||||
self.logger = logger
|
||||
|
||||
self.config = config
|
||||
|
||||
self.kipris_api_key = self.config.get("Kipris_API", "API_KEY")
|
||||
self.kiprisapi = Kipris_API(logger=self.logger, apikey=self.kipris_api_key)
|
||||
|
||||
kw_db_path = os.path.join("ForbiddenKeyword.db")
|
||||
self.kw_db_manager = KeywordDBManager(kw_db_path, self.logger)
|
||||
|
||||
self.setWindowTitle("금지어 관리")
|
||||
self.setGeometry(200, 200, 300, 400)
|
||||
|
||||
self._keyword_items = [] # 모든 키워드를 저장할 리스트
|
||||
self.loaded_items = {} # 화면에 표시된 항목만 저장
|
||||
self.batch_size = 50 # 한 번에 로드할 항목 수
|
||||
|
||||
# 정렬 상태 및 순서 초기화
|
||||
self.sort_ascending = True # 기본 오름차순
|
||||
self.current_sort_order = "default" # 기본 정렬 순서: default, korean, english, number
|
||||
|
||||
# 레이아웃 구성
|
||||
self.layout = QVBoxLayout()
|
||||
|
||||
# 설명 텍스트 추가
|
||||
self.description_label = QLabel(
|
||||
"1.금지: 상품 자체가 등록되지 않습니다.\n"
|
||||
"2.비허용: 상품은 등록되지만, 해당 키워드는 상품명이나 태그에서 제거됩니다.\n"
|
||||
"3.등급은 더블클릭, 좌,우키로 토글됩니다."
|
||||
)
|
||||
self.description_label.setWordWrap(True) # 텍스트 줄바꿈 활성화
|
||||
self.description_label.setStyleSheet("font-weight: bold; color: #555; margin-bottom: 12px;")
|
||||
|
||||
# 키워드 개수 라벨 추가
|
||||
self.count_label = QLabel("현재 키워드 수: 0")
|
||||
|
||||
# 금지어 목록 및 추가/삭제 위젯
|
||||
self.keyword_list = QListWidget()
|
||||
self.keyword_list.setFocusPolicy(Qt.StrongFocus) # 키보드 이벤트를 수신
|
||||
self.new_keyword_input = QLineEdit()
|
||||
self.new_keyword_input.setPlaceholderText("금지어를 입력하세요")
|
||||
self.add_button = QPushButton("추가")
|
||||
self.delete_button = QPushButton("삭제")
|
||||
self.sort_button = QPushButton("정렬")
|
||||
self.order_button = QPushButton("정렬 순서 변경")
|
||||
|
||||
"""필터링 버튼 설정"""
|
||||
self.filter_all_button = QPushButton("필터:All") # 기본 텍스트는 "All"
|
||||
self.filter_modes = ["필터:All", "필터:금지", "필터:비허용"] # 필터링 모드 리스트
|
||||
self.current_filter_index = 0 # 현재 필터링 모드의 인덱스
|
||||
|
||||
"""전체 로딩 버튼 설정"""
|
||||
self.load_all_button = QPushButton("전체 로딩")
|
||||
self.load_all_button.clicked.connect(self.load_all_keywords)
|
||||
|
||||
# 버튼 클릭 이벤트 연결
|
||||
self.filter_all_button.clicked.connect(self.toggle_filter_mode)
|
||||
|
||||
# 버튼 이벤트 연결
|
||||
self.add_button.clicked.connect(self.handle_keyword_input)
|
||||
self.delete_button.clicked.connect(self.delete_keyword)
|
||||
self.new_keyword_input.returnPressed.connect(self.handle_keyword_input)
|
||||
self.keyword_list.itemDoubleClicked.connect(self.change_grade) # 더블클릭으로 등급 변경
|
||||
self.new_keyword_input.textChanged.connect(self.highlight_keyword)
|
||||
self.sort_button.clicked.connect(self.sort_keywords)
|
||||
self.order_button.clicked.connect(self.change_sort_order)
|
||||
|
||||
# 레이아웃에 위젯 추가
|
||||
button_layout = QHBoxLayout()
|
||||
button_layout.addWidget(self.add_button)
|
||||
button_layout.addWidget(self.delete_button)
|
||||
button_layout.addWidget(self.sort_button)
|
||||
button_layout.addWidget(self.order_button)
|
||||
button_layout.addWidget(self.filter_all_button)
|
||||
button_layout.addWidget(self.load_all_button)
|
||||
|
||||
self.layout.addWidget(self.description_label) # 설명 추가
|
||||
self.layout.addWidget(self.count_label)
|
||||
self.layout.addWidget(self.keyword_list)
|
||||
self.layout.addWidget(self.new_keyword_input)
|
||||
self.layout.addLayout(button_layout)
|
||||
self.setLayout(self.layout)
|
||||
|
||||
# # 금지어 로드
|
||||
# self.load_keywords()
|
||||
|
||||
# # 스크롤 이벤트 감지
|
||||
# self.keyword_list.viewport().installEventFilter(self)
|
||||
|
||||
# 생성자에서 스크롤 이벤트 연결
|
||||
self.setup_scroll_event()
|
||||
|
||||
|
||||
def showEvent(self, event):
|
||||
"""다이얼로그가 표시될 때 입력 필드에 포커스 설정"""
|
||||
super().showEvent(event)
|
||||
|
||||
self.keyword_list.clear() # 기존 리스트 초기화
|
||||
self.loaded_items = {} # 로드된 항목 추적 초기화
|
||||
self.load_keywords() # 키워드 로드
|
||||
|
||||
self.new_keyword_input.setFocus() # 입력 필드에 포커스 설정
|
||||
|
||||
|
||||
def keyPressEvent(self, event):
|
||||
"""키보드 이벤트 처리"""
|
||||
if event.key() == Qt.Key_Delete:
|
||||
self.delete_keyword()
|
||||
# elif event.key() == Qt.Key_Return or event.key() == Qt.Key_Enter: # 엔터 키 처리
|
||||
elif event.key() == Qt.Key_Left or event.key() == Qt.Key_Right: # 엔터 키 처리
|
||||
self.change_grade_from_selection() # 스페이스 키로 등급 변경
|
||||
else:
|
||||
super().keyPressEvent(event)
|
||||
|
||||
def load_all_keywords(self):
|
||||
"""모든 키워드를 리스트에 로드"""
|
||||
try:
|
||||
# 모든 데이터를 한 번에 로드
|
||||
self.display_keywords(0, len(self._keyword_items))
|
||||
self.logger.log("전체 키워드 로딩 완료", level=logging.INFO)
|
||||
except Exception as e:
|
||||
self.logger.log(f"전체 키워드 로딩 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
|
||||
def load_keywords(self):
|
||||
"""금지어를 데이터베이스에서 로드"""
|
||||
try:
|
||||
# 데이터베이스에서 금지어를 가져와 딕셔너리에 저장
|
||||
keywords = self.kw_db_manager.get_all_keywords()
|
||||
self._keyword_items = {} # 기존 구조 유지
|
||||
for keyword, grade in keywords:
|
||||
self._keyword_items[keyword] = {
|
||||
"item": None, # 아직 리스트 위젯에 추가되지 않음
|
||||
"grade": grade,
|
||||
"status": "미검증", # 초기 상태 설정
|
||||
"index": None # 인덱스는 아직 없음
|
||||
}
|
||||
|
||||
self.logger.log("금지어 목록 로드 완료", level=logging.DEBUG)
|
||||
|
||||
# 처음 몇 개 항목만 표시
|
||||
self.loaded_items = {} # 로드된 항목 추적 딕셔너리
|
||||
self.display_keywords(0, self.batch_size)
|
||||
except Exception as e:
|
||||
self.logger.log(f"금지어 로드 중 오류 발생: {e}", level=logging.ERROR)
|
||||
|
||||
def display_keywords(self, start, end):
|
||||
"""지정된 범위의 키워드를 리스트에 추가"""
|
||||
try:
|
||||
# 중복 방지를 위해 기존에 로드되지 않은 키워드만 추가
|
||||
keyword_keys = list(self._keyword_items.keys()) # 딕셔너리 키 목록
|
||||
|
||||
for i in range(start, min(end, len(keyword_keys))):
|
||||
keyword = keyword_keys[i]
|
||||
# 이미 리스트에 로드된 아이템은 건너뜀
|
||||
if self._keyword_items[keyword]["item"] is not None:
|
||||
continue # 이미 로드된 항목은 추가하지 않음
|
||||
|
||||
# 새로운 아이템 추가
|
||||
keyword_data = self._keyword_items[keyword]
|
||||
item_widget = self.create_item_widget(keyword, keyword_data["grade"])
|
||||
if item_widget is not None:
|
||||
self._keyword_items[keyword]["item"] = item_widget # 로드된 항목 추적
|
||||
|
||||
self.logger.log(f"키워드 표시 완료: {start}부터 {end}까지", level=logging.DEBUG)
|
||||
|
||||
except Exception as e:
|
||||
self.logger.log(f"키워드 추가 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
def create_item_widget(self, keyword, grade):
|
||||
"""키워드를 리스트에 추가하고 딕셔너리에 저장"""
|
||||
try:
|
||||
self.logger.log(f"self.keyword_list: {self.keyword_list}", level=logging.DEBUG)
|
||||
|
||||
# 기존 상태 가져오기
|
||||
status = self.kw_db_manager.get_keyword_status(keyword) or "미검증"
|
||||
|
||||
# 리스트 아이템 위젯 생성
|
||||
item_widget = QWidget()
|
||||
layout = QHBoxLayout()
|
||||
|
||||
# 키워드와 등급 표시 라벨
|
||||
keyword_label = QLabel(f"{keyword} ({grade}) [{status}]")
|
||||
layout.addWidget(keyword_label)
|
||||
|
||||
# 검증 버튼 추가
|
||||
verify_button = QPushButton("검증" if status != "등록" else "내용보기")
|
||||
verify_button.setToolTip("키워드를 키프리스로 검증" if status != "등록" else "Kipris 세부 정보 보기")
|
||||
verify_button.setFixedWidth(60)
|
||||
layout.addWidget(verify_button)
|
||||
|
||||
# 버튼 이벤트 연결
|
||||
if status == "등록":
|
||||
# 내용보기 버튼으로 연결
|
||||
keyword_id = self.kw_db_manager.get_keyword_id(keyword)
|
||||
verify_button.clicked.connect(lambda: self.show_kipris_details(keyword, keyword_id))
|
||||
else:
|
||||
# 검증 버튼으로 연결
|
||||
verify_button.clicked.connect(lambda: self.verify_keyword(
|
||||
keyword=keyword,
|
||||
grade=grade,
|
||||
keyword_label=keyword_label,
|
||||
verify_button=verify_button
|
||||
))
|
||||
|
||||
# 레이아웃 설정
|
||||
layout.setContentsMargins(0, 0, 0, 0)
|
||||
item_widget.setLayout(layout)
|
||||
|
||||
# 리스트 아이템 생성
|
||||
list_item = QListWidgetItem(self.keyword_list)
|
||||
list_item.setSizeHint(item_widget.sizeHint())
|
||||
|
||||
# 데이터 설정
|
||||
list_item.setData(1, {"keyword": keyword, "grade": grade})
|
||||
|
||||
# 색상 업데이트
|
||||
self.set_item_colors(list_item, grade)
|
||||
|
||||
# 리스트에 위젯 및 아이템 추가
|
||||
self.keyword_list.setItemWidget(list_item, item_widget)
|
||||
|
||||
# 현재 인덱스를 저장
|
||||
current_index = self.keyword_list.row(list_item)
|
||||
|
||||
# 딕셔너리에 저장 (기존 구조 유지)
|
||||
self._keyword_items[keyword] = {
|
||||
"item": list_item,
|
||||
"grade": grade,
|
||||
"status": status,
|
||||
"index": current_index # 인덱스 저장
|
||||
}
|
||||
|
||||
# 키워드 개수 갱신
|
||||
self.update_keyword_count()
|
||||
|
||||
return item_widget
|
||||
except Exception as e:
|
||||
self.logger.log(f"키워드 추가 오류: {e}", level=logging.ERROR, exc_info=True)
|
||||
return None
|
||||
|
||||
def setup_scroll_event(self):
|
||||
"""스크롤바 시그널 연결"""
|
||||
scroll_bar = self.keyword_list.verticalScrollBar()
|
||||
scroll_bar.valueChanged.connect(self.on_scroll)
|
||||
|
||||
def on_scroll(self):
|
||||
"""스크롤바가 움직일 때 호출"""
|
||||
scroll_bar = self.keyword_list.verticalScrollBar()
|
||||
if scroll_bar.value() == scroll_bar.maximum(): # 스크롤이 끝까지 내려갔을 때
|
||||
current_count = len([item for item in self._keyword_items.values() if item["item"] is not None])
|
||||
self.display_keywords(current_count, current_count + self.batch_size)
|
||||
|
||||
|
||||
def toggle_filter_mode(self):
|
||||
"""필터 모드 토글"""
|
||||
# 현재 필터 모드 인덱스를 업데이트
|
||||
self.current_filter_index = (self.current_filter_index + 1) % len(self.filter_modes)
|
||||
current_mode = self.filter_modes[self.current_filter_index]
|
||||
|
||||
# 버튼 텍스트 업데이트
|
||||
self.filter_all_button.setText(current_mode)
|
||||
|
||||
# 필터링 적용
|
||||
if current_mode == "필터:All":
|
||||
self.filter_keywords("필터:All")
|
||||
elif current_mode == "필터:금지":
|
||||
self.filter_keywords("필터:금지")
|
||||
elif current_mode == "필터:비허용":
|
||||
self.filter_keywords("필터:비허용")
|
||||
|
||||
# 로그 기록
|
||||
self.logger.log(f"필터 모드 변경: {current_mode}", level=logging.INFO)
|
||||
|
||||
def filter_keywords(self, filter_mode):
|
||||
"""현재 로딩된 리스트 항목에 대해서만 필터링 모드 적용"""
|
||||
try:
|
||||
# 현재 로딩된 리스트 아이템 탐색
|
||||
for i in range(self.keyword_list.count()):
|
||||
list_item = self.keyword_list.item(i)
|
||||
item_data = list_item.data(1) # 키워드와 등급 정보 포함
|
||||
|
||||
# item_data가 None이면 건너뜀
|
||||
if not item_data:
|
||||
self.logger.log("필터링 중 NoneType 데이터 발견. 건너뜁니다.", level=logging.WARNING)
|
||||
continue
|
||||
|
||||
keyword_grade = item_data.get("grade", None)
|
||||
|
||||
# 필터링 조건에 따라 숨기거나 표시
|
||||
if filter_mode == "필터:금지" and keyword_grade != "금지":
|
||||
list_item.setHidden(True)
|
||||
elif filter_mode == "필터:비허용" and keyword_grade != "비허용":
|
||||
list_item.setHidden(True)
|
||||
else:
|
||||
list_item.setHidden(False)
|
||||
|
||||
self.logger.log(f"현재 로딩된 항목에 대해 필터링 완료: {filter_mode}", level=logging.INFO)
|
||||
except Exception as e:
|
||||
self.logger.log(f"필터링 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
def keyword_set_to_list(self, keyword, grade="금지"):
|
||||
"""키워드를 리스트에 추가하고 딕셔너리에 저장"""
|
||||
try:
|
||||
self.logger.log(f"self.keyword_list: {self.keyword_list}", level=logging.DEBUG)
|
||||
|
||||
# 기존 상태 가져오기
|
||||
status = self.kw_db_manager.get_keyword_status(keyword) or "미검증"
|
||||
|
||||
# 리스트 아이템 위젯 생성
|
||||
item_widget = QWidget()
|
||||
layout = QHBoxLayout()
|
||||
|
||||
# 키워드와 등급 표시 라벨
|
||||
keyword_label = QLabel(f"{keyword} ({grade}) [{status}]")
|
||||
layout.addWidget(keyword_label)
|
||||
|
||||
# 검증 버튼 추가
|
||||
verify_button = QPushButton("검증" if status != "등록" else "내용보기")
|
||||
verify_button.setToolTip("키워드를 키프리스로 검증" if status != "등록" else "Kipris 세부 정보 보기")
|
||||
verify_button.setFixedWidth(60)
|
||||
layout.addWidget(verify_button)
|
||||
|
||||
# 버튼 이벤트 연결
|
||||
if status == "등록":
|
||||
# 내용보기 버튼으로 연결
|
||||
keyword_id = self.kw_db_manager.get_keyword_id(keyword)
|
||||
verify_button.clicked.connect(lambda: self.show_kipris_details(keyword, keyword_id))
|
||||
else:
|
||||
# 검증 버튼으로 연결
|
||||
verify_button.clicked.connect(lambda: self.verify_keyword(
|
||||
keyword=keyword,
|
||||
grade=grade,
|
||||
keyword_label=keyword_label,
|
||||
verify_button=verify_button
|
||||
))
|
||||
|
||||
# 레이아웃 설정
|
||||
layout.setContentsMargins(0, 0, 0, 0)
|
||||
item_widget.setLayout(layout)
|
||||
|
||||
# 리스트 아이템 생성
|
||||
list_item = QListWidgetItem(self.keyword_list)
|
||||
list_item.setSizeHint(item_widget.sizeHint())
|
||||
|
||||
# 데이터 설정
|
||||
list_item.setData(1, {"keyword": keyword, "grade": grade})
|
||||
|
||||
# 색상 업데이트
|
||||
self.set_item_colors(list_item, grade)
|
||||
|
||||
# 리스트에 위젯 및 아이템 추가
|
||||
self.keyword_list.setItemWidget(list_item, item_widget)
|
||||
|
||||
# 현재 인덱스를 저장
|
||||
current_index = self.keyword_list.row(list_item)
|
||||
|
||||
# 딕셔너리에 저장 (인덱스 포함)
|
||||
self._keyword_items[keyword] = {
|
||||
"item": list_item,
|
||||
"grade": grade,
|
||||
"status": status,
|
||||
"index": current_index # 인덱스 저장
|
||||
}
|
||||
|
||||
# 키워드 개수 갱신
|
||||
self.update_keyword_count()
|
||||
except Exception as e:
|
||||
self.logger.log(f"키워드 추가 오류: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
def set_item_colors(self, item, grade):
|
||||
"""
|
||||
리스트 아이템의 색상을 등급에 따라 설정합니다.
|
||||
:param item: QListWidgetItem 객체
|
||||
:param grade: 키워드 등급 (금지/비허용/기타)
|
||||
"""
|
||||
if grade == "비허용":
|
||||
item.setBackground(Qt.yellow) # 배경색 노란색
|
||||
item.setForeground(Qt.black) # 텍스트 색상 검정색
|
||||
elif grade == "금지":
|
||||
# item.setBackground(Qt.red) # 배경색 빨간색
|
||||
item.setBackground(QColor("#FFA500")) # 오렌지색 배경 (HTML 코드)
|
||||
item.setForeground(Qt.white) # 텍스트 색상 흰색
|
||||
else:
|
||||
item.setBackground(Qt.white) # 기본 배경색
|
||||
item.setForeground(Qt.black) # 기본 텍스트 색상
|
||||
|
||||
def add_keyword_to_list(self, keyword: str, grade: str = "금지"):
|
||||
"""키워드를 리스트와 데이터베이스에 추가"""
|
||||
try:
|
||||
# 중복 검사
|
||||
if keyword in self._keyword_items:
|
||||
QMessageBox.warning(self, "추가 실패", "이미 존재하는 금지어입니다!")
|
||||
self.logger.log(f"중복 금지어 추가 시도: {keyword}", level=logging.WARNING)
|
||||
return
|
||||
|
||||
# 데이터베이스에 추가
|
||||
self.kw_db_manager.add_keyword(keyword, grade)
|
||||
|
||||
# _keyword_items에 추가
|
||||
self._keyword_items[keyword] = {
|
||||
"item": None, # 아직 로드되지 않음
|
||||
"grade": grade,
|
||||
"status": "미검증",
|
||||
"index": None
|
||||
}
|
||||
|
||||
# 현재 로드된 상태에서만 리스트에 추가
|
||||
if len(self.loaded_items) < self.batch_size:
|
||||
self.display_keywords(len(self.loaded_items), len(self.loaded_items) + 1)
|
||||
|
||||
# 키워드 개수 갱신
|
||||
self.update_keyword_count()
|
||||
|
||||
self.logger.log(f"키워드 추가: {keyword} ({grade})", level=logging.INFO)
|
||||
except Exception as e:
|
||||
self.logger.log(f"키워드 추가 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
def update_keyword_count(self):
|
||||
"""전체 키워드 수와 현재 로딩된 아이템 수를 표시"""
|
||||
try:
|
||||
# 데이터베이스에서 현재 금지어, 비허용 키워드 개수 가져오기
|
||||
total_count = len(self._keyword_items) # 전체 키워드 수
|
||||
ban_count = len([item for item in self._keyword_items.values() if item.get("grade") == "금지"])
|
||||
restricted_count = len([item for item in self._keyword_items.values() if item.get("grade") == "비허용"])
|
||||
|
||||
# 현재 로딩된 키워드 수 계산
|
||||
loaded_count = self.keyword_list.count()
|
||||
|
||||
# 라벨 업데이트
|
||||
self.count_label.setText(
|
||||
f"전체 키워드 수: {total_count} (금지: {ban_count}, 비허용: {restricted_count}), 현재 로딩된 키워드 수: {loaded_count}"
|
||||
)
|
||||
|
||||
# 로그 출력
|
||||
self.logger.log(
|
||||
f"전체 키워드 개수 업데이트: 총 {total_count}개 (금지: {ban_count}, 비허용: {restricted_count}), 현재 로딩된: {loaded_count}개",
|
||||
level=logging.DEBUG,
|
||||
)
|
||||
except Exception as e:
|
||||
self.logger.log(f"키워드 개수 갱신 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
def handle_keyword_input(self):
|
||||
"""엔터 입력 시 키워드 추가"""
|
||||
new_keyword = self.new_keyword_input.text().strip() # 입력된 키워드 가져오기
|
||||
if not new_keyword:
|
||||
return
|
||||
|
||||
try:
|
||||
# 기존 키워드 확인
|
||||
existing_keywords = list(self._keyword_items.keys())
|
||||
|
||||
# 콤마, 엔터, 공백으로 구분된 키워드 처리
|
||||
new_keywords = [kw.strip() for kw in re.split(r'[,\n\s]+', new_keyword) if kw.strip()]
|
||||
duplicate_words = []
|
||||
|
||||
for keyword in new_keywords:
|
||||
if keyword in existing_keywords:
|
||||
duplicate_words.append(keyword) # 중복된 키워드 기록
|
||||
else:
|
||||
# 새 키워드를 리스트와 DB에 추가
|
||||
self.add_keyword_to_list(keyword, "금지")
|
||||
# self.kw_db_manager.add_keyword(keyword, "금지") # DB에 저장
|
||||
self.logger.log(f"키워드 추가: {keyword} (금지)", level=logging.INFO)
|
||||
|
||||
# 중복 키워드 경고 메시지
|
||||
if duplicate_words:
|
||||
duplicates_str = ", ".join(set(duplicate_words))
|
||||
QMessageBox.warning(self, "중복 키워드", f"다음 키워드는 이미 존재합니다: {duplicates_str}")
|
||||
self.logger.log(f"중복 키워드 발견: {duplicates_str}", level=logging.WARNING)
|
||||
|
||||
# 입력 필드 초기화
|
||||
self.new_keyword_input.clear()
|
||||
self.new_keyword_input.setFocus()
|
||||
|
||||
# 키워드 개수 갱신
|
||||
self.update_keyword_count()
|
||||
|
||||
except Exception as e:
|
||||
self.logger.log(f"키워드 입력 처리 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
def show_existing_keyword_dialog(self, keyword):
|
||||
"""기존 키워드가 있을 때 메시지 박스 표시"""
|
||||
msg = QMessageBox(self)
|
||||
msg.setWindowTitle("키워드 존재")
|
||||
msg.setText(f"키워드 '{keyword}'가 이미 존재합니다. 삭제할까요?")
|
||||
msg.setIcon(QMessageBox.Question)
|
||||
delete_button = msg.addButton("삭제", QMessageBox.AcceptRole)
|
||||
cancel_button = msg.addButton("취소", QMessageBox.RejectRole)
|
||||
cancel_button.setDefault(True)
|
||||
|
||||
msg.exec()
|
||||
|
||||
if msg.clickedButton() == delete_button:
|
||||
self.delete_keyword()
|
||||
|
||||
def show_delete_confirmation_dialog(self, keyword):
|
||||
"""키워드 삭제 전에 확인 메시지를 표시"""
|
||||
msg = QMessageBox(self)
|
||||
msg.setWindowTitle("키워드 삭제 확인")
|
||||
msg.setText(f"키워드 '{keyword}'를 삭제하시겠습니까?")
|
||||
msg.setIcon(QMessageBox.Warning)
|
||||
delete_button = msg.addButton("삭제", QMessageBox.AcceptRole)
|
||||
cancel_button = msg.addButton("취소", QMessageBox.RejectRole)
|
||||
cancel_button.setDefault(True)
|
||||
|
||||
msg.exec()
|
||||
|
||||
return msg.clickedButton() == delete_button
|
||||
|
||||
def delete_keyword(self):
|
||||
"""선택된 키워드 삭제"""
|
||||
selected_items = self.keyword_list.selectedItems()
|
||||
if selected_items:
|
||||
for item in selected_items:
|
||||
keyword = item.data(1)["keyword"]
|
||||
# 삭제 확인 다이얼로그 표시
|
||||
if self.show_delete_confirmation_dialog(keyword):
|
||||
# 데이터베이스와 _keyword_items에서 삭제
|
||||
self.kw_db_manager.delete_keyword(keyword)
|
||||
del self._keyword_items[keyword]
|
||||
|
||||
# 리스트에서 제거
|
||||
self.keyword_list.takeItem(self.keyword_list.row(item))
|
||||
|
||||
# 키워드 개수 갱신
|
||||
self.update_keyword_count()
|
||||
self.logger.log(f"키워드 삭제: {keyword}", level=logging.INFO)
|
||||
else:
|
||||
self.logger.log(f"키워드 삭제 취소: {keyword}", level=logging.INFO)
|
||||
else:
|
||||
QMessageBox.warning(self, "삭제 실패", "삭제할 키워드를 선택해주세요!")
|
||||
|
||||
def highlight_keyword(self, text):
|
||||
"""입력한 텍스트와 일치하는 키워드로 이동"""
|
||||
|
||||
self.logger.log(f"self.keyword_list: {self.keyword_list}", level=logging.DEBUG)
|
||||
|
||||
for i in range(self.keyword_list.count()):
|
||||
item_data = self.keyword_list.item(i).data(1)
|
||||
self.logger.log(f"item_data: {item_data}", level=logging.DEBUG)
|
||||
if text in item_data["keyword"]:
|
||||
self.keyword_list.setCurrentRow(i)
|
||||
break
|
||||
|
||||
def change_sort_order(self):
|
||||
"""정렬 순서 변경"""
|
||||
sort_orders = ["default", "korean", "english", "number"]
|
||||
current_index = sort_orders.index(self.current_sort_order)
|
||||
self.current_sort_order = sort_orders[(current_index + 1) % len(sort_orders)]
|
||||
self.logger.log(f"정렬 순서 변경: {self.current_sort_order}", level=logging.INFO)
|
||||
|
||||
def sort_keywords(self):
|
||||
"""금지어 목록 정렬 (인덱스를 조정하여 순서를 변경)"""
|
||||
try:
|
||||
# 현재 리스트의 키워드 데이터를 가져옴
|
||||
keywords = list(self._keyword_items.values())
|
||||
|
||||
# 정렬 기준 설정
|
||||
def sort_criteria(item):
|
||||
keyword = item["item"].data(1)["keyword"]
|
||||
if self.current_sort_order == "default":
|
||||
return keyword
|
||||
elif self.current_sort_order == "korean":
|
||||
return (not self.is_korean(keyword), keyword)
|
||||
elif self.current_sort_order == "english":
|
||||
return (not keyword.isalpha(), keyword)
|
||||
elif self.current_sort_order == "number":
|
||||
return (not keyword.isdigit(), keyword)
|
||||
return keyword # 기본값
|
||||
|
||||
# 정렬된 키워드 리스트
|
||||
sorted_keywords = sorted(keywords, key=sort_criteria)
|
||||
|
||||
# 리스트의 순서를 재배치
|
||||
for new_index, data in enumerate(sorted_keywords):
|
||||
old_index = data["index"]
|
||||
item = data["item"]
|
||||
|
||||
# 기존 위치에서 제거하고 새 위치에 삽입
|
||||
self.keyword_list.takeItem(old_index)
|
||||
self.keyword_list.insertItem(new_index, item)
|
||||
|
||||
# 인덱스 업데이트
|
||||
self._keyword_items[data["item"].data(1)["keyword"]]["index"] = new_index
|
||||
|
||||
self.logger.log(f"키워드 정렬 완료: {self.current_sort_order}", level=logging.INFO)
|
||||
|
||||
except Exception as e:
|
||||
self.logger.log(f"정렬 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
def is_korean(self, text):
|
||||
"""문자열이 한글인지 확인"""
|
||||
return all("가" <= char <= "힣" for char in text)
|
||||
|
||||
def change_grade_from_selection(self):
|
||||
"""선택된 키워드의 등급 변경"""
|
||||
selected_items = self.keyword_list.selectedItems()
|
||||
if not selected_items:
|
||||
return
|
||||
self.change_grade(selected_items[0])
|
||||
|
||||
def change_grade(self, item):
|
||||
"""선택된 키워드의 등급 변경"""
|
||||
try:
|
||||
# 리스트 아이템의 데이터를 가져옴
|
||||
item_data = item.data(1)
|
||||
if not item_data:
|
||||
return
|
||||
|
||||
# 등급 변경
|
||||
current_grade = item_data["grade"]
|
||||
new_grade = "비허용" if current_grade == "금지" else "금지"
|
||||
item_data["grade"] = new_grade
|
||||
|
||||
# _keyword_items에 등급 업데이트
|
||||
keyword = item_data["keyword"]
|
||||
self._keyword_items[keyword]["grade"] = new_grade
|
||||
|
||||
# 데이터베이스 업데이트
|
||||
self.kw_db_manager.update_keyword_grade(keyword, new_grade)
|
||||
|
||||
# UI 업데이트: 현재 아이템만 다시 그리기
|
||||
item_widget = self.keyword_list.itemWidget(item)
|
||||
if item_widget:
|
||||
keyword_label = item_widget.findChild(QLabel)
|
||||
if keyword_label:
|
||||
# 라벨 업데이트
|
||||
status = self._keyword_items[keyword].get("status", "미검증")
|
||||
keyword_label.setText(f"{keyword} ({new_grade}) [{status}]")
|
||||
|
||||
# 색상 업데이트
|
||||
self.set_item_colors(item, new_grade)
|
||||
item.setData(1, item_data)
|
||||
|
||||
# 키워드 개수 갱신
|
||||
self.update_keyword_count()
|
||||
|
||||
self.logger.log(f"키워드 등급 변경: {keyword} ({current_grade} -> {new_grade})", level=logging.INFO)
|
||||
except Exception as e:
|
||||
self.logger.log(f"키워드 등급 변경 중 오류: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
def get_ban_list(self):
|
||||
"""
|
||||
데이터베이스에서 금지 등급 키워드만 반환합니다.
|
||||
"""
|
||||
try:
|
||||
keywords = self.kw_db_manager.get_all_keywords() # 모든 키워드 가져오기
|
||||
ban_list = [keyword for keyword, grade in keywords if grade == "금지"]
|
||||
return ban_list
|
||||
except Exception as e:
|
||||
self.logger.log(f"금지 키워드 가져오기 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
return []
|
||||
|
||||
def get_restricted_list(self):
|
||||
"""
|
||||
데이터베이스에서 비허용 등급 키워드만 반환합니다.
|
||||
"""
|
||||
try:
|
||||
keywords = self.kw_db_manager.get_all_keywords() # 모든 키워드 가져오기
|
||||
restricted_list = [keyword for keyword, grade in keywords if grade == "비허용"]
|
||||
return restricted_list
|
||||
except Exception as e:
|
||||
self.logger.log(f"비허용 키워드 가져오기 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
return []
|
||||
|
||||
def verify_keyword(self, keyword, grade, keyword_label, verify_button):
|
||||
"""Kipris API를 사용해 키워드를 검증하고 DB에 결과를 저장"""
|
||||
try:
|
||||
self.logger.log(f"키워드 검증 시작: {keyword}", level=logging.INFO)
|
||||
result_list = self.kipris_api.search_trademark(keyword)
|
||||
|
||||
if result_list:
|
||||
# 키워드 ID 조회 또는 생성
|
||||
keyword_id = self.kw_db_manager.get_keyword_id(keyword)
|
||||
if not keyword_id:
|
||||
self.logger.log(f"키워드 ID 조회 실패: {keyword}", level=logging.ERROR)
|
||||
return
|
||||
|
||||
# 모든 결과 저장
|
||||
for result in result_list:
|
||||
self.kw_db_manager.add_kipris_result(keyword_id, result)
|
||||
|
||||
# 상태 업데이트
|
||||
status = "등록"
|
||||
self.kw_db_manager.update_status(keyword, status)
|
||||
|
||||
# 라벨 및 버튼 업데이트
|
||||
updated_text = f"{keyword} ({grade}) [{status}]"
|
||||
keyword_label.setText(updated_text)
|
||||
verify_button.setText("내용보기")
|
||||
verify_button.clicked.disconnect()
|
||||
verify_button.clicked.connect(lambda: self.show_kipris_details(keyword, keyword_id))
|
||||
self.show_kipris_details(keyword, keyword_id)
|
||||
else:
|
||||
status = "미등록"
|
||||
self.kw_db_manager.update_status(keyword, status)
|
||||
updated_text = f"{keyword} ({grade}) [{status}]"
|
||||
keyword_label.setText(updated_text)
|
||||
|
||||
self.logger.log(f"키워드 '{keyword}' 검증 완료: {status}", level=logging.INFO)
|
||||
except Exception as e:
|
||||
self.logger.log(f"키워드 검증 중 오류: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
|
||||
def show_kipris_details(self, keyword, keyword_id):
|
||||
"""키워드의 Kipris 결과를 다이얼로그로 표시"""
|
||||
try:
|
||||
# DB에서 키워드와 Kipris 결과 가져오기
|
||||
results = self.kw_db_manager.get_kipris_results(keyword_id)
|
||||
|
||||
if not results:
|
||||
QMessageBox.information(self, "Kipris 결과", "해당 키워드의 Kipris 결과가 없습니다.")
|
||||
return
|
||||
|
||||
# 다이알로그 생성
|
||||
dialog = QDialog(self)
|
||||
dialog.setWindowTitle("Kipris 세부 정보")
|
||||
dialog.setMinimumSize(600, 400)
|
||||
dialog.setMaximumSize(800, 600)
|
||||
|
||||
layout = QVBoxLayout(dialog)
|
||||
|
||||
# 키워드 이름 표시 (굵고 큰 글씨)
|
||||
keyword_label = QLabel(keyword)
|
||||
keyword_label.setAlignment(Qt.AlignCenter)
|
||||
keyword_label.setStyleSheet("font-size: 18px; font-weight: bold;")
|
||||
layout.addWidget(keyword_label)
|
||||
|
||||
# 탭 위젯 생성
|
||||
tab_widget = QTabWidget()
|
||||
layout.addWidget(tab_widget)
|
||||
|
||||
scroll_areas = [] # 스크롤 영역 참조를 저장
|
||||
|
||||
for idx, result in enumerate(results):
|
||||
# 각 결과에 대한 탭 생성
|
||||
tab = QWidget()
|
||||
|
||||
# 탭에 스크롤 영역 적용
|
||||
scroll_area = QScrollArea()
|
||||
scroll_area.setWidgetResizable(True)
|
||||
|
||||
# 스크롤 내용 위젯 생성
|
||||
content_widget = QWidget()
|
||||
content_layout = QVBoxLayout(content_widget)
|
||||
content_layout.setAlignment(Qt.AlignTop)
|
||||
|
||||
# drawing 이미지
|
||||
if result.get('drawing'):
|
||||
drawing_label = QLabel("Drawing:")
|
||||
drawing_image = QLabel()
|
||||
pixmap = QPixmap()
|
||||
pixmap.loadFromData(requests.get(result['drawing']).content)
|
||||
drawing_image.setPixmap(pixmap.scaled(200, 200, Qt.KeepAspectRatio))
|
||||
content_layout.addWidget(drawing_label)
|
||||
content_layout.addWidget(drawing_image)
|
||||
|
||||
# 텍스트 정보 생성
|
||||
result_text = (
|
||||
f"상태: {result['application_status']}\n"
|
||||
f"등록일: {result['registration_date']}\n"
|
||||
f"출원인: {result['applicant_name']}\n"
|
||||
f"분류 코드: {result['classification_code']}\n"
|
||||
"카테고리 설명:\n"
|
||||
)
|
||||
|
||||
# 카테고리 설명 정렬
|
||||
category_description_lines = []
|
||||
for line in result['category_description'].split(';'):
|
||||
line = line.strip()
|
||||
if line.startswith('['):
|
||||
category_description_lines.append(f"\n{line}")
|
||||
else:
|
||||
category_description_lines.append(f" {line}")
|
||||
formatted_category_description = "\n".join(category_description_lines)
|
||||
result_text += formatted_category_description
|
||||
|
||||
# 텍스트 라벨 추가
|
||||
label = QLabel(result_text)
|
||||
label.setWordWrap(True)
|
||||
content_layout.addWidget(label)
|
||||
|
||||
# bigDrawing 이미지
|
||||
if result.get('bigDrawing'):
|
||||
big_drawing_label = QLabel("Big Drawing:")
|
||||
big_drawing_image = QLabel()
|
||||
pixmap = QPixmap()
|
||||
pixmap.loadFromData(requests.get(result['bigDrawing']).content)
|
||||
big_drawing_image.setPixmap(pixmap.scaled(300, 300, Qt.KeepAspectRatio))
|
||||
content_layout.addWidget(big_drawing_label)
|
||||
content_layout.addWidget(big_drawing_image)
|
||||
|
||||
# 레이아웃 설정
|
||||
content_widget.setLayout(content_layout)
|
||||
scroll_area.setWidget(content_widget)
|
||||
|
||||
# 스크롤 영역을 탭에 추가
|
||||
tab_layout = QVBoxLayout(tab)
|
||||
tab_layout.addWidget(scroll_area)
|
||||
tab.setLayout(tab_layout)
|
||||
|
||||
tab_widget.addTab(tab, f"결과 {idx + 1}")
|
||||
scroll_areas.append(scroll_area) # 스크롤 영역 저장
|
||||
|
||||
# 닫기 버튼 추가
|
||||
close_button = QPushButton("닫기")
|
||||
close_button.clicked.connect(dialog.close)
|
||||
layout.addWidget(close_button)
|
||||
|
||||
dialog.setLayout(layout)
|
||||
|
||||
# 키보드 이벤트 처리
|
||||
def keyPressEvent(event):
|
||||
"""키보드 이벤트 처리"""
|
||||
current_index = tab_widget.currentIndex()
|
||||
total_tabs = tab_widget.count()
|
||||
|
||||
if event.key() in range(Qt.Key_1, Qt.Key_9 + 1):
|
||||
# 숫자 키를 누르면 해당 탭으로 이동
|
||||
tab_index = event.key() - Qt.Key_1
|
||||
if tab_index < total_tabs:
|
||||
tab_widget.setCurrentIndex(tab_index)
|
||||
elif event.key() == Qt.Key_Left:
|
||||
# 왼쪽 화살표 키로 이전 탭으로 이동
|
||||
tab_widget.setCurrentIndex((current_index - 1) % total_tabs)
|
||||
elif event.key() == Qt.Key_Right:
|
||||
# 오른쪽 화살표 키로 다음 탭으로 이동
|
||||
tab_widget.setCurrentIndex((current_index + 1) % total_tabs)
|
||||
elif event.key() in (Qt.Key_Up, Qt.Key_Down):
|
||||
# 현재 탭의 스크롤 영역을 탐색하고 스크롤
|
||||
current_scroll_area = scroll_areas[current_index]
|
||||
if current_scroll_area:
|
||||
scroll_bar = current_scroll_area.verticalScrollBar()
|
||||
scroll_bar.setValue(scroll_bar.value() - 20 if event.key() == Qt.Key_Up else scroll_bar.value() + 20)
|
||||
elif event.key() == Qt.Key_PageUp:
|
||||
# 현재 탭의 스크롤 영역을 탐색하고 위로 한 페이지 스크롤
|
||||
current_scroll_area = scroll_areas[current_index]
|
||||
if current_scroll_area:
|
||||
scroll_bar = current_scroll_area.verticalScrollBar()
|
||||
scroll_bar.setValue(scroll_bar.value() - scroll_bar.pageStep())
|
||||
elif event.key() == Qt.Key_PageDown:
|
||||
# 현재 탭의 스크롤 영역을 탐색하고 아래로 한 페이지 스크롤
|
||||
current_scroll_area = scroll_areas[current_index]
|
||||
if current_scroll_area:
|
||||
scroll_bar = current_scroll_area.verticalScrollBar()
|
||||
scroll_bar.setValue(scroll_bar.value() + scroll_bar.pageStep())
|
||||
elif event.key() == Qt.Key_Escape:
|
||||
dialog.close()
|
||||
else:
|
||||
super(QDialog, dialog).keyPressEvent(event)
|
||||
|
||||
dialog.keyPressEvent = keyPressEvent
|
||||
dialog.exec_()
|
||||
|
||||
except Exception as e:
|
||||
self.logger.log(f"Kipris 세부 정보 표시 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
|
@ -0,0 +1,178 @@
|
|||
import xml.etree.ElementTree as ET
|
||||
import requests, json
|
||||
import os, sys
|
||||
import logging
|
||||
|
||||
class Kipris_API:
|
||||
def __init__(self, logger, apikey=None):
|
||||
self.logger = logger
|
||||
self.url = 'http://kipo-api.kipi.or.kr/openapi/service/trademarkInfoSearchService/getWordSearch'
|
||||
self.apikey = apikey
|
||||
self.results = []
|
||||
|
||||
self.base_dir = self.get_base_dir()
|
||||
self.kipris_cat_path = os.path.join(self.base_dir, 'kiprisCategories.json')
|
||||
|
||||
self.category_description = self.load_category_descriptions(self.kipris_cat_path)
|
||||
|
||||
|
||||
def get_base_dir(self):
|
||||
"""
|
||||
실행 환경에 따라 base_dir을 설정하는 메서드.
|
||||
cx_Freeze로 패키징된 경우 실행 파일의 경로, 일반 Python 환경일 경우 __file__을 기준으로 설정.
|
||||
"""
|
||||
if getattr(sys, 'frozen', False): # 패키징된 경우
|
||||
base_dir = os.path.dirname(sys.executable)
|
||||
internal_dir = os.path.join(base_dir, '_internal') # _internal 디렉토리 포함
|
||||
if os.path.exists(internal_dir): # _internal 디렉토리가 존재하면 base_dir로 설정
|
||||
return internal_dir
|
||||
|
||||
else: # 일반 Python 실행 환경
|
||||
base_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
return base_dir
|
||||
|
||||
def fetch_and_decode(self, params):
|
||||
# API 요청 및 응답 받기
|
||||
try:
|
||||
response = requests.get(self.url, params=params)
|
||||
decoded_data = response.content.decode('utf-8')
|
||||
return decoded_data
|
||||
except Exception as e:
|
||||
self.logger.log(f'키프리스 요청 중 에러발생 : {e}', level=logging.ERROR, exc_info=True)
|
||||
|
||||
def parse_xml(self, xml_data, status):
|
||||
"""XML 데이터 파싱 및 결과 저장"""
|
||||
try:
|
||||
root = ET.fromstring(xml_data)
|
||||
for i, item in enumerate(root.findall('.//body/items/item')):
|
||||
application_status = (
|
||||
item.find('applicationStatus').text
|
||||
if item.find('applicationStatus') is not None else None
|
||||
)
|
||||
|
||||
# 필요한 데이터 필드 추출
|
||||
if application_status in status:
|
||||
title = item.find('title').text if item.find('title') is not None else None
|
||||
registration_date = (
|
||||
item.find('registrationDate').text
|
||||
if item.find('registrationDate') is not None else None
|
||||
)
|
||||
applicant_name = (
|
||||
item.find('applicantName').text
|
||||
if item.find('applicantName') is not None else None
|
||||
)
|
||||
classification_code = (
|
||||
item.find('classificationCode').text
|
||||
if item.find('classificationCode') is not None else None
|
||||
)
|
||||
category_desc = (
|
||||
self.add_category_description(classification_code)
|
||||
if classification_code else None
|
||||
)
|
||||
drawing = (
|
||||
item.find('drawing').text
|
||||
if item.find('drawing') is not None else None
|
||||
)
|
||||
big_drawing = (
|
||||
item.find('bigDrawing').text
|
||||
if item.find('bigDrawing') is not None else None
|
||||
)
|
||||
|
||||
# 결과 리스트에 추가
|
||||
self.results.append({
|
||||
"title": title,
|
||||
"registration_date": registration_date,
|
||||
"applicant_name": applicant_name,
|
||||
"classification_code": classification_code,
|
||||
"category_description": category_desc,
|
||||
"application_status": application_status,
|
||||
"drawing": drawing,
|
||||
"bigDrawing": big_drawing,
|
||||
})
|
||||
|
||||
self.logger.log(f"총 {len(self.results)}개의 결과를 파싱했습니다.", level=logging.INFO)
|
||||
|
||||
except ET.ParseError as e:
|
||||
self.logger.log(f"XML 파싱 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
except Exception as e:
|
||||
self.logger.log(f"알 수 없는 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
# 결과 확인용 로그 출력
|
||||
self.logger.log(f"총 {len(self.results)}개의 결과가 '등록' 또는 '공개' 상태로 검색됨.", level=logging.ERROR)
|
||||
def get_results(self):
|
||||
"""결과 반환"""
|
||||
return self.results
|
||||
|
||||
|
||||
# if application_status in ["등록", "공개"]:
|
||||
# if application_status in status: # status는 self.set_status 리스트를 참조
|
||||
|
||||
# self.results[f"result_{i+1}"] = {
|
||||
# "index_no": item.find('indexNo').text if item.find('indexNo') is not None else None,
|
||||
# "application_number": item.find('applicationNumber').text if item.find('applicationNumber') is not None else None,
|
||||
# "application_date": item.find('applicationDate').text if item.find('applicationDate') is not None else None,
|
||||
# "publication_number": item.find('publicationNumber').text if item.find('publicationNumber') is not None else None,
|
||||
# "publication_date": item.find('publicationDate').text if item.find('publicationDate') is not None else None,
|
||||
# "registration_date": item.find('registrationDate').text if item.find('registrationDate') is not None else None,
|
||||
# "registration_number": item.find('registrationNumber').text if item.find('registrationNumber') is not None else None,
|
||||
# "applicant_name": item.find('applicantName').text if item.find('applicantName') is not None else None,
|
||||
# "agent_name": item.find('agentName').text if item.find('agentName') is not None else None,
|
||||
# "title": item.find('title').text if item.find('title') is not None else None,
|
||||
# "drawing_url": item.find('drawing').text if item.find('drawing') is not None else None,
|
||||
# "big_drawing_url": item.find('bigDrawing').text if item.find('bigDrawing') is not None else None,
|
||||
# "full_text": item.find('fullText').text if item.find('fullText') is not None else None,
|
||||
# "application_status": application_status,
|
||||
# "classification_code": item.find('classificationCode').text if item.find('classificationCode') is not None else None,
|
||||
# "category_description": category_desc
|
||||
# }
|
||||
|
||||
|
||||
def search_trademark(self, keyword, status=['등록', '공개']):
|
||||
"""키워드로 상표 검색 후 결과 반환"""
|
||||
|
||||
# 검색 시작 전에 self.results 초기화
|
||||
self.results = []
|
||||
|
||||
|
||||
params = {
|
||||
'serviceKey': self.apikey,
|
||||
'searchString': keyword,
|
||||
'searchRecentYear': '0',
|
||||
'title': '',
|
||||
'fullText': '',
|
||||
'drawing': '',
|
||||
'bigDrawing': ''
|
||||
}
|
||||
self.logger.log(f'Search params: {params}', level=logging.ERROR)
|
||||
|
||||
try:
|
||||
xml_data = self.fetch_and_decode(params)
|
||||
if xml_data:
|
||||
self.parse_xml(xml_data, status)
|
||||
else:
|
||||
self.logger.log(f'API 응답이 없습니다.', level=logging.ERROR)
|
||||
|
||||
return {}
|
||||
except Exception as e:
|
||||
self.logger.log(f'API 요청 중 에러 발생: {e}', level=logging.ERROR, exc_info=True)
|
||||
return {}
|
||||
|
||||
return self.get_results()
|
||||
|
||||
def close_Kipris(self):
|
||||
pass
|
||||
|
||||
def load_category_descriptions(self, filename):
|
||||
"""JSON 파일에서 카테고리 설명을 로드합니다."""
|
||||
with open(filename, 'r', encoding='utf-8') as file:
|
||||
return json.load(file)
|
||||
|
||||
def add_category_description(self, category_code):
|
||||
"""각 분류 코드를 설명과 함께 포맷합니다."""
|
||||
descriptions = []
|
||||
codes = category_code.split('|')
|
||||
for code in codes:
|
||||
description = self.category_description.get(code, "카테고리 설명을 찾을 수 없습니다.")
|
||||
descriptions.append(f"[{code}] - {description}")
|
||||
return "; ".join(descriptions)
|
||||
|
||||
|
|
@ -0,0 +1,47 @@
|
|||
{
|
||||
"01": "공업/과학 및 사진용 및 농업/원예 및 임업용 화학제; 미가공 인조수지, 미가공 플라스틱; 소화 및 화재예방용 조성물; 조질제 및 땜납용 조제; 수피용 무두질제; 공업용 접착제; 퍼티 및 기타 페이스트 충전제; 퇴비, 거름, 비료; 산업용 및 과학용 생물학적 제제",
|
||||
"02": "페인트, 니스, 래커; 방청제 및 목재 보존제; 착색제, 염료; 인쇄, 표시 및 판화용 잉크; 미가공 천연수지; 도장용/장식용/인쇄용/미술용 금속박(箔) 및 금속분(紛)",
|
||||
"03": "비의료용 화장품 및 세면용품; 비의료용 치약; 향료, 에센셜 오일; 표백제 및 기타 세탁용 제제; 세정/광택 및 연마재",
|
||||
"04": "공업용 오일 및 그리스, 왁스; 윤활제; 먼지흡수제, 먼지습윤제 및 먼지흡착제; 연료 및 발광체; 조명용 양초 및 심지",
|
||||
"05": "약제, 의료용 및 수의과용 제제; 의료용 위생제; 의료용 또는 수의과용 식이요법 식품 및 제제, 유아용 식품; 인체용 및 동물용 식이보충제; 플라스터, 외상치료용 재료; 치과용 충전재료, 치과용 왁스; 소독제; 해충구제제; 살균제, 제초제",
|
||||
"06": "일반금속 및 그 합금, 광석; 금속제 건축 및 구축용 재료; 금속제 이동식 건축물; 비전기용 일반금속제 케이블 및 와이어; 소형금속제품; 저장 또는 운반용 금속제 용기; 금고",
|
||||
"07": "기계, 공작기계, 전동공구; 모터 및 엔진(육상차량용은 제외); 기계 커플링 및 전동장치 부품(육상차량용은 제외); 농기구(수동식 수공구는 제외); 부란기(孵卵器); 자동판매기",
|
||||
"08": "수동식 수공구 및 수동기구; 커틀러리; 휴대 무기(화기는 제외); 면도기",
|
||||
"09": "과학, 연구, 항법, 측량, 사진, 영화, 시청각, 광학, 계량, 측정, 신호, 탐지, 시험, 검사, 구명 및 교육용 기기; 전기 분배 또는 전기 사용의 전도, 전환, 변형, 축적, 조절 또는 통제를 위한 기기; 음향/영상 또는 데이터의 기록/전송/재생 또는 처리용 장치 및 기구; 기록 및 내려받기 가능한 미디어, 컴퓨터 소프트웨어, 빈 디지털 또는 아날로그 기록 및 저장매체; 동전작동식 기계장치; 금전등록기, 계산기; 컴퓨터 및 컴퓨터주변기기; 잠수복, 잠수마스크, 잠수용 귀마개, 다이버 및 수영용 노즈클립, 잠수용 장갑, 잠수용 호흡장치; 소화기기",
|
||||
"10": "외과용, 내과용, 치과용 및 수의과용 기계기구; 의지(義肢), 의안(義眼) 및 의치(義齒); 정형외과용품; 봉합용 재료; 장애인용 치료 및 재활보조장치; 안마기; 유아수유용 기기 및 용품; 성활동용 기기 및 용품",
|
||||
"11": "조명용, 가열용, 냉각용, 증기발생용, 조리용, 건조용, 환기용, 급수용, 위생용 장치 및 설비",
|
||||
"12": "수송기계기구; 육상, 항공 또는 해상을 통해 이동하는 수송수단",
|
||||
"13": "화기(火器); 탄약 및 발사체; 폭약; 폭죽",
|
||||
"14": "귀금속 및 그 합금; 보석, 귀석 및 반귀석; 시계용구",
|
||||
"15": "악기; 악보대 및 악기용 받침대; 지휘봉",
|
||||
"16": "종이 및 판지; 인쇄물; 제본재료; 사진; 문방구 및 사무용품(가구는 제외); 문방구용 또는 가정용 접착제; 제도용구 및 미술용 재료; 회화용 솔; 교재; 포장용 플라스틱제 시트, 필름 및 가방; 인쇄활자, 프린팅블록",
|
||||
"17": "미가공 및 반가공 고무, 구타페르카, 고무액(gum), 석면, 운모(雲母) 및 이들의 제품; 제조용 압출성형형태의 플라스틱 및 수지; 충전용, 마개용 및 절연용 재료; 비금속제 신축관, 튜브 및 호스",
|
||||
"18": "가죽 및 모조가죽; 수피; 수하물가방 및 운반용 가방; 우산 및 파라솔; 걷기용 지팡이; 채찍 및 마구(馬具); 동물용 목걸이, 가죽끈 및 의류",
|
||||
"19": "건축용 및 구축용 비금속제 건축재료; 건축용 비금속제 경질관(硬質管); 아스팔트, 피치, 타르 및 역청; 비금속제 이동식 건축물; 비금속제 기념물",
|
||||
"20": "가구, 거울, 액자; 보관 또는 운송용 비금속제 컨테이너; 미가공 또는 반가공 뼈, 뿔, 고래수염 또는 나전(螺鈿); 패각; 해포석(海泡石); 호박(琥珀)(원석)",
|
||||
"21": "가정용 또는 주방용 기구 및 용기; 조리기구 및 식기(포크, 나이프 및 스푼은 제외); 빗 및 스펀지; 솔(페인트 솔은 제외); 솔 제조용 재료; 청소용구; 비건축용 미가공 또는 반가공 유리; 유리제품, 도자기제품 및 토기제품",
|
||||
"22": "로프 및 노끈; 망(網); 텐트 및 타폴린; 직물제 또는 합성재료제 차양; 돛; 하역물운반용 및 보관용 포대; 충전재료(고무/플라스틱/종이 및 판지제는 제외); 직물용 미가공 섬유 및 그 대용품",
|
||||
"23": "직물용 실(絲)",
|
||||
"24": "직물 및 직물대용품; 가정용 린넨; 직물 또는 플라스틱제 커튼",
|
||||
"25": "의류, 신발, 모자",
|
||||
"26": "레이스, 장식용 끈 및 자수포, 의류장식용 리본 및 나비매듭리본; 단추, 훅 및 아이(hooks and eyes), 핀 및 바늘; 조화(造花); 머리장식품; 가발",
|
||||
"27": "카펫, 융단, 매트, 리놀륨 및 기타 바닥깔개용 재료; 비직물제 벽걸이",
|
||||
"28": "오락용구, 장난감; 비디오게임장치; 체조 및 스포츠용품; 크리스마스트리용 장식품",
|
||||
"29": "식육, 생선, 가금 및 엽조수; 고기진액; 보존처리/냉동/건조 및 조리된 과일 및 채소; 젤리, 잼, 콤폿; 달걀; 우유, 치즈, 버터, 요구르트 및 기타 유제품; 식용 유지(油脂)",
|
||||
"30": "커피, 차(茶), 코코아 및 그 대용물; 쌀, 파스타 및 국수; 타피오카 및 사고(sago); 곡분 및 곡물 조제품; 빵, 페이스트리 및 과자; 초콜릿; 아이스크림, 셔벗 및 기타 식용 얼음; 설탕, 꿀, 당밀(糖蜜); 식품용 이스트, 베이킹 파우더; 소금, 조미료, 향신료, 보존처리된 허브; 식초, 소스 및 기타 조미료; 얼음",
|
||||
"31": "미가공 농업, 수산양식, 원예 및 임업 생산물; 미가공 곡물 및 종자; 신선한 과실 및 채소, 신선한 허브; 살아 있는 식물 및 꽃; 구근(球根), 모종 및 재배용 곡물종자; 살아있는 동물; 동물용 사료 및 음료; 맥아",
|
||||
"32": "맥주; 비알코올성 음료; 광천수 및 탄산수; 과실음료 및 과실주스; 시럽 및 비알코올성 음료용 제제",
|
||||
"33": "알코올성 음료(맥주는 제외); 음료제조용 알코올성 제제",
|
||||
"34": "담배 및 대용담배; 권연 및 여송연; 흡연자용 전자담배 및 기화기; 흡연용구; 성냥",
|
||||
"35": "광고업; 사업관리/조직 및 경영업; 사무처리업",
|
||||
"36": "금융, 통화 및 은행업; 보험서비스업; 부동산업",
|
||||
"37": "건축서비스업; 설치 및 수리서비스업; 채광업/석유 및 가스 시추업",
|
||||
"38": "통신서비스업",
|
||||
"39": "운송업; 상품의 포장 및 보관업; 여행알선업",
|
||||
"40": "재료처리업; 폐기물 재생업; 공기 정화 및 물 처리업; 인쇄 서비스업; 음식 및 음료수 보존업",
|
||||
"41": "교육업; 훈련제공업; 연예오락업; 스포츠 및 문화활동업",
|
||||
"42": "과학적, 기술적 서비스업 및 관련 연구, 디자인업; 산업분석, 산업연구 및 산업디자인 서비스업; 품질 관리 및 인증 서비스업; 컴퓨터 하드웨어 및 소프트웨어의 디자인 및 개발업",
|
||||
"43": "식음료제공서비스업; 임시숙박시설업",
|
||||
"44": "의료업; 수의업; 인간 또는 동물을 위한 위생 및 미용업; 농업, 수산양식, 원예 및 임업 서비스업",
|
||||
"45": "법무서비스업; 유형의 재산 및 개인을 물리적으로 보호하기 위한 보안서비스업; 이성(異性) 소개업, 온라인 소셜 네트워킹 서비스업; 장례업; 베이비시팅업"
|
||||
}
|
||||
|
|
@ -0,0 +1,362 @@
|
|||
import sqlite3
|
||||
import os
|
||||
import logging
|
||||
|
||||
|
||||
class KeywordDBManager:
|
||||
def __init__(self, db_path, logger):
|
||||
"""
|
||||
KeywordDBManager 초기화 메서드.
|
||||
기존 DB 파일이 있으면 활용하고, 없으면 새로 생성합니다.
|
||||
:param logger: 로깅 인스턴스
|
||||
:param db_path: 데이터베이스 파일 경로
|
||||
"""
|
||||
self.logger = logger
|
||||
self.db_path = db_path
|
||||
|
||||
self.conn = sqlite3.connect(self.db_path, check_same_thread=False)
|
||||
self.cursor = self.conn.cursor()
|
||||
|
||||
if os.path.exists(self.db_path):
|
||||
self.logger.log(f"기존 데이터베이스 파일 '{self.db_path}'을(를) 사용합니다.", level=logging.INFO)
|
||||
else:
|
||||
self.logger.log(f"데이터베이스 파일 '{self.db_path}'이(가) 없습니다. 새 파일을 생성합니다.", level=logging.INFO)
|
||||
|
||||
self._initialize_db()
|
||||
|
||||
def _initialize_db(self):
|
||||
"""
|
||||
데이터베이스 초기화 및 테이블 생성.
|
||||
"""
|
||||
try:
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
# keywords 테이블 생성
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS keywords (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
keyword TEXT UNIQUE NOT NULL,
|
||||
grade TEXT NOT NULL CHECK(grade IN ('금지', '비허용')), -- 허용 등급만 저장
|
||||
status TEXT DEFAULT NULL, -- 검증 결과 상태 ('등록', '미등록', NULL 등)
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
)
|
||||
""")
|
||||
|
||||
# kipris_results 테이블 생성
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS kipris_results (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
keyword_id INTEGER NOT NULL, -- keywords 테이블의 ID와 연결
|
||||
application_status TEXT, -- 등록/공개 상태
|
||||
registration_date TEXT,
|
||||
applicant_name TEXT,
|
||||
classification_code TEXT,
|
||||
category_description TEXT,
|
||||
drawing TEXT,
|
||||
bigDrawing TEXT,
|
||||
FOREIGN KEY (keyword_id) REFERENCES keywords(id) ON DELETE CASCADE
|
||||
)
|
||||
""")
|
||||
|
||||
conn.commit()
|
||||
self.logger.log("DB 테이블 초기화 완료", level=logging.DEBUG)
|
||||
except Exception as e:
|
||||
self.logger.log(f"DB 테이블 초기화 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
def add_keyword(self, keyword, grade):
|
||||
"""
|
||||
새로운 키워드 추가.
|
||||
:param keyword: 키워드 (문자열)
|
||||
:param grade: 등급 ('금지' 또는 '비허용')
|
||||
:return: 성공 여부 (True/False)
|
||||
"""
|
||||
if grade not in ["금지", "비허용"]:
|
||||
self.logger.log(f"유효하지 않은 등급: {grade}", level=logging.WARNING)
|
||||
return False
|
||||
|
||||
try:
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("INSERT INTO keywords (keyword, grade) VALUES (?, ?)", (keyword.strip(), grade.strip()))
|
||||
conn.commit()
|
||||
self.logger.log(f"키워드 '{keyword}' 추가 성공 (등급: {grade})", level=logging.INFO)
|
||||
return True
|
||||
except sqlite3.IntegrityError:
|
||||
self.logger.log(f"키워드 '{keyword}'는 이미 존재합니다.", level=logging.WARNING)
|
||||
return False
|
||||
|
||||
def delete_keyword(self, keyword):
|
||||
"""
|
||||
키워드 삭제.
|
||||
:param keyword: 삭제할 키워드 (문자열)
|
||||
:return: 성공 여부 (True/False)
|
||||
"""
|
||||
try:
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("DELETE FROM keywords WHERE keyword = ?", (keyword,))
|
||||
if cursor.rowcount > 0:
|
||||
conn.commit()
|
||||
self.logger.log(f"키워드 '{keyword}' 삭제 성공.", level=logging.INFO)
|
||||
return True
|
||||
else:
|
||||
self.logger.log(f"키워드 '{keyword}'가 존재하지 않습니다.", level=logging.WARNING)
|
||||
return False
|
||||
except Exception as e:
|
||||
self.logger.log(f"키워드 삭제 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
return False
|
||||
|
||||
def update_keyword_grade(self, keyword, new_grade):
|
||||
"""
|
||||
키워드의 등급 업데이트.
|
||||
:param keyword: 수정할 키워드
|
||||
:param new_grade: 새로운 등급 ('금지' 또는 '비허용')
|
||||
:return: 성공 여부 (True/False)
|
||||
"""
|
||||
if new_grade not in ["금지", "비허용"]:
|
||||
self.logger.log(f"유효하지 않은 등급: {new_grade}", level=logging.WARNING)
|
||||
return False
|
||||
|
||||
try:
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("UPDATE keywords SET grade = ? WHERE keyword = ?", (new_grade, keyword))
|
||||
if cursor.rowcount > 0:
|
||||
conn.commit()
|
||||
self.logger.log(f"키워드 '{keyword}' 등급 업데이트 성공 (새 등급: {new_grade})", level=logging.INFO)
|
||||
return True
|
||||
else:
|
||||
self.logger.log(f"키워드 '{keyword}'가 존재하지 않습니다.", level=logging.WARNING)
|
||||
return False
|
||||
except Exception as e:
|
||||
self.logger.log(f"키워드 등급 업데이트 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
return False
|
||||
|
||||
def get_all_keywords(self):
|
||||
"""
|
||||
모든 키워드와 등급 조회.
|
||||
:return: 키워드와 등급 리스트 [(keyword, grade), ...]
|
||||
"""
|
||||
try:
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT keyword, grade FROM keywords ORDER BY id ASC")
|
||||
keywords = cursor.fetchall()
|
||||
self.logger.log(f"모든 키워드 로드 완료: {keywords}", level=logging.DEBUG)
|
||||
return keywords
|
||||
except Exception as e:
|
||||
self.logger.log(f"키워드 로드 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
return []
|
||||
|
||||
def get_keywords_with_grades(self):
|
||||
"""
|
||||
데이터베이스에서 모든 키워드와 등급을 가져옵니다.
|
||||
:return: {키워드: 등급} 형태의 딕셔너리
|
||||
"""
|
||||
try:
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT keyword, grade FROM keywords")
|
||||
rows = cursor.fetchall()
|
||||
return {row[0]: row[1] for row in rows}
|
||||
except Exception as e:
|
||||
self.logger.log(f"키워드 가져오기 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
return {}
|
||||
|
||||
def search_keyword(self, keyword):
|
||||
"""
|
||||
특정 키워드 검색.
|
||||
:param keyword: 검색할 키워드
|
||||
:return: 키워드와 등급 또는 None
|
||||
"""
|
||||
try:
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT keyword, grade FROM keywords WHERE keyword = ?", (keyword,))
|
||||
result = cursor.fetchone()
|
||||
self.logger.log(f"키워드 검색 결과: {result}", level=logging.DEBUG)
|
||||
return result
|
||||
except Exception as e:
|
||||
self.logger.log(f"키워드 검색 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
return None
|
||||
|
||||
def clear_keywords(self):
|
||||
"""
|
||||
모든 키워드 삭제 (테이블 초기화).
|
||||
"""
|
||||
try:
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("DELETE FROM keywords")
|
||||
conn.commit()
|
||||
self.logger.log("모든 키워드가 삭제되었습니다.", level=logging.INFO)
|
||||
except Exception as e:
|
||||
self.logger.log(f"키워드 초기화 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
def bulk_add_keywords(self, keywords_with_grades):
|
||||
"""
|
||||
키워드와 등급을 한 번에 여러 개 추가.
|
||||
:param keywords_with_grades: [(keyword, grade), ...] 형식의 리스트
|
||||
:return: 추가된 키워드 수
|
||||
"""
|
||||
try:
|
||||
count = 0
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
for keyword, grade in keywords_with_grades:
|
||||
try:
|
||||
cursor.execute("INSERT INTO keywords (keyword, grade) VALUES (?, ?)", (keyword.strip(), grade.strip()))
|
||||
count += 1
|
||||
except sqlite3.IntegrityError:
|
||||
self.logger.log(f"키워드 '{keyword}'는 이미 존재합니다.", level=logging.WARNING)
|
||||
conn.commit()
|
||||
self.logger.log(f"{count}개의 키워드가 추가되었습니다.", level=logging.INFO)
|
||||
return count
|
||||
except Exception as e:
|
||||
self.logger.log(f"키워드와 등급을 한 번에 여러 개 추가 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
|
||||
def update_status(self, keyword, status):
|
||||
"""키워드 검증 상태 업데이트"""
|
||||
try:
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
"UPDATE keywords SET status = ? WHERE keyword = ?",
|
||||
(status, keyword)
|
||||
)
|
||||
conn.commit()
|
||||
self.logger.log(f"키워드 '{keyword}' 상태 업데이트: {status}", level=logging.INFO)
|
||||
except Exception as e:
|
||||
self.logger.log(f"상태 업데이트 중 오류: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
def get_keyword_status(self, keyword):
|
||||
"""키워드의 검증 상태 가져오기"""
|
||||
try:
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT status FROM keywords WHERE keyword = ?", (keyword,))
|
||||
result = cursor.fetchone()
|
||||
return result[0] if result else None
|
||||
except Exception as e:
|
||||
self.logger.log(f"키워드의 검증 상태 가져오기 중 오류: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
def add_kipris_result(self, keyword_id, result):
|
||||
"""
|
||||
키프리스 결과를 테이블에 추가
|
||||
:param keyword_id: 키워드의 ID (keywords 테이블)
|
||||
:param result: 키프리스 결과 딕셔너리
|
||||
"""
|
||||
try:
|
||||
query = """
|
||||
INSERT INTO kipris_results (
|
||||
keyword_id,
|
||||
application_status,
|
||||
registration_date,
|
||||
applicant_name,
|
||||
classification_code,
|
||||
category_description,
|
||||
drawing,
|
||||
bigDrawing
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||||
"""
|
||||
self.cursor.execute(query, (
|
||||
keyword_id,
|
||||
result.get("application_status"),
|
||||
result.get("registration_date"),
|
||||
result.get("applicant_name"),
|
||||
result.get("classification_code"),
|
||||
result.get("category_description"),
|
||||
result.get("drawing"),
|
||||
result.get("bigDrawing")
|
||||
))
|
||||
self.conn.commit()
|
||||
self.logger.log(f"키프리스 결과 추가 성공: {result}", level=logging.INFO)
|
||||
except Exception as e:
|
||||
self.logger.log(f"키프리스 결과 추가 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
def get_kipris_results(self, keyword_id):
|
||||
"""
|
||||
특정 키워드의 모든 키프리스 결과를 반환
|
||||
:param keyword_id: 키워드의 ID (keywords 테이블)
|
||||
:return: 결과 리스트
|
||||
"""
|
||||
try:
|
||||
query = """
|
||||
SELECT application_status, registration_date, applicant_name,
|
||||
classification_code, category_description, drawing, bigDrawing
|
||||
FROM kipris_results
|
||||
WHERE keyword_id = ?
|
||||
"""
|
||||
self.cursor.execute(query, (keyword_id,))
|
||||
rows = self.cursor.fetchall()
|
||||
results = [
|
||||
{
|
||||
"application_status": row[0],
|
||||
"registration_date": row[1],
|
||||
"applicant_name": row[2],
|
||||
"classification_code": row[3],
|
||||
"category_description": row[4],
|
||||
"drawing": row[5],
|
||||
"bigDrawing": row[6],
|
||||
}
|
||||
for row in rows
|
||||
]
|
||||
return results
|
||||
except Exception as e:
|
||||
self.logger.log(f"키프리스 결과 조회 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
return []
|
||||
|
||||
# def get_keyword_details(self, keyword):
|
||||
# """
|
||||
# 키프리스 검증 결과 세부 정보를 반환
|
||||
# :param keyword: 대상 키워드
|
||||
# :return: 키프리스 세부 정보 딕셔너리
|
||||
# """
|
||||
# try:
|
||||
# query = """
|
||||
# SELECT status, registration_date, applicant_name, classification_code, category_description
|
||||
# FROM keywords
|
||||
# WHERE keyword = ?
|
||||
# """
|
||||
# self.cursor.execute(query, (keyword,))
|
||||
# row = self.cursor.fetchone()
|
||||
# if row:
|
||||
# return {
|
||||
# "status": row[0],
|
||||
# "registration_date": row[1],
|
||||
# "applicant_name": row[2],
|
||||
# "classification_code": row[3],
|
||||
# "category_description": row[4],
|
||||
# }
|
||||
# return None
|
||||
# except Exception as e:
|
||||
# self.logger.log(f"키워드 '{keyword}' 세부 정보 조회 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
# return None
|
||||
|
||||
def get_keyword_id(self, keyword):
|
||||
"""
|
||||
키워드의 ID를 반환합니다. 키워드가 없으면 새로 생성 후 ID를 반환합니다.
|
||||
:param keyword: 조회할 키워드
|
||||
:return: 키워드 ID
|
||||
"""
|
||||
try:
|
||||
# 키워드 조회
|
||||
query = "SELECT id FROM keywords WHERE keyword = ?"
|
||||
self.cursor.execute(query, (keyword,))
|
||||
result = self.cursor.fetchone()
|
||||
|
||||
if result:
|
||||
return result[0] # 기존 ID 반환
|
||||
|
||||
# 키워드가 없으면 새로 삽입
|
||||
insert_query = "INSERT INTO keywords (keyword, grade) VALUES (?, '비허용')"
|
||||
self.cursor.execute(insert_query, (keyword,))
|
||||
self.conn.commit()
|
||||
|
||||
# 새로 삽입된 ID 반환
|
||||
return self.cursor.lastrowid
|
||||
except Exception as e:
|
||||
self.logger.log(f"키워드 ID 조회/생성 중 오류: {e}", level=logging.ERROR, exc_info=True)
|
||||
return None
|
||||
|
|
@ -1,13 +0,0 @@
|
|||
{
|
||||
"type": "service_account",
|
||||
"project_id": "igneous-primacy-409723",
|
||||
"private_key_id": "9a9816ba7d7bcde45bc1f0f0f984586ad753022d",
|
||||
"private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCzx0uJieV+r8PU\ntdVrWrbpbMOXR0SIhCVKkM1GWEOF+p6c5Hd3WZz09ALdBWtIQkGNIodzwO4nKh3h\n9F0dNviY6zB86co6UkpdivOm+w6EOBA0qQSF5dCQsvsJ6FyWqZgNPBfZonlk96Hy\nCNtNbkUqxTEPXJQ751WiH0Ke9e6jhPQ9g2ORmDrW1ANNGD4r1dz9GUEegWkFouGE\nsjPgwXEO62i3cFxUuRNMFEE/bvIB/VNLYMqbo4osGfTXopl9N+lW0MJ0sINVnZaI\nImnU6I6F9NMRb1PjkXD/FsLl6lbE82dbZDTdq11rjA7KQqBNXne58DVwE3/dmowi\n3NaGdqkjAgMBAAECggEAAZbnNTXPeaBoxR5kKBbUFyw93Cv4kEoj82g/w8wGQ6og\nKE5hYATCz6MK8ScHrzhv4snX75hYgrAiWlvtltNkc9o75vEcgJwSyfWBr7BBsb4k\nAffBKDJ6P9+MFl3EIYMpxiHlT4Pag3sob/Wq8Y6ZK6sQXkOLR2VzD80G9AxEBZTt\n1Asa9lBXSnK5NCYLsRRX765YysrsQXd8tk6oqsLmcigWCr75nKgbELxiszOTKtLT\npb50y8cb0kDfMUXOltWojGIkhOKetbzmWA3EgXIIa75s9z8tUVE+vO6kSq3TuQXH\nwL+IwatWRlrco3G1LGsHtw2nPR2mljcCVZjvT/Z0gQKBgQDmOJIeobqku/bVDxcq\neLYrMV1/X7zTihox33C7w+ruVXB/edJxR+gROWVO9ufTIgfPTzkDkxZYvTiHA9hC\nz56vu1b2RNLgiGT9ASOvfR00xRSCE/FfaaN5VlWzGskuFUyWDthkbld78wEFLf19\nutsaV/9+RCGZNoGtRjjw4symSQKBgQDH6MI+wXejhAJyZrr2S5jvIlSKtY12QdU6\n+JhD+3OEBl+OndyfucD5HgjSJnMjnzRMML+mPFlcwqU1VmDeeTqSE/mmvyRAr9rm\nG6Xdh+dOngpqwpq9OGsqc+JZ8JF0bdn6V/g26LjkQgNnRCIFnARQ2UHBDoS2/wHn\n72ShlP9kCwKBgGlinADJp9ag9Gyza7dVao57GoGkIZv0K+mIjuJk3LYdBlJUQbD5\naZH45Bcxjw1nFowfh8nLGv+kHqwvZl+vCsUGzNgOyTlfNltamitK6oOtc6XX2zYB\n9YMlsjU6nb0qotROF2Bh4korAtyMIO3dC08T2TDDn12zRck7y/T43RWBAoGALEEO\nny3c+knC8OhlAxkBJg8HgB1oz4ELXx6hNot3qwZuKPgxWvqYCY3ojf0NCBm6ThOM\nmZRKhApi4Efa8eUMXkIlxhASSm+jmcUNFtl7DyBVVgT2lGTk9GTq+tYSnR+kXZMT\n07P5Gi6y6i1fCrbbDbrKn55DKu+Q0HNiZ5LAZrkCgYAp7Aa1rlbXbyoSXNWTA4Nc\nTZIRKj9Ra2hD2Y2EKjLWqLa9RVK+D9a9I/v1gW59PeRpUY9w674IEZXqOq5jx0D9\nFmwL00Omtfv96q+syq7pqSUmI7hDSd1CfLDaxCGHzGykI98GkjQDz6xPEgTAKRIL\njcB1KaEd55AuovwONS3qKA==\n-----END PRIVATE KEY-----\n",
|
||||
"client_email": "service-account@igneous-primacy-409723.iam.gserviceaccount.com",
|
||||
"client_id": "102875157826238718143",
|
||||
"auth_uri": "https://accounts.google.com/o/oauth2/auth",
|
||||
"token_uri": "https://oauth2.googleapis.com/token",
|
||||
"auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
|
||||
"client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/service-account%40igneous-primacy-409723.iam.gserviceaccount.com",
|
||||
"universe_domain": "googleapis.com"
|
||||
}
|
||||
|
|
@ -47,6 +47,9 @@ class Logger(QObject):
|
|||
console_handler.setFormatter(formatter)
|
||||
self.logger.addHandler(console_handler)
|
||||
|
||||
def set_gui_logger(self, gui_logger):
|
||||
self.gui_logger = gui_logger
|
||||
|
||||
def _add_file_handler(self, log_file, level):
|
||||
"""파일 핸들러 추가"""
|
||||
file_handler = RotatingFileHandler(
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ from playwright.async_api import async_playwright, Page
|
|||
|
||||
class PlaywrightThread(QThread):
|
||||
data_collected = Signal(bool, str)
|
||||
progress_signal = Signal(int) # 진행률을 전달하는 시그널
|
||||
|
||||
def __init__(self, logger, db_manager):
|
||||
super().__init__()
|
||||
|
|
@ -59,11 +60,11 @@ class PlaywrightThread(QThread):
|
|||
# 페이지 열기
|
||||
page = await context.new_page()
|
||||
await page.goto("https://world.taobao.com")
|
||||
self.logger.log(f"타오바오 사이트에 접속했습니다.", level=logging.INFO)
|
||||
self.logger.log(f"접속 완료.", level=logging.INFO)
|
||||
|
||||
# 페이지 로딩 확인 및 pagedown
|
||||
await page.wait_for_selector(".tb-pick-content-item") # 상품 카드 로딩 대기
|
||||
self.logger.log(f"페이지 로딩 완료 - Pagedown을 두 번 누릅니다.", level=logging.INFO)
|
||||
self.logger.log(f"페이지 로딩 완료", level=logging.INFO)
|
||||
await page.keyboard.press("PageDown")
|
||||
await page.wait_for_timeout(1000) # 1초 대기
|
||||
await page.keyboard.press("PageDown")
|
||||
|
|
@ -90,6 +91,7 @@ class PlaywrightThread(QThread):
|
|||
async def scrape_items(self, page: Page):
|
||||
try:
|
||||
items = await page.query_selector_all("div#ice-container div.tb-pick-feeds-container > div")
|
||||
total_items = len(items)
|
||||
self.logger.log(f"총 {len(items)}개의 상품 카드가 발견되었습니다.", level=logging.DEBUG)
|
||||
|
||||
items_data = []
|
||||
|
|
@ -140,6 +142,11 @@ class PlaywrightThread(QThread):
|
|||
|
||||
items_data.append((item_id, pc_url, name, float(price), image_url, sales))
|
||||
|
||||
# 프로그레스바 업데이트
|
||||
progress = int(((idx + 1) / total_items) * 100)
|
||||
self.progress_signal.emit(progress)
|
||||
|
||||
|
||||
self.logger.log(f"수집된 상품 수 : {len(items_data)}", level=logging.DEBUG)
|
||||
return items_data
|
||||
except Exception as e:
|
||||
|
|
|
|||
|
|
@ -3,14 +3,9 @@ import pandas as pd
|
|||
from typing import Dict, List
|
||||
import os, sys
|
||||
from PySide6.QtWidgets import QFileDialog
|
||||
from PySide6.QtCore import QObject, Signal
|
||||
|
||||
# from pywinauto import Application, findwindows, timings
|
||||
# from pywinauto.controls.hwndwrapper import HwndWrapper
|
||||
import configparser
|
||||
# from src.shoppingLens import ShoppingLensScraper
|
||||
from src.titleManager import TitleManager
|
||||
from src.categoryManager import CategoryManager
|
||||
# from src.naver_parser import NaverParser
|
||||
from src.gpt_client import GPTClient
|
||||
from src.xlsSerachThread import XlsSerachThread
|
||||
from src.wh_con import WhaleController
|
||||
|
|
@ -20,44 +15,36 @@ from bs4 import BeautifulSoup
|
|||
|
||||
import openpyxl
|
||||
|
||||
class PostProcessor:
|
||||
def __init__(self, logger, db_manager):
|
||||
class PostProcessor(QObject):
|
||||
progress_signal = Signal(int) # 진행률을 전달하는 시그널
|
||||
|
||||
def __init__(self, logger, db_manager, config, categoryManager):
|
||||
super().__init__() # 부모 클래스의 __init__ 호출
|
||||
self.logger = logger
|
||||
self.db_manager = db_manager
|
||||
self.categoryManager = categoryManager
|
||||
self.config = config
|
||||
self.gpt_api_key = self.config.get("GPT_API", "API_KEY")
|
||||
|
||||
base_xls_path = 'baseXLS_Percenty.xlsx'
|
||||
config_path = 'config.ini'
|
||||
|
||||
self.gpt = GPTClient(self.logger, api_key='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')
|
||||
self.gpt = GPTClient(self.logger, api_key=self.gpt_api_key)
|
||||
|
||||
# self.shopping_lens = ShoppingLensScraper(self.logger)
|
||||
self.wh_con = WhaleController(self.logger)
|
||||
self.title_manager = TitleManager(self.logger, self.gpt)
|
||||
self.categoryManager = CategoryManager(self.logger, base_xls_path)
|
||||
# self.naver_parser = NaverParser(self.logger, client_id=client_id, client_secret=client_secret)
|
||||
|
||||
self.xlThread = XlsSerachThread(self.logger, self.db_manager)
|
||||
|
||||
# 설정 파일 로드
|
||||
self.config = configparser.ConfigParser()
|
||||
self.read_config(config_path)
|
||||
# # 필터 데이터 설정
|
||||
self.banned_words = None
|
||||
self.disallowed_words = None
|
||||
|
||||
# 필터 데이터 로드
|
||||
self.banned_tags = set(self.config.get("Filters", "banned_tags", fallback="").split(","))
|
||||
self.banned_words = set(self.config.get("Filters", "banned_words", fallback="").split(","))
|
||||
self.disallowed_words = set(self.config.get("Filters", "disallowed_words", fallback="").split(","))
|
||||
def set_keyword_manager(self, keyword_manager):
|
||||
self.banned_words = keyword_manager.get_ban_list()
|
||||
self.disallowed_words = keyword_manager.get_restricted_list()
|
||||
|
||||
def read_config(self, config_path):
|
||||
try:
|
||||
# 파일을 UTF-8로 열어서 ConfigParser로 읽기
|
||||
with open(config_path, 'r', encoding='utf-8') as config_file:
|
||||
self.config.read_file(config_file)
|
||||
except UnicodeDecodeError as e:
|
||||
self.logger.error(f"Config 파일 읽기 중 인코딩 오류 발생: {e}")
|
||||
raise
|
||||
except FileNotFoundError:
|
||||
self.logger.error(f"Config 파일을 찾을 수 없습니다: {config_path}")
|
||||
raise
|
||||
self.logger.log(f"self.banned_words : {self.banned_words}", level=logging.DEBUG)
|
||||
self.logger.log(f"self.disallowed_words : {self.disallowed_words}", level=logging.DEBUG)
|
||||
|
||||
def get_base_dir(self):
|
||||
"""
|
||||
|
|
@ -118,7 +105,7 @@ class PostProcessor:
|
|||
# except Exception as e:
|
||||
# self.logger.log(f"웨일 창 탐색 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
# return None
|
||||
|
||||
|
||||
async def post_by_DB(self):
|
||||
# 1. DB에서 처리되지 않은 상품 가져오기
|
||||
# products = self.db_manager.fetch_all().query("is_export == 0").to_dict('records')
|
||||
|
|
@ -126,118 +113,11 @@ class PostProcessor:
|
|||
# is_valid가 1이고 generated_Title이 비어 있으며 is_export가 1인 항목 가져오기
|
||||
custom_conditions = "is_valid = 1 AND (generated_Title IS NULL OR generated_Title = '') AND is_export = 0"
|
||||
products = self.db_manager.fetch_filtered(conditions=custom_conditions).to_dict('records')
|
||||
|
||||
self.logger.log(f"처리 대상 상품 {len(products)}개 로드 완료", level=logging.INFO)
|
||||
|
||||
self.logger.log(f"처리 대상 상품 {len(products)}개 로드 완료", level=logging.DEBUG)
|
||||
self.process_products(products)
|
||||
|
||||
await self.xlThread.start_br()
|
||||
|
||||
await self.process_products(products)
|
||||
|
||||
# def post_by_XLS(self):
|
||||
# default_folder = os.path.join(os.getcwd(), 'XLS')
|
||||
# selected_folder = QFileDialog.getExistingDirectory(None, "XLS 폴더 선택", default_folder)
|
||||
|
||||
# if not selected_folder:
|
||||
# self.logger.warning("폴더 선택이 취소되었습니다.")
|
||||
# return
|
||||
|
||||
# self.logger.info(f"선택된 폴더: {selected_folder}")
|
||||
# self._post_by_XLS(selected_folder)
|
||||
|
||||
|
||||
async def post_by_XLS(self, folder_path):
|
||||
"""
|
||||
주어진 폴더 경로에서 모든 엑셀 파일을 순회하며 데이터를 수집 및 DB에 저장.
|
||||
:param folder_path: 엑셀 파일이 위치한 폴더 경로
|
||||
"""
|
||||
try:
|
||||
# PlaywrightThread 초기화 및 브라우저 시작
|
||||
page = await self.xlThread.start_br()
|
||||
|
||||
# 폴더 내 모든 엑셀 파일 가져오기
|
||||
excel_files = [f for f in os.listdir(folder_path) if f.endswith('.xls') or f.endswith('.xlsx')]
|
||||
|
||||
if not excel_files:
|
||||
self.logger.log(f"엑셀 파일이 폴더 '{folder_path}'에 없습니다.", level=logging.WARNING)
|
||||
return
|
||||
|
||||
self.logger.log(f"총 {len(excel_files)}개의 엑셀 파일을 발견했습니다.", level=logging.DEBUG)
|
||||
|
||||
# DB 초기화
|
||||
db_name = "xls_db.db"
|
||||
self.db_manager.create_table(db_name)
|
||||
|
||||
for excel_file in excel_files:
|
||||
file_path = os.path.join(folder_path, excel_file)
|
||||
self.logger.log(f"엑셀 파일 처리 중: {file_path}", level=logging.DEBUG)
|
||||
|
||||
try:
|
||||
# 엑셀 파일 열기
|
||||
workbook = openpyxl.load_workbook(file_path, data_only=True)
|
||||
sheet = workbook.active
|
||||
|
||||
# 데이터 추출 (B4~B53, C4~C53)
|
||||
items = []
|
||||
url_list = []
|
||||
row_map = {}
|
||||
|
||||
for row in range(4, 54): # 4번 행부터 53번 행까지
|
||||
pc_url = sheet[f"B{row}"].value # PC_URL
|
||||
name = sheet[f"C{row}"].value # 상품명
|
||||
|
||||
if not pc_url or not name:
|
||||
self.logger.log(f"필수 데이터 누락: 행 {row} (PC_URL: {pc_url}, Name: {name})", level=logging.WARNING)
|
||||
continue
|
||||
|
||||
id_value = self.parse_id_from_url(pc_url) # URL에서 ID 추출
|
||||
if id_value:
|
||||
url_list.append(pc_url)
|
||||
row_map[pc_url] = (id_value, name, row)
|
||||
|
||||
# URL 목록 처리하여 가격 및 이미지 URL 수집
|
||||
for url in url_list:
|
||||
result = await self.xlThread.goto_url_and_parsing(db_name, page, id_value, url)
|
||||
|
||||
price = result.get("price") if result else None
|
||||
image_url = result.get("image_url") if result else None
|
||||
|
||||
if not price or not image_url:
|
||||
self.logger.log(f"데이터 수집 실패: URL {url}", level=logging.WARNING)
|
||||
continue
|
||||
|
||||
self.logger.log(f"url: {url}", level=logging.DEBUG)
|
||||
self.logger.log(f"price: {price}", level=logging.DEBUG)
|
||||
self.logger.log(f"image_url: {image_url}", level=logging.DEBUG)
|
||||
|
||||
id_value, name, row = row_map[url]
|
||||
items.append((id_value, url, name, price, image_url, 0)) # sales는 0으로 설정
|
||||
|
||||
# 수집된 데이터 DB 저장
|
||||
self.logger.log(f"수집된 items: {items}", level=logging.DEBUG)
|
||||
|
||||
if items:
|
||||
self.db_manager.insert_items(items, db_name)
|
||||
self.logger.log(f"{file_path}의 데이터를 DB에 저장했습니다.", level=logging.DEBUG)
|
||||
|
||||
# 처리 완료 후 브라우저 종료
|
||||
await self.xlThread.close_br()
|
||||
|
||||
except Exception as e:
|
||||
self.logger.log(f"엑셀 파일 처리 중 오류 발생: {file_path}, 오류: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
# 처리되지 않은 상품 로드 및 후처리
|
||||
# products = self.db_manager.fetch_all(db_path=db_name).query("is_export == 0").to_dict('records')
|
||||
# products = self.db_manager.fetch_filtered().to_dict('records')
|
||||
|
||||
# is_valid가 1이고 generated_Title이 비어 있으며 is_export가 1인 항목 가져오기
|
||||
custom_conditions = "is_valid = 1 AND (generated_Title IS NULL OR generated_Title = '') AND is_export = 1"
|
||||
products = self.db_manager.fetch_filtered(conditions=custom_conditions).to_dict('records')
|
||||
|
||||
self.logger.log(f"총 {len(products)}개의 처리되지 않은 상품 로드 완료.", level=logging.DEBUG)
|
||||
self.process_products(products)
|
||||
|
||||
except Exception as e:
|
||||
self.logger.log(f"XLS 데이터 처리 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
def parse_id_from_url(self, url):
|
||||
"""
|
||||
|
|
@ -297,19 +177,17 @@ class PostProcessor:
|
|||
|
||||
return None, None
|
||||
|
||||
async def process_products(self, products):
|
||||
def process_products(self, products):
|
||||
|
||||
total_products = len(products) # 총 상품 개수
|
||||
if total_products == 0:
|
||||
self.logger.log("처리할 상품이 없습니다.", level=logging.WARNING)
|
||||
self.progress_signal.emit(100)
|
||||
return
|
||||
|
||||
# 쇼핑렌즈를 위한 웹브라우저 준비
|
||||
# whale_window = self.start_whale_browser()
|
||||
self.wh_con.start_whale_Browser()
|
||||
|
||||
# time.sleep(600)
|
||||
|
||||
# # 1. DB에서 처리되지 않은 상품 가져오기
|
||||
# products = self.db_manager.fetch_all().query("is_export == 0").to_dict('records')
|
||||
# self.logger.log(f"처리 대상 상품 {len(products)}개 로드 완료", level=logging.DEBUG)
|
||||
|
||||
for product in products:
|
||||
for idx, product in enumerate(products, start=1):
|
||||
try:
|
||||
# 2. 쇼핑렌즈로 상품 정보 수집
|
||||
self.logger.log(f"상품 {product['id']}에 대한 쇼핑렌즈 검색 시작", level=logging.DEBUG)
|
||||
|
|
@ -339,7 +217,7 @@ class PostProcessor:
|
|||
titles = [product["title"] for product in scraped_data if "title" in product]
|
||||
|
||||
# 4. 상품명 생성
|
||||
final_title = self.title_manager.generate_product_name(titles, product['name'])
|
||||
final_title = self.title_manager.generate_product_name(titles, product['name'], self.banned_words, self.disallowed_words)
|
||||
self.logger.log(f"상품명 생성 완료: {final_title}", level=logging.DEBUG)
|
||||
|
||||
# 태그 필터링 및 병합
|
||||
|
|
@ -359,9 +237,17 @@ class PostProcessor:
|
|||
self.db_manager.update_item(product)
|
||||
self.logger.log(f"상품 {product['id']} 처리 완료", level=logging.DEBUG)
|
||||
|
||||
# 진행률 계산 및 업데이트
|
||||
progress = int((idx / total_products) * 100)
|
||||
self.progress_signal.emit(progress)
|
||||
|
||||
except Exception as e:
|
||||
self.logger.log(f"상품 {product['id']} 처리 중 오류 발생: {e}", level=logging.ERROR, exc_info=True)
|
||||
|
||||
# 작업 완료 후 100%로 설정
|
||||
self.progress_signal.emit(100)
|
||||
self.logger.log("모든 상품 처리가 완료되었습니다.", level=logging.INFO)
|
||||
|
||||
def filter_and_merge_tags(self, scraped_data) -> str:
|
||||
"""
|
||||
태그 필터링 및 병합 메서드
|
||||
|
|
@ -382,7 +268,7 @@ class PostProcessor:
|
|||
unique_tags = sorted(set(split_tags))
|
||||
|
||||
# 금지 태그 제거
|
||||
filtered_tags = [tag for tag in unique_tags if tag not in self.banned_tags]
|
||||
filtered_tags = [tag for tag in unique_tags if tag not in self.disallowed_words]
|
||||
|
||||
# 최종 태그 문자열 반환
|
||||
result = ", ".join(filtered_tags)
|
||||
|
|
|
|||
|
|
@ -2,9 +2,6 @@ import re
|
|||
import logging
|
||||
|
||||
class TitleManager:
|
||||
# 허용불가 목록과 금지어 목록 정의
|
||||
disallowed_words = ["불가단어1", "불가단어2"] # 허용불가 단어 리스트
|
||||
banned_words = ["금지단어1", "금지단어2"] # 금지어 단어 리스트
|
||||
|
||||
def __init__(self, logger, gpt):
|
||||
"""
|
||||
|
|
@ -14,7 +11,7 @@ class TitleManager:
|
|||
self.logger = logger
|
||||
self.gpt = gpt
|
||||
|
||||
def generate_product_name(self, titles, original_name):
|
||||
def generate_product_name(self, titles, original_name, banned_words, disallowed_words):
|
||||
"""
|
||||
상품명을 생성하는 메서드
|
||||
:param processed_data: 처리된 상품 데이터 리스트
|
||||
|
|
@ -52,10 +49,10 @@ class TitleManager:
|
|||
]
|
||||
|
||||
# 허용불가 목록에 있는 단어 제거
|
||||
alloed_words = [word for word in filtered_words if word not in self.disallowed_words]
|
||||
alloed_words = [word for word in filtered_words if word not in disallowed_words]
|
||||
|
||||
# 금지어 목록에 있는 단어 제거
|
||||
final_filtered_words = [word for word in alloed_words if word not in self.banned_words]
|
||||
final_filtered_words = [word for word in alloed_words if word not in banned_words]
|
||||
|
||||
generated_title = self.gpt.generate_product_name_next(final_filtered_words, original_name, titles, unique_first_two_words)
|
||||
self.logger.log(f"gpt를 이용한 상품명 제작 : {generated_title}", level=logging.DEBUG)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,150 @@
|
|||
from PySide6.QtWidgets import QDialog, QVBoxLayout, QHBoxLayout, QPushButton, QCheckBox, QLineEdit, QMessageBox, QFormLayout
|
||||
from PySide6.QtCore import Qt, QSettings
|
||||
|
||||
|
||||
class UserInfoDialog(QDialog):
|
||||
def __init__(self, parent=None):
|
||||
super().__init__(parent)
|
||||
self.setWindowTitle("사용자 정보")
|
||||
self.setMinimumSize(400, 300)
|
||||
|
||||
self.settings = QSettings("내차는언제타냐", "역소싱기")
|
||||
|
||||
# 메인 레이아웃
|
||||
main_layout = QVBoxLayout()
|
||||
|
||||
# 사용자 정보 입력 레이아웃
|
||||
form_layout = QFormLayout()
|
||||
|
||||
# 이메일 (아이디)
|
||||
self.email_input = QLineEdit()
|
||||
self.email_input.setPlaceholderText("example@example.com")
|
||||
form_layout.addRow("이메일:", self.email_input)
|
||||
|
||||
# 비밀번호
|
||||
self.password_input = QLineEdit()
|
||||
self.password_input.setPlaceholderText("비밀번호 입력")
|
||||
self.password_input.setEchoMode(QLineEdit.Password)
|
||||
form_layout.addRow("비밀번호:", self.password_input)
|
||||
|
||||
# API Key
|
||||
self.api_key_input = QLineEdit()
|
||||
self.api_key_input.setPlaceholderText("API Key 입력")
|
||||
form_layout.addRow("API Key:", self.api_key_input)
|
||||
|
||||
# 저장 토글 버튼
|
||||
self.save_credentials_checkbox = QCheckBox("로그인 정보 저장")
|
||||
form_layout.addWidget(self.save_credentials_checkbox)
|
||||
|
||||
main_layout.addLayout(form_layout)
|
||||
|
||||
# 버튼 레이아웃
|
||||
button_layout = QHBoxLayout()
|
||||
|
||||
# 회원가입 버튼
|
||||
self.signup_button = QPushButton("회원가입")
|
||||
self.signup_button.clicked.connect(self.signup)
|
||||
button_layout.addWidget(self.signup_button)
|
||||
|
||||
# 로그인 버튼
|
||||
self.login_button = QPushButton("로그인")
|
||||
self.login_button.clicked.connect(self.handle_login)
|
||||
button_layout.addWidget(self.login_button)
|
||||
|
||||
self.cancel_button = QPushButton("취소")
|
||||
self.cancel_button.clicked.connect(self.reject)
|
||||
button_layout.addWidget(self.cancel_button)
|
||||
|
||||
# 비밀번호 찾기 버튼
|
||||
self.forgot_password_button = QPushButton("비밀번호 찾기")
|
||||
self.forgot_password_button.clicked.connect(self.forgot_password)
|
||||
button_layout.addWidget(self.forgot_password_button)
|
||||
|
||||
main_layout.addLayout(button_layout)
|
||||
|
||||
# 레이아웃 설정
|
||||
self.setLayout(main_layout)
|
||||
|
||||
# 이전 저장된 정보 불러오기
|
||||
self.load_credentials()
|
||||
|
||||
# 초기 포커스 설정: 로그인 버튼
|
||||
self.login_button.setFocus()
|
||||
|
||||
def load_credentials(self):
|
||||
"""
|
||||
QSettings에서 저장된 이메일, 비밀번호, API 키를 불러옵니다.
|
||||
"""
|
||||
email = self.settings.value("email", "")
|
||||
password = self.settings.value("password", "")
|
||||
api_key = self.settings.value("api_key", "")
|
||||
save_credentials = self.settings.value("save_credentials", False, type=bool)
|
||||
|
||||
self.email_input.setText(email)
|
||||
self.password_input.setText(password)
|
||||
self.api_key_input.setText(api_key)
|
||||
self.save_credentials_checkbox.setChecked(save_credentials)
|
||||
|
||||
def save_credentials(self):
|
||||
"""
|
||||
QSettings에 이메일, 비밀번호, API 키를 저장합니다.
|
||||
"""
|
||||
if self.save_credentials_checkbox.isChecked():
|
||||
self.settings.setValue("email", self.email_input.text())
|
||||
self.settings.setValue("password", self.password_input.text())
|
||||
self.settings.setValue("api_key", self.api_key_input.text())
|
||||
self.settings.setValue("save_credentials", True)
|
||||
else:
|
||||
self.settings.remove("email")
|
||||
self.settings.remove("password")
|
||||
self.settings.remove("api_key")
|
||||
self.settings.setValue("save_credentials", False)
|
||||
|
||||
def signup(self):
|
||||
"""
|
||||
회원가입 버튼 클릭 시 호출
|
||||
"""
|
||||
email = self.email_input.text()
|
||||
password = self.password_input.text()
|
||||
api_key = self.api_key_input.text()
|
||||
|
||||
if not email or not password or not api_key:
|
||||
QMessageBox.warning(self, "입력 오류", "모든 필드를 입력해 주세요.")
|
||||
return
|
||||
|
||||
# 간단한 이메일 형식 확인
|
||||
if "@" not in email or "." not in email:
|
||||
QMessageBox.warning(self, "입력 오류", "올바른 이메일 형식이 아닙니다.")
|
||||
return
|
||||
|
||||
# 회원가입 로직 추가 (API 요청 등)
|
||||
QMessageBox.information(self, "회원가입", "회원가입이 완료되었습니다!")
|
||||
|
||||
def handle_login(self):
|
||||
"""
|
||||
로그인 로직 처리 및 저장 기능
|
||||
"""
|
||||
email = self.email_input.text()
|
||||
password = self.password_input.text()
|
||||
api_key = self.api_key_input.text()
|
||||
|
||||
# 로그인 검증 (간단한 예제)
|
||||
if email == "admin@example.com" and password == "password" and api_key == "API123":
|
||||
QMessageBox.information(self, "로그인 성공", "로그인에 성공했습니다.")
|
||||
self.save_credentials() # 로그인 성공 시 자격 증명 저장
|
||||
self.accept() # 다이얼로그 닫기
|
||||
else:
|
||||
QMessageBox.warning(self, "로그인 실패", "이메일, 비밀번호 또는 API 키가 올바르지 않습니다.")
|
||||
|
||||
def forgot_password(self):
|
||||
"""
|
||||
비밀번호 찾기 버튼 클릭 시 호출
|
||||
"""
|
||||
email = self.email_input.text()
|
||||
|
||||
if not email:
|
||||
QMessageBox.warning(self, "입력 오류", "이메일을 입력해 주세요.")
|
||||
return
|
||||
|
||||
# 비밀번호 찾기 로직 추가 (API 요청 등)
|
||||
QMessageBox.information(self, "비밀번호 찾기", "비밀번호 재설정 이메일을 전송했습니다!")
|
||||
BIN
taobao_items.db
BIN
taobao_items.db
Binary file not shown.
Loading…
Reference in New Issue