157 lines
6.1 KiB
Python
157 lines
6.1 KiB
Python
import sqlite3
|
|
from datetime import datetime
|
|
|
|
class VOCDatabase:
|
|
def __init__(self, db_name="voc.db"):
|
|
self.conn = sqlite3.connect(db_name, check_same_thread=False)
|
|
self.conn.row_factory = sqlite3.Row # 기본적으로 모든 쿼리 결과를 Row 객체로 반환
|
|
self.create_table()
|
|
|
|
def create_table(self):
|
|
"""데이터베이스 테이블 생성 및 스키마 업데이트"""
|
|
cursor = self.conn.cursor()
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS posts (
|
|
id TEXT PRIMARY KEY,
|
|
title TEXT,
|
|
writer TEXT,
|
|
date TEXT,
|
|
department TEXT,
|
|
is_public INTEGER, -- 0:X, 1:O
|
|
status TEXT,
|
|
content TEXT,
|
|
is_related INTEGER,
|
|
|
|
-- 추가된 상세 필드들
|
|
station TEXT, -- 역명
|
|
channel TEXT, -- 접수채널
|
|
attachment TEXT, -- 고객첨부파일
|
|
answer TEXT, -- 답변내용
|
|
voc_type TEXT, -- VOC유형
|
|
response_type TEXT, -- 응답구분
|
|
summary TEXT, -- 요약 (시스템/AI용)
|
|
|
|
created_at TIMESTAMP,
|
|
updated_at TIMESTAMP,
|
|
last_checked_at TIMESTAMP
|
|
)
|
|
''')
|
|
self.conn.commit()
|
|
|
|
def upsert_post(self, data):
|
|
"""기본 목록 정보를 저장하거나 업데이트 (is_new, is_updated 반환)"""
|
|
cursor = self.conn.cursor()
|
|
now = datetime.now()
|
|
|
|
cursor.execute("SELECT is_public, department, status, title FROM posts WHERE id=?", (data['id'],))
|
|
row = cursor.fetchone()
|
|
|
|
if row is None:
|
|
# [신규]
|
|
cursor.execute('''
|
|
INSERT INTO posts (
|
|
id, title, writer, date, department, is_public, status,
|
|
is_related, created_at, updated_at, last_checked_at
|
|
)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
''', (
|
|
data['id'], data['title'], data['writer'], data.get('date', ''),
|
|
data['department'], data['is_public'], data['status'],
|
|
data['is_related'], now, now, now
|
|
))
|
|
self.conn.commit()
|
|
return True, False
|
|
|
|
else:
|
|
# [업데이트] 제목이 '...'으로 끝나는 경우 상세 제목으로 덮어씌워질 수 있도록 처리
|
|
old_public, old_dept, old_status, old_title = row
|
|
changed = False
|
|
|
|
if old_public != data['is_public'] or old_dept != data['department'] or old_status != data['status']:
|
|
changed = True
|
|
|
|
# 제목이 '...'으로 요약된 경우보다 더 긴 제목이 들어오면 업데이트
|
|
if len(data['title']) > len(old_title) and not data['title'].endswith('...'):
|
|
changed = True
|
|
|
|
if changed:
|
|
cursor.execute('''
|
|
UPDATE posts SET
|
|
title=?, department=?, is_public=?, status=?, is_related=?,
|
|
updated_at=?, last_checked_at=?
|
|
WHERE id=?
|
|
''', (
|
|
data['title'], data['department'], data['is_public'],
|
|
data['status'], data['is_related'], now, now, data['id']
|
|
))
|
|
self.conn.commit()
|
|
return False, True
|
|
else:
|
|
cursor.execute("UPDATE posts SET last_checked_at=? WHERE id=?", (now, data['id']))
|
|
self.conn.commit()
|
|
return False, False
|
|
|
|
def update_detail(self, voc_id, detail_data):
|
|
"""상세 페이지에서 추출한 모든 정보를 업데이트"""
|
|
cursor = self.conn.cursor()
|
|
now = datetime.now()
|
|
|
|
# 상세 데이터에는 전체 제목이 포함되어 있을 것이므로 제목도 함께 업데이트
|
|
cursor.execute('''
|
|
UPDATE posts SET
|
|
title = ?,
|
|
content = ?,
|
|
date = ?,
|
|
station = ?,
|
|
channel = ?,
|
|
attachment = ?,
|
|
answer = ?,
|
|
voc_type = ?,
|
|
response_type = ?,
|
|
summary = ?,
|
|
updated_at = ?,
|
|
last_checked_at = ?
|
|
WHERE id = ?
|
|
''', (
|
|
detail_data.get('title'),
|
|
detail_data.get('content'),
|
|
detail_data.get('date'),
|
|
detail_data.get('station'),
|
|
detail_data.get('channel'),
|
|
detail_data.get('attachment'),
|
|
detail_data.get('answer'),
|
|
detail_data.get('voc_type'),
|
|
detail_data.get('response_type'),
|
|
detail_data.get('summary'),
|
|
now, now, voc_id
|
|
))
|
|
self.conn.commit()
|
|
|
|
def get_all_posts(self):
|
|
"""전체 목록 조회 (접수번호 기준 내림차순 정렬)"""
|
|
cursor = self.conn.cursor()
|
|
# 번호(id)가 문자열이므로 캐스팅하여 정렬하거나, 최신 등록순으로 정렬
|
|
cursor.execute("SELECT * FROM posts ORDER BY id DESC LIMIT 500")
|
|
return cursor.fetchall()
|
|
|
|
def get_post_by_id(self, voc_id):
|
|
"""특정 ID의 상세 데이터 전체 조회"""
|
|
cursor = self.conn.cursor()
|
|
cursor.execute("SELECT * FROM posts WHERE id=?", (voc_id,))
|
|
return cursor.fetchone()
|
|
|
|
def get_posts_needing_detail(self, recheck_hours=3):
|
|
"""상세 정보 수집이 필요한 게시글 조회"""
|
|
# 전략:
|
|
# 1. '내용이 없는 글'을 최우선으로 찾는다.
|
|
# 2. 그 중에서도 '공개(is_public=1)' 된 글을 먼저 처리한다. (비공개->공개 전환 즉시 반영)
|
|
# 3. 그 다음 최신순으로 처리한다.
|
|
|
|
cursor = self.conn.cursor()
|
|
cursor.execute('''
|
|
SELECT id, title, is_related FROM posts
|
|
WHERE content IS NULL OR content = ''
|
|
ORDER BY is_public DESC, id DESC
|
|
LIMIT 10
|
|
''')
|
|
return cursor.fetchall() |