Initial commit: Flask stats API (v1 PharmIT3000, v2 PMPLUS20)
This commit is contained in:
청춘약국
1
queries/__init__.py
Normal file
1
queries/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
200
queries/v1_pharmit3000.py
Normal file
200
queries/v1_pharmit3000.py
Normal file
@@ -0,0 +1,200 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
PharmIT3000 통계 쿼리 (v1)
|
||||
|
||||
테이블:
|
||||
- PM_PRES.PS_MAIN: 처방 헤더
|
||||
- PM_PRES.PS_SUB_PHARM: 처방 상세
|
||||
- PM_PRES.CD_SUNAB: 수납 정보
|
||||
- PM_DRUG.CD_GOODS: 약품 마스터
|
||||
|
||||
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,
|
||||
m.PRICE_N,
|
||||
ISNULL(n.Appr_Gubun, '') AS Appr_Gubun,
|
||||
ISNULL(n.nAPPROVAL_NUM, '') AS nAPPROVAL_NUM
|
||||
FROM PM_PRES..PS_MAIN m
|
||||
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)
|
||||
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
|
||||
|
||||
# 급여/비급여 분리
|
||||
if gubun == '9':
|
||||
ins_prep, ins_drug = 0, 0
|
||||
nonins_prep = max(0, price_n - drug_t4 - price_p) if drug_t4 > 0 else 0
|
||||
nonins_drug = drug_t4
|
||||
else:
|
||||
ins_prep, ins_drug = s_prep, price_t - s_prep
|
||||
nonins_prep = max(0, price_n - drug_t4 - price_p) if drug_t4 > 0 else 0
|
||||
nonins_drug = drug_t4
|
||||
|
||||
args = (sales_amt, ins_prep, ins_drug, nonins_prep, nonins_drug, 0, price_c, price_p, price_n)
|
||||
|
||||
# 전체
|
||||
_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)
|
||||
|
||||
|
||||
def _add(d, sales_amt, ins_prep, ins_drug, nonins_prep, nonins_drug,
|
||||
nonins_margin, claim_amt, copay, receipt):
|
||||
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
|
||||
|
||||
|
||||
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
|
||||
207
queries/v2_pmplus20.py
Normal file
207
queries/v2_pmplus20.py
Normal file
@@ -0,0 +1,207 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
PMPLUS20 통계 쿼리 (v2)
|
||||
|
||||
테이블 매핑 (PharmIT3000 → PMPLUS20):
|
||||
- PM_PRES.PS_MAIN → PM_MAIN.TBSIM040_28
|
||||
- PM_PRES.PS_SUB_PHARM → PM_MAIN.TBSIM040_29
|
||||
- PM_PRES.CD_SUNAB → PM_MAIN.TBSIR000_01
|
||||
- PM_DRUG.CD_GOODS → PM_MAIN.TBSIM040_01
|
||||
|
||||
주의:
|
||||
- DrugCode vs Drug_Code 대소문자 차이
|
||||
- TBSIR000_01.RECP_DT는 datetime (변환 필요)
|
||||
"""
|
||||
import pyodbc
|
||||
from datetime import date
|
||||
from config import PMPLUS20_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:
|
||||
"""
|
||||
매출 통계 조회 (PMPLUS20 버전)
|
||||
|
||||
Args:
|
||||
date_from: 시작일 (YYYYMMDD)
|
||||
date_to: 종료일 (YYYYMMDD)
|
||||
|
||||
Returns:
|
||||
dict: {total, by_gubun, by_age, by_time, by_pay, by_hosp}
|
||||
|
||||
Note:
|
||||
컬럼명/테이블명은 실제 PMPLUS20 스키마 확인 후 수정 필요
|
||||
"""
|
||||
# TODO: 실제 PMPLUS20 스키마 확인 후 쿼리 수정
|
||||
# 현재는 PharmIT3000과 동일한 컬럼명 가정
|
||||
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,
|
||||
m.PRICE_N,
|
||||
ISNULL(n.Appr_Gubun, '') AS Appr_Gubun,
|
||||
ISNULL(n.nAPPROVAL_NUM, '') AS nAPPROVAL_NUM
|
||||
FROM PM_MAIN..TBSIM040_28 m
|
||||
LEFT JOIN PM_MAIN..TBSIR000_01 n ON n.PRESERIAL = m.PreSerial
|
||||
WHERE m.INDATE BETWEEN ? AND ?
|
||||
"""
|
||||
|
||||
try:
|
||||
rows = query(sql, (date_from, date_to))
|
||||
except Exception as e:
|
||||
return {
|
||||
'error': str(e),
|
||||
'note': 'PMPLUS20 스키마 확인 필요. 컬럼명/테이블명이 다를 수 있음.'
|
||||
}
|
||||
|
||||
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)
|
||||
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
|
||||
|
||||
if gubun == '9':
|
||||
ins_prep, ins_drug = 0, 0
|
||||
nonins_prep = max(0, price_n - drug_t4 - price_p) if drug_t4 > 0 else 0
|
||||
nonins_drug = drug_t4
|
||||
else:
|
||||
ins_prep, ins_drug = s_prep, price_t - s_prep
|
||||
nonins_prep = max(0, price_n - drug_t4 - price_p) if drug_t4 > 0 else 0
|
||||
nonins_drug = drug_t4
|
||||
|
||||
args = (sales_amt, ins_prep, ins_drug, nonins_prep, nonins_drug, 0, price_c, price_p, price_n)
|
||||
|
||||
_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)
|
||||
|
||||
|
||||
def _add(d, sales_amt, ins_prep, ins_drug, nonins_prep, nonins_drug,
|
||||
nonins_margin, claim_amt, copay, receipt):
|
||||
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
|
||||
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user