8bcea3040f
- /api/admin/qr/generate에서 client_items 파라미터 추가 - 클라이언트 전달 items 우선, 없으면 MSSQL 조회 - get_sale_items 쿼리 컬럼명 수정 (DrugCode, GoodsName 등)
304 lines
9.3 KiB
Python
304 lines
9.3 KiB
Python
"""
|
|
QR Claim Token 생성 모듈
|
|
후향적 적립을 위한 1회성 토큰 생성
|
|
|
|
v2 (2026-03-29): 서버 즉시 전송 추가 (pos.pharmq.kr)
|
|
"""
|
|
|
|
import hashlib
|
|
import secrets
|
|
import logging
|
|
import requests
|
|
from datetime import datetime, timedelta
|
|
import sys
|
|
import os
|
|
|
|
# DB 연결
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
|
from db.dbsetup import DatabaseManager
|
|
|
|
# 설정값
|
|
MILEAGE_RATE = 0.03 # 3% 적립
|
|
TOKEN_EXPIRY_DAYS = 30 # 30일 유효기간
|
|
|
|
# 서버 설정 (v2) - config.json에서 읽기
|
|
import json
|
|
_config_path = os.path.join(os.path.dirname(__file__), '..', 'config.json')
|
|
try:
|
|
with open(_config_path, 'r', encoding='utf-8') as f:
|
|
_config = json.load(f)
|
|
CLOUD_API_URL = _config.get('cloud_api_url', 'https://pos.pharmq.kr')
|
|
PHARMACY_CODE = _config.get('pharmacy_code', 'P0001')
|
|
except:
|
|
CLOUD_API_URL = "https://pos.pharmq.kr"
|
|
PHARMACY_CODE = "P0001"
|
|
|
|
QR_BASE_URL = f"{CLOUD_API_URL}/{PHARMACY_CODE}/claim"
|
|
|
|
# 로거
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def generate_claim_token(transaction_id, total_amount, pharmacy_id="YANGGU001"):
|
|
"""
|
|
Claim Token 생성 (SHA256 해시 기반)
|
|
|
|
Args:
|
|
transaction_id (str): POS 거래 ID (SALE_MAIN.SL_NO_order)
|
|
total_amount (float): 총 판매 금액
|
|
pharmacy_id (str): 약국 코드
|
|
|
|
Returns:
|
|
dict: {
|
|
'qr_url': QR 코드 URL,
|
|
'token_raw': 토큰 원문 (QR에만 포함),
|
|
'token_hash': SHA256 해시 (DB 저장용),
|
|
'claimable_points': 적립 가능 포인트,
|
|
'expires_at': 만료일시
|
|
}
|
|
|
|
보안 정책:
|
|
- token_raw는 DB에 저장하지 않음 (QR 코드에만 포함)
|
|
- token_hash만 DB 저장 (SHA256, 64자)
|
|
- nonce 사용으로 동일 거래도 매번 다른 토큰
|
|
|
|
URL 최적화:
|
|
- nonce: 64자 -> 12자로 단축 (6바이트 = 충분한 보안)
|
|
- 타임스탬프: 제거 (DB에만 저장)
|
|
- 결과: 약 80자 -> 빠른 QR 인식
|
|
"""
|
|
# 1. 랜덤 nonce 생성 (6바이트 = 12자 hex) - 대폭 단축!
|
|
nonce = secrets.token_hex(6)
|
|
|
|
# 2. 타임스탬프 (DB 저장용)
|
|
timestamp = datetime.now().isoformat()
|
|
|
|
# 3. 토큰 원문 생성 (검증용 - DB 저장 X)
|
|
token_raw = f"{transaction_id}:{nonce}:{timestamp}"
|
|
|
|
# 4. SHA256 해시 생성 (DB 저장용)
|
|
token_hash = hashlib.sha256(token_raw.encode('utf-8')).hexdigest()
|
|
|
|
# 5. QR URL 생성 (짧게!) - 타임스탬프 제외
|
|
qr_token_short = f"{transaction_id}:{nonce}"
|
|
qr_url = f"{QR_BASE_URL}?t={qr_token_short}"
|
|
|
|
# 6. 적립 포인트 계산
|
|
claimable_points = calculate_claimable_points(total_amount)
|
|
|
|
# 7. 만료일 계산 (30일 후)
|
|
expires_at = datetime.now() + timedelta(days=TOKEN_EXPIRY_DAYS)
|
|
|
|
return {
|
|
'qr_url': qr_url,
|
|
'token_raw': token_raw, # 전체 토큰 (해시 생성용)
|
|
'token_hash': token_hash,
|
|
'claimable_points': claimable_points,
|
|
'expires_at': expires_at,
|
|
'pharmacy_id': pharmacy_id,
|
|
'transaction_id': transaction_id,
|
|
'total_amount': total_amount,
|
|
'nonce': nonce, # 서버 전송용
|
|
}
|
|
|
|
|
|
def calculate_claimable_points(total_amount):
|
|
"""
|
|
적립 포인트 계산 (3% 정책)
|
|
|
|
Args:
|
|
total_amount (float): 판매 금액
|
|
|
|
Returns:
|
|
int: 적립 포인트 (정수, 소수점 절사)
|
|
"""
|
|
return int(total_amount * MILEAGE_RATE)
|
|
|
|
|
|
def save_token_to_db(transaction_id, token_hash, total_amount, claimable_points,
|
|
expires_at, pharmacy_id):
|
|
"""
|
|
생성된 토큰을 SQLite DB에 저장
|
|
|
|
Args:
|
|
transaction_id (str): 거래 ID
|
|
token_hash (str): SHA256 해시
|
|
total_amount (float): 판매 금액
|
|
claimable_points (int): 적립 포인트
|
|
expires_at (datetime): 만료일시
|
|
pharmacy_id (str): 약국 코드
|
|
|
|
Returns:
|
|
tuple: (성공 여부, 에러 메시지 or None)
|
|
|
|
중복 방지:
|
|
- transaction_id가 이미 존재하면 실패
|
|
- token_hash가 이미 존재하면 실패 (UNIQUE 제약)
|
|
"""
|
|
try:
|
|
from db.dbsetup import db_manager as _db_manager
|
|
conn = _db_manager.get_sqlite_connection()
|
|
cursor = conn.cursor()
|
|
|
|
# 중복 체크 (transaction_id)
|
|
cursor.execute("""
|
|
SELECT id FROM claim_tokens WHERE transaction_id = ?
|
|
""", (transaction_id,))
|
|
|
|
if cursor.fetchone():
|
|
return (False, f"이미 QR이 생성된 거래입니다: {transaction_id}")
|
|
|
|
# INSERT
|
|
cursor.execute("""
|
|
INSERT INTO claim_tokens (
|
|
transaction_id, pharmacy_id, token_hash,
|
|
total_amount, claimable_points, expires_at
|
|
) VALUES (?, ?, ?, ?, ?, ?)
|
|
""", (
|
|
transaction_id,
|
|
pharmacy_id,
|
|
token_hash,
|
|
int(total_amount), # INTEGER 타입
|
|
claimable_points,
|
|
expires_at.strftime('%Y-%m-%d %H:%M:%S')
|
|
))
|
|
|
|
conn.commit()
|
|
|
|
return (True, None)
|
|
|
|
except Exception as e:
|
|
return (False, f"DB 저장 실패: {str(e)}")
|
|
|
|
|
|
def sync_token_to_server(transaction_id, total_amount, pharmacy_code=None, items=None):
|
|
"""
|
|
토큰을 서버(pos.pharmq.kr)에 즉시 전송 (품목 상세 포함)
|
|
|
|
Args:
|
|
transaction_id: 거래 ID
|
|
total_amount: 판매 금액
|
|
pharmacy_code: 약국 코드 (기본값: P0001)
|
|
items: 품목 리스트 [{'item_code': ..., 'item_name': ..., 'quantity': ..., 'unit_price': ..., 'total_price': ...}]
|
|
|
|
Returns:
|
|
tuple: (성공 여부, 서버 응답 or 에러 메시지)
|
|
"""
|
|
pharmacy_code = pharmacy_code or PHARMACY_CODE
|
|
|
|
payload = {
|
|
'pharmacy_code': pharmacy_code,
|
|
'transaction_id': str(transaction_id),
|
|
'total_amount': int(total_amount),
|
|
}
|
|
|
|
# 품목 상세 추가
|
|
if items:
|
|
payload['items'] = items
|
|
|
|
try:
|
|
response = requests.post(
|
|
f"{CLOUD_API_URL}/api/v1/tokens/create",
|
|
json=payload,
|
|
timeout=5
|
|
)
|
|
|
|
if response.ok:
|
|
result = response.json()
|
|
logger.info(f"[QR] 서버 전송 성공: {transaction_id} → {result.get('points', 0)}P")
|
|
return (True, result)
|
|
else:
|
|
logger.warning(f"[QR] 서버 응답 오류: {response.status_code} - {response.text[:100]}")
|
|
return (False, f"서버 오류: {response.status_code}")
|
|
|
|
except requests.Timeout:
|
|
logger.warning(f"[QR] 서버 타임아웃 (오프라인?): {transaction_id}")
|
|
return (False, "타임아웃")
|
|
except Exception as e:
|
|
logger.warning(f"[QR] 서버 전송 실패: {e}")
|
|
return (False, str(e))
|
|
|
|
|
|
def generate_and_sync_token(transaction_id, total_amount, pharmacy_id="P0001", items=None):
|
|
"""
|
|
토큰 생성 + 로컬 저장 + 서버 즉시 전송 (통합 함수)
|
|
|
|
Args:
|
|
transaction_id: 거래 ID
|
|
total_amount: 판매 금액
|
|
pharmacy_id: 약국 코드
|
|
items: 품목 리스트 [{'item_code': ..., 'item_name': ..., 'quantity': ..., 'unit_price': ..., 'total_price': ...}]
|
|
|
|
Returns:
|
|
dict: 토큰 정보 + synced 플래그
|
|
"""
|
|
# 1. 토큰 생성
|
|
token_info = generate_claim_token(transaction_id, total_amount, pharmacy_id)
|
|
|
|
# 2. 로컬 DB 저장
|
|
local_success, local_error = save_token_to_db(
|
|
transaction_id,
|
|
token_info['token_hash'],
|
|
total_amount,
|
|
token_info['claimable_points'],
|
|
token_info['expires_at'],
|
|
pharmacy_id
|
|
)
|
|
|
|
token_info['local_saved'] = local_success
|
|
if not local_success:
|
|
token_info['local_error'] = local_error
|
|
|
|
# 3. ⚡ 서버 즉시 전송 (품목 포함)
|
|
sync_success, sync_result = sync_token_to_server(
|
|
transaction_id, total_amount, pharmacy_id, items=items
|
|
)
|
|
|
|
token_info['synced'] = sync_success
|
|
if sync_success and isinstance(sync_result, dict):
|
|
token_info['server_token_id'] = sync_result.get('token_id')
|
|
token_info['server_points'] = sync_result.get('points')
|
|
|
|
return token_info
|
|
|
|
|
|
# 테스트 코드
|
|
if __name__ == "__main__":
|
|
# 테스트
|
|
test_tx_id = "20251024000042"
|
|
test_amount = 50000.0
|
|
|
|
print("=" * 80)
|
|
print("Claim Token 생성 테스트")
|
|
print("=" * 80)
|
|
|
|
token_info = generate_claim_token(test_tx_id, test_amount)
|
|
|
|
print(f"거래 ID: {test_tx_id}")
|
|
print(f"판매 금액: {test_amount:,}원")
|
|
print(f"적립 포인트: {token_info['claimable_points']}P")
|
|
print(f"토큰 원문: {token_info['token_raw'][:80]}...")
|
|
print(f"토큰 해시: {token_info['token_hash']}")
|
|
print(f"QR URL: {token_info['qr_url'][:80]}...")
|
|
print(f"만료일: {token_info['expires_at']}")
|
|
print("=" * 80)
|
|
|
|
# DB 저장 테스트
|
|
print("\nDB 저장 테스트...")
|
|
success, error = save_token_to_db(
|
|
test_tx_id,
|
|
token_info['token_hash'],
|
|
test_amount,
|
|
token_info['claimable_points'],
|
|
token_info['expires_at'],
|
|
token_info['pharmacy_id']
|
|
)
|
|
|
|
if success:
|
|
print("[OK] DB 저장 성공")
|
|
print(f"\nSQLite DB 경로: backend/db/mileage.db")
|
|
print("다음 명령으로 확인:")
|
|
print(" sqlite3 backend/db/mileage.db \"SELECT * FROM claim_tokens\"")
|
|
else:
|
|
print(f"[ERROR] {error}")
|