thug0bin/pharmacy-stats-api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(v1): QT-POS sales_stats_dialog.py 와 1:1 동기화

Browse Source 
pharmon-web/sales_stats_dialog.py 의 _query / _build_margin_cache
로직을 그대로 이식하여 QT-POS 와 stats-api 의 통계 수치 불일치
문제를 해결.

주요 변경:
- CD_SUNAB LEFT JOIN → OUTER APPLY TOP 1
  (1:N 곱증으로 행 중복되어 매출 이중계산되던 버그 제거)
- PS_SUB_BOJO JOIN 추가 (보훈 GITA_GUBUN 세분화 지원)
- SELECT 확장: S_FASTMON, S_TEMP3, SE_BOHUN_C, DRUG_T1~T3,
  S_S_PHOL_0~3, SUGA_ZE_PRICE, GITA_GUBUN
- GPPOS2 공식으로 재작성
  sales_amt = PRICE_C + PRICE_P + S_FASTMON + S_TEMP3
            + SE_PRICE_C + SE_PRICE_P + SE_BOHUN_C
- 보훈(PreGubun='4') + GITA_GUBUN ∈ {1,2,3,4} → '4_1'~'4_4'
- 자동차보험 할증금액, 선별급여 청구액 (SE_PRICE_P*0.25) 반영
- 비급여 마진 캐시 (PS_SUB_PHARM × WH_sub/CD_GOODS)
- config.py: DB 서버 192.168.0.201 → 192.168.0.69 (테스트 서버)

2026-04-08 1/3 스크린샷 기준 QT-POS 와 1원 단위까지 수치 일치 확인.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
청춘약국
2026-04-08 20:53:46 +09:00
parent 34374c5626
commit ef23830a77
2 changed files with 253 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

344
queries/v1_pharmit3000.py
View File

@@ -2,18 +2,33 @@
"""
PharmIT3000 통계 쿼리 (v1)
테이블:
- PM_PRES.PS_MAIN: 처방 헤더
- PM_PRES.PS_Main_Sub: 처방 부가정보 (선별급여/상한제 등)
- PM_PRES.PS_SUB_PHARM: 처방 상세
- PM_PRES.CD_SUNAB: 수납 정보
- PM_DRUG.CD_GOODS: 약품 마스터
**QT-POS sales_stats_dialog.py 와 완전 동기화된 버전** (2026-04-08 rewrite).
주의:
- PRICE_N은 선별급여/상한제초과 미포함 (버그)
- PS_Main_Sub.SE_PRICE_P를 합산해야 진짜 수납액
기준 파일: pharmon-web/sales_stats_dialog.py (_query / _build_margin_cache / _calc_age / _empty_row / _add)
QT-POS sales_stats_dialog.py 기반
변경 포인트 (구버전 대비):
1. CD_SUNAB LEFT JOIN → OUTER APPLY TOP 1 (1:N 곱증으로 행 중복 → 금액 이중계산되던 버그 제거)
2. PS_SUB_BOJO JOIN 추가 (보훈 세분화 GITA_GUBUN)
3. SELECT 컬럼 확장: S_FASTMON, S_TEMP3, SE_BOHUN_C, DRUG_T1~T3,
S_S_PHOL_0~3, SUGA_ZE_PRICE, GITA_GUBUN
4. 매출공식 GPPOS2 기준으로 재작성
- sales_amt = PRICE_C + PRICE_P + S_FASTMON + S_TEMP3 + SE_PRICE_C + SE_PRICE_P + SE_BOHUN_C
- ins_prep = S_Prep + S_S_PHOL_0~3 + SUGA_ZE_PRICE
- ins_drug = ((DRUG_T1+T2+T3 + SE_PRICE_C + SE_BOHUN_C) // 10) * 10 (10원 절사)
- nonins_prep= S_FASTMON - Drug_T4 (S_FASTMON>0 일 때)
- nonins_drug= Drug_T4
- claim_amt = PRICE_C + SE_PRICE_P * 0.25 (선별급여 보험부담 20%)
5. 보훈(PreGubun='4') + GITA_GUBUN ∈ {1,2,3,4} → '4_1'~'4_4' 세분화 키
6. 자동차보험(PreGubun='3') 할증금액 car_surcharge = PRICE_T - (ins_drug + ins_prep)
7. 비급여 마진 = Drug_T4 - 입고총액 (PS_SUB_PHARM × WH_sub/CD_GOODS 캐시)
8. 결제수단 분류
- card : Appr_Gubun IN ('1','5','9')
- paper: nAPPROVAL_NUM 보유 (현금영수증 번호)
- cash : 나머지
9. 시간가산 분류
- hd_add (Holiday IN '2','4') → saturday(공휴가산)
- overtime (Holiday IN '3','4') → overtime (hd_add 없고 overtime인 경우)
- normal (그 외)
"""
import pyodbc
from datetime import date
@@ -34,26 +49,105 @@ def get_connection():
def query(sql, params=None):
"""쿼리 실행 후 dict 리스트 반환"""
"""쿼리 실행 후 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
try:
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()]
return rows
finally:
conn.close()
# ─── 비급여 마진 캐시 ──────────────────────────────────────────
def _build_margin_cache(rows) -> dict:
"""비급여 처방들의 입고총액을 계산하여 캐시로 반환.
마진 = Drug_T4 - cost_cache[serial]
QT-POS sales_stats_dialog.py::_build_margin_cache 동일 로직.
- PS_SUB_PHARM.UnitCode != 1 (비급여 약품)만
- 입고가 우선순위: WH_sub 최근 입고가 → CD_GOODS.Price
(최근 입고가가 판매가의 5배 이상이면 데이터 오류로 보고 CD_GOODS.Price 사용)
- SQL Server 파라미터 제한(2100) 고려하여 500건 청크 처리
"""
noncov_serials = [r['PreSerial'] for r in rows if (r.get('Drug_T4') or 0) > 0]
if not noncov_serials:
return {}
CHUNK_SIZE = 500
sub_rows = []
for i in range(0, len(noncov_serials), CHUNK_SIZE):
chunk = noncov_serials[i:i + CHUNK_SIZE]
placeholders = ','.join(['?'] * len(chunk))
sub_sql = f"""
SELECT s.PreSerial, s.DrugCode, s.USERPRICE, s.QUAN, s.QUAN_TIME, s.Days
FROM PM_PRES..PS_SUB_PHARM s
WHERE s.PreSerial IN ({placeholders})
AND s.UnitCode != 1
"""
sub_rows.extend(query(sub_sql, tuple(chunk)))
drug_codes = list({r['DrugCode'] for r in sub_rows if r.get('DrugCode')})
if not drug_codes:
return {}
cost_rows = []
for i in range(0, len(drug_codes), CHUNK_SIZE):
chunk = drug_codes[i:i + CHUNK_SIZE]
placeholders = ','.join(['?'] * len(chunk))
cost_sql = f"""
SELECT g.DrugCode, g.Price AS GoodsPrice,
(SELECT TOP 1 WH_MY_unit_a FROM PM_DRUG..WH_sub w
WHERE w.DrugCode = g.DrugCode
ORDER BY w.WH_DT_appl DESC) AS LastCost
FROM PM_DRUG..CD_GOODS g
WHERE g.DrugCode IN ({placeholders})
"""
cost_rows.extend(query(cost_sql, tuple(chunk)))
cost_map = {}
for c in cost_rows:
code = c['DrugCode']
last_cost = float(c.get('LastCost') or 0)
goods_price = float(c.get('GoodsPrice') or 0)
cost_map[code] = {'last_cost': last_cost, 'goods_price': goods_price}
cost_cache: dict = {}
for s in sub_rows:
serial = s['PreSerial']
code = s['DrugCode']
user_price = float(s.get('USERPRICE') or 0)
qty = float(s.get('QUAN') or 0)
qty_time = float(s.get('QUAN_TIME') or 1)
days = float(s.get('Days') or 1)
info = cost_map.get(code, {})
last_cost = info.get('last_cost', 0)
goods_price = info.get('goods_price', 0)
if last_cost > user_price * 5 and goods_price > 0:
cost_price = goods_price
elif last_cost > 0:
cost_price = last_cost
else:
cost_price = goods_price
cost_cache[serial] = cost_cache.get(serial, 0) + cost_price * qty * qty_time * days
return cost_cache
# ─── 메인 통계 ────────────────────────────────────────────────
def get_sales_stats(date_from: str, date_to: str) -> dict:
"""
매출 통계 조회 (QT-POS sales_stats_dialog.py 기준)
Args:
date_from: 시작일 (YYYYMMDD)
date_to: 종료일 (YYYYMMDD)
"""매출 통계 조회. QT-POS sales_stats_dialog.py::_query() 이식.
Returns:
dict: {total, by_gubun, by_age, by_time, by_pay, by_hosp}
{total, by_gubun, by_age, by_time, by_pay, by_hosp}
"""
sql = """
SELECT
@@ -67,129 +161,191 @@ def get_sales_stats(date_from: str, date_to: str) -> dict:
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,
-- 수납금액 = PRICE_N + 선별급여/상한제초과
(m.PRICE_N + ISNULL(sub.SE_PRICE_P, 0)) AS PRICE_N,
-- GPPOS 매출계산용
ISNULL(m.S_FASTMON, 0) AS S_FASTMON,
ISNULL(sub.S_TEMP3, 0) AS S_TEMP3,
ISNULL(sub.SE_PRICE_C, 0) AS SE_PRICE_C,
ISNULL(sub.SE_PRICE_P, 0) AS SE_PRICE_P,
ISNULL(sub.SE_BOHUN_C, 0) AS SE_BOHUN_C,
-- GPPOS 급여약가 계산용 (DRUG_T1 + T2 + T3)
ISNULL(sub.DRUG_T1, 0) AS DRUG_T1,
ISNULL(sub.DRUG_T2, 0) AS DRUG_T2,
ISNULL(sub.DRUG_T3, 0) AS DRUG_T3,
-- 급여조제료 가산 (약국관리료)
ISNULL(sub.S_S_PHOL_0, 0) AS S_S_PHOL_0,
ISNULL(sub.S_S_PHOL_1, 0) AS S_S_PHOL_1,
ISNULL(sub.S_S_PHOL_2, 0) AS S_S_PHOL_2,
ISNULL(sub.S_S_PHOL_3, 0) AS S_S_PHOL_3,
-- 명절/휴일 특수가산
ISNULL(sub.SUGA_ZE_PRICE, 0) AS SUGA_ZE_PRICE,
-- 보훈 세분화용 (1=100%, 2=60%, 3=50%, 4=30%)
ISNULL(bojo.GITA_GUBUN, '') AS GITA_GUBUN,
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
LEFT JOIN PM_PRES..PS_Main_Sub sub ON sub.PreSerial = m.PreSerial
LEFT JOIN PM_PRES..PS_SUB_BOJO bojo ON bojo.PreSerial = m.PreSerial
OUTER APPLY (
SELECT TOP 1 *
FROM PM_PRES..CD_SUNAB WITH (NOLOCK)
WHERE PRESERIAL = m.PreSerial
ORDER BY CASE WHEN ISNULL(APPR_DATE,'') = '' THEN 1 ELSE 0 END,
APPR_DATE DESC
) n
WHERE m.INDATE BETWEEN ? AND ?
"""
rows = query(sql, (date_from, date_to))
cost_cache = _build_margin_cache(rows)
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': {},
'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')
gubun = str(r.get('PreGubun') or '0')
gita_gubun = str(r.get('GITA_GUBUN') or '')
# 보훈 세분화: PreGubun='4' + GITA_GUBUN ∈ {1,2,3,4}
if gubun == '4' and gita_gubun in ('1', '2', '3', '4'):
gubun = f'4_{gita_gubun}'
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)
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)
s_fastmon = int(r.get('S_FASTMON') or 0)
s_temp3 = int(r.get('S_TEMP3') 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)
se_price_p = int(r.get('SE_PRICE_P') or 0)
se_bohun_c = int(r.get('SE_BOHUN_C') or 0)
drug_t1 = int(r.get('DRUG_T1') or 0)
drug_t2 = int(r.get('DRUG_T2') or 0)
drug_t3 = int(r.get('DRUG_T3') or 0)
s_s_phol = (int(r.get('S_S_PHOL_0') or 0) + int(r.get('S_S_PHOL_1') or 0)
+ int(r.get('S_S_PHOL_2') or 0) + int(r.get('S_S_PHOL_3') or 0))
suga_ze_price = int(r.get('SUGA_ZE_PRICE') or 0)
appr_gubun = str(r.get('Appr_Gubun') or '')
nappr_num = str(r.get('nAPPROVAL_NUM') or '').strip()
# ─── GPPOS 공식 ───────────────────────────────────────
sales_amt = price_c + price_p + s_fastmon + s_temp3 + se_price_c + se_price_p + se_bohun_c
calc_nonins_prep = s_fastmon - drug_t4 if s_fastmon > 0 else 0
pre_serial = r.get('PreSerial')
cost_total = cost_cache.get(pre_serial, 0)
nonins_margin = int(drug_t4 - cost_total) if drug_t4 > 0 else 0
ins_prep = s_prep + s_s_phol + suga_ze_price
ins_drug = ((drug_t1 + drug_t2 + drug_t3 + se_price_c + se_bohun_c) // 10) * 10
nonins_prep, nonins_drug = calc_nonins_prep, drug_t4
# 청구액 = PRICE_C + SE_PRICE_P * 0.25 (선별급여 보험부담 20% — SE_PRICE_P는 80% 본인부담)
claim_amt = price_c + int(se_price_p * 0.25)
# 자동차보험 할증금액
if gubun == '3':
base_total = ins_drug + ins_prep
car_surcharge = price_t - base_total if price_t > base_total else 0
else:
car_surcharge = 0
args = (sales_amt, ins_prep, ins_drug,
nonins_prep, nonins_drug, se_price_p, nonins_margin,
claim_amt, price_p, price_n, car_surcharge)
# 전체
_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)
age = _calc_age(str(r.get('PaNum') or ''), 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') # 시간외
# 시간가산별 (Unit_Statistics.pas 기준)
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':
if appr_gubun in ('1', '5', '9'):
_add(result['by_pay']['card'], *args)
elif appr_gubun == '2':
_add(result['by_pay']['cash'], *args)
else:
elif nappr_num:
_add(result['by_pay']['paper'], *args)
else:
_add(result['by_pay']['cash'], *args)
# 병원별
if order_name not in result['by_hosp']:
result['by_hosp'][order_name] = _empty_row()
_add(result['by_hosp'][order_name], *args)
hosp = str(r.get('OrderName') or '').strip() or '(미상)'
if hosp not in result['by_hosp']:
result['by_hosp'][hosp] = _empty_row()
_add(result['by_hosp'][hosp], *args)
return result
def _empty_row():
# ─── 유틸 ─────────────────────────────────────────────────────
def _empty_row() -> dict:
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)
nonins_prep=0, nonins_drug=0, se_price_p=0, nonins_margin=0,
claim_amt=0, copay=0, receipt=0, car_surcharge=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
def _add(d: dict, sales_amt, ins_prep, ins_drug,
nonins_prep, nonins_drug, se_price_p, nonins_margin,
claim_amt, copay, receipt, car_surcharge=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['se_price_p'] += se_price_p
d['nonins_margin'] += nonins_margin
d['claim_amt'] += claim_amt
d['copay'] += copay
d['receipt'] += receipt
d['se_price_p'] += se_price_p
d['claim_amt'] += claim_amt
d['copay'] += copay
d['receipt'] += receipt
d['car_surcharge'] += car_surcharge
def _calc_age(panum: str, ref_date: date) -> int | None:
"""주민번호 → 나이"""
def _calc_age(panum: str, ref_date: date):
"""주민번호 → 나이. 실패 시 None."""
if not panum or len(panum) < 7:
return None
birth6 = panum[:6]