212 lines
6.7 KiB
Python
212 lines
6.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
PharmIT3000 통계 쿼리 (v1)
|
|
|
|
테이블:
|
|
- PM_PRES.PS_MAIN: 처방 헤더
|
|
- PM_PRES.PS_Main_Sub: 처방 부가정보 (선별급여/상한제 등)
|
|
- PM_PRES.PS_SUB_PHARM: 처방 상세
|
|
- PM_PRES.CD_SUNAB: 수납 정보
|
|
- PM_DRUG.CD_GOODS: 약품 마스터
|
|
|
|
주의:
|
|
- PRICE_N은 선별급여/상한제초과 미포함 (버그)
|
|
- PS_Main_Sub.SE_PRICE_P를 합산해야 진짜 수납액
|
|
|
|
QT-POS sales_stats_dialog.py 기반
|
|
"""
|
|
import pyodbc
|
|
from datetime import date
|
|
from config import PHARMIT3000_CONFIG as CFG
|
|
|
|
|
|
def get_connection():
|
|
conn_str = (
|
|
f"DRIVER={{ODBC Driver 17 for SQL Server}};"
|
|
f"SERVER={CFG['server']};"
|
|
f"DATABASE={CFG['database']};"
|
|
f"UID={CFG['username']};"
|
|
f"PWD={CFG['password']};"
|
|
f"TrustServerCertificate=yes;"
|
|
f"Connection Timeout=30;"
|
|
)
|
|
return pyodbc.connect(conn_str)
|
|
|
|
|
|
def query(sql, params=None):
|
|
"""쿼리 실행 후 dict 리스트 반환"""
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute(sql, params or ())
|
|
columns = [col[0] for col in cursor.description]
|
|
rows = [dict(zip(columns, row)) for row in cursor.fetchall()]
|
|
conn.close()
|
|
return rows
|
|
|
|
|
|
def get_sales_stats(date_from: str, date_to: str) -> dict:
|
|
"""
|
|
매출 통계 조회 (QT-POS sales_stats_dialog.py 기준)
|
|
|
|
Args:
|
|
date_from: 시작일 (YYYYMMDD)
|
|
date_to: 종료일 (YYYYMMDD)
|
|
|
|
Returns:
|
|
dict: {total, by_gubun, by_age, by_time, by_pay, by_hosp}
|
|
"""
|
|
sql = """
|
|
SELECT
|
|
m.PreSerial,
|
|
m.PreGubun,
|
|
m.PaNum,
|
|
m.OrderName,
|
|
m.Holiday,
|
|
m.PRICE_T,
|
|
m.Drug_T4,
|
|
m.S_Prep,
|
|
m.PRICE_C,
|
|
m.PRICE_P,
|
|
-- 진짜 수납액: PRICE_N + 선별급여/상한제초과 (PS_Main_Sub)
|
|
(m.PRICE_N + ISNULL(s.SE_PRICE_P, 0)) AS PRICE_N,
|
|
-- 선별급여 (QT-POS 동기화)
|
|
ISNULL(s.SE_PRICE_P, 0) AS SE_PRICE_P,
|
|
ISNULL(s.SE_PRICE_C, 0) AS SE_PRICE_C,
|
|
ISNULL(n.Appr_Gubun, '') AS Appr_Gubun,
|
|
ISNULL(n.nAPPROVAL_NUM, '') AS nAPPROVAL_NUM
|
|
FROM PM_PRES..PS_MAIN m
|
|
LEFT JOIN PM_PRES..PS_Main_Sub s ON s.PreSerial = m.PreSerial
|
|
LEFT JOIN PM_PRES..CD_SUNAB n ON n.PRESERIAL = m.PreSerial
|
|
WHERE m.INDATE BETWEEN ? AND ?
|
|
"""
|
|
rows = query(sql, (date_from, date_to))
|
|
|
|
ref_date = date(
|
|
int(date_to[:4]),
|
|
int(date_to[4:6]),
|
|
int(date_to[6:8]),
|
|
)
|
|
|
|
# 집계
|
|
result = {
|
|
'total': _empty_row(),
|
|
'by_gubun': {},
|
|
'by_age': {'old': _empty_row(), 'infant': _empty_row(), 'mid': _empty_row()},
|
|
'by_time': {'overtime': _empty_row(), 'saturday': _empty_row(), 'normal': _empty_row()},
|
|
'by_pay': {'card': _empty_row(), 'cash': _empty_row(), 'paper': _empty_row()},
|
|
'by_hosp': {},
|
|
}
|
|
|
|
for r in rows:
|
|
gubun = str(r.get('PreGubun') or '0')
|
|
holiday = str(r.get('Holiday') or '1')
|
|
price_t = int(r.get('PRICE_T') or 0)
|
|
drug_t4 = int(r.get('Drug_T4') or 0)
|
|
s_prep = int(r.get('S_Prep') or 0)
|
|
price_c = int(r.get('PRICE_C') or 0)
|
|
price_p = int(r.get('PRICE_P') or 0)
|
|
price_n = int(r.get('PRICE_N') or 0)
|
|
se_price_p = int(r.get('SE_PRICE_P') or 0)
|
|
se_price_c = int(r.get('SE_PRICE_C') or 0)
|
|
appr_gubun = str(r.get('Appr_Gubun') or '')
|
|
order_name = str(r.get('OrderName') or '기타')
|
|
panum = str(r.get('PaNum') or '')
|
|
|
|
sales_amt = price_t + drug_t4
|
|
|
|
# 급여/비급여 분리 (비급여에서도 급여조제료/약가 있을 수 있음 - GPPOS 동일)
|
|
ins_prep, ins_drug = s_prep, price_t - s_prep
|
|
nonins_prep = max(0, price_n - drug_t4 - se_price_p - price_p) if drug_t4 > 0 else 0
|
|
nonins_drug = drug_t4
|
|
|
|
# 청구액 = PRICE_C + SE_PRICE_C (선별급여 보험부담분)
|
|
claim_amt = price_c + se_price_c
|
|
|
|
args = (sales_amt, ins_prep, ins_drug, nonins_prep, nonins_drug, 0, claim_amt, price_p, price_n, se_price_p)
|
|
|
|
# 전체
|
|
_add(result['total'], *args)
|
|
|
|
# 보험별
|
|
if gubun not in result['by_gubun']:
|
|
result['by_gubun'][gubun] = _empty_row()
|
|
_add(result['by_gubun'][gubun], *args)
|
|
|
|
# 연령가산별
|
|
age = _calc_age(panum, ref_date)
|
|
if age is None or (6 <= age < 65):
|
|
_add(result['by_age']['mid'], *args)
|
|
elif age >= 65:
|
|
_add(result['by_age']['old'], *args)
|
|
else:
|
|
_add(result['by_age']['infant'], *args)
|
|
|
|
# 시간가산별
|
|
hd_add = holiday in ('2', '4') # 공휴가산
|
|
overtime = holiday in ('3', '4') # 시간외
|
|
if hd_add:
|
|
_add(result['by_time']['saturday'], *args)
|
|
elif overtime:
|
|
_add(result['by_time']['overtime'], *args)
|
|
else:
|
|
_add(result['by_time']['normal'], *args)
|
|
|
|
# 결제수단별
|
|
if appr_gubun == '1':
|
|
_add(result['by_pay']['card'], *args)
|
|
elif appr_gubun == '2':
|
|
_add(result['by_pay']['cash'], *args)
|
|
else:
|
|
_add(result['by_pay']['paper'], *args)
|
|
|
|
# 병원별
|
|
if order_name not in result['by_hosp']:
|
|
result['by_hosp'][order_name] = _empty_row()
|
|
_add(result['by_hosp'][order_name], *args)
|
|
|
|
return result
|
|
|
|
|
|
def _empty_row():
|
|
return dict(cnt=0, sales_amt=0, ins_prep=0, ins_drug=0,
|
|
nonins_prep=0, nonins_drug=0, nonins_margin=0,
|
|
claim_amt=0, copay=0, receipt=0, se_price_p=0)
|
|
|
|
|
|
def _add(d, sales_amt, ins_prep, ins_drug, nonins_prep, nonins_drug,
|
|
nonins_margin, claim_amt, copay, receipt, se_price_p=0):
|
|
d['cnt'] += 1
|
|
d['sales_amt'] += sales_amt
|
|
d['ins_prep'] += ins_prep
|
|
d['ins_drug'] += ins_drug
|
|
d['nonins_prep'] += nonins_prep
|
|
d['nonins_drug'] += nonins_drug
|
|
d['nonins_margin'] += nonins_margin
|
|
d['claim_amt'] += claim_amt
|
|
d['copay'] += copay
|
|
d['receipt'] += receipt
|
|
d['se_price_p'] += se_price_p
|
|
|
|
|
|
def _calc_age(panum: str, ref_date: date) -> int | None:
|
|
"""주민번호 → 나이"""
|
|
if not panum or len(panum) < 7:
|
|
return None
|
|
birth6 = panum[:6]
|
|
gender = panum[6]
|
|
if not birth6.isdigit():
|
|
return None
|
|
yy, mm, dd = int(birth6[:2]), int(birth6[2:4]), int(birth6[4:6])
|
|
if gender in ('1', '2'):
|
|
year = 1900 + yy
|
|
elif gender in ('3', '4'):
|
|
year = 2000 + yy
|
|
else:
|
|
year = 1900 + yy
|
|
try:
|
|
bday = date(year, mm, dd)
|
|
return ref_date.year - bday.year - (
|
|
(ref_date.month, ref_date.day) < (bday.month, bday.day))
|
|
except ValueError:
|
|
return None
|