65 lines
2.1 KiB
Python
65 lines
2.1 KiB
Python
import sqlite3
|
|
# from collections import defaultdict
|
|
import logging
|
|
|
|
# 로거 인스턴스 가져오기
|
|
logger = logging.getLogger('default_logger')
|
|
def connect_database(db_path):
|
|
"""SQLite 데이터베이스 파일에 연결합니다."""
|
|
conn = None
|
|
|
|
try:
|
|
conn = sqlite3.connect(db_path)
|
|
conn.row_factory = sqlite3.Row # 이 부분을 추가합니다.
|
|
logger.debug("데이터베이스 연결 성공")
|
|
return conn
|
|
except sqlite3.Error as e:
|
|
logger.debug(f"데이터베이스 연결 실패: {e}")
|
|
return None
|
|
|
|
def fetch_grouped_data(conn):
|
|
"""'NaverShopping' 테이블에서 'keyword_id'로 그룹화된 레코드를 dict로 가져옵니다."""
|
|
data_dict = {}
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT * FROM NaverShopping")
|
|
rows = cur.fetchall()
|
|
|
|
for row in rows:
|
|
# row는 이제 딕셔너리처럼 접근할 수 있습니다.
|
|
keyword_id = row['keyword_id']
|
|
if keyword_id not in data_dict:
|
|
data_dict[keyword_id] = []
|
|
data_dict[keyword_id].append(row)
|
|
|
|
except sqlite3.Error as e:
|
|
logger.debug(f"데이터 가져오기 실패: {e}")
|
|
|
|
return data_dict
|
|
|
|
def print_keyword_products(data_dict, keyword_id):
|
|
"""특정 'keyword_id'를 가진 레코드의 'productTitle'을 순차적으로 출력합니다."""
|
|
if keyword_id in data_dict:
|
|
for record in data_dict[keyword_id]:
|
|
logger.debug(record['productTitle'])
|
|
else:
|
|
logger.debug(f"'keyword_id': {keyword_id}를 가진 데이터가 없습니다.")
|
|
|
|
def main(id):
|
|
database = "2024-03-04-2210.db"
|
|
|
|
# 데이터베이스 연결 생성
|
|
conn = connect_database(database)
|
|
if conn is not None:
|
|
data_dict = fetch_grouped_data(conn)
|
|
# 'keyword_id'가 1인 그룹의 'productTitle' 출력
|
|
if 1 in data_dict:
|
|
for item in data_dict[1]:
|
|
logger.debug(item['productTitle'])
|
|
conn.close()
|
|
else:
|
|
logger.debug("Error! cannot create the database connection.")
|
|
|
|
if __name__ == '__main__':
|
|
main(id)
|