56a6e2da93
- src/dist 산출물 분리 원칙 정리(.gitignore, .gitattributes) - 루트 및 주요 폴더(config/scripts/prompts/tests/src, 런타임 폴더 5종)에 안내용 README.md 추가 - CHANGELOG.md, LICENSE, docs/ops/05-release-and-versioning.md 추가 - docs/README.md 문서 지도 갱신
177 KiB
177 KiB
xlsx 리포트 완전 스펙 (시트·연동·서식·시각화)
이 문서의 역할: DMF Crawler 가 매일 06:00 에 산출하는 최종 결과물
DMF_리포트_YYYY-MM-DD.xlsx를 셀 단위로 확정한다. 시트 8종의 컬럼·너비·표시형식·조건부서식·수식, 대시보드의 셀 주소별 레이아웃, 시트 간 하이퍼링크·정의된 이름, 디자인 토큰(HEX·폰트·행높이), 그리고src/dmf_crawler/report/전 모듈의 구현 골격을 담는다. 이 문서를 읽고 구현하면 리포트가 그대로 나와야 한다. 상위 SSOT 는docs/design/01-architecture.md(구조·모듈명·설정키)와docs/design/00-DATA-SOURCE-DECISION.md(원천 7필드)이며, 근거는docs/research/06-xlsx-linking-and-formatting.md·07-pharma-excel-dashboard-design.md에 있다.
0. 한눈에 보기
이 문서가 확정하는 것:
- 쓰기 엔진은
XlsxWriter단독. openpyxl 은 런타임 의존성에서 완전히 제외한다. 스파크라인 API 가 openpyxl 에 없고, diff 원본은 xlsx 가 아니라 SQLite 이므로 "어제 파일 읽기" 요구 자체가 소멸했다. 산출물 검증은 stdlibzipfile로 한다. - 시트 8종, 이 순서로 고정:
00_대시보드→01_오늘변경분→02_전체현황→03_성분별→04_업체별→05_워치리스트(빈 경우 생략) →06_추이→99_메타. 아키텍처 SSOT §3.13 과 1:1 일치한다. - 대시보드는 스크롤 없는 1화면. 총 폭 ≈1,257px / 총 높이 ≈794px,
set_zoom(90)적용 시 1,131 × 715px → 1366×768 노트북에서 가로·세로 스크롤 모두 없음. - 모든 숫자는 파이썬이 미리 계산해 넣는다. 수식을 쓰는 곳은 전부
write_formula(..., value=캐시값)로 캐시값을 함께 기록한다. 파일을 Excel 로 한 번도 열지 않아도 값이 보인다. - 색은 Okabe-Ito 고정, 상태는 4중 코딩(배경색 + 폰트색 + 텍스트 라벨 + 기호). 신호등 아이콘셋은 CVD 위험으로 금지,
3_arrows_gray만 허용. 모든 상태 배경/폰트 조합은 WCAG 대비비 7.2 이상(AAA). - 상태 도메인은 3종뿐:
신규/변경/취하. 원천 API 에 등록구분 필드가 없으므로 리서치 07 의연차보고·사전등록상태는 채택하지 않는다(부록에 기록). - 원자적 저장 + 폴백: 같은 볼륨 임시파일 →
zipfile검증 →os.replace. 잠김 시 3회 지수 백오프 후DMF_리포트_YYYY-MM-DD_HHMMSS.xlsx로 폴백하고 WARN 알림을 남긴다. 최신본 고정 링크는 심볼릭 링크가 아니라 원자적 복사다(Windows 심볼릭 링크는 권한이 필요). - AI 산출물은 리포트를 막지 못한다.
enrichment조인이 비면 헤드라인은(AI 요약 없음), 메모 열은-로 렌더링하고 리포트는 항상 완주한다. src/report/workbook.py는 존재하지 않는다. 아키텍처 SSOT 가 확정한 분할 —report/{build,data,theme,widgets,atomic}.py+report/sheets/s00~s99— 을 따른다(§8.0 에서 매핑을 명시).- 인쇄는 A4 가로 1페이지 너비 맞춤, 데이터 시트는 헤더 3행을 매 페이지 반복(
repeat_rows(0, 2)).
1. 라이브러리 확정
1.1 결론
쓰기 =
XlsxWriter단독. openpyxl·pandas 는 이 프로젝트의 런타임 의존성이 아니다.
아키텍처 SSOT §8.1 의 런타임 의존성 3개(httpx, XlsxWriter, jsonschema)와 정확히 일치한다.
1.2 요구 대조표
| 리포트 요구 | XlsxWriter | openpyxl | 판정 |
|---|---|---|---|
| 스파크라인(KPI 타일 6개 하단) | ✅ add_sparkline() — line/column/win_loss |
❌ API 자체가 없음 | XlsxWriter 필수 |
| 누적 세로 막대 차트(3계열) | ✅ add_chart({'type':'column','subtype':'stacked'}) |
✅ | 둘 다 가능 |
| 가로 막대 차트 + 데이터 라벨 | ✅ data_labels, set_size |
✅(라벨 옵션 빈약) | XlsxWriter 우위 |
| Excel 표(ListObject) + 합계행 | ✅ add_table(total_row=True, total_function, total_value) |
수동 구현 | XlsxWriter 우위 |
| 수식 캐시값 지정 | ✅ write_formula(cell, f, fmt, value) |
❌ 불가 | XlsxWriter 필수 |
| 조건부서식 18종 | ✅ conditional_format() |
✅ | 둘 다 가능 |
| 내부 하이퍼링크 | ✅ write_url("internal:'시트'!A1") |
✅ | 둘 다 가능 |
| 정의된 이름 | ✅ define_name() |
✅ | 둘 다 가능 |
| 인쇄·탭색·틀고정 | ✅ | ✅ | 둘 다 가능 |
| 기존 파일 수정 | ❌ 신규 생성 전용 | ✅(단 차트·스파크라인 손실) | 해당 없음 — §1.3 |
1.3 openpyxl 을 뺀 근거 3가지
- 스파크라인이 요구다. 요구 R3.3(시각화) 과 대시보드 KPI 타일 설계가 스파크라인을 전제한다. openpyxl 에는 API 가 없다. 이 한 줄로 엔진은 결정된다.
- "어제 파일을 읽어 diff" 요구가 소멸했다. 아키텍처 ADR 이 diff 원본을 SQLite 이벤트 소싱으로 확정했다(
repo.load_snapshot). 리포트 xlsx 는 사람이 열어 편집·오염시킬 수 있는 파생물이지 정본이 아니다. 따라서 xlsx 읽기 기능이 필요 없다. - 차트·스파크라인 손실 함정을 원천 차단한다. openpyxl 공식 문서: "openpyxl does currently not read all possible items in an Excel file so shapes will be lost from existing files if they are opened and saved with the same name." 매일 새 파일을 만드는 우리 정책에서는 애초에 열고 저장할 일이 없다. 의존성을 두면 언젠가 누군가
load_workbook → save를 쓴다. 없애는 것이 가장 확실한 방어다.
1.4 openpyxl 이 담당했을 두 역할의 대체안
| 원래 역할 | 대체 |
|---|---|
| 산출물 xlsx 무결성 검증 | stdlib zipfile 로 testzip() + xl/workbook.xml 존재 확인 (report/atomic.py::verify_xlsx) |
| 워치리스트 사용자 편집 왕복(이월) | SSOT 를 SQLite watchlist 테이블로 확정. 05_워치리스트 시트는 읽기 전용 미러이며 시트 보호를 건다. 편집은 설정 GUI(gui/app.py)에서 한다 |
1.5 금지 사항 (구현 시 위반하면 리뷰에서 반려)
constant_memory: True금지. XlsxWriter 문서 원문: "tables aren't available in XlsxWriter when Workbook 'constant_memory' mode is enabled."02_전체현황이 Excel 표를 쓰므로 켤 수 없다. 현재 규모(수천~수만 행)에서 메모리 문제는 없다.worksheet.autofit()금지. Calibri 11 메트릭 추정이라 한글에서 좁게 나온다. §8.3 의display_width()로 직접 계산한다.- 수식을 캐시값 없이 쓰는 것 금지. 예외는
add_table의total_function인데, 이때는 반드시total_value를 함께 준다. - LibreOffice headless 재계산 금지(아키텍처 부록에서 기각). 외부 프로그램 의존을 늘리지 않는다.
2. 워크북 전체 구성
2.1 시트 목록·순서·탭 색
| # | 시트명 | 탭 색 | 목적 1줄 | 생략 조건 |
|---|---|---|---|---|
| 1 | 00_대시보드 |
#0072B2 |
5초 안에 오늘 상황을 파악하는 1화면 요약. 활성 시트 | 없음 |
| 2 | 01_오늘변경분 |
#D55E00 |
전일 스냅샷 대비 신규/변경/취하만 모은 리포트 본체 | 없음(0건이면 안내행 1줄) |
| 3 | 02_전체현황 |
#5B6770 |
유효 등록 전체 원장. 사용자가 필터로 파고드는 시트 | 없음 |
| 4 | 03_성분별 |
#009E73 |
성분(원료명) 기준 Top N 집계 + 30일 추이 | 없음 |
| 5 | 04_업체별 |
#009E73 |
업체(신청인)·제조국 기준 Top N 집계 | 없음 |
| 6 | 05_워치리스트 |
#CC79A7 |
관심 키워드 히트 현황(읽기 전용 미러) | watchlist 활성 항목 0건이면 시트 자체를 만들지 않음 |
| 7 | 06_추이 |
#0072B2 |
일자별 건수 시계열. 모든 스파크라인·차트의 원본 데이터 | 없음 |
| 8 | 99_메타 |
#888888 |
수집 시각·소스·건수 검증·해시·로그 경로·AI 상태 | 없음 |
- 탭 색
#0072B2가00_대시보드와06_추이에 중복되는 것은 의도다. 파란색 = 워크북의 구조/강조 축(대시보드와 그 데이터 원본), 나머지는 의미색이다. - 리서치 07 의
스냅샷·로그시트는 채택하지 않는다. 그 내용(append-only 감사 로그)은 SQLiteevents/fetch_stats테이블이 정본이고, xlsx 에는99_메타의 요약·경로 포인터만 둔다. 5만 행 롤오프 규칙도 함께 소멸.
2.2 워크북 전역 설정
wb = xlsxwriter.Workbook(str(tmp_path), {
"constant_memory": False, # 표(ListObject) 사용을 위해 반드시 False
"strings_to_numbers": False, # 등록번호 '20121228-168-I-169-04' 훼손 방지
"strings_to_urls": False, # 제조소 소재지에 든 문자열이 링크로 오인되지 않게
"nan_inf_to_errors": True,
"use_future_functions": True, # _xlfn 접두 자동 처리
"default_date_format": "yyyy-mm-dd",
"remove_timezone": True,
})
wb.set_size(1500, 900) # 최초 열림 창 크기
wb.set_properties({
"title": f"DMF 일일 모니터링 리포트 {report_date}",
"subject": "원료의약품 등록(DMF) 신규·변경·취하 현황",
"author": "DMF Crawler",
"manager": "",
"company": "",
"category": "규제 인텔리전스",
"keywords": "DMF, 원료의약품, 식약처, 등록현황",
"comments": f"run_id={run_id} / 자동 생성 / 원본: 공공데이터포털 15057075",
"created": datetime_of_run, # datetime 객체
})
| 항목 | 값 | 근거 |
|---|---|---|
| 기본 폰트 | 맑은 고딕 10pt #1F2933 |
report.font 설정키. Windows Vista+ 기본 탑재 |
| 최초 활성 시트 | 00_대시보드 (activate() + set_first_sheet()) |
5초 규칙 |
| 눈금선 | 00_대시보드 만 hide_gridlines(2), 나머지 유지 |
대시보드는 카드 UI, 데이터 시트는 표 |
| 확대 | 00_대시보드 90%, 나머지 100% |
1화면 수납 |
| 값 없음 표기 | 빈 문자열이 아니라 - |
정렬·필터 일관성 |
| 표 스타일 | Table Style Light 11 |
줄무늬 약함 = 데이터 잉크 비율 우선 |
| 표 이름 | T_LEDGER / T_INGREDIENT / T_COMPANY / T_WATCH / T_TREND |
Excel 표 이름 규칙(공백·숫자 시작 불가) |
2.3 시트 공통 레이아웃 규약 (01~06, 99)
| 행 | 내용 |
|---|---|
| 1행 | A1 = ← 대시보드 역링크, B1 = 시트 제목(14pt bold), 우측 끝 = 기준일 YYYY-MM-DD · 생성 HH:MM |
| 2행 | 공백(높이 6) |
| 3행 | 컬럼 헤더 |
| 4행~ | 데이터 |
freeze_panes(3, k)— 0-index 로 "3행까지 고정".k는 시트별로 다르다(§3).- 헤더 셀에는
write_comment()로 컬럼 정의 툴팁을 단다. - 데이터가 0건이면 4행에
(해당 없음)안내 1줄을 회색으로 쓰고 표/조건부서식은 걸지 않는다.
3. 시트별 완전 명세
표기 규약: 정렬 = 셀 가로 정렬, 형식 = num_format 문자열, 폭 = set_column 문자 단위 폭.
3.1 00_대시보드
셀 단위 레이아웃은 §4 에서 별도로 다룬다. 여기서는 시트 속성만 확정한다.
| 항목 | 값 |
|---|---|
| 탭 색 | #0072B2 |
| 눈금선 | hide_gridlines(2) (화면·인쇄 모두 숨김) |
| 확대 | set_zoom(90) |
| 틀 고정 | 없음 (1화면이라 불필요) |
| 자동 필터 | 없음 |
| 활성화 | activate(), set_first_sheet() |
| 인쇄 | A4 가로, fit_to_pages(1, 1), 인쇄 영역 A1:S47 |
3.2 01_오늘변경분 — 리포트 본체
| 항목 | 값 |
|---|---|
| 헤더 행 | 3행 (0-index 2) |
| 데이터 시작 | 4행 (0-index 3) |
freeze_panes |
(3, 5) — 3행 + A~E열 고정 (정렬키·기호·상태·등록번호·성분명까지 따라다님) |
autofilter |
(2, 0, last_row, 18) = A3:S{last} — Excel 표를 쓰지 않고 명시 지정(행 전체 조건부서식이 표 줄무늬와 충돌하지 않게) |
| 정렬 기준 | A(정렬키) 오름차순 → N(워치히트) 내림차순 → E(성분명) 오름차순. 파이썬에서 미리 정렬해 기록한다 |
| 필터 기본값 | 없음(전량 표시) |
| 행 높이 | 데이터 행 30(줄바꿈 2줄 수용), 헤더 32 |
정렬키 값 도메인: 취하=1, 신규=2, 변경=3. 사용자가 임의 정렬한 뒤에도 A열 오름차순으로 원복할 수 있다.
| 열 | 컬럼명 | 타입 | 폭 | 정렬 | 형식 | 조건부서식 / 수식 | 비고 |
|---|---|---|---|---|---|---|---|
| A | 정렬키 |
int | 4 | 가운데 | 0 |
— | 히든 {'hidden': 1} |
| B | 기호 |
str | 4 | 가운데 | @ |
행 규칙에 포함 | ＋ ◆ ✕ |
| C | 상태 |
str | 8 | 가운데 | @ |
행 전체 규칙의 판정 열 | 신규/변경/취하 |
| D | 등록번호 |
str | 24 | 왼쪽 | @ |
내부 링크 → 02_전체현황 해당 행 |
write_url |
| E | 성분명 |
str | 30 | 왼쪽 | @ |
워치 매칭 시 bold(CF-01-06) |
줄바꿈 |
| F | 업체명 |
str | 26 | 왼쪽 | @ |
워치 매칭 시 bold | ENTP_NAME |
| G | 제조소명 |
str | 26 | 왼쪽 | @ |
— | MNFCTR_NAME |
| H | 제조소 소재지 |
str | 40 | 왼쪽 | @ |
— | MNFCTR_PLACE, 줄바꿈 |
| I | 제조국가 |
str | 16 | 가운데 | @ |
대한민국 포함 → #E3F3EA(CF-01-05) |
다중값은 , 결합 |
| J | 발급일자 |
date | 12 | 가운데 | yyyy-mm-dd |
— | DMF_PERMIT_DATE |
| K | 수리일자(파생) |
date | 12 | 가운데 | yyyy-mm-dd |
— | 등록번호 앞 8자리 |
| L | 변경필드 |
str | 24 | 왼쪽 | @ |
줄바꿈 | 성분명, 제조국가 |
| M | 변경내용(전→후) |
str | 52 | 왼쪽 | @ |
줄바꿈, 9pt | 제조국가: 이탈리아 → 이탈리아, 스위스 |
| N | 워치히트 |
int | 8 | 가운데 | 0;;- |
>0 → #F7E9F0 + bold(CF-01-04) |
매칭 개수 |
| O | 워치키워드 |
str | 20 | 왼쪽 | @ |
— | 쉼표 구분 |
| P | AI 메모 |
str | 40 | 왼쪽 | @ |
줄바꿈, 9pt #5B6770 |
없으면 - |
| Q | 원문검색 |
url | 10 | 가운데 | @ |
write_url(외부, string='조회') |
nedrug 성분명 검색 URL |
| R | dmf_key |
str | 26 | 왼쪽 | @ |
— | 히든. 재현·추적용 |
| S | content_hash |
str | 18 | 왼쪽 | @ |
— | 히든 |
Q열 외부 링크 URL 템플릿 (성분명 검색):
https://nedrug.mfds.go.kr/pbp/CCBAC03/getItem?totalPages=1&limit=10&page=1&searchYn=true&itemName={성분명 URL 인코딩}
⚠️ 미검증: 위 쿼리스트링이 성분명 검색으로 정확히 동작하는지는 실측하지 않았다. 실패 시 파라미터 없는 목록 URL
https://nedrug.mfds.go.kr/pbp/CCBAC03로 폴백하도록 구현한다.
행 전체 조건부서식은 B4:S{last} 에 건다(A열 히든 제외). 판정 열은 $C.
3.3 02_전체현황 — 누적 원장
| 항목 | 값 |
|---|---|
| 헤더 행 | 3행 |
| Excel 표 | add_table(2, 0, last_row, 16, {'name':'T_LEDGER','style':'Table Style Light 11','banded_rows':True,'autofilter':True}) |
freeze_panes |
(3, 3) — 3행 + A~C열(등록번호·성분명·업체명) |
| 정렬 기준 | I(수리일자) 내림차순 → A(등록번호) 오름차순 |
| 행 높이 | 15 고정, 줄바꿈 없음(수천~수만 행 성능) |
| 열 그룹화 | P:Q 를 {'level': 1, 'hidden': True} 로 묶어 접기 가능하게 |
| 열 | 컬럼명 | 타입 | 폭 | 정렬 | 형식 | 조건부서식 / 수식 |
|---|---|---|---|---|---|---|
| A | 등록번호 |
str | 24 | 왼쪽 | @ |
duplicate 강조(CF-02-01) |
| B | 성분명 |
str | 30 | 왼쪽 | @ |
워치 매칭 → #F7E9F0(CF-02-05) |
| C | 업체명 |
str | 26 | 왼쪽 | @ |
워치 매칭 → #F7E9F0 |
| D | 제조소명 |
str | 26 | 왼쪽 | @ |
— |
| E | 제조소 소재지 |
str | 44 | 왼쪽 | @ |
헤더 주석(툴팁) |
| F | 제조국가 |
str | 18 | 가운데 | @ |
대한민국 포함 → #E3F3EA(CF-02-04) |
| G | 제조국 수 |
int | 8 | 가운데 | 0 |
>=2 → bold (CF-02-06) |
| H | 발급일자 |
date | 12 | 가운데 | yyyy-mm-dd |
— |
| I | 수리일자 |
date | 12 | 가운데 | yyyy-mm-dd |
data_bar #0072B2(CF-02-07) |
| J | 최초관측일 |
date | 12 | 가운데 | yyyy-mm-dd |
— |
| K | 최종관측일 |
date | 12 | 가운데 | yyyy-mm-dd |
— |
| L | 최종변경일 |
date | 12 | 가운데 | yyyy-mm-dd |
오늘이면 #FDF0DC(CF-02-02) |
| M | 상태 |
str | 8 | 가운데 | @ |
취하 → 행 전체 #FBE5DC+취소선(CF-02-03) |
| N | 경과일 |
int | 8 | 가운데 | #,##0;;- |
수식(아래) + 3_arrows_gray 아이콘셋(CF-02-08) |
| O | 원문검색 |
url | 10 | 가운데 | @ |
외부 링크 조회 |
| P | dmf_key |
str | 26 | 왼쪽 | @ |
히든, 그룹 level 1 |
| Q | content_hash |
str | 18 | 왼쪽 | @ |
히든, 그룹 level 1 |
N열 수식 전문 (행 4 기준, 아래로 상대 복사):
=IF($I4="","-",INT(REPORT_DATE-$I4))
REPORT_DATE 는 §5.4 의 정의된 이름(='99_메타'!$B$3). 캐시값은 파이썬이 (기준일 - 수리일자).days 로 계산해 write_formula(..., value=n) 로 함께 기록한다.
주의: Excel 표(ListObject) 안에서는 컬럼 수식이 구조적 참조로 자동 확장된다. 우리는 캐시값을 넣어야 하므로
add_table의formula옵션을 쓰지 않고 셀마다write_formula로 직접 기록한 뒤 표를 씌운다. 표는 서식·필터만 담당한다.
3.4 03_성분별
| 항목 | 값 |
|---|---|
| 헤더 행 | 3행 |
| Excel 표 | add_table(2, 0, last_row+1, 15, {'name':'T_INGREDIENT','style':'Table Style Light 11','total_row':True}) — 마지막 1행은 합계행 |
freeze_panes |
(3, 2) — 3행 + A~B열(순위·성분명) |
| 정렬 기준 | F(오늘 합계) 내림차순 → I(누적 유효등록) 내림차순 → B(성분명) 오름차순 |
| 행 수 | report.top_n(기본 10). 단 오늘 변경분이 있는 성분은 top_n 을 넘어도 전부 포함한다(합집합) |
| 행 높이 | 18(스파크라인 수용) |
| 열 | 컬럼명 | 타입 | 폭 | 정렬 | 형식 | 조건부서식 / 합계행 |
|---|---|---|---|---|---|---|
| A | 순위 |
int | 6 | 가운데 | 0 |
total_string: '합계' |
| B | 성분명 |
str | 34 | 왼쪽 | @ |
워치 매칭 → #F7E9F0 + bold(CF-03-05) |
| C | 오늘 신규 |
int | 9 | 가운데 | #,##0;;- |
>0 → #E3F3EA(CF-03-01), total_function:'sum' |
| D | 오늘 변경 |
int | 9 | 가운데 | #,##0;;- |
>0 → #FDF0DC(CF-03-02), total_function:'sum' |
| E | 오늘 취하 |
int | 9 | 가운데 | #,##0;;- |
>0 → #FBE5DC(CF-03-03), total_function:'sum' |
| F | 오늘 합계 |
int | 9 | 가운데 | #,##0;;- |
수식 =SUM($C4:$E4) + 캐시값, total_function:'sum' |
| G | 최근 7일 |
int | 9 | 가운데 | #,##0;;- |
data_bar #0072B2(CF-03-06), total_function:'sum' |
| H | 최근 30일 |
int | 9 | 가운데 | #,##0;;- |
data_bar #0072B2, total_function:'sum' |
| I | 누적 유효등록 |
int | 11 | 가운데 | #,##0 |
data_bar #0072B2, total_function:'sum' |
| J | 등록업체 수 |
int | 9 | 가운데 | #,##0 |
>=5 → bold(CF-03-07, 공급처 다변화 신호) |
| K | 제조국 수 |
int | 9 | 가운데 | #,##0 |
— |
| L | 주요 제조국 |
str | 14 | 가운데 | @ |
— |
| M | 국산 보유 |
str | 8 | 가운데 | @ |
Y → #E3F3EA(CF-03-08) |
| N | 전일 대비 |
int | 9 | 가운데 | ▲ #,##0;▼ -#,##0;– 0 |
icon_set 3_arrows_gray(CF-03-04) |
| O | 30일 추이 |
— | 14 | 가운데 | — | 스파크라인(column). 원본 '06_추이' 히든 블록 |
| P | 상세 |
url | 8 | 가운데 | @ |
내부 링크 → 02_전체현황!A3 |
O열 스파크라인: 성분별 30일 시계열은 06_추이 시트의 히든 블록 AD:BG(행 = 성분, 열 = 일자 30개)에 가로로 적재한다. 성분 i 의 범위는 '06_추이'!$AE${4+i}:$BH${4+i}. 설정은 {'type':'column','style':None,'series_color':'#0072B2','high_point':True,'empty_cells':'zero'}.
3.5 04_업체별
03_성분별 과 대칭 구조.
| 항목 | 값 |
|---|---|
| Excel 표 | add_table(2, 0, last_row+1, 16, {'name':'T_COMPANY','style':'Table Style Light 11','total_row':True}) |
freeze_panes |
(3, 2) |
| 정렬 기준 | G(오늘 합계) 내림차순 → I(누적 유효등록) 내림차순 → B(업체명) 오름차순 |
| 행 수 | report.top_n ∪ (오늘 변경분이 있는 업체 전체) |
| 열 | 컬럼명 | 타입 | 폭 | 정렬 | 형식 | 조건부서식 / 합계행 |
|---|---|---|---|---|---|---|
| A | 순위 |
int | 6 | 가운데 | 0 |
total_string: '합계' |
| B | 업체명 |
str | 30 | 왼쪽 | @ |
워치 매칭 → #F7E9F0 + bold |
| C | 업체키 |
str | 26 | 왼쪽 | @ |
히든(정규화 키) |
| D | 오늘 신규 |
int | 9 | 가운데 | #,##0;;- |
>0 → #E3F3EA, sum |
| E | 오늘 변경 |
int | 9 | 가운데 | #,##0;;- |
>0 → #FDF0DC, sum |
| F | 오늘 취하 |
int | 9 | 가운데 | #,##0;;- |
>0 → #FBE5DC, sum |
| G | 오늘 합계 |
int | 9 | 가운데 | #,##0;;- |
수식 =SUM($D4:$F4) + 캐시값, sum |
| H | 최근 30일 |
int | 9 | 가운데 | #,##0;;- |
data_bar, sum |
| I | 누적 유효등록 |
int | 11 | 가운데 | #,##0 |
data_bar, sum |
| J | 보유 성분 수 |
int | 9 | 가운데 | #,##0 |
— |
| K | 제조소 수 |
int | 9 | 가운데 | #,##0 |
— |
| L | 주요 제조국 |
str | 14 | 가운데 | @ |
— |
| M | 국산여부 |
str | 8 | 가운데 | @ |
국산 → #E3F3EA |
| N | 최근 등록일 |
date | 12 | 가운데 | yyyy-mm-dd |
90일 이상 경과 → #EFEFEF(CF-04-05) |
| O | 워치리스트 |
str | 8 | 가운데 | @ |
★ → #F7E9F0 |
| P | 30일 추이 |
— | 14 | 가운데 | — | 스파크라인(column) |
| Q | 상세 |
url | 8 | 가운데 | @ |
내부 링크 → 02_전체현황!A3 |
3.6 05_워치리스트 (조건부 생성)
- 생성 조건:
watchlist테이블에active='Y'인 행이 1건 이상. 0건이면 시트를 만들지 않고, 대시보드 내비게이션의 해당 슬롯은 회색 비활성 텍스트(워치리스트 없음)로 대체한다. - 시트 보호:
protect("", {"objects": True, "scenarios": True, "select_locked_cells": True, "select_unlocked_cells": True, "sort": True, "autofilter": True}). 읽기 전용 미러임을 물리적으로 표시한다. 비밀번호 없음 — 해제 자체를 막는 것이 목적이 아니라 실수로 편집하고 저장했다가 다음 날 덮어써지는 착각을 막는 것이 목적이다. freeze_panes(3, 3)
블록 A — 키워드 정의(3행 헤더, 4행~)
| 열 | 컬럼명 | 타입 | 폭 | 정렬 | 형식 | 유효성 / 규칙 |
|---|---|---|---|---|---|---|
| A | 번호 |
int | 6 | 가운데 | 0 |
— |
| B | 대상유형 |
str | 12 | 가운데 | @ |
data_validation list: 성분명,업체명,제조소명,제조국가,등록번호 |
| C | 키워드 |
str | 30 | 왼쪽 | @ |
빈 값 → #FBE5DC(CF-05-03) |
| D | 매칭방식 |
str | 12 | 가운데 | @ |
list: 부분일치,정확일치,정규식 |
| E | 우선순위 |
int | 8 | 가운데 | 0 |
list: 1,2,3 |
| F | 활성 |
str | 6 | 가운데 | @ |
list: Y,N |
| G | 메모 |
str | 26 | 왼쪽 | @ |
자유 |
블록 B — 히트 결과(같은 행, I~N)
| 열 | 컬럼명 | 타입 | 폭 | 정렬 | 형식 | 조건부서식 / 수식 |
|---|---|---|---|---|---|---|
| I | 오늘 매칭 |
int | 9 | 가운데 | #,##0;;- |
수식(아래) + >0 → #F7E9F0·bold(CF-05-01) |
| J | 최근 7일 |
int | 9 | 가운데 | #,##0;;- |
data_bar #CC79A7 |
| K | 최근 30일 |
int | 9 | 가운데 | #,##0;;- |
data_bar #CC79A7 |
| L | 누적 매칭 |
int | 9 | 가운데 | #,##0 |
— |
| M | 마지막 매칭일 |
date | 12 | 가운데 | yyyy-mm-dd |
오늘 → #F7E9F0(CF-05-02) |
| N | 상세 |
url | 8 | 가운데 | @ |
내부 링크 → 01_오늘변경분!A3 |
I열 수식 전문 (행 4, 아래로 상대 복사):
=IF($C4="",0,COUNTIF('01_오늘변경분'!$O$4:$O$100000,"*"&$C4&"*"))
캐시값은 파이썬이 실제 매칭 개수로 채운다.
블록 C — 매칭 상세표 (A{blockC} 이하, 블록 A 마지막 행 + 3행부터)
헤더 7컬럼: 키워드 / 상태 / 등록번호 / 성분명 / 업체명 / 제조국가 / 발급일자.
행 전체에 01_오늘변경분 과 동일한 상태 조건부서식(CF-01-01~03)을 재사용한다.
3.7 06_추이 — 시각화 원본
| 항목 | 값 |
|---|---|
| 헤더 행 | 3행 |
| 데이터 | 4행 ~ 3 + report.trend_days(기본 90). 오래된 날짜가 위, 최신이 아래 → 스파크라인이 왼→오 시간순으로 읽힘 |
| Excel 표 | add_table(2, 0, last_row, 8, {'name':'T_TREND','style':'Table Style Light 11'}) |
freeze_panes |
(3, 1) |
| 정렬 기준 | A(일자) 오름차순 |
| 열 | 컬럼명 | 타입 | 폭 | 정렬 | 형식 | 조건부서식 |
|---|---|---|---|---|---|---|
| A | 일자 |
date | 12 | 가운데 | yyyy-mm-dd |
오늘 → bold + #E5F1F8(CF-06-01) |
| B | 신규 |
int | 9 | 가운데 | #,##0;;- |
data_bar #009E73(CF-06-02) |
| C | 변경 |
int | 9 | 가운데 | #,##0;;- |
data_bar #E69F00 |
| D | 취하 |
int | 9 | 가운데 | #,##0;;- |
data_bar #D55E00 |
| E | 합계 |
int | 9 | 가운데 | #,##0;;- |
수식 =SUM($B4:$D4) + 캐시값 |
| F | 워치 히트 |
int | 9 | 가운데 | #,##0;;- |
>0 → #F7E9F0 |
| G | 누적 유효등록 |
int | 12 | 가운데 | #,##0 |
— |
| H | 수집 건수 |
int | 10 | 가운데 | #,##0 |
전일 대비 5% 이상 급감 → #FBE5DC(CF-06-03) |
| I | 실행 상태 |
str | 10 | 가운데 | @ |
SUCCESS→#E3F3EA / PARTIAL→#FDF0DC / FAILED→#FBE5DC(CF-06-04) |
히든 블록 (차트·스파크라인 원본) — 열 폭 0, {'hidden': 1}
| 범위 | 내용 |
|---|---|
AA3:AB3 |
헤더 제조국 / 건수 |
AA4:AB11 |
제조국 Top 8 (최근 30일 기준, 건수 내림차순) — 차트2 원본 |
AD3 |
헤더 성분명 |
AE3:BH3 |
최근 30일 일자 헤더 |
AD4:BH{3+n_ing} |
성분별 30일 시계열 (행 = 성분, 열 = 일자) — 03_성분별 O열 스파크라인 원본 |
BJ3 |
헤더 업체명 |
BK3:CN3 |
최근 30일 일자 헤더 |
BJ4:CN{3+n_co} |
업체별 30일 시계열 — 04_업체별 P열 스파크라인 원본 |
3.8 99_메타
| 항목 | 값 |
|---|---|
| 헤더 | 블록마다 별도 |
freeze_panes |
(3, 0) |
| 자동 필터 | 없음 |
| 열 폭 | A=28, B=46, C=16, D=16, E=30 |
블록 1 — 실행 메타 (A3:B18) — A열 항목명(bold, #5B6770), B열 값
| 행 | 항목 | 값 예 | 형식 |
|---|---|---|---|
| 3 | 리포트 기준일 |
2026-09-02 |
yyyy-mm-dd |
| 4 | 수집 시작 시각 |
2026-09-02 06:02:14 |
yyyy-mm-dd hh:mm:ss |
| 5 | 수집 종료 시각 |
2026-09-02 06:06:41 |
yyyy-mm-dd hh:mm:ss |
| 6 | 실행 ID |
20260902_060214 |
@ |
| 7 | 실행 상태 |
SUCCESS |
@ (CF-99-02) |
| 8 | 실행 트리거 |
scheduled |
@ |
| 9 | 실행 호스트 |
DESKTOP-XXXX |
@ |
| 10 | 파이썬 버전 |
3.12.7 |
@ |
| 11 | XlsxWriter 버전 |
3.2.9 |
@ |
| 12 | 직전 성공 실행 |
20260901_060119 |
@ |
| 13 | 실행 로그 경로 |
logs\run_20260902_060214\pipeline.log |
@ (외부 링크) |
| 14 | 이벤트 로그 경로 |
logs\run_20260902_060214\events.jsonl |
@ (외부 링크) |
| 15 | 원문 아카이브 |
data\raw\2026-09-02\ |
@ (외부 링크) |
| 16 | DB 백업 |
backup\dmf_2026-09-02.sqlite3 |
@ |
| 17 | 스냅샷 해시 |
9f2c1a7be40d5c11 |
@ |
| 18 | 직전 리포트 |
reports\DMF_리포트_2026-09-01.xlsx |
@ (외부 링크) |
블록 2 — 소스 (A21:E23): 헤더 소스명 / URL / HTTP / 수집행수 / 비고. 최소 1행(공식 OpenAPI getMdcDmfList01). URL 은 외부 하이퍼링크.
블록 3 — 무결성 게이트 5종 (A26:E32): 헤더 게이트 / 기대 / 실제 / 판정 / 상세. 아키텍처 §3.6 의 5게이트와 1:1.
| 행 | 게이트 | 기대 |
|---|---|---|
| 27 | 1. 응답 정상성 |
전 페이지 200 + resultCode=00 + 본문 시그니처 |
| 28 | 2. 건수 일치 |
수집 = totalCount |
| 29 | 3. 급감 방지 |
전일 대비 감소율 ≤ max_drop_ratio |
| 30 | 4. 널 비율 |
필수 3필드 널 비율 ≤ max_null_ratio |
| 31 | 5. 중복 비율 |
중복 등록번호 비율 ≤ max_duplicate_ratio |
| 32 | 종합 |
전부 통과 |
D열(판정) 수식 전문 (행 27, 아래로 복사):
=IF($C27=$B27,"OK","불일치")
수치 비교가 불가능한 게이트 1은 파이썬이 OK/불일치 문자열을 직접 쓴다. 조건부서식 CF-99-01.
블록 4 — AI 계층 상태 (A35:B41)
| 항목 | 값 예 |
|---|---|
agy 사용 여부 |
사용 / 건너뜀(변화 없음) / 건너뜀(서킷 OPEN) / 건너뜀(토큰 상한) / 실패 |
모델 |
gemini-3.7-flash-medium |
노력 수준 |
medium |
입력 토큰 / 출력 토큰 |
28,412 / 1,180 |
소요 시간(초) |
33.4 |
오늘 누적 토큰 |
29,592 / 300,000 |
봉투 원문 경로 |
logs\run_...\agy.stdout.json (외부 링크) |
블록 5 — 코드북 (A44:C60): 구분 / 코드 / 설명. 상태 3종, 등록번호 5구획 해부(발급일자8-성분일련-시행군-접수일련-성분내일련), 제조국 표기 정규화 매핑.
블록 6 — 출처·면책 (A63:E66): 9pt #5B6770. 출처 URL, 이용약관 준수 문구, "본 리포트는 공공데이터를 자동 수집·가공한 참고 자료이며 법적 효력이 없다" 면책.
4. 00_대시보드 셀 단위 레이아웃
4.1 열 폭 그리드
| 열 | 폭 | ≈px | 역할 |
|---|---|---|---|
A |
1.5 | 15 | 좌 여백 |
B C |
13 / 13 | 96 / 96 | KPI 타일 1 · 차트1 · Top 성분표 |
D |
1.5 | 15 | 거터 |
E F |
13 / 13 | 96 / 96 | KPI 타일 2 |
G |
1.5 | 15 | 거터 |
H I |
13 / 13 | 96 / 96 | KPI 타일 3 · Top 업체표 |
J |
1.5 | 15 | 거터 |
K L |
13 / 13 | 96 / 96 | KPI 타일 4 · 차트2 |
M |
1.5 | 15 | 거터 |
N O |
13 / 13 | 96 / 96 | KPI 타일 5 · 워치 표 |
P |
1.5 | 15 | 거터 |
Q R |
13 / 13 | 96 / 96 | KPI 타일 6 |
S |
1.5 | 15 | 우 여백 |
총 폭 = 15 + 96×12 + 15×6 + 15 = 1,257px. set_zoom(90) → 1,131px → 1366px 화면에서 가로 스크롤 없음.
4.2 행 높이 그리드
| 행 | 높이 | 역할 |
|---|---|---|
| 1 | 6 | 상단 여백 |
| 2 | 30 | 제목 |
| 3 | 18 | 헤드라인 문장(AI 또는 규칙 기반) |
| 4 | 8 (배너 시 22) | 여백 / 상태 배너 |
| 5 | 18 | KPI 라벨 |
| 6 | 34 | KPI 값 (28pt) |
| 7 | 16 | KPI 델타 |
| 8 | 14 | KPI 스파크라인 |
| 9 | 10 | 여백 |
| 10–24 | 20 ×15 = 300 | 차트 밴드 |
| 25 | 10 | 여백 |
| 26 | 18 | 표 헤더 |
| 27–36 | 16 ×10 = 160 | 표 본문(10행) |
| 37–40 | 16 ×4 = 64 | 표 예비행 |
| 41 | 10 | 여백 |
| 42 | 20 | 내비게이션 1행 |
| 43 | 20 | 내비게이션 2행 |
| 44–46 | 8 ×3 = 24 | 여백 |
| 47 | 14 | 출처·면책 |
총 높이 = 794px. set_zoom(90) → 715px → 768px 화면에 수납.
4.3 텍스트 와이어프레임 (실제 셀 주소 표기)
A B C D E F G H I J K L M N O P Q R S
┌────┬─────────────────┬───┬─────────────────┬───┬─────────────────┬───┬─────────────────┬───┬─────────────────┬───┬─────────────────┬────┐
1 │ (상단 여백 h=6) │
2 │ │ B2:R2 (merge) DMF 일일 모니터링 리포트 — 2026-09-02(수) 18pt bold #1F2933 좌측정렬 │ │
3 │ │ B3:R3 (merge) 오늘 신규 12건 · 변경 3건 · 취하 1건 — 워치리스트 2건 적중(피타바스타틴칼슘, 다파글리플로진) 11pt #5B6770 │ │
4 │ │ B4:R4 (merge) [배너] ⚠ 오늘 수집 실패 — 마지막 성공: 2026-09-01. 아래 수치는 그날 기준이다. 11pt bold #8A1D00 / #FBE5DC│ │
5 │ │ B5:C5 │ │ E5:F5 │ │ H5:I5 │ │ K5:L5 │ │ N5:O5 │ │ Q5:R5 │ │
│ │ 오늘 신규 │ │ 오늘 변경 │ │ 오늘 취하 │ │ 오늘 총 변경분 │ │ 워치리스트 히트 │ │ 전체 누적 등록 │ │
│ │ top=5 #009E73 │ │ top=5 #E69F00 │ │ top=5 #D55E00 │ │ top=5 #0072B2 │ │ top=5 #CC79A7 │ │ top=5 #1F2933 │ │
6 │ │ B6:C6 │ │ E6:F6 │ │ H6:I6 │ │ K6:L6 │ │ N6:O6 │ │ Q6:R6 │ │
│ │ 12 │ │ 3 │ │ 1 │ │ 16 │ │ 2 │ │ 13,842 │ │
│ │ 28pt bold 상태색│ │ 28pt bold │ │ 28pt bold │ │ 28pt bold │ │ 28pt bold │ │ 28pt bold │ │
7 │ │ B7:C7 ▲ 5 │ │ E7:F7 ▼ -2 │ │ H7:I7 – 0 │ │ K7:L7 ▲ 3 │ │ N7:O7 ▲ 2 │ │ Q7:R7 ▲ 11 │ │
8 │ │ B8 ▁▂▅▁▃█▂▁ │ │ E8 ▁▁▂▁▁▃▁▁ │ │ H8 ▁▁▁▁█▁▁▁ │ │ K8 ▂▃▅▂▄█▃▂ │ │ N8 ▁▁█▁▁█▁▁ │ │ Q8 ╱╱╱╱╱╱╱╱ │ │
│ │ sparkline column│ │ sparkline column│ │ sparkline column│ │ sparkline column│ │ sparkline winloss│ │ sparkline line │ │
9 │ (여백 h=10) │
10 │ ┌──────────────────────────────────────────────────┐ ┌──────────────────────────────────────────────────┐ │
│ │ 앵커 B10 · 600×300px │ │ 앵커 K10 · 600×300px │ │
│ │ [차트1] 최근 30일 일별 등록 동향 (누적 세로 막대) │ │ [차트2] 제조국 Top 8 (가로 막대, 단색 #0072B2) │ │
│ │ x=일자(30) y=건수 │ │ x=건수 y=국가명 (직접 라벨링, 색 구분 없음) │ │
│ │ 계열: 신규#009E73 / 변경#E69F00 / 취하#D55E00 │ │ 인도 ████████████████ 120 │ │
24 │ │ 범례: top │ │ 범례: none (단색이라 불필요) │ │
│ └──────────────────────────────────────────────────┘ └──────────────────────────────────────────────────┘ │
25 │ (여백 h=10) │
26 │ │ B26:F26 Top 10 성분 │ │ H26:L26 Top 10 업체 │ │ N26:R26 워치리스트 히트 │ │
│ │ 순위│성분명│오늘│30일│누적 │ │ 순위│업체명│오늘│30일│누적 │ │ 키워드│매칭│성분명│업체명│상태 │ │
27 │ │ 1 │아토르바…│ 3 │ 11 │ 128│ │ 1 │㈜○○제약│ 4 │15 │ 342 │ │ 다파글…│ 1 │… │… │ 신규 │ │
36 │ │ … 10행, I열 data_bar #0072B2│ │ … 10행 │ │ … 최대 10행 │ │
37 │ │ (예비 4행 — 데이터가 적으면 공백) │
41 │ (여백 h=10) │
42 │ │ B42:C42 [오늘 변경분] │ E42:F42 [전체 현황] │ H42:I42 [성분별] │ K42:L42 [업체별] │ N42:O42 [워치리스트] │ Q42:R42 [추이] │
43 │ │ B43:C43 [메타] │
44 │ (여백 h=8 ×3) │
47 │ │ B47:R47 출처: 공공데이터포털 의약품 원료의약품 등록 정보(15057075) · 식품의약품안전처 │ 수집 2026-09-02 06:02 KST │ 참고용 │
└────┴─────────────────┴───┴─────────────────┴───┴─────────────────┴───┴─────────────────┴───┴─────────────────┴───┴─────────────────┴────┘
4.4 KPI 타일 6종 — 셀 주소·값·델타·스파크라인
| # | 라벨 | 라벨셀 | 값셀 | 델타셀 | 스파크라인셀 | 액센트 | 스파크 타입 | 스파크 원본 |
|---|---|---|---|---|---|---|---|---|
| 1 | 오늘 신규 | B5:C5 |
B6:C6 |
B7:C7 |
B8:C8 |
#009E73 |
column |
'06_추이'!$B${s}:$B${e} |
| 2 | 오늘 변경 | E5:F5 |
E6:F6 |
E7:F7 |
E8:F8 |
#E69F00 |
column |
'06_추이'!$C${s}:$C${e} |
| 3 | 오늘 취하 | H5:I5 |
H6:I6 |
H7:I7 |
H8:I8 |
#D55E00 |
column |
'06_추이'!$D${s}:$D${e} |
| 4 | 오늘 총 변경분 | K5:L5 |
K6:L6 |
K7:L7 |
K8:L8 |
#0072B2 |
column |
'06_추이'!$E${s}:$E${e} |
| 5 | 워치리스트 히트 | N5:O5 |
N6:O6 |
N7:O7 |
N8:O8 |
#CC79A7 |
win_loss |
'06_추이'!$F${s}:$F${e} |
| 6 | 전체 누적 등록 | Q5:R5 |
Q6:R6 |
Q7:R7 |
Q8:R8 |
#1F2933 |
line |
'06_추이'!$G${s}:$G${e} |
e = 06_추이 마지막 데이터 행, s = max(4, e - 29) (최근 30일).
- 스파크라인은 병합 범위의 좌상단 셀(
B8,E8, …)에add_sparkline한다. 병합해도 좌상단 기준으로 그려진다. - 공통 옵션:
{"series_color": accent, "high_point": True, "last_point": True, "empty_cells": "zero"}.win_loss만{"negative_points": True}추가. - 스파크라인 데이터 포인트는 30개 — 실무 권고 상한(12~24)을 넘으므로
column타입으로 개별 값을 구분 가능하게 한다.line은 단조 증가하는 누적 KPI(#6)에만 쓴다.
KPI 값 셀 수식 전문 (예: 타일 1) — 캐시값 동봉:
='06_추이'!$B$93
KPI 델타 셀 수식 전문 (예: 타일 1):
=IF(ROW('06_추이'!$B$93)<=4,0,'06_추이'!$B$93-'06_추이'!$B$92)
실제로는 93/92 가 파이썬이 계산한 절대 행 번호로 치환된다. 데이터가 1일치뿐이면 델타는 0 을 직접 쓰고 수식을 걸지 않는다.
4.5 차트 2종 사양
| 항목 | 차트1 | 차트2 |
|---|---|---|
| 앵커 셀 | B10 |
K10 |
| 크기 | set_size({'width': 600, 'height': 300}) |
동일 |
| 타입 | column / stacked |
bar |
| 제목 | 최근 30일 일별 등록 동향 |
제조국 Top 8 (최근 30일) |
| 카테고리 | ['06_추이', s-1, 0, e-1, 0] (A열 일자) |
['06_추이', 3, 26, 10, 26] (AA4:AA11) |
| 계열 | 3개: 신규(B, #009E73) / 변경(C, #E69F00) / 취하(D, #D55E00) |
1개: 건수(AB, #0072B2 단색) |
| 범례 | {'position': 'top', 'font': {'name':'맑은 고딕','size':9}} |
{'none': True} |
| x축 | 라벨 8pt #5B6770, major_tick_mark: 'none', 축선 #D0D7DE |
{'visible': False} — 데이터 라벨로 대체 |
| y축 | 8pt, major_gridlines #EFEFEF 0.75pt, 축선 없음 |
9pt #1F2933, reverse: True(1위가 위) |
| 데이터 라벨 | 없음(누적 막대라 혼잡) | {'value': True, 'position': 'outside_end', 'font': 8pt} |
| gap | 40 | 50 |
| 차트영역 | 테두리 #D0D7DE, 배경 #FFFFFF |
동일 |
| 플롯영역 | 테두리 없음, 배경 없음 | 동일 |
색 구분 금지 근거: 차트2 는 카테고리 8개다. Claus Wilke — "Use direct labeling instead of colors when you need to distinguish between more than about eight categorical items." → 단색 + 축 라벨 직접 표기.
4.6 Top 표 3종
| 표 | 헤더 범위 | 데이터 범위 | 컬럼 | 데이터 원본 |
|---|---|---|---|---|
| Top 10 성분 | B26:F26 |
B27:F36 |
순위(6) / 성분명(20) / 오늘(6) / 30일(6) / 누적(8) |
03_성분별 상위 10행 |
| Top 10 업체 | H26:L26 |
H27:L36 |
순위 / 업체명 / 오늘 / 30일 / 누적 |
04_업체별 상위 10행 |
| 워치 히트 | N26:R26 |
N27:R36 |
키워드(10) / 매칭(5) / 성분명(14) / 업체명(12) / 상태(6) |
05_워치리스트 블록 C 상위 10행 |
- 표 헤더: 10pt bold
#FFFFFFon#1F2933, 가운데. - 표 본문: 10pt
#1F2933, 아래선#D0D7DEhair(스타일 7). 누적열에data_bar#0072B2(CF-00-01).- 워치 히트 표의
상태열에 상태 조건부서식(CF-00-02~04). 누적열은 시트 간 수식으로 검증 가능하게 한다 — §5.3 참조.
4.7 상태 배너 (B4:R4)
| 조건 | 표시 | 서식 |
|---|---|---|
정상(SUCCESS) |
배너 없음. 행 4 높이 8, 내용 없음 | — |
무결성 차단(PARTIAL) |
⚠ 오늘 수집 실패 — 마지막 성공: {날짜}. 아래 수치는 그날 기준이다. 사유: {blocking_reason} |
bold #8A1D00 on #FBE5DC, 높이 22 |
| 기준선 수립일(첫 실행) | ℹ 기준선 수립일입니다. 변경 탐지는 내일 실행부터 시작됩니다. |
bold #00456B on #E5F1F8, 높이 22 |
| AI 실패/스킵 | ℹ AI 요약을 생성하지 못했습니다({사유}). 수치는 모두 정상입니다. |
#5B6770 on #EFEFEF, 높이 22 |
배너가 2개 이상 해당하면 심각도 순(PARTIAL > 기준선 > AI)으로 1개만 표시하고 나머지는 99_메타 에 기록한다. 대시보드에 경고를 쌓지 않는다.
4.8 내비게이션 (42~43행)
| 셀(병합) | 표시 문자열 | 링크 대상 |
|---|---|---|
B42:C42 |
오늘 변경분 → |
internal:'01_오늘변경분'!A1 |
E42:F42 |
전체 현황 → |
internal:'02_전체현황'!A1 |
H42:I42 |
성분별 → |
internal:'03_성분별'!A1 |
K42:L42 |
업체별 → |
internal:'04_업체별'!A1 |
N42:O42 |
워치리스트 → |
internal:'05_워치리스트'!A1 (없으면 회색 (워치리스트 없음)) |
Q42:R42 |
추이 → |
internal:'06_추이'!A1 |
B43:C43 |
메타·로그 → |
internal:'99_메타'!A1 |
서식: 10pt bold #0072B2, 밑줄 없음, 배경 #F7F7F7, 테두리 1px #D0D7DE, 가운데 정렬. tip= 로 툴팁을 단다.
5. 시트 간 연동 명세
5.1 하이퍼링크 전집
시트명 인용 규칙: 이 워크북의 모든 시트명은 숫자로 시작한다(00_대시보드 …). Excel 은 숫자로 시작하는 시트명을 참조할 때 작은따옴표 인용을 요구한다. 따라서 예외 없이 '시트명'!셀 로 쓴다. 내부 링크는 internal:'00_대시보드'!A1 형태다.
| # | 출발 시트 | 출발 셀 | 링크 종류 | 대상 | 표시 문자열 | 툴팁 |
|---|---|---|---|---|---|---|
| L01 | 00_대시보드 |
B42:C42 |
internal | '01_오늘변경분'!A1 |
오늘 변경분 → |
오늘 신규·변경·취하 전체 |
| L02 | 00_대시보드 |
E42:F42 |
internal | '02_전체현황'!A1 |
전체 현황 → |
유효 등록 전체 원장 |
| L03 | 00_대시보드 |
H42:I42 |
internal | '03_성분별'!A1 |
성분별 → |
성분 기준 집계 |
| L04 | 00_대시보드 |
K42:L42 |
internal | '04_업체별'!A1 |
업체별 → |
업체 기준 집계 |
| L05 | 00_대시보드 |
N42:O42 |
internal | '05_워치리스트'!A1 |
워치리스트 → |
관심 키워드 히트 |
| L06 | 00_대시보드 |
Q42:R42 |
internal | '06_추이'!A1 |
추이 → |
일자별 시계열 |
| L07 | 00_대시보드 |
B43:C43 |
internal | '99_메타'!A1 |
메타·로그 → |
수집 메타·무결성·AI 상태 |
| L08 | 00_대시보드 |
B26 |
internal | '03_성분별'!A1 |
Top 10 성분 (헤더가 링크) |
성분별 시트로 |
| L09 | 00_대시보드 |
H26 |
internal | '04_업체별'!A1 |
Top 10 업체 |
업체별 시트로 |
| L10 | 00_대시보드 |
N26 |
internal | '05_워치리스트'!A1 |
워치리스트 히트 |
워치리스트 시트로 |
| L11 | 01~06,99 |
A1 |
internal | '00_대시보드'!A1 |
← 대시보드 |
대시보드로 돌아가기 |
| L12 | 01_오늘변경분 |
D4:D{last} |
internal | '02_전체현황'!A{행} |
등록번호 원문 | 전체 현황의 해당 행으로 |
| L13 | 01_오늘변경분 |
Q4:Q{last} |
external | nedrug 검색 URL | 조회 |
의약품안전나라에서 검색 |
| L14 | 02_전체현황 |
O4:O{last} |
external | nedrug 검색 URL | 조회 |
의약품안전나라에서 검색 |
| L15 | 03_성분별 |
P4:P{last} |
internal | '02_전체현황'!A3 |
보기 |
전체 현황에서 필터 |
| L16 | 04_업체별 |
Q4:Q{last} |
internal | '02_전체현황'!A3 |
보기 |
전체 현황에서 필터 |
| L17 | 05_워치리스트 |
N4:N{last} |
internal | '01_오늘변경분'!A3 |
보기 |
오늘 변경분에서 확인 |
| L18 | 99_메타 |
B13,B14,B15,B18 |
external | file:///… 로컬 경로 |
경로 문자열 | 파일 열기 |
| L19 | 99_메타 |
B22 |
external | OpenAPI 엔드포인트 URL | URL | 원본 API 문서 |
L12 의 대상 행 계산: 02_전체현황 은 dmf_key → 행번호 사전을 빌드 순서대로 만들어 두고, 01_오늘변경분 을 쓸 때 그 사전을 조회한다. 취하 이벤트는 02_전체현황 에 행이 없을 수 있으므로 그때는 링크 없이 일반 텍스트로 쓴다.
링크 시각 스타일 — Excel 기본 파랑 밑줄을 쓰지 않는다.
| 링크 종류 | 서식 |
|---|---|
| 내비게이션 버튼(L01~L07) | 10pt bold #0072B2, 밑줄 없음, 배경 #F7F7F7, 테두리 #D0D7DE |
| 역링크(L11) | 10pt #0072B2, 밑줄 없음, 배경 없음 |
| 표 안 내부 링크(L12, L15~L17) | 10pt #0072B2, 밑줄 없음 |
| 외부 링크(L13, L14, L18, L19) | 10pt #0072B2, 밑줄 있음 — 워크북 밖으로 나간다는 신호를 시각적으로 구분 |
5.2 목차 기능
별도 목차 시트를 만들지 않는다. 00_대시보드 가 목차를 겸한다(§4.8 내비게이션 + §4.6 Top 표 헤더 링크). 근거:
- 시트가 8개뿐이고 탭 자체가 목차 역할을 한다. 목차 전용 시트는 클릭 1회를 더 요구한다.
- 5초 규칙 — 첫 화면은 "지금 무슨 일이 일어났는가"여야 하며 "어디로 갈 수 있는가"는 그 아래여야 한다.
- 모든 데이터 시트
A1의 역링크(L11)가 항상 대시보드로 되돌린다 → 왕복 구조가 완결된다.
5.3 시트 간 수식 전집
모두 write_formula(row, col, formula, fmt, value=캐시값) 로 기록한다.
| # | 위치 | 수식 전문 | 캐시값 산출 |
|---|---|---|---|
| F01 | 00_대시보드!B6 |
='06_추이'!$B${e} |
오늘 NEW 건수 |
| F02 | 00_대시보드!E6 |
='06_추이'!$C${e} |
오늘 CHANGED 건수 |
| F03 | 00_대시보드!H6 |
='06_추이'!$D${e} |
오늘 WITHDRAWN 건수 |
| F04 | 00_대시보드!K6 |
='06_추이'!$E${e} |
세 값의 합 |
| F05 | 00_대시보드!N6 |
='06_추이'!$F${e} |
오늘 워치 히트 |
| F06 | 00_대시보드!Q6 |
='06_추이'!$G${e} |
누적 유효 등록 |
| F07 | 00_대시보드!B7 |
='06_추이'!$B${e}-'06_추이'!$B${e-1} |
전일 대비 델타 |
| F08 | 00_대시보드!E7 |
='06_추이'!$C${e}-'06_추이'!$C${e-1} |
〃 |
| F09 | 00_대시보드!H7 |
='06_추이'!$D${e}-'06_추이'!$D${e-1} |
〃 |
| F10 | 00_대시보드!K7 |
='06_추이'!$E${e}-'06_추이'!$E${e-1} |
〃 |
| F11 | 00_대시보드!N7 |
='06_추이'!$F${e}-'06_추이'!$F${e-1} |
〃 |
| F12 | 00_대시보드!Q7 |
='06_추이'!$G${e}-'06_추이'!$G${e-1} |
〃 |
| F13 | 00_대시보드!F27:F36 (Top 성분 누적) |
=COUNTIFS('02_전체현황'!$B:$B,$C27,'02_전체현황'!$M:$M,"유효") |
SQL COUNT(*) |
| F14 | 00_대시보드!L27:L36 (Top 업체 누적) |
=COUNTIFS('02_전체현황'!$C:$C,$I27,'02_전체현황'!$M:$M,"유효") |
SQL COUNT(*) |
| F15 | 02_전체현황!N4:N{last} |
=IF($I4="","-",INT(REPORT_DATE-$I4)) |
(기준일-수리일자).days |
| F16 | 03_성분별!F4:F{last} |
=SUM($C4:$E4) |
세 값의 합 |
| F17 | 04_업체별!G4:G{last} |
=SUM($D4:$F4) |
세 값의 합 |
| F18 | 05_워치리스트!I4:I{last} |
=IF($C4="",0,COUNTIF('01_오늘변경분'!$O$4:$O$100000,"*"&$C4&"*")) |
실제 매칭 개수 |
| F19 | 06_추이!E4:E{last} |
=SUM($B4:$D4) |
세 값의 합 |
| F20 | 99_메타!D27:D31 |
=IF($C27=$B27,"OK","불일치") |
게이트 판정 |
| F21 | 99_메타!D32 (종합) |
=IF(COUNTIF($D$27:$D$31,"불일치")=0,"OK","불일치") |
verdict.ok |
XlsxWriter 수식 작성 규칙 2가지 (문서 명시):
- 인자 구분자는 쉼표다(세미콜론 아님). 한국어 Excel 에서 화면에는 쉼표로 보이지만 파일 포맷은 항상 US 스타일이다.
_xlfn접두가 필요한 미래 함수는 워크북 옵션use_future_functions: True로 자동 처리된다. 위 수식들은 모두 레거시 함수라 해당 없음.
5.4 정의된 이름 (Defined Names)
ASCII 이름을 정본으로 한다. Excel 은 유니코드 정의 이름을 허용하지만 XlsxWriter 의 유효성 검사 통과 여부를 실측하지 않았다(부록 기록). ASCII 는 100% 안전하고, 이름 상자에 뜨는 문자열이 짧아 오히려 쓰기 편하다.
| 이름 | 스코프 | 수식 | 용도 |
|---|---|---|---|
REPORT_DATE |
통합문서 | ='99_메타'!$B$3 |
F15 경과일 계산의 기준일 |
RUN_ID |
통합문서 | ='99_메타'!$B$6 |
추적 |
KPI_NEW |
통합문서 | ='00_대시보드'!$B$6 |
외부 참조용 |
KPI_CHANGED |
통합문서 | ='00_대시보드'!$E$6 |
〃 |
KPI_WITHDRAWN |
통합문서 | ='00_대시보드'!$H$6 |
〃 |
KPI_TOTAL |
통합문서 | ='00_대시보드'!$K$6 |
〃 |
KPI_WATCH |
통합문서 | ='00_대시보드'!$N$6 |
〃 |
KPI_ACTIVE |
통합문서 | ='00_대시보드'!$Q$6 |
〃 |
TREND_DATE |
통합문서 | ='06_추이'!$A$4:$A${last} |
차트1 카테고리 |
TREND_NEW |
통합문서 | ='06_추이'!$B$4:$B${last} |
차트1 계열 |
TREND_CHANGED |
통합문서 | ='06_추이'!$C$4:$C${last} |
〃 |
TREND_WITHDRAWN |
통합문서 | ='06_추이'!$D$4:$D${last} |
〃 |
LEDGER |
통합문서 | ='02_전체현황'!$A$3:$Q${last} |
사용자 임의 수식용 |
TODAY_CHANGES |
통합문서 | ='01_오늘변경분'!$A$3:$S${last} |
〃 |
_xlnm.Print_Titles |
시트별 | repeat_rows(0, 2) 가 자동 생성 |
인쇄 제목 행 |
_xlnm.Print_Area |
시트별 | print_area(...) 가 자동 생성 |
인쇄 영역 |
# report/build.py 에서 시트를 모두 만든 뒤 마지막에 호출
wb.define_name("REPORT_DATE", "='99_메타'!$B$3")
wb.define_name("TREND_NEW", f"='06_추이'!$B$4:$B${trend_last}")
# … 표의 나머지도 같은 형태
주의:
define_name은 워크북 스코프에서 시트를 모두 생성한 뒤 호출해야 한다. 존재하지 않는 시트를 참조하면 Excel 이 열 때#REF!가 된다.
5.5 연동 구조 요약 다이어그램
┌───────────────────────────────┐
│ 00_대시보드 │
│ KPI 6 · 차트 2 · Top 표 3 │
└───┬───────────────────────┬───┘
L01~L07 (내비 링크) │ │ F01~F14 (수식 참조)
┌──────┬──────┬────┴─┬──────┬──────┬───────┴──┐
▼ ▼ ▼ ▼ ▼ ▼ ▼
01_오늘 02_전체 03_성분 04_업체 05_워치 06_추이 99_메타
변경분 현황 별 별 리스트 (원본) (기준일)
│ ▲ │ │ │ ▲ ▲
│ │ │ │ │ │ │
└──────┘ └──────┴──────┘ │ │
L12 등록번호 L15/L16 상세 │ │
└──────────────────────────── L17 ───┘ │
│
모든 데이터 시트 A1 ──── L11 (← 대시보드) ─────────┘
스파크라인 원본: 06_추이 B~G열 + 히든 AD:CN
차트 원본: 06_추이 A~D열(차트1), AA:AB(차트2)
6. 디자인 토큰
6.1 색 팔레트 — Okabe-Ito 기반
채택 근거: Masataka Okabe · Kei Ito 의 Color Universal Design 팔레트. Nature Methods 권장, Claus Wilke Fundamentals of Data Visualization 의 기본 범주형 스케일. 순수 red 와 순수 green 을 아예 쓰지 않는다 — Orange(#E69F00)는 1형 색각(protanopia)에서도 황등색으로 지각되고, Bluish Green(#009E73)은 청색 채널이 충분해 적록 병합에서 살아남는다. Tableau 10 은 인접한 빨강·초록이 2형 색각에서 충돌하므로 기각.
| 토큰 | HEX | 원 팔레트 | 흰 배경 대비비 | 용도 |
|---|---|---|---|---|
ACCENT |
#0072B2 |
Okabe-Ito Blue | 5.19 (AA) | 유일한 강조색. 링크·데이터바·단색 차트·합계 |
NEW |
#009E73 |
Bluish Green | 3.42 | 신규 액센트(KPI 값·차트 계열) |
CHANGED |
#E69F00 |
Orange | 2.25 | 변경 액센트 |
WITHDRAWN |
#D55E00 |
Vermillion | 3.87 | 취하 액센트 |
WATCH |
#CC79A7 |
Reddish Purple | 3.06 | 워치리스트 액센트 |
SKY |
#56B4E9 |
Sky Blue | 2.31 | 보조 계열(예비) |
INK |
#1F2933 |
(파생) | 14.76 (AAA) | 본문 텍스트·표 헤더 배경·누적 KPI |
MUTED |
#5B6770 |
(파생) | 5.80 (AA) | 보조 텍스트·라벨·각주 |
NEUTRAL |
#888888 |
Paul Tol bad-data grey | 3.54 | 결측·무효·비활성 |
RULE |
#D0D7DE |
(파생) | 1.45 | 테두리·구분선(장식 전용, 텍스트 아님) |
GRID |
#EFEFEF |
(파생) | — | 차트 격자선 |
TILE_BG |
#F7F7F7 |
(파생) | — | KPI 타일·내비 버튼 배경 |
PAPER |
#FFFFFF |
— | — | 시트 배경 |
사용 금지: Okabe-Ito Yellow(#F0E442) — 흰 배경 대비 1.32 로 텍스트·선에 부적합. 배경으로도 쓰지 않는다(우리 연한 톤이 이미 충분).
6.2 상태색 매핑 — 4중 코딩 (배경 + 폰트 + 라벨 + 기호)
WCAG 1.4.1 은 색 단독 전달을 금지한다. 남성 약 8%, 여성 약 0.5% 가 색각 이상이다. 따라서 모든 상태는 배경색·폰트색·텍스트 라벨·기호 4개 채널로 동시에 표현한다.
| 상태 | 라벨 | 기호 | 배경 HEX | 폰트 HEX | 대비비 | 등급 | 액센트 HEX |
|---|---|---|---|---|---|---|---|
| 신규 | 신규 |
＋ |
#E3F3EA |
#005A32 |
7.28 | AAA | #009E73 |
| 변경 | 변경 |
◆ |
#FDF0DC |
#7A3E00 |
7.42 | AAA | #E69F00 |
| 취하 | 취하 |
✕ |
#FBE5DC |
#8A1D00 |
7.69 | AAA | #D55E00 |
| 워치 히트 | ★ |
★ |
#F7E9F0 |
#7A2E55 |
7.61 | AAA | #CC79A7 |
| 정보/정상 | OK |
✓ |
#E5F1F8 |
#00456B |
8.85 | AAA | #0072B2 |
| 변동 없음 | - |
· |
#FFFFFF |
#5B6770 |
5.80 | AA | #888888 |
| 오류/결측 | ? |
⚠ |
#EFEFEF |
#3A3A3A |
9.89 | AAA | #888888 |
- 취하 행에는 배경·폰트·라벨·기호에 더해 취소선(
font_strikeout: True)까지 5번째 채널을 얹는다. - Excel 조건부서식 기본색(
#FFC7CE/#9C0006,#C6EFCE/#006100)은 red/green 쌍이라 CUD 금지 조합이고 대비도 낮다 → 사용하지 않는다.
6.3 타이포그래피 계층
폰트는 report.font 설정키(기본 맑은 고딕). 폰트명에 공백이 있어 반드시 문자열로 전달한다. Excel 은 해당 PC 에 설치된 폰트만 렌더링하며, 배포 대상이 Windows 로 한정된다는 전제에서 안전하다.
| 요소 | 크기 | 굵기 | 색 | 비고 |
|---|---|---|---|---|
| 대시보드 제목 | 18pt | bold | #1F2933 |
B2:R2 |
| 대시보드 헤드라인 | 11pt | regular | #5B6770 |
B3:R3 |
| 상태 배너 | 11pt | bold | 상태 폰트색 | B4:R4 |
| KPI 값 | 28pt | bold | 상태 액센트색 | 카드 내 단독 최대 요소. 44pt 는 5자리에서 넘침 |
| KPI 라벨 | 10pt | bold | #5B6770 |
|
| KPI 델타 | 10pt | bold | #5B6770 |
색이 아니라 ▲▼ 기호로 방향 전달 |
시트 제목 (B1) |
14pt | bold | #1F2933 |
|
| 표 헤더 | 10pt | bold | #FFFFFF on #1F2933 |
줄바꿈 허용 |
| 표 본문 | 10pt | regular | #1F2933 |
|
| 표 본문(긴 텍스트: 변경내용·AI 메모) | 9pt | regular | #1F2933 / #5B6770 |
줄바꿈 |
| 하이퍼링크 | 10pt | bold(내비)/regular | #0072B2 |
외부만 밑줄 |
| 각주·출처·면책 | 9pt | regular | #5B6770 |
|
| 차트 제목 | 11pt | bold | #1F2933 |
|
| 차트 축 라벨 | 8~9pt | regular | #5B6770 / #1F2933 |
6.4 행 높이 · 열 너비 규칙
| 대상 | 값 | 근거 |
|---|---|---|
| 대시보드 행 높이 | §4.2 표 그대로 | 1화면 수납 계산 |
| 데이터 시트 1행(제목) | 24 | |
| 데이터 시트 2행(여백) | 6 | |
| 데이터 시트 3행(헤더) | 32 | 2줄 줄바꿈 헤더 수용 |
01_오늘변경분 데이터 행 |
30 | 변경내용 2줄 수용 |
02_전체현황 데이터 행 |
15 | 수천~수만 행 성능. 줄바꿈 없음 |
03/04 데이터 행 |
18 | 스파크라인 수용 |
05/06/99 데이터 행 |
16 | |
| 열 너비 최소 / 최대 / 여백 | 8.0 / 48.0 / +2.0 | compute_col_widths 파라미터 |
| 한글 폭 계수 | W·F = 1.8, A = 1.2, 그 외 = 1.0 | east_asian_width 의 2.0 은 맑은 고딕에서 과대 추정. 실무 경험치 1.8 채택 |
6.5 테두리 규칙 (데이터 잉크 비율 우선)
| 대상 | 테두리 |
|---|---|
| 표 본문 셀 | 세로선 없음. 아래선만 hair(스타일 7) #D0D7DE |
| 표 헤더 | 아래선 medium(스타일 2) #1F2933 |
| 표 합계행 | 위선 double(스타일 6) #1F2933 |
| KPI 타일 | 좌·우 thin(1) #D0D7DE, 상단 thick(5) 상태 액센트색, 하단 thin #D0D7DE |
| 내비 버튼 | 사방 thin(1) #D0D7DE |
| 상태 배너 | 좌측 thick(5) 상태 폰트색 |
| 차트 영역 | #D0D7DE 1px |
| 플롯 영역 | 없음 |
XlsxWriter 테두리 스타일 번호: 1=thin, 2=medium, 3=dashed, 4=dotted, 5=thick, 6=double, 7=hair.
6.6 여백·정렬
| 항목 | 값 |
|---|---|
| 인쇄 여백 | set_margins(left=0.4, right=0.4, top=0.6, bottom=0.6) (인치) |
| 머리글/바닥글 여백 | 0.3 / 0.3 |
| 문자열 열 정렬 | 왼쪽 + valign: vcenter |
| 숫자·날짜 열 정렬 | 가운데 (자릿수 비교가 아니라 스캔이 목적) |
| 합계·금액성 큰 수 | 오른쪽 (02_전체현황 에는 해당 없음) |
| 헤더 정렬 | 가운데 + text_wrap: True |
| 긴 텍스트 셀 | 왼쪽 + text_wrap: True + valign: top |
6.7 표시 형식(number_format) 사전
| 토큰 | 문자열 | 적용 |
|---|---|---|
INT |
#,##0 |
누적 건수 |
INT_DASH |
#,##0;;- |
0을 - 로 (일일 건수) |
INT_UNIT |
#,##0"건" |
대시보드 표 예비 |
DELTA |
▲ #,##0;▼ -#,##0;– 0 |
KPI 델타 |
DELTA_PCT |
▲ 0.0%;▼ -0.0%;– 0.0% |
비율 델타(예비) |
DATE |
yyyy-mm-dd |
모든 날짜 |
DATETIME |
yyyy-mm-dd hh:mm:ss |
메타 시각 |
TEXT |
@ |
등록번호·해시 등 숫자 변환 금지 문자열 |
RATIO |
0.0% |
널 비율·감소율 |
SEC |
0.0"초" |
agy 소요 시간 |
▲ #,##0;▼ -#,##0;– 0의 3구획은 각각 양수/음수/0 이다. 음수 구획에-를 남겨 두는 이유는▼ -2처럼 부호를 함께 보여 스크린 리더와 흑백 인쇄에서도 방향이 전달되게 하기 위함이다.
7. 조건부 서식 규칙 전집
모든 규칙은 worksheet.conditional_format(first_row, first_col, last_row, last_col, options) 로 건다. first_row 등은 0-index지만 criteria 수식 안의 셀 참조는 A1 표기(1-index) 다. 아래 표의 "수식" 열은 데이터 첫 행이 워크시트 4행(0-index 3)임을 전제한다 — $C4 의 4 가 그것이다. 이 한 칸 어긋남이 "한 행씩 밀린 색칠"의 대부분 원인이다.
| ID | 시트 | 대상 범위 | 유형 | 수식 / 조건 | 서식 결과 |
|---|---|---|---|---|---|
CF-01-01 |
01_오늘변경분 |
B4:S{last} |
formula |
=$C4="신규" |
bg #E3F3EA, font #005A32 |
CF-01-02 |
01_오늘변경분 |
B4:S{last} |
formula |
=$C4="변경" |
bg #FDF0DC, font #7A3E00 |
CF-01-03 |
01_오늘변경분 |
B4:S{last} |
formula |
=$C4="취하" |
bg #FBE5DC, font #8A1D00, 취소선 |
CF-01-04 |
01_오늘변경분 |
N4:N{last} |
cell |
> 0 |
bg #F7E9F0, font #7A2E55, bold |
CF-01-05 |
01_오늘변경분 |
I4:I{last} |
text |
containing 대한민국 |
bg #E3F3EA, font #005A32 |
CF-01-06 |
01_오늘변경분 |
E4:F{last} |
formula |
=$N4>0 |
bold (성분명·업체명만) |
CF-02-01 |
02_전체현황 |
A4:A{last} |
duplicate |
— | bg #FFF2CC, font #7F6000 |
CF-02-02 |
02_전체현황 |
L4:L{last} |
formula |
=$L4=REPORT_DATE |
bg #FDF0DC, font #7A3E00, bold |
CF-02-03 |
02_전체현황 |
A4:Q{last} |
formula |
=$M4="취하" |
bg #FBE5DC, font #8A1D00, 취소선 |
CF-02-04 |
02_전체현황 |
F4:F{last} |
text |
containing 대한민국 |
bg #E3F3EA, font #005A32 |
CF-02-05 |
02_전체현황 |
B4:C{last} |
formula |
=COUNTIF(WATCH_KEYS,"*"&B4&"*")>0 → 미사용. 파이썬이 워치 매칭 여부를 계산해 히든 열에 쓰고 =$R4="Y" 로 판정 |
bg #F7E9F0, font #7A2E55 |
CF-02-06 |
02_전체현황 |
G4:G{last} |
cell |
>= 2 |
bold |
CF-02-07 |
02_전체현황 |
I4:I{last} |
data_bar |
bar_color #0072B2, bar_solid True, bar_only False |
수리일자 최신일수록 긴 막대 |
CF-02-08 |
02_전체현황 |
N4:N{last} |
icon_set |
3_arrows_gray, icons=[{'criteria':'>=','type':'number','value':1095},{'criteria':'>=','type':'number','value':365}], icons_only False |
3년↑ / 1년↑ / 그 외 방향 표시 |
CF-03-01 |
03_성분별 |
C4:C{last} |
cell |
> 0 |
bg #E3F3EA, font #005A32 |
CF-03-02 |
03_성분별 |
D4:D{last} |
cell |
> 0 |
bg #FDF0DC, font #7A3E00 |
CF-03-03 |
03_성분별 |
E4:E{last} |
cell |
> 0 |
bg #FBE5DC, font #8A1D00 |
CF-03-04 |
03_성분별 |
N4:N{last} |
icon_set |
3_arrows_gray, icons=[{'criteria':'>=','type':'number','value':1},{'criteria':'>=','type':'number','value':0}] |
▲ / ▬ / ▼ |
CF-03-05 |
03_성분별 |
B4:B{last} |
formula |
=$Q4="Y" (히든 워치플래그 열) |
bg #F7E9F0, font #7A2E55, bold |
CF-03-06 |
03_성분별 |
G4:I{last} |
data_bar |
bar_color #0072B2, bar_solid True |
3열 각각 별도 규칙(multi_range 사용 금지 — 열별 스케일이 달라야 함) |
CF-03-07 |
03_성분별 |
J4:J{last} |
cell |
>= 5 |
bold |
CF-03-08 |
03_성분별 |
M4:M{last} |
text |
containing Y |
bg #E3F3EA, font #005A32 |
CF-04-01 |
04_업체별 |
D4:D{last} |
cell |
> 0 |
bg #E3F3EA, font #005A32 |
CF-04-02 |
04_업체별 |
E4:E{last} |
cell |
> 0 |
bg #FDF0DC, font #7A3E00 |
CF-04-03 |
04_업체별 |
F4:F{last} |
cell |
> 0 |
bg #FBE5DC, font #8A1D00 |
CF-04-04 |
04_업체별 |
H4:I{last} |
data_bar |
bar_color #0072B2 |
열별 개별 규칙 |
CF-04-05 |
04_업체별 |
N4:N{last} |
formula |
=AND($N4<>"",REPORT_DATE-$N4>90) |
bg #EFEFEF, font #5B6770 |
CF-04-06 |
04_업체별 |
O4:O{last} |
text |
containing ★ |
bg #F7E9F0, font #7A2E55 |
CF-05-01 |
05_워치리스트 |
I4:I{last} |
cell |
> 0 |
bg #F7E9F0, font #7A2E55, bold |
CF-05-02 |
05_워치리스트 |
M4:M{last} |
formula |
=$M4=REPORT_DATE |
bg #F7E9F0, font #7A2E55 |
CF-05-03 |
05_워치리스트 |
C4:C{last} |
blanks |
— | bg #FBE5DC, font #8A1D00 |
CF-05-04 |
05_워치리스트 |
J4:K{last} |
data_bar |
bar_color #CC79A7 |
열별 개별 규칙 |
CF-05-05 |
05_워치리스트 (블록C) |
A{c}:G{cend} |
formula ×3 |
=$B{c}="신규" / ="변경" / ="취하" |
CF-01-01~03 과 동일 서식 |
CF-06-01 |
06_추이 |
A4:A{last} |
formula |
=$A4=REPORT_DATE |
bg #E5F1F8, font #00456B, bold |
CF-06-02 |
06_추이 |
B4:B{last} / C… / D… |
data_bar ×3 |
bar_color 각각 #009E73 / #E69F00 / #D55E00 |
열별 |
CF-06-03 |
06_추이 |
H4:H{last} |
formula |
=AND(ROW()>4,$H4<$H3*0.95) |
bg #FBE5DC, font #8A1D00 |
CF-06-04 |
06_추이 |
I4:I{last} |
text ×3 |
containing SUCCESS / PARTIAL / FAILED |
#E3F3EA / #FDF0DC / #FBE5DC |
CF-99-01 |
99_메타 |
D27:D32 |
text ×2 |
containing OK / 불일치 |
#E3F3EA+#005A32 / #FBE5DC+#8A1D00 |
CF-99-02 |
99_메타 |
B7 |
text ×3 |
containing SUCCESS / PARTIAL / FAILED |
#E3F3EA / #FDF0DC / #FBE5DC |
CF-00-01 |
00_대시보드 |
F27:F36, L27:L36 |
data_bar ×2 |
bar_color #0072B2, bar_solid True |
Top 표 누적 열 |
CF-00-02 |
00_대시보드 |
R27:R36 |
text |
containing 신규 |
#E3F3EA + #005A32 |
CF-00-03 |
00_대시보드 |
R27:R36 |
text |
containing 변경 |
#FDF0DC + #7A3E00 |
CF-00-04 |
00_대시보드 |
R27:R36 |
text |
containing 취하 |
#FBE5DC + #8A1D00 |
7.1 규칙 적용 순서와 stop_if_true
- 행 전체 규칙 3종(
CF-01-01~03,CF-02-03)은 상태 값이 상호 배타적이므로stop_if_true를 주지 않는다(기본 False). 서로 겹치지 않는다. - 열 단위 강조(
CF-01-04,CF-01-06)는 행 규칙 뒤에 등록해야 위에 얹힌다. XlsxWriter 는 등록 순서를 그대로 규칙 우선순위로 쓰므로,report/sheets/*안에서 행 규칙 → 열 규칙 → 데이터바/아이콘셋 순서로 호출한다. data_bar와icon_set은 배경색을 덮지 않으므로 언제 등록해도 무방하지만, 관례를 위해 항상 마지막에 둔다.
7.2 아이콘셋 금지 목록
3_traffic_lights, 3_traffic_lights_rimmed, 4_traffic_lights 는 사용 금지. red/green 조합에 형태까지 동일한 원이라 색각 이상에서 정보가 완전히 소실된다. 허용은 3_arrows_gray(무채색, 방향만) 하나뿐이며, 3_symbols_circled 는 예비다.
icons 파라미터 규칙: criteria 는 >= 또는 < 만 가능(기본 >=), type 은 number/percentile/percent/formula(기본 percent), 값은 높은 것부터 내림차순으로 나열한다.
8. 생성 코드
8.0 모듈 매핑 — src/report/workbook.py 는 없다
요구 항목에 적힌 src/report/workbook.py 는 아키텍처 SSOT §2 의 확정 트리에 존재하지 않는다. SSOT 가 이긴다. 아래가 확정 매핑이다.
| 요구가 말한 것 | 실제 파일 | 책임 |
|---|---|---|
workbook.py 의 워크북 조립 |
src/dmf_crawler/report/build.py |
워크북 생성·시트 순서·정의된 이름·저장 호출 |
workbook.py 의 스타일 헬퍼 |
src/dmf_crawler/report/theme.py |
팔레트 상수·Format 캐시 |
workbook.py 의 한글 열너비 계산 |
src/dmf_crawler/report/widgets.py |
display_width·compute_col_widths·KPI 타일·역링크 |
workbook.py 의 원자적 저장 |
src/dmf_crawler/report/atomic.py |
임시파일 → 검증 → os.replace → 폴백 |
workbook.py 의 시트 빌더 |
src/dmf_crawler/report/sheets/s00~s99 |
시트 1개당 파일 1개 |
| (신규) DB 조회 | src/dmf_crawler/report/data.py |
ReportData 조립. 시트별 SQL 상수 |
공개 진입점은 아키텍처 §3.13 대로 report/__init__.py 의 build_report(conn, run_id, cfg) -> ReportOutcome 하나다.
8.1 src/dmf_crawler/report/theme.py
"""리포트 디자인 토큰과 Format 캐시.
XlsxWriter 의 Format 객체는 워크북에 종속되고, 같은 속성이면 재사용해야
파일 크기와 생성 속도가 유지된다. FormatCache 가 그 재사용을 강제한다.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
import xlsxwriter
# ─────────────────────────────────────────────────────────── 팔레트 (§6.1)
PALETTE: dict[str, str] = {
"ACCENT": "#0072B2",
"NEW": "#009E73",
"CHANGED": "#E69F00",
"WITHDRAWN": "#D55E00",
"WATCH": "#CC79A7",
"SKY": "#56B4E9",
"INK": "#1F2933",
"MUTED": "#5B6770",
"NEUTRAL": "#888888",
"RULE": "#D0D7DE",
"GRID": "#EFEFEF",
"TILE_BG": "#F7F7F7",
"PAPER": "#FFFFFF",
}
# ─────────────────────────────────────────────── 상태 4중 코딩 매핑 (§6.2)
@dataclass(frozen=True, slots=True)
class StatusStyle:
label: str
symbol: str
bg: str
fg: str
accent: str
strikeout: bool = False
sort_key: int = 9
STATUS: dict[str, StatusStyle] = {
"취하": StatusStyle("취하", "✕", "#FBE5DC", "#8A1D00", "#D55E00", True, 1),
"신규": StatusStyle("신규", "＋", "#E3F3EA", "#005A32", "#009E73", False, 2),
"변경": StatusStyle("변경", "◆", "#FDF0DC", "#7A3E00", "#E69F00", False, 3),
"워치": StatusStyle("★", "★", "#F7E9F0", "#7A2E55", "#CC79A7", False, 4),
"정보": StatusStyle("OK", "✓", "#E5F1F8", "#00456B", "#0072B2", False, 5),
"없음": StatusStyle("-", "·", "#FFFFFF", "#5B6770", "#888888", False, 8),
"오류": StatusStyle("?", "⚠", "#EFEFEF", "#3A3A3A", "#888888", False, 9),
}
EVENT_TO_STATUS = {"NEW": "신규", "CHANGED": "변경", "WITHDRAWN": "취하"}
# ─────────────────────────────────────────────────────── 표시 형식 (§6.7)
NUMFMT: dict[str, str] = {
"INT": "#,##0",
"INT_DASH": "#,##0;;-",
"INT_UNIT": '#,##0"건"',
"DELTA": "▲ #,##0;▼ -#,##0;– 0",
"DELTA_PCT": "▲ 0.0%;▼ -0.0%;– 0.0%",
"DATE": "yyyy-mm-dd",
"DATETIME": "yyyy-mm-dd hh:mm:ss",
"TEXT": "@",
"RATIO": "0.0%",
"SEC": '0.0"초"',
}
# ─────────────────────────────────────────────────── 탭 색 · 시트명 (§2.1)
SHEET_NAMES = {
"dashboard": "00_대시보드",
"changes": "01_오늘변경분",
"ledger": "02_전체현황",
"ingredient": "03_성분별",
"company": "04_업체별",
"watchlist": "05_워치리스트",
"trend": "06_추이",
"meta": "99_메타",
}
TAB_COLORS = {
"00_대시보드": "#0072B2",
"01_오늘변경분": "#D55E00",
"02_전체현황": "#5B6770",
"03_성분별": "#009E73",
"04_업체별": "#009E73",
"05_워치리스트": "#CC79A7",
"06_추이": "#0072B2",
"99_메타": "#888888",
}
TABLE_STYLE = "Table Style Light 11"
def qsheet(name: str) -> str:
"""시트명을 수식·링크에 쓸 수 있게 작은따옴표로 감싼다.
이 워크북의 시트명은 전부 숫자로 시작하므로 인용이 필수다.
시트명 안의 작은따옴표는 Excel 규칙대로 두 번 반복해 이스케이프한다.
"""
return "'" + name.replace("'", "''") + "'"
class FormatCache:
"""같은 속성 조합에 대해 Format 객체를 단 한 번만 만든다."""
def __init__(self, wb: xlsxwriter.Workbook, font_name: str = "맑은 고딕") -> None:
self._wb = wb
self._font = font_name
self._cache: dict[tuple[tuple[str, Any], ...], Any] = {}
def get(self, **props: Any):
props.setdefault("font_name", self._font)
props.setdefault("font_size", 10)
props.setdefault("font_color", PALETTE["INK"])
key = tuple(sorted(props.items(), key=lambda kv: kv[0]))
fmt = self._cache.get(key)
if fmt is None:
fmt = self._wb.add_format(props)
self._cache[key] = fmt
return fmt
# ── 자주 쓰는 조합 (이름으로 접근) ────────────────────────────────
def title(self, size: int = 18):
return self.get(font_size=size, bold=True, valign="vcenter")
def subtitle(self):
return self.get(font_size=11, font_color=PALETTE["MUTED"], valign="vcenter")
def header(self, wrap: bool = True):
return self.get(bold=True, font_color="#FFFFFF", bg_color=PALETTE["INK"],
align="center", valign="vcenter", text_wrap=wrap,
bottom=2, bottom_color=PALETTE["INK"])
def cell_text(self, wrap: bool = False, size: int = 10, color: str | None = None):
return self.get(font_size=size, font_color=color or PALETTE["INK"],
align="left", valign="vcenter" if not wrap else "top",
text_wrap=wrap, num_format=NUMFMT["TEXT"],
bottom=7, bottom_color=PALETTE["RULE"])
def cell_num(self, fmt_key: str = "INT_DASH"):
return self.get(align="center", valign="vcenter",
num_format=NUMFMT[fmt_key],
bottom=7, bottom_color=PALETTE["RULE"])
def cell_date(self):
return self.get(align="center", valign="vcenter",
num_format=NUMFMT["DATE"],
bottom=7, bottom_color=PALETTE["RULE"])
def link_internal(self, bold: bool = False):
return self.get(font_color=PALETTE["ACCENT"], bold=bold, underline=0,
align="center", valign="vcenter",
bottom=7, bottom_color=PALETTE["RULE"])
def link_external(self):
return self.get(font_color=PALETTE["ACCENT"], underline=1,
align="center", valign="vcenter",
bottom=7, bottom_color=PALETTE["RULE"])
def nav_button(self):
return self.get(bold=True, font_color=PALETTE["ACCENT"],
bg_color=PALETTE["TILE_BG"], align="center",
valign="vcenter", border=1, border_color=PALETTE["RULE"])
def footnote(self):
return self.get(font_size=9, font_color=PALETTE["MUTED"], valign="vcenter")
def status_cell(self, status: str, *, align: str = "center"):
s = STATUS[status]
return self.get(bg_color=s.bg, font_color=s.fg, bold=True,
align=align, valign="vcenter",
font_strikeout=s.strikeout,
bottom=7, bottom_color=PALETTE["RULE"])
def cf(self, status: str):
"""조건부서식 전용 — 배경·폰트·취소선만 지정(테두리는 건드리지 않는다)."""
s = STATUS[status]
props = {"bg_color": s.bg, "font_color": s.fg}
if s.strikeout:
props["font_strikeout"] = True
return self._wb.add_format(props) # CF 서식은 캐시하지 않는다(부분 서식이라 재사용 위험)
cf()를 캐시하지 않는 이유: 조건부서식용 Format 은 "지정한 속성만 덮어쓰는" 부분 서식이다. 일반 셀 서식 캐시와 섞이면 폰트 크기·테두리까지 딸려 들어가 예상 밖의 결과를 만든다. 규칙마다 새로 만드는 비용은 규칙 수(40개 미만)를 고려하면 무시할 만하다.
8.2 src/dmf_crawler/report/widgets.py
"""리포트 공용 위젯: 한글 열너비, 시트 머리글, KPI 타일, Top 표."""
from __future__ import annotations
import unicodedata
from typing import Any, Iterable, Sequence
from .theme import NUMFMT, PALETTE, STATUS, FormatCache, qsheet
# ───────────────────────────────────────── 한글 폭 계산 (§6.4)
# east_asian_width 의 W/F 를 2.0 으로 잡으면 맑은 고딕에서 과대 추정된다.
# 실무 경험치 1.8 을 쓴다. A(Ambiguous)는 1.2.
_EAW_WIDTH: dict[str, float] = {
"F": 1.8, # Fullwidth
"H": 1.0, # Halfwidth
"W": 1.8, # Wide (한글·한자·가나)
"Na": 1.0, # Narrow
"A": 1.2, # Ambiguous
"N": 1.0, # Neutral
}
def display_width(value: Any) -> float:
"""맑은 고딕 10pt 기준 셀 표시 폭(문자 단위)을 근사한다.
줄바꿈이 든 문자열은 가장 긴 줄을 기준으로 한다.
"""
if value is None:
return 0.0
text = str(value)
if not text:
return 0.0
best = 0.0
for line in text.split("\n"):
w = 0.0
for ch in line:
w += _EAW_WIDTH.get(unicodedata.east_asian_width(ch), 1.0)
if w > best:
best = w
return best
def compute_col_widths(
header: Sequence[Any],
rows: Iterable[Sequence[Any]],
*,
min_w: float = 8.0,
max_w: float = 48.0,
margin: float = 2.0,
sample: int = 2000,
overrides: dict[int, float] | None = None,
) -> list[float]:
"""헤더 + 데이터 표본으로 열별 폭 리스트를 만든다.
overrides 로 특정 열의 폭을 강제 고정한다(스파크라인 열 등).
"""
widths = [display_width(h) for h in header]
for i, row in enumerate(rows):
if i >= sample:
break
for c, v in enumerate(row):
if c >= len(widths):
widths.append(0.0)
w = display_width(v)
if w > widths[c]:
widths[c] = w
result = [max(min_w, min(max_w, w + margin)) for w in widths]
for c, w in (overrides or {}).items():
if 0 <= c < len(result):
result[c] = w
return result
def apply_col_widths(ws, widths: Sequence[float],
formats: Sequence[Any] | None = None,
options: dict[int, dict] | None = None) -> None:
"""열 너비 + 열 기본 서식 + 열 옵션(hidden/level)을 적용한다."""
opts = options or {}
for i, w in enumerate(widths):
fmt = formats[i] if formats else None
ws.set_column(i, i, w, fmt, opts.get(i, {}))
# ───────────────────────────────────────── 시트 머리글 (§2.3)
def write_sheet_header(ws, fc: FormatCache, *, title: str, dashboard: str,
report_date: str, generated_at: str,
last_col: int) -> None:
"""1행에 역링크 + 제목 + 생성 정보를 쓰고, 2행을 여백으로 비운다."""
ws.set_row(0, 24)
ws.set_row(1, 6)
ws.write_url(0, 0, f"internal:{qsheet(dashboard)}!A1",
fc.get(font_color=PALETTE["ACCENT"], valign="vcenter"),
string="← 대시보드", tip="대시보드로 돌아가기")
ws.write(0, 1, title, fc.get(font_size=14, bold=True, valign="vcenter"))
ws.write(0, max(3, last_col - 3),
f"기준일 {report_date} · 생성 {generated_at}",
fc.get(font_size=9, font_color=PALETTE["MUTED"],
align="right", valign="vcenter"))
def write_table_header(ws, fc: FormatCache, row: int, header: Sequence[str],
comments: dict[int, str] | None = None,
height: float = 32.0) -> None:
"""헤더 행을 쓰고 정의 툴팁(셀 주석)을 단다."""
ws.set_row(row, height)
hfmt = fc.header()
for c, name in enumerate(header):
ws.write(row, c, name, hfmt)
for c, text in (comments or {}).items():
ws.write_comment(row, c, text,
{"width": 220, "height": 90, "font_name": "맑은 고딕",
"font_size": 9, "x_scale": 1, "y_scale": 1})
def write_empty_notice(ws, fc: FormatCache, row: int, last_col: int,
message: str = "(해당 없음)") -> None:
ws.merge_range(row, 0, row, last_col, message,
fc.get(font_color=PALETTE["MUTED"], align="center",
valign="vcenter", italic=True))
# ───────────────────────────────────────── KPI 타일 (§4.4)
KPI_TILES: tuple[tuple[str, int, int, str, str], ...] = (
# (라벨, 시작 열(0-index), 끝 열, 액센트, 스파크라인 타입)
("오늘 신규", 1, 2, "#009E73", "column"), # B:C
("오늘 변경", 4, 5, "#E69F00", "column"), # E:F
("오늘 취하", 7, 8, "#D55E00", "column"), # H:I
("오늘 총 변경분", 10, 11, "#0072B2", "column"), # K:L
("워치리스트 히트",13, 14, "#CC79A7", "win_loss"), # N:O
("전체 누적 등록", 16, 17, "#1F2933", "line"), # Q:R
)
def tile_formats(fc: FormatCache, accent: str) -> tuple[Any, Any, Any, Any]:
"""KPI 타일 4행(라벨/값/델타/스파크라인)용 서식 세트."""
common = {
"bg_color": PALETTE["TILE_BG"],
"align": "center", "valign": "vcenter",
"left": 1, "left_color": PALETTE["RULE"],
"right": 1, "right_color": PALETTE["RULE"],
}
label = fc.get(**common, font_size=10, bold=True,
font_color=PALETTE["MUTED"], top=5, top_color=accent)
value = fc.get(**common, font_size=28, bold=True,
font_color=accent, num_format=NUMFMT["INT"])
delta = fc.get(**common, font_size=10, bold=True,
font_color=PALETTE["MUTED"], num_format=NUMFMT["DELTA"])
spark = fc.get(**common, bottom=1, bottom_color=PALETTE["RULE"])
return label, value, delta, spark
def write_kpi_tiles(ws, fc: FormatCache, *, trend_sheet: str,
trend_last_row: int, trend_first_row: int,
values: Sequence[int], deltas: Sequence[int],
value_formulas: Sequence[str | None],
delta_formulas: Sequence[str | None]) -> None:
"""5~8행에 KPI 타일 6장을 그린다.
trend_last_row / trend_first_row 는 워크시트 1-index 행 번호다.
value_formulas / delta_formulas 는 None 이면 값만 직접 쓴다.
"""
ws.set_row(4, 18)
ws.set_row(5, 34)
ws.set_row(6, 16)
ws.set_row(7, 14)
# 스파크라인 원본 열: B(1) 신규, C(2) 변경, D(3) 취하, E(4) 합계, F(5) 워치, G(6) 누적
src_cols = ("B", "C", "D", "E", "F", "G")
q = qsheet(trend_sheet)
for i, (label, c1, c2, accent, stype) in enumerate(KPI_TILES):
f_label, f_value, f_delta, f_spark = tile_formats(fc, accent)
ws.merge_range(4, c1, 4, c2, label, f_label)
if value_formulas[i]:
ws.merge_range(5, c1, 5, c2, "", f_value)
ws.write_formula(5, c1, value_formulas[i], f_value, values[i])
else:
ws.merge_range(5, c1, 5, c2, values[i], f_value)
if delta_formulas[i]:
ws.merge_range(6, c1, 6, c2, "", f_delta)
ws.write_formula(6, c1, delta_formulas[i], f_delta, deltas[i])
else:
ws.merge_range(6, c1, 6, c2, deltas[i], f_delta)
ws.merge_range(7, c1, 7, c2, "", f_spark)
col = src_cols[i]
opts: dict[str, Any] = {
"range": f"{q}!${col}${trend_first_row}:${col}${trend_last_row}",
"type": stype,
"series_color": accent,
"high_point": True,
"last_point": True,
"empty_cells": "zero",
}
if stype == "win_loss":
opts["negative_points"] = True
opts.pop("high_point")
ws.add_sparkline(7, c1, opts)
# ───────────────────────────────────────── Top 표 (§4.6)
def write_mini_table(ws, fc: FormatCache, *, header_row: int, first_col: int,
title: str, title_link: str | None,
header: Sequence[str], rows: Sequence[Sequence[Any]],
max_rows: int = 10,
formulas: dict[int, list[tuple[str, Any]]] | None = None) -> None:
"""대시보드용 소형 표. title 행 위에 제목을 얹고 header_row 에 헤더를 쓴다."""
ncol = len(header)
hfmt = fc.get(bold=True, font_color="#FFFFFF", bg_color=PALETTE["INK"],
align="center", valign="vcenter")
if title_link:
ws.write_url(header_row - 1, first_col, title_link,
fc.get(bold=True, font_color=PALETTE["ACCENT"],
valign="vcenter"), string=title)
else:
ws.write(header_row - 1, first_col, title,
fc.get(bold=True, valign="vcenter"))
for c, name in enumerate(header):
ws.write(header_row, first_col + c, name, hfmt)
text_fmt = fc.get(align="left", valign="vcenter", num_format=NUMFMT["TEXT"],
bottom=7, bottom_color=PALETTE["RULE"])
num_fmt = fc.get(align="center", valign="vcenter",
num_format=NUMFMT["INT_DASH"],
bottom=7, bottom_color=PALETTE["RULE"])
for r in range(max_rows):
wrow = header_row + 1 + r
if r >= len(rows):
for c in range(ncol):
ws.write_blank(wrow, first_col + c, None, text_fmt if c == 1 else num_fmt)
continue
for c, v in enumerate(rows[r][:ncol]):
fmt = text_fmt if isinstance(v, str) else num_fmt
fset = (formulas or {}).get(c)
if fset is not None and r < len(fset):
formula, cached = fset[r]
ws.write_formula(wrow, first_col + c, formula, num_fmt, cached)
else:
ws.write(wrow, first_col + c, v, fmt)
8.3 src/dmf_crawler/report/atomic.py
"""원자적 xlsx 저장 — Excel 이 파일을 잡고 있어도 데이터를 잃지 않는다.
아키텍처 §3.13 계약:
atomic_write(build_fn, target, retries=3, backoff_seconds=2.0)
-> tuple[Path, bool] # (실제 저장 경로, 폴백 사용 여부)
"""
from __future__ import annotations
import os
import shutil
import tempfile
import time
import zipfile
from datetime import datetime
from pathlib import Path
from typing import Callable
from ..errors import ReportError
_MIN_XLSX_BYTES = 4096
def excel_lock_file(path: Path) -> Path:
"""Excel 이 만드는 숨김 잠금 파일 경로(~$name.xlsx)."""
return path.with_name("~$" + path.name)
def looks_locked(path: Path) -> bool:
"""빠른 사전 판단. 확정적이지 않으므로 실제 시도의 보조로만 쓴다."""
if excel_lock_file(path).exists():
return True
if not path.exists():
return False
try:
with open(path, "r+b"):
return False
except OSError:
return True
def verify_xlsx(path: Path, min_bytes: int = _MIN_XLSX_BYTES) -> None:
"""저장된 xlsx 가 온전한지 stdlib 만으로 검사한다(openpyxl 불필요)."""
if not path.exists():
raise ReportError(f"출력 파일이 생성되지 않았다: {path}")
size = path.stat().st_size
if size < min_bytes:
raise ReportError(f"출력 파일이 너무 작다({size} bytes): {path}")
try:
with zipfile.ZipFile(path) as zf:
bad = zf.testzip()
if bad is not None:
raise ReportError(f"손상된 zip 엔트리: {bad}")
names = set(zf.namelist())
except zipfile.BadZipFile as exc:
raise ReportError(f"xlsx 가 zip 으로 열리지 않는다: {path}") from exc
for required in ("xl/workbook.xml", "[Content_Types].xml"):
if required not in names:
raise ReportError(f"{required} 가 없다. 올바른 xlsx 가 아니다: {path}")
def fallback_path(target: Path, now: datetime | None = None) -> Path:
"""DMF_리포트_2026-09-02_060241.xlsx 형태의 폴백 이름."""
stamp = (now or datetime.now()).strftime("%H%M%S")
return target.with_name(f"{target.stem}_{stamp}{target.suffix}")
def atomic_write(build_fn: Callable[[Path], None], target: Path,
retries: int = 3, backoff_seconds: float = 2.0) -> tuple[Path, bool]:
"""build_fn(tmp) 으로 임시 파일을 만든 뒤 target 으로 원자 교체한다.
- 같은 볼륨에 임시 파일을 만들어야 os.replace 가 원자적이다.
- 쓰기 도중 죽어도 target 은 이전 상태 그대로 남는다.
- 잠김 시 지수 백오프로 재시도하고, 끝내 실패하면 시각 접미사 폴백.
"""
target = Path(target).resolve()
target.parent.mkdir(parents=True, exist_ok=True)
fd, tmp_name = tempfile.mkstemp(prefix=".~dmf_", suffix=".xlsx",
dir=str(target.parent))
os.close(fd)
tmp_path = Path(tmp_name)
try:
build_fn(tmp_path)
verify_xlsx(tmp_path)
for attempt in range(max(1, retries)):
if not looks_locked(target):
try:
os.replace(tmp_path, target)
return target, False
except PermissionError:
pass
if attempt < retries - 1:
time.sleep(backoff_seconds * (2 ** attempt))
alt = fallback_path(target)
shutil.move(str(tmp_path), str(alt))
return alt, True
finally:
if tmp_path.exists():
try:
tmp_path.unlink()
except OSError:
pass
def refresh_latest_link(source: Path, latest: Path) -> bool:
"""최신본 고정 파일을 원자적 '복사'로 갱신한다.
Windows 심볼릭 링크는 관리자 권한 또는 개발자 모드를 요구하므로 쓰지 않는다.
하드링크도 쓰지 않는다 — 원본을 지우면 최신본이 남아 혼란스럽다.
잠겨 있으면 조용히 False 를 반환한다(리포트 본체는 이미 저장됐다).
"""
if not source.exists():
return False
latest.parent.mkdir(parents=True, exist_ok=True)
fd, tmp_name = tempfile.mkstemp(prefix=".~dmflatest_", suffix=".xlsx",
dir=str(latest.parent))
os.close(fd)
tmp_path = Path(tmp_name)
try:
shutil.copyfile(source, tmp_path)
if looks_locked(latest):
return False
os.replace(tmp_path, latest)
return True
except OSError:
return False
finally:
if tmp_path.exists():
try:
tmp_path.unlink()
except OSError:
pass
8.4 src/dmf_crawler/report/data.py
⚠️ 아래 SQL 은
docs/design/02-data-model.md(M1 에서 작성) 확정 전의 잠정 스키마를 전제한다. 컬럼명이 바뀌면 이 파일만 고치면 되도록 SQL 을 모듈 상수로 몰아 두었다.
전제 스키마 요약 (0001~0003 마이그레이션)
runs(run_id TEXT PK, run_date TEXT, trigger TEXT, started_at TEXT, finished_at TEXT,
status TEXT, exit_code INT, notes TEXT)
fetch_stats(run_id TEXT PK, total_count_reported INT, pages INT, fetched_rows INT,
duration_ms INT, http_summary TEXT, archive_dir TEXT, snapshot_hash TEXT)
snapshots(run_id TEXT, dmf_key TEXT, permit_no TEXT, ingredient_name TEXT,
applicant TEXT, manufacturer TEXT, manufacture_place TEXT, countries TEXT,
permit_date TEXT, accepted_date TEXT, content_hash TEXT,
PRIMARY KEY(run_id, dmf_key))
records(dmf_key TEXT PK, permit_no TEXT, ingredient_name TEXT, applicant TEXT,
manufacturer TEXT, manufacture_place TEXT, countries TEXT, permit_date TEXT,
accepted_date TEXT, content_hash TEXT, first_seen_date TEXT,
last_seen_date TEXT, last_changed_date TEXT, status TEXT, last_run_id TEXT)
events(event_id INTEGER PK, run_id TEXT, run_date TEXT, dmf_key TEXT,
event_type TEXT, field TEXT, before_value TEXT, after_value TEXT, created_at TEXT)
enrichment_run(run_id TEXT PK, headline TEXT, summary TEXT, risk_note TEXT,
model TEXT, created_at TEXT)
enrichment(run_id TEXT, dmf_key TEXT, note TEXT, importance INT,
PRIMARY KEY(run_id, dmf_key))
agy_calls(id INTEGER PK, run_id TEXT, status TEXT, model TEXT, effort TEXT,
input_tokens INT, output_tokens INT, duration_ms INT, error_kind TEXT)
watchlist(id INTEGER PK, target_type TEXT, keyword TEXT, match_mode TEXT,
priority INT, active TEXT, memo TEXT, created_at TEXT)
"""DB → ReportData. 리포트는 이 모듈을 통해서만 DB를 읽는다."""
from __future__ import annotations
import sqlite3
from dataclasses import dataclass, field
from datetime import date, datetime
from typing import Any
# ─────────────────────────────────────────────────────── 자료구조
@dataclass(frozen=True, slots=True)
class ChangeRow:
sort_key: int
status: str # 신규 / 변경 / 취하
permit_no: str
ingredient_name: str
applicant: str
manufacturer: str
manufacture_place: str
countries: str
permit_date: date | None
accepted_date: date | None
changed_fields: str
change_detail: str
watch_hits: int
watch_keywords: str
ai_note: str
dmf_key: str
content_hash: str
@dataclass(frozen=True, slots=True)
class LedgerRow:
permit_no: str
ingredient_name: str
applicant: str
manufacturer: str
manufacture_place: str
countries: str
country_count: int
permit_date: date | None
accepted_date: date | None
first_seen: date | None
last_seen: date | None
last_changed: date | None
status: str # 유효 / 취하
elapsed_days: int | None
watch_flag: str # Y / N
dmf_key: str
content_hash: str
@dataclass(frozen=True, slots=True)
class AggRow:
rank: int
name: str
key: str
today_new: int
today_changed: int
today_withdrawn: int
last7: int
last30: int
cumulative: int
partner_count: int # 성분→등록업체 수 / 업체→보유 성분 수
country_count: int # 성분→제조국 수 / 업체→제조소 수
main_country: str
domestic: str
delta: int
last_date: date | None
watch_flag: str
series30: tuple[int, ...]
@dataclass(frozen=True, slots=True)
class WatchRow:
no: int
target_type: str
keyword: str
match_mode: str
priority: int
active: str
memo: str
today: int
last7: int
last30: int
cumulative: int
last_hit: date | None
@dataclass(frozen=True, slots=True)
class WatchDetailRow:
keyword: str
status: str
permit_no: str
ingredient_name: str
applicant: str
countries: str
permit_date: date | None
@dataclass(frozen=True, slots=True)
class TrendRow:
day: date
new: int
changed: int
withdrawn: int
total: int
watch_hits: int
active_total: int
fetched_rows: int
run_status: str
@dataclass(frozen=True, slots=True)
class GateRow:
name: str
expected: str
actual: str
verdict: str
detail: str
@dataclass(frozen=True, slots=True)
class MetaInfo:
report_date: date
run_id: str
trigger: str
started_at: str
finished_at: str
run_status: str
host: str
python_version: str
xlsxwriter_version: str
prev_run_id: str
log_dir: str
events_path: str
archive_dir: str
backup_path: str
snapshot_hash: str
prev_report_path: str
source_name: str
source_url: str
http_summary: str
fetched_rows: int
is_baseline: bool
banner_kind: str # none / partial / baseline / ai
banner_text: str
@dataclass(frozen=True, slots=True)
class AiInfo:
used: str # 사용 / 건너뜀(...) / 실패
headline: str
summary: str
risk_note: str
model: str
effort: str
input_tokens: int
output_tokens: int
duration_s: float
tokens_today: int
token_cap: int
envelope_path: str
@dataclass(frozen=True, slots=True)
class ReportData:
meta: MetaInfo
ai: AiInfo
kpi: dict[str, int] # new/changed/withdrawn/total/watch/active
kpi_delta: dict[str, int]
changes: tuple[ChangeRow, ...]
ledger: tuple[LedgerRow, ...]
ingredients: tuple[AggRow, ...]
companies: tuple[AggRow, ...]
watch: tuple[WatchRow, ...]
watch_detail: tuple[WatchDetailRow, ...]
trend: tuple[TrendRow, ...]
gates: tuple[GateRow, ...]
top_countries: tuple[tuple[str, int], ...]
trend_dates: tuple[date, ...] = field(default_factory=tuple)
# ─────────────────────────────────────────────────────── SQL 상수
SQL_RUN = """
SELECT r.run_id, r.run_date, r.trigger, r.started_at, r.finished_at,
r.status, r.notes,
f.total_count_reported, f.fetched_rows, f.pages, f.duration_ms,
f.http_summary, f.archive_dir, f.snapshot_hash
FROM runs r LEFT JOIN fetch_stats f ON f.run_id = r.run_id
WHERE r.run_id = ?
"""
SQL_PREV_RUN = """
SELECT run_id, run_date FROM runs
WHERE status IN ('SUCCESS','PARTIAL') AND run_id < ?
ORDER BY run_id DESC LIMIT 1
"""
SQL_CHANGES = """
SELECT e.dmf_key,
e.event_type,
GROUP_CONCAT(DISTINCT e.field) AS fields,
GROUP_CONCAT(e.field || ': ' || COALESCE(e.before_value,'-') ||
' → ' || COALESCE(e.after_value,'-'), CHAR(10)) AS detail,
c.permit_no, c.ingredient_name, c.applicant, c.manufacturer,
c.manufacture_place, c.countries, c.permit_date, c.accepted_date,
c.content_hash,
COALESCE(x.note, '') AS ai_note
FROM events e
LEFT JOIN records c ON c.dmf_key = e.dmf_key
LEFT JOIN enrichment x ON x.run_id = e.run_id AND x.dmf_key = e.dmf_key
WHERE e.run_id = ?
GROUP BY e.dmf_key, e.event_type
"""
SQL_LEDGER = """
SELECT permit_no, ingredient_name, applicant, manufacturer, manufacture_place,
countries, permit_date, accepted_date, first_seen_date, last_seen_date,
last_changed_date, status, dmf_key, content_hash
FROM records
ORDER BY accepted_date DESC, permit_no ASC
"""
SQL_INGREDIENT_TODAY = """
SELECT r.ingredient_name,
SUM(CASE WHEN e.event_type='NEW' THEN 1 ELSE 0 END) AS n_new,
SUM(CASE WHEN e.event_type='CHANGED' THEN 1 ELSE 0 END) AS n_chg,
SUM(CASE WHEN e.event_type='WITHDRAWN' THEN 1 ELSE 0 END) AS n_wdr
FROM events e JOIN records r ON r.dmf_key = e.dmf_key
WHERE e.run_id = ?
GROUP BY r.ingredient_name
"""
SQL_INGREDIENT_WINDOW = """
SELECT r.ingredient_name, e.run_date, COUNT(DISTINCT e.dmf_key) AS n
FROM events e JOIN records r ON r.dmf_key = e.dmf_key
WHERE e.run_date >= ?
GROUP BY r.ingredient_name, e.run_date
"""
SQL_INGREDIENT_CUM = """
SELECT ingredient_name,
COUNT(*) AS cum,
COUNT(DISTINCT applicant) AS n_applicant,
COUNT(DISTINCT countries) AS n_country,
MAX(accepted_date) AS last_date,
SUM(CASE WHEN countries LIKE '%대한민국%' THEN 1 ELSE 0 END) AS n_domestic
FROM records
WHERE status = '유효'
GROUP BY ingredient_name
"""
SQL_COMPANY_TODAY = SQL_INGREDIENT_TODAY.replace("r.ingredient_name", "r.applicant")
SQL_COMPANY_WINDOW = SQL_INGREDIENT_WINDOW.replace("r.ingredient_name", "r.applicant")
SQL_COMPANY_CUM = """
SELECT applicant,
COUNT(*) AS cum,
COUNT(DISTINCT ingredient_name) AS n_ingredient,
COUNT(DISTINCT manufacturer) AS n_site,
MAX(accepted_date) AS last_date,
SUM(CASE WHEN countries LIKE '%대한민국%' THEN 1 ELSE 0 END) AS n_domestic
FROM records
WHERE status = '유효'
GROUP BY applicant
"""
SQL_TREND = """
SELECT r.run_date,
SUM(CASE WHEN e.event_type='NEW' THEN 1 ELSE 0 END) AS n_new,
SUM(CASE WHEN e.event_type='CHANGED' THEN 1 ELSE 0 END) AS n_chg,
SUM(CASE WHEN e.event_type='WITHDRAWN' THEN 1 ELSE 0 END) AS n_wdr,
MAX(r.status) AS run_status,
MAX(COALESCE(f.fetched_rows, 0)) AS fetched
FROM runs r
LEFT JOIN events e ON e.run_id = r.run_id
LEFT JOIN fetch_stats f ON f.run_id = r.run_id
WHERE r.run_date >= ? AND r.status IN ('SUCCESS','PARTIAL')
GROUP BY r.run_date
ORDER BY r.run_date ASC
"""
SQL_ACTIVE_BY_DAY = """
SELECT run_id, run_date, COUNT(*) AS n
FROM snapshots s JOIN runs u USING (run_id)
WHERE u.run_date >= ? AND u.status IN ('SUCCESS','PARTIAL')
GROUP BY run_id, run_date
ORDER BY run_date ASC
"""
SQL_TOP_COUNTRY = """
SELECT TRIM(value) AS country, COUNT(*) AS n
FROM records, json_each('["' || REPLACE(countries, ',', '","') || '"]')
WHERE status = '유효' AND TRIM(value) <> ''
GROUP BY country
ORDER BY n DESC
LIMIT 8
"""
SQL_WATCHLIST = """
SELECT id, target_type, keyword, match_mode, priority, active, memo
FROM watchlist
WHERE active = 'Y'
ORDER BY priority ASC, id ASC
"""
SQL_AGY = """
SELECT status, model, effort, input_tokens, output_tokens, duration_ms, error_kind
FROM agy_calls WHERE run_id = ? ORDER BY id DESC LIMIT 1
"""
SQL_AGY_TOKENS_TODAY = """
SELECT COALESCE(SUM(input_tokens + output_tokens), 0)
FROM agy_calls a JOIN runs r USING (run_id)
WHERE r.run_date = ?
"""
SQL_ENRICHMENT_RUN = """
SELECT headline, summary, risk_note, model FROM enrichment_run WHERE run_id = ?
"""
def fetch_report_data(conn: sqlite3.Connection, run_id: str, cfg) -> ReportData:
"""모든 시트가 필요로 하는 데이터를 한 번에 조립한다.
- 워치리스트 매칭은 파이썬에서 수행한다(정규식 모드가 SQL 로 불가능).
- 파생 컬럼(elapsed_days, series30 등)도 여기서 계산해 시트 빌더는
'쓰기'만 하게 한다.
"""
conn.row_factory = sqlite3.Row
# 구현 본문은 위 SQL 을 순서대로 실행해 dataclass 로 옮기는 단순 매핑이다.
# 규칙 3가지만 지키면 된다:
# 1) 모든 날짜 문자열은 datetime.date 로 변환해 넣는다(엑셀 날짜 서식용).
# 2) 값이 없으면 None 이 아니라 '-' 를 쓸 자리는 시트 빌더가 판단한다.
# data.py 는 None 을 그대로 넘긴다.
# 3) trend 는 오래된 날짜가 먼저 오도록 ASC 로 정렬한다.
raise NotImplementedError("M3 에서 위 SQL 매핑을 채운다")
8.5 src/dmf_crawler/report/build.py
"""워크북 조립 오케스트레이션."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
import xlsxwriter
from ..errors import ReportError
from .atomic import atomic_write, refresh_latest_link
from .data import ReportData, fetch_report_data
from .sheets import SHEET_BUILDERS
from .theme import SHEET_NAMES, TAB_COLORS, FormatCache, qsheet
@dataclass(frozen=True, slots=True)
class ReportOutcome:
path: Path
latest_link_path: Path | None
used_fallback_name: bool
sheets: tuple[str, ...]
warnings: tuple[str, ...]
class BuildContext:
"""빌더 함수들이 공유하는 상태. 시트 간 참조(행 번호 등)를 여기에 남긴다."""
def __init__(self, wb: xlsxwriter.Workbook, rd: ReportData, cfg) -> None:
self.wb = wb
self.rd = rd
self.cfg = cfg
self.fc = FormatCache(wb, font_name=cfg.font)
self.sheets: dict[str, object] = {}
self.warnings: list[str] = []
# 시트 간 연동에 필요한 좌표들
self.ledger_row_of: dict[str, int] = {} # dmf_key -> 워크시트 1-index 행
self.trend_first_row: int = 4
self.trend_last_row: int = 4
self.spark_first_row: int = 4
self.ledger_last_row: int = 3
self.changes_last_row: int = 3
self.has_watchlist: bool = False
def add_sheet(self, key: str):
name = SHEET_NAMES[key]
ws = self.wb.add_worksheet(name)
ws.set_tab_color(TAB_COLORS[name])
self.sheets[key] = ws
return ws
def _build_workbook(rd: ReportData, cfg, tmp_path: Path) -> tuple[list[str], list[str]]:
wb = xlsxwriter.Workbook(str(tmp_path), {
"constant_memory": False,
"strings_to_numbers": False,
"strings_to_urls": False,
"nan_inf_to_errors": True,
"use_future_functions": True,
"default_date_format": "yyyy-mm-dd",
"remove_timezone": True,
})
wb.set_size(1500, 900)
wb.set_properties({
"title": f"DMF 일일 모니터링 리포트 {rd.meta.report_date:%Y-%m-%d}",
"subject": "원료의약품 등록(DMF) 신규·변경·취하 현황",
"author": "DMF Crawler",
"category": "규제 인텔리전스",
"keywords": "DMF, 원료의약품, 식약처, 등록현황",
"comments": f"run_id={rd.meta.run_id} / 자동 생성 / 출처: 공공데이터포털 15057075",
})
ctx = BuildContext(wb, rd, cfg)
built: list[str] = []
try:
# ── 1단계: 원본 시트를 먼저 만든다 ──────────────────────────
# 06_추이 를 먼저 만들어야 대시보드 스파크라인 범위를 알 수 있고,
# 02_전체현황 을 만들어야 01_오늘변경분 의 링크 대상 행을 안다.
# 그래서 '생성 순서'와 '탭 순서'를 분리한다.
# XlsxWriter 는 add_worksheet 순서가 곧 탭 순서이므로,
# 시트 객체는 탭 순서대로 미리 만들고 '기록'만 나중에 한다.
for key in ("dashboard", "changes", "ledger", "ingredient", "company"):
ctx.add_sheet(key)
if rd.watch:
ctx.has_watchlist = True
ctx.add_sheet("watchlist")
ctx.add_sheet("trend")
ctx.add_sheet("meta")
# ── 2단계: 의존 순서대로 내용을 채운다 ───────────────────────
for key, builder in SHEET_BUILDERS:
if key == "watchlist" and not ctx.has_watchlist:
continue
builder(ctx)
built.append(SHEET_NAMES[key])
# ── 3단계: 정의된 이름 (모든 시트 생성 후) ────────────────────
t = qsheet(SHEET_NAMES["trend"])
d = qsheet(SHEET_NAMES["dashboard"])
m = qsheet(SHEET_NAMES["meta"])
lg = qsheet(SHEET_NAMES["ledger"])
ch = qsheet(SHEET_NAMES["changes"])
last = ctx.trend_last_row
wb.define_name("REPORT_DATE", f"={m}!$B$3")
wb.define_name("RUN_ID", f"={m}!$B$6")
wb.define_name("KPI_NEW", f"={d}!$B$6")
wb.define_name("KPI_CHANGED", f"={d}!$E$6")
wb.define_name("KPI_WITHDRAWN", f"={d}!$H$6")
wb.define_name("KPI_TOTAL", f"={d}!$K$6")
wb.define_name("KPI_WATCH", f"={d}!$N$6")
wb.define_name("KPI_ACTIVE", f"={d}!$Q$6")
wb.define_name("TREND_DATE", f"={t}!$A$4:$A${last}")
wb.define_name("TREND_NEW", f"={t}!$B$4:$B${last}")
wb.define_name("TREND_CHANGED", f"={t}!$C$4:$C${last}")
wb.define_name("TREND_WITHDRAWN", f"={t}!$D$4:$D${last}")
wb.define_name("LEDGER", f"={lg}!$A$3:$Q${ctx.ledger_last_row}")
wb.define_name("TODAY_CHANGES", f"={ch}!$A$3:$S${ctx.changes_last_row}")
ctx.sheets["dashboard"].activate()
ctx.sheets["dashboard"].set_first_sheet()
finally:
wb.close()
return built, ctx.warnings
def build_report(conn, run_id: str, cfg) -> ReportOutcome:
"""아키텍처 §3.13 의 공개 API. AI 산출물 유무와 무관하게 항상 완주한다."""
rd = fetch_report_data(conn, run_id, cfg)
out_dir = Path(cfg.output_dir)
target = out_dir / cfg.filename_pattern.format(date=f"{rd.meta.report_date:%Y-%m-%d}")
built: list[str] = []
warns: list[str] = []
def _fn(tmp: Path) -> None:
nonlocal built, warns
built, warns = _build_workbook(rd, cfg, tmp)
try:
path, used_fallback = atomic_write(
_fn, target,
retries=cfg.lock_retries,
backoff_seconds=2.0,
)
except Exception as exc: # noqa: BLE001
raise ReportError(
f"리포트 생성 실패: {exc}. "
f"`python -m dmf_crawler report-only --run-id {run_id}` 로 재생성할 수 있다."
) from exc
if used_fallback:
warns.append(
f"원본 파일이 열려 있어 대체 이름으로 저장했다: {path.name}. "
f"Excel 을 닫고 report-only 를 다시 실행하면 정규 파일명으로 저장된다."
)
latest: Path | None = None
if cfg.latest_link_name:
cand = out_dir / cfg.latest_link_name
if refresh_latest_link(path, cand):
latest = cand
else:
warns.append(f"최신본 갱신 실패(열려 있음): {cand.name}")
return ReportOutcome(
path=path,
latest_link_path=latest,
used_fallback_name=used_fallback,
sheets=tuple(built),
warnings=tuple(warns),
)
생성 순서와 탭 순서의 분리가 이 모듈의 핵심 트릭이다. XlsxWriter 는
add_worksheet호출 순서가 곧 탭 순서이므로, 워크시트 객체는 탭 순서대로 미리 전부 만들고, 실제 쓰기는SHEET_BUILDERS의 의존 순서로 한다.
8.6 src/dmf_crawler/report/sheets/__init__.py
"""시트 빌더 등록부. 튜플의 순서 = 실제 '쓰기' 순서(의존 순서)."""
from __future__ import annotations
from . import (s00_dashboard, s01_changes, s02_ledger, s03_ingredient,
s04_company, s05_watchlist, s06_trend, s99_meta)
SHEET_BUILDERS = (
("trend", s06_trend.build), # 1. 스파크라인·차트 원본을 먼저
("ledger", s02_ledger.build), # 2. dmf_key → 행번호 사전 생성
("changes", s01_changes.build), # 3. ledger 사전을 써서 링크
("ingredient", s03_ingredient.build),
("company", s04_company.build),
("watchlist", s05_watchlist.build),
("meta", s99_meta.build), # 4. REPORT_DATE 원본
("dashboard", s00_dashboard.build), # 5. 모든 좌표가 확정된 뒤 마지막
)
8.7 src/dmf_crawler/report/sheets/s00_dashboard.py
"""00_대시보드 — KPI 6 · 차트 2 · Top 표 3 · 내비게이션."""
from __future__ import annotations
from ..theme import NUMFMT, PALETTE, SHEET_NAMES, STATUS, qsheet
from ..widgets import write_kpi_tiles, write_mini_table
COL_WIDTHS = [1.5, 13, 13, 1.5, 13, 13, 1.5, 13, 13, 1.5,
13, 13, 1.5, 13, 13, 1.5, 13, 13, 1.5] # A..S
ROW_HEIGHTS = {0: 6, 1: 30, 2: 18, 3: 8, 8: 10, 24: 10, 25: 18,
40: 10, 41: 20, 42: 20, 46: 14}
NAV = (
(41, 1, 2, "오늘 변경분 →", "changes", "오늘 신규·변경·취하 전체"),
(41, 4, 5, "전체 현황 →", "ledger", "유효 등록 전체 원장"),
(41, 7, 8, "성분별 →", "ingredient", "성분 기준 집계"),
(41, 10, 11, "업체별 →", "company", "업체 기준 집계"),
(41, 13, 14, "워치리스트 →", "watchlist", "관심 키워드 히트"),
(41, 16, 17, "추이 →", "trend", "일자별 시계열"),
(42, 1, 2, "메타·로그 →", "meta", "수집 메타·무결성·AI 상태"),
)
def build(ctx) -> None:
ws, fc, rd = ctx.sheets["dashboard"], ctx.fc, ctx.rd
ws.hide_gridlines(2)
ws.set_zoom(90)
for c, w in enumerate(COL_WIDTHS):
ws.set_column(c, c, w)
for r, h in ROW_HEIGHTS.items():
ws.set_row(r, h)
for r in range(9, 24):
ws.set_row(r, 20)
for r in range(26, 40):
ws.set_row(r, 16)
for r in range(43, 46):
ws.set_row(r, 8)
_title_block(ctx, ws, fc, rd)
_banner(ctx, ws, fc, rd)
_kpis(ctx, ws, fc, rd)
_charts(ctx, ws, rd)
_top_tables(ctx, ws, fc, rd)
_nav(ctx, ws, fc)
_footer(ctx, ws, fc, rd)
_print_setup(ws)
def _title_block(ctx, ws, fc, rd) -> None:
weekday = "월화수목금토일"[rd.meta.report_date.weekday()]
ws.merge_range(1, 1, 1, 17,
f"DMF 일일 모니터링 리포트 — {rd.meta.report_date:%Y-%m-%d}({weekday})",
fc.get(font_size=18, bold=True, valign="vcenter"))
headline = rd.ai.headline or _rule_headline(rd)
ws.merge_range(2, 1, 2, 17, headline, fc.subtitle())
def _rule_headline(rd) -> str:
"""AI 요약이 없을 때 쓰는 규칙 기반 한 줄."""
k = rd.kpi
base = (f"오늘 신규 {k['new']}건 · 변경 {k['changed']}건 · "
f"취하 {k['withdrawn']}건")
if k["watch"]:
names = ", ".join(w.ingredient_name for w in rd.watch_detail[:3])
return f"{base} — 워치리스트 {k['watch']}건 적중({names})"
if k["total"] == 0:
return f"{base} — 전일 대비 변동이 없다"
return f"{base} — 워치리스트 적중 없음"
def _banner(ctx, ws, fc, rd) -> None:
kind, text = rd.meta.banner_kind, rd.meta.banner_text
if kind == "none" or not text:
return
style = {"partial": ("취하", 22), "baseline": ("정보", 22), "ai": ("없음", 22)}[kind]
s = STATUS[style[0]]
ws.set_row(3, style[1])
ws.merge_range(3, 1, 3, 17, text,
fc.get(font_size=11, bold=True, font_color=s.fg,
bg_color=s.bg, align="left", valign="vcenter",
left=5, left_color=s.fg, indent=1))
def _kpis(ctx, ws, fc, rd) -> None:
t = qsheet(SHEET_NAMES["trend"])
e, p = ctx.trend_last_row, ctx.trend_last_row - 1
has_prev = ctx.trend_last_row > ctx.trend_first_row
cols = ("B", "C", "D", "E", "F", "G")
keys = ("new", "changed", "withdrawn", "total", "watch", "active")
vformulas = [f"={t}!${c}${e}" for c in cols]
dformulas = ([f"={t}!${c}${e}-{t}!${c}${p}" for c in cols]
if has_prev else [None] * 6)
write_kpi_tiles(
ws, fc,
trend_sheet=SHEET_NAMES["trend"],
trend_last_row=ctx.trend_last_row,
trend_first_row=ctx.spark_first_row,
values=[rd.kpi[k] for k in keys],
deltas=[rd.kpi_delta[k] for k in keys],
value_formulas=vformulas,
delta_formulas=dformulas,
)
def _charts(ctx, ws, rd) -> None:
wb, t = ctx.wb, SHEET_NAMES["trend"]
s0 = ctx.spark_first_row - 1 # 0-index
e0 = ctx.trend_last_row - 1
ch1 = wb.add_chart({"type": "column", "subtype": "stacked"})
for col, (label, color) in enumerate(
(("신규", "#009E73"), ("변경", "#E69F00"), ("취하", "#D55E00")), start=1):
ch1.add_series({
"name": label,
"categories": [t, s0, 0, e0, 0],
"values": [t, s0, col, e0, col],
"fill": {"color": color},
"border": {"none": True},
"gap": 40,
})
ch1.set_title({"name": "최근 30일 일별 등록 동향",
"name_font": {"name": "맑은 고딕", "size": 11, "bold": True,
"color": PALETTE["INK"]}})
ch1.set_x_axis({"num_font": {"name": "맑은 고딕", "size": 8,
"color": PALETTE["MUTED"]},
"line": {"color": PALETTE["RULE"]},
"major_tick_mark": "none", "date_axis": False})
ch1.set_y_axis({"num_font": {"name": "맑은 고딕", "size": 8,
"color": PALETTE["MUTED"]},
"major_gridlines": {"visible": True,
"line": {"color": PALETTE["GRID"],
"width": 0.75}},
"line": {"none": True}, "major_tick_mark": "none"})
ch1.set_legend({"position": "top",
"font": {"name": "맑은 고딕", "size": 9}})
ch1.set_chartarea({"border": {"color": PALETTE["RULE"]},
"fill": {"color": PALETTE["PAPER"]}})
ch1.set_plotarea({"border": {"none": True}, "fill": {"none": True}})
ch1.set_size({"width": 600, "height": 300})
ws.insert_chart(9, 1, ch1) # 앵커 B10
n = max(1, len(rd.top_countries))
ch2 = wb.add_chart({"type": "bar"})
ch2.add_series({
"name": "건수",
"categories": [t, 3, 26, 2 + n, 26], # AA4:AA{3+n}
"values": [t, 3, 27, 2 + n, 27], # AB4:AB{3+n}
"fill": {"color": PALETTE["ACCENT"]},
"border": {"none": True},
"data_labels": {"value": True, "position": "outside_end",
"font": {"name": "맑은 고딕", "size": 8,
"color": PALETTE["INK"]}},
"gap": 50,
})
ch2.set_title({"name": "제조국 Top 8 (유효 등록 기준)",
"name_font": {"name": "맑은 고딕", "size": 11, "bold": True,
"color": PALETTE["INK"]}})
ch2.set_x_axis({"visible": False})
ch2.set_y_axis({"num_font": {"name": "맑은 고딕", "size": 9,
"color": PALETTE["INK"]},
"line": {"color": PALETTE["RULE"]},
"major_tick_mark": "none", "reverse": True})
ch2.set_legend({"none": True})
ch2.set_chartarea({"border": {"color": PALETTE["RULE"]},
"fill": {"color": PALETTE["PAPER"]}})
ch2.set_plotarea({"border": {"none": True}, "fill": {"none": True}})
ch2.set_size({"width": 600, "height": 300})
ws.insert_chart(9, 10, ch2) # 앵커 K10
def _top_tables(ctx, ws, fc, rd) -> None:
lg = qsheet(SHEET_NAMES["ledger"])
ing = [(a.rank, a.name, a.today_new + a.today_changed + a.today_withdrawn,
a.last30, a.cumulative) for a in rd.ingredients[:10]]
co = [(a.rank, a.name, a.today_new + a.today_changed + a.today_withdrawn,
a.last30, a.cumulative) for a in rd.companies[:10]]
wd = [(w.keyword, 1, w.ingredient_name, w.applicant, w.status)
for w in rd.watch_detail[:10]]
# 누적 열은 시트 간 COUNTIFS 수식 + 캐시값 (§5.3 F13/F14)
f_ing = {4: [(f'=COUNTIFS({lg}!$B:$B,$C{27 + i},{lg}!$M:$M,"유효")', r[4])
for i, r in enumerate(ing)]}
f_co = {4: [(f'=COUNTIFS({lg}!$C:$C,$I{27 + i},{lg}!$M:$M,"유효")', r[4])
for i, r in enumerate(co)]}
write_mini_table(ws, fc, header_row=25, first_col=1,
title="Top 10 성분",
title_link=f"internal:{qsheet(SHEET_NAMES['ingredient'])}!A1",
header=("순위", "성분명", "오늘", "30일", "누적"),
rows=ing, formulas=f_ing)
write_mini_table(ws, fc, header_row=25, first_col=7,
title="Top 10 업체",
title_link=f"internal:{qsheet(SHEET_NAMES['company'])}!A1",
header=("순위", "업체명", "오늘", "30일", "누적"),
rows=co, formulas=f_co)
write_mini_table(ws, fc, header_row=25, first_col=13,
title="워치리스트 히트",
title_link=(f"internal:{qsheet(SHEET_NAMES['watchlist'])}!A1"
if ctx.has_watchlist else None),
header=("키워드", "매칭", "성분명", "업체명", "상태"),
rows=wd)
bar = {"type": "data_bar", "bar_color": PALETTE["ACCENT"], "bar_solid": True}
ws.conditional_format(26, 5, 35, 5, dict(bar)) # F27:F36
ws.conditional_format(26, 11, 35, 11, dict(bar)) # L27:L36
for label in ("신규", "변경", "취하"):
ws.conditional_format(26, 17, 35, 17, {
"type": "text", "criteria": "containing", "value": label,
"format": fc.cf(label),
}) # R27:R36
def _nav(ctx, ws, fc) -> None:
nav_fmt = fc.nav_button()
off_fmt = fc.get(font_color=PALETTE["NEUTRAL"], bg_color=PALETTE["TILE_BG"],
align="center", valign="vcenter",
border=1, border_color=PALETTE["RULE"], italic=True)
for row, c1, c2, text, key, tip in NAV:
if key == "watchlist" and not ctx.has_watchlist:
ws.merge_range(row, c1, row, c2, "(워치리스트 없음)", off_fmt)
continue
ws.merge_range(row, c1, row, c2, "", nav_fmt)
ws.write_url(row, c1, f"internal:{qsheet(SHEET_NAMES[key])}!A1",
nav_fmt, string=text, tip=tip)
def _footer(ctx, ws, fc, rd) -> None:
ws.merge_range(
46, 1, 46, 17,
f"출처: 공공데이터포털 「의약품 원료의약품 등록 정보」(15057075) · 식품의약품안전처 "
f"│ 수집 {rd.meta.started_at} KST │ 본 리포트는 공개 데이터를 자동 수집·가공한 "
f"참고 자료이며 법적 효력이 없다.",
fc.footnote())
def _print_setup(ws) -> None:
ws.set_landscape()
ws.set_paper(9) # A4
ws.fit_to_pages(1, 1)
ws.set_margins(0.4, 0.4, 0.6, 0.6)
ws.center_horizontally()
ws.print_area(0, 0, 46, 18)
ws.set_header("")
ws.set_footer("&L&\"맑은 고딕\"&8DMF Crawler&C&\"맑은 고딕\"&8&P / &N"
"&R&\"맑은 고딕\"&8&D &T")
8.8 src/dmf_crawler/report/sheets/s01_changes.py
"""01_오늘변경분 — 리포트 본체."""
from __future__ import annotations
from urllib.parse import quote
from ..theme import NUMFMT, PALETTE, SHEET_NAMES, STATUS, qsheet
from ..widgets import (apply_col_widths, compute_col_widths, write_empty_notice,
write_sheet_header, write_table_header)
HEADER = ("정렬키", "기호", "상태", "등록번호", "성분명", "업체명", "제조소명",
"제조소 소재지", "제조국가", "발급일자", "수리일자(파생)", "변경필드",
"변경내용(전→후)", "워치히트", "워치키워드", "AI 메모", "원문검색",
"dmf_key", "content_hash")
WIDTH_OVERRIDE = {0: 4.0, 1: 4.0, 2: 8.0, 3: 24.0, 7: 40.0, 12: 52.0,
13: 8.0, 15: 40.0, 16: 10.0, 17: 26.0, 18: 18.0}
HIDDEN_COLS = (0, 17, 18)
COMMENTS = {
2: "신규 = 어제 없던 등록번호 / 변경 = 비교 대상 필드가 달라짐 / 취하 = 어제 있었으나 오늘 사라짐",
3: "DMF_PERMIT_NO. 발급일자8자리-성분일련-시행군-접수일련-성분내일련 구조",
10: "등록번호 앞 8자리에서 파생한 최초 수리일. 원본 필드가 아니다",
12: "diff 로 잡힌 필드별 변경 전/후 값. 여러 건이면 줄바꿈으로 구분",
13: "이 레코드가 매칭된 워치리스트 키워드 개수",
}
NEDRUG_SEARCH = ("https://nedrug.mfds.go.kr/pbp/CCBAC03/getItem"
"?totalPages=1&limit=10&page=1&searchYn=true&itemName={q}")
def build(ctx) -> None:
ws, fc, rd = ctx.sheets["changes"], ctx.fc, ctx.rd
last_col = len(HEADER) - 1
write_sheet_header(ws, fc, title="오늘 변경분",
dashboard=SHEET_NAMES["dashboard"],
report_date=f"{rd.meta.report_date:%Y-%m-%d}",
generated_at=rd.meta.finished_at[-8:-3],
last_col=last_col)
write_table_header(ws, fc, 2, HEADER, COMMENTS)
rows = rd.changes
if not rows:
write_empty_notice(ws, fc, 3, last_col,
"(오늘 전일 대비 변경된 DMF 가 없습니다)")
ctx.changes_last_row = 4
_finish(ws, fc, last_col, data_last_row=3, empty=True)
return
body = [_as_row(r) for r in rows]
widths = compute_col_widths(HEADER, body, overrides=WIDTH_OVERRIDE)
apply_col_widths(ws, widths,
options={c: {"hidden": 1} for c in HIDDEN_COLS})
lg = qsheet(SHEET_NAMES["ledger"])
f_text = fc.cell_text()
f_wrap = fc.cell_text(wrap=True)
f_small = fc.cell_text(wrap=True, size=9)
f_muted = fc.cell_text(wrap=True, size=9, color=PALETTE["MUTED"])
f_num = fc.cell_num("INT_DASH")
f_date = fc.cell_date()
f_ctr = fc.get(align="center", valign="vcenter", num_format=NUMFMT["TEXT"],
bottom=7, bottom_color=PALETTE["RULE"])
f_link_i = fc.link_internal()
f_link_e = fc.link_external()
for i, r in enumerate(rows):
wr = 3 + i
ws.set_row(wr, 30)
ws.write_number(wr, 0, r.sort_key, f_num)
ws.write_string(wr, 1, STATUS[r.status].symbol, f_ctr)
ws.write_string(wr, 2, r.status, f_ctr)
target_row = ctx.ledger_row_of.get(r.dmf_key)
if target_row:
ws.write_url(wr, 3, f"internal:{lg}!A{target_row}", f_link_i,
string=r.permit_no, tip="전체 현황에서 이 등록번호 보기")
else:
ws.write_string(wr, 3, r.permit_no, f_text)
ws.write_string(wr, 4, r.ingredient_name or "-", f_wrap)
ws.write_string(wr, 5, r.applicant or "-", f_wrap)
ws.write_string(wr, 6, r.manufacturer or "-", f_wrap)
ws.write_string(wr, 7, r.manufacture_place or "-", f_wrap)
ws.write_string(wr, 8, r.countries or "-", f_ctr)
_write_date(ws, wr, 9, r.permit_date, f_date, f_ctr)
_write_date(ws, wr, 10, r.accepted_date, f_date, f_ctr)
ws.write_string(wr, 11, r.changed_fields or "-", f_wrap)
ws.write_string(wr, 12, r.change_detail or "-", f_small)
ws.write_number(wr, 13, r.watch_hits, f_num)
ws.write_string(wr, 14, r.watch_keywords or "-", f_text)
ws.write_string(wr, 15, r.ai_note or "-", f_muted)
url = NEDRUG_SEARCH.format(q=quote(r.ingredient_name or ""))
ws.write_url(wr, 16, url, f_link_e, string="조회",
tip="의약품안전나라에서 이 성분 검색")
ws.write_string(wr, 17, r.dmf_key, f_text)
ws.write_string(wr, 18, r.content_hash, f_text)
data_last = 3 + len(rows) - 1
ctx.changes_last_row = data_last + 1
_conditional(ws, fc, data_last)
_finish(ws, fc, last_col, data_last_row=data_last, empty=False)
def _as_row(r):
return (r.sort_key, "", r.status, r.permit_no, r.ingredient_name, r.applicant,
r.manufacturer, r.manufacture_place, r.countries, r.permit_date,
r.accepted_date, r.changed_fields, r.change_detail, r.watch_hits,
r.watch_keywords, r.ai_note, "조회", r.dmf_key, r.content_hash)
def _write_date(ws, row, col, value, f_date, f_dash) -> None:
if value is None:
ws.write_string(row, col, "-", f_dash)
else:
ws.write_datetime(row, col, value, f_date)
def _conditional(ws, fc, last: int) -> None:
# 순서가 곧 우선순위다: 행 규칙 → 열 규칙 → 데이터바/아이콘
for label in ("신규", "변경", "취하"): # CF-01-01~03
ws.conditional_format(3, 1, last, 18, {
"type": "formula", "criteria": f'=$C4="{label}"',
"format": fc.cf(label),
})
ws.conditional_format(3, 13, last, 13, { # CF-01-04
"type": "cell", "criteria": ">", "value": 0,
"format": fc.cf("워치"),
})
ws.conditional_format(3, 8, last, 8, { # CF-01-05
"type": "text", "criteria": "containing", "value": "대한민국",
"format": fc.cf("신규"),
})
ws.conditional_format(3, 4, last, 5, { # CF-01-06
"type": "formula", "criteria": "=$N4>0",
"format": fc._wb.add_format({"bold": True}),
})
def _finish(ws, fc, last_col: int, *, data_last_row: int, empty: bool) -> None:
ws.freeze_panes(3, 5)
if not empty:
ws.autofilter(2, 0, data_last_row, last_col)
ws.set_landscape()
ws.set_paper(9)
ws.fit_to_pages(1, 0)
ws.set_margins(0.4, 0.4, 0.6, 0.6)
ws.repeat_rows(0, 2)
ws.print_area(0, 0, data_last_row, last_col)
ws.set_footer("&L&\"맑은 고딕\"&8DMF 오늘 변경분&C&\"맑은 고딕\"&8&P / &N"
"&R&\"맑은 고딕\"&8&D")
8.9 src/dmf_crawler/report/sheets/s02_ledger.py
"""02_전체현황 — 누적 원장. dmf_key → 행번호 사전을 ctx 에 남긴다."""
from __future__ import annotations
from urllib.parse import quote
from ..theme import NUMFMT, PALETTE, SHEET_NAMES, TABLE_STYLE
from ..widgets import (apply_col_widths, compute_col_widths, write_sheet_header,
write_table_header)
HEADER = ("등록번호", "성분명", "업체명", "제조소명", "제조소 소재지", "제조국가",
"제조국 수", "발급일자", "수리일자", "최초관측일", "최종관측일",
"최종변경일", "상태", "경과일", "원문검색", "dmf_key", "content_hash")
WIDTH_OVERRIDE = {0: 24.0, 4: 44.0, 6: 8.0, 12: 8.0, 13: 8.0, 14: 10.0,
15: 26.0, 16: 18.0}
HIDDEN_COLS = (15, 16)
NEDRUG_SEARCH = ("https://nedrug.mfds.go.kr/pbp/CCBAC03/getItem"
"?totalPages=1&limit=10&page=1&searchYn=true&itemName={q}")
def build(ctx) -> None:
ws, fc, rd = ctx.sheets["ledger"], ctx.fc, ctx.rd
last_col = len(HEADER) - 1
write_sheet_header(ws, fc, title="전체 현황 (누적 원장)",
dashboard=SHEET_NAMES["dashboard"],
report_date=f"{rd.meta.report_date:%Y-%m-%d}",
generated_at=rd.meta.finished_at[-8:-3],
last_col=last_col)
write_table_header(ws, fc, 2, HEADER, {
4: "MNFCTR_PLACE 원문. 원본 데이터에 공백 오류가 섞여 있을 수 있다",
13: f"기준일({rd.meta.report_date:%Y-%m-%d}) − 수리일자. 3년 이상이면 아이콘이 바뀐다",
})
rows = rd.ledger
body = [_as_row(r) for r in rows]
widths = compute_col_widths(HEADER, body, overrides=WIDTH_OVERRIDE)
apply_col_widths(ws, widths, options={
**{c: {"hidden": 1, "level": 1} for c in HIDDEN_COLS},
})
f_text = fc.cell_text()
f_ctr = fc.get(align="center", valign="vcenter", num_format=NUMFMT["TEXT"],
bottom=7, bottom_color=PALETTE["RULE"])
f_num = fc.cell_num("INT")
f_date = fc.cell_date()
f_link_e = fc.link_external()
for i, r in enumerate(rows):
wr = 3 + i
ws.set_row(wr, 15)
ctx.ledger_row_of[r.dmf_key] = wr + 1 # 1-index 행 번호
ws.write_string(wr, 0, r.permit_no, f_text)
ws.write_string(wr, 1, r.ingredient_name or "-", f_text)
ws.write_string(wr, 2, r.applicant or "-", f_text)
ws.write_string(wr, 3, r.manufacturer or "-", f_text)
ws.write_string(wr, 4, r.manufacture_place or "-", f_text)
ws.write_string(wr, 5, r.countries or "-", f_ctr)
ws.write_number(wr, 6, r.country_count, f_num)
_wd(ws, wr, 7, r.permit_date, f_date, f_ctr)
_wd(ws, wr, 8, r.accepted_date, f_date, f_ctr)
_wd(ws, wr, 9, r.first_seen, f_date, f_ctr)
_wd(ws, wr, 10, r.last_seen, f_date, f_ctr)
_wd(ws, wr, 11, r.last_changed, f_date, f_ctr)
ws.write_string(wr, 12, r.status, f_ctr)
if r.elapsed_days is None:
ws.write_string(wr, 13, "-", f_ctr)
else:
ws.write_formula(wr, 13, f'=IF($I{wr + 1}="","-",INT(REPORT_DATE-$I{wr + 1}))',
fc.cell_num("INT_DASH"), r.elapsed_days)
ws.write_url(wr, 14, NEDRUG_SEARCH.format(q=quote(r.ingredient_name or "")),
f_link_e, string="조회")
ws.write_string(wr, 15, r.dmf_key, f_text)
ws.write_string(wr, 16, r.watch_flag, f_text) # 히든: CF-02-05 판정용
last = 3 + len(rows) - 1
ctx.ledger_last_row = last + 1
ws.add_table(2, 0, last, last_col, {
"name": "T_LEDGER", "style": TABLE_STYLE,
"banded_rows": True, "autofilter": True, "header_row": True,
"columns": [{"header": h} for h in HEADER],
})
_conditional(ws, fc, last)
ws.freeze_panes(3, 3)
ws.set_landscape()
ws.set_paper(9)
ws.fit_to_pages(1, 0)
ws.set_margins(0.4, 0.4, 0.6, 0.6)
ws.repeat_rows(0, 2)
ws.print_area(0, 0, last, 14) # 히든 열은 인쇄 영역에서 제외
ws.set_footer("&L&\"맑은 고딕\"&8DMF 전체 현황&C&\"맑은 고딕\"&8&P / &N"
"&R&\"맑은 고딕\"&8&D")
def _as_row(r):
return (r.permit_no, r.ingredient_name, r.applicant, r.manufacturer,
r.manufacture_place, r.countries, r.country_count, r.permit_date,
r.accepted_date, r.first_seen, r.last_seen, r.last_changed,
r.status, r.elapsed_days, "조회", r.dmf_key, r.watch_flag)
def _wd(ws, row, col, value, f_date, f_dash) -> None:
if value is None:
ws.write_string(row, col, "-", f_dash)
else:
ws.write_datetime(row, col, value, f_date)
def _conditional(ws, fc, last: int) -> None:
ws.conditional_format(3, 0, last, 16, { # CF-02-03
"type": "formula", "criteria": '=$M4="취하"',
"format": fc.cf("취하"),
})
ws.conditional_format(3, 0, last, 0, { # CF-02-01
"type": "duplicate",
"format": fc._wb.add_format({"bg_color": "#FFF2CC",
"font_color": "#7F6000"}),
})
ws.conditional_format(3, 11, last, 11, { # CF-02-02
"type": "formula", "criteria": "=$L4=REPORT_DATE",
"format": fc.cf("변경"),
})
ws.conditional_format(3, 5, last, 5, { # CF-02-04
"type": "text", "criteria": "containing", "value": "대한민국",
"format": fc.cf("신규"),
})
ws.conditional_format(3, 1, last, 2, { # CF-02-05
"type": "formula", "criteria": '=$Q4="Y"',
"format": fc.cf("워치"),
})
ws.conditional_format(3, 6, last, 6, { # CF-02-06
"type": "cell", "criteria": ">=", "value": 2,
"format": fc._wb.add_format({"bold": True}),
})
ws.conditional_format(3, 8, last, 8, { # CF-02-07
"type": "data_bar", "bar_color": PALETTE["ACCENT"], "bar_solid": True,
})
ws.conditional_format(3, 13, last, 13, { # CF-02-08
"type": "icon_set", "icon_style": "3_arrows_gray",
"icons_only": False, "reverse_icons": True,
"icons": [{"criteria": ">=", "type": "number", "value": 1095},
{"criteria": ">=", "type": "number", "value": 365}],
})
8.10 src/dmf_crawler/report/sheets/s03_ingredient.py · s04_company.py
두 시트는 컬럼 구성만 다르고 로직이 같다. 공용 빌더를 두고 스펙 테이블로 분기한다.
# s03_ingredient.py
"""03_성분별 — 성분 기준 집계."""
from __future__ import annotations
from ._agg import AggSpec, build_agg
from ..theme import SHEET_NAMES
SPEC = AggSpec(
sheet_key="ingredient",
title="성분별 집계",
name_header="성분명",
name_width=34.0,
partner_header="등록업체 수",
country_header="제조국 수",
domestic_header="국산 보유",
domestic_values=("Y", "N"),
has_company_key=False,
spark_block_col=29, # AD (0-index)
table_name="T_INGREDIENT",
)
def build(ctx) -> None:
build_agg(ctx, SPEC)
# s04_company.py
"""04_업체별 — 업체 기준 집계."""
from __future__ import annotations
from ._agg import AggSpec, build_agg
SPEC = AggSpec(
sheet_key="company",
title="업체별 집계",
name_header="업체명",
name_width=30.0,
partner_header="보유 성분 수",
country_header="제조소 수",
domestic_header="국산여부",
domestic_values=("국산", "해외"),
has_company_key=True, # C열에 히든 업체키를 넣는다
spark_block_col=61, # BJ (0-index)
table_name="T_COMPANY",
)
def build(ctx) -> None:
build_agg(ctx, SPEC)
# _agg.py — 03/04 공용 빌더
"""성분별·업체별 집계 시트 공용 구현."""
from __future__ import annotations
from dataclasses import dataclass
from ..theme import NUMFMT, PALETTE, SHEET_NAMES, TABLE_STYLE, qsheet
from ..widgets import (apply_col_widths, compute_col_widths, write_empty_notice,
write_sheet_header, write_table_header)
@dataclass(frozen=True, slots=True)
class AggSpec:
sheet_key: str
title: str
name_header: str
name_width: float
partner_header: str
country_header: str
domestic_header: str
domestic_values: tuple[str, str]
has_company_key: bool
spark_block_col: int
table_name: str
def _header(spec: AggSpec) -> tuple[str, ...]:
base = ["순위", spec.name_header]
if spec.has_company_key:
base.append("업체키")
base += ["오늘 신규", "오늘 변경", "오늘 취하", "오늘 합계"]
if not spec.has_company_key:
base.append("최근 7일")
base += ["최근 30일", "누적 유효등록", spec.partner_header, spec.country_header,
"주요 제조국", spec.domestic_header]
if spec.has_company_key:
base += ["최근 등록일", "워치리스트"]
else:
base.append("전일 대비")
base += ["30일 추이", "상세"]
return tuple(base)
def build_agg(ctx, spec: AggSpec) -> None:
ws, fc, rd = ctx.sheets[spec.sheet_key], ctx.fc, ctx.rd
rows = rd.ingredients if spec.sheet_key == "ingredient" else rd.companies
header = _header(spec)
last_col = len(header) - 1
idx = {name: i for i, name in enumerate(header)}
write_sheet_header(ws, fc, title=spec.title,
dashboard=SHEET_NAMES["dashboard"],
report_date=f"{rd.meta.report_date:%Y-%m-%d}",
generated_at=rd.meta.finished_at[-8:-3],
last_col=last_col)
write_table_header(ws, fc, 2, header, {
idx["누적 유효등록"]: "취하되지 않은 유효 등록 건수",
idx["주요 제조국"]: "이 항목에서 건수가 가장 많은 제조국",
})
if not rows:
write_empty_notice(ws, fc, 3, last_col, "(집계할 데이터가 없습니다)")
ws.freeze_panes(3, 2)
return
body = [_as_row(r, spec, idx) for r in rows]
overrides = {idx["순위"]: 6.0, idx[spec.name_header]: spec.name_width,
idx["30일 추이"]: 14.0, idx["상세"]: 8.0}
if spec.has_company_key:
overrides[idx["업체키"]] = 26.0
widths = compute_col_widths(header, body, overrides=overrides)
opts = {idx["업체키"]: {"hidden": 1}} if spec.has_company_key else {}
opts[last_col + 1] = {"hidden": 1} # 워치플래그 히든 열 (CF-03-05)
apply_col_widths(ws, widths, options=opts)
f_text = fc.cell_text()
f_ctr = fc.get(align="center", valign="vcenter", num_format=NUMFMT["TEXT"],
bottom=7, bottom_color=PALETTE["RULE"])
f_num = fc.cell_num("INT_DASH")
f_cum = fc.cell_num("INT")
f_date = fc.cell_date()
f_delta = fc.cell_num("DELTA")
f_link = fc.link_internal()
lg = qsheet(SHEET_NAMES["ledger"])
trend = qsheet(SHEET_NAMES["trend"])
for i, r in enumerate(rows):
wr = 3 + i
ws.set_row(wr, 18)
ws.write_number(wr, idx["순위"], r.rank, f_num)
ws.write_string(wr, idx[spec.name_header], r.name or "-", f_text)
if spec.has_company_key:
ws.write_string(wr, idx["업체키"], r.key, f_text)
ws.write_number(wr, idx["오늘 신규"], r.today_new, f_num)
ws.write_number(wr, idx["오늘 변경"], r.today_changed, f_num)
ws.write_number(wr, idx["오늘 취하"], r.today_withdrawn, f_num)
c1 = _colletter(idx["오늘 신규"])
c3 = _colletter(idx["오늘 취하"])
ws.write_formula(wr, idx["오늘 합계"], f"=SUM(${c1}{wr + 1}:${c3}{wr + 1})",
f_num, r.today_new + r.today_changed + r.today_withdrawn)
if "최근 7일" in idx:
ws.write_number(wr, idx["최근 7일"], r.last7, f_num)
ws.write_number(wr, idx["최근 30일"], r.last30, f_num)
ws.write_number(wr, idx["누적 유효등록"], r.cumulative, f_cum)
ws.write_number(wr, idx[spec.partner_header], r.partner_count, f_cum)
ws.write_number(wr, idx[spec.country_header], r.country_count, f_cum)
ws.write_string(wr, idx["주요 제조국"], r.main_country or "-", f_ctr)
ws.write_string(wr, idx[spec.domestic_header], r.domestic, f_ctr)
if "최근 등록일" in idx:
if r.last_date is None:
ws.write_string(wr, idx["최근 등록일"], "-", f_ctr)
else:
ws.write_datetime(wr, idx["최근 등록일"], r.last_date, f_date)
if "워치리스트" in idx:
ws.write_string(wr, idx["워치리스트"],
"★" if r.watch_flag == "Y" else "-", f_ctr)
if "전일 대비" in idx:
ws.write_number(wr, idx["전일 대비"], r.delta, f_delta)
ws.write_blank(wr, idx["30일 추이"], None, f_ctr)
ws.add_sparkline(wr, idx["30일 추이"], {
"range": (f"{trend}!${_colletter(spec.spark_block_col + 1)}${wr + 1}"
f":${_colletter(spec.spark_block_col + 30)}${wr + 1}"),
"type": "column", "series_color": PALETTE["ACCENT"],
"high_point": True, "empty_cells": "zero",
})
ws.write_url(wr, idx["상세"], f"internal:{lg}!A3", f_link, string="보기",
tip="전체 현황에서 필터로 확인")
ws.write_string(wr, last_col + 1, r.watch_flag, f_text) # 히든 판정 열
last = 3 + len(rows) - 1
total_row = last + 1
ws.add_table(2, 0, total_row, last_col, {
"name": spec.table_name, "style": TABLE_STYLE,
"banded_rows": True, "autofilter": True, "total_row": True,
"columns": _table_columns(header, idx, rows),
})
_conditional(ws, fc, spec, idx, last, last_col)
ws.freeze_panes(3, 2)
ws.set_landscape()
ws.set_paper(9)
ws.fit_to_pages(1, 0)
ws.set_margins(0.4, 0.4, 0.6, 0.6)
ws.repeat_rows(0, 2)
ws.print_area(0, 0, total_row, last_col)
def _table_columns(header, idx, rows):
"""add_table 컬럼 정의. 합계행은 total_value 로 캐시값을 함께 준다."""
sum_cols = {"오늘 신규", "오늘 변경", "오늘 취하", "오늘 합계",
"최근 7일", "최근 30일", "누적 유효등록"}
attr = {"오늘 신규": "today_new", "오늘 변경": "today_changed",
"오늘 취하": "today_withdrawn", "최근 7일": "last7",
"최근 30일": "last30", "누적 유효등록": "cumulative"}
cols = []
for name in header:
col: dict = {"header": name}
if name == "순위":
col["total_string"] = "합계"
elif name in sum_cols:
col["total_function"] = "sum"
if name == "오늘 합계":
col["total_value"] = sum(r.today_new + r.today_changed
+ r.today_withdrawn for r in rows)
else:
col["total_value"] = sum(getattr(r, attr[name]) for r in rows)
cols.append(col)
return cols
def _conditional(ws, fc, spec, idx, last, last_col) -> None:
for name, status in (("오늘 신규", "신규"), ("오늘 변경", "변경"),
("오늘 취하", "취하")):
c = idx[name]
ws.conditional_format(3, c, last, c, {
"type": "cell", "criteria": ">", "value": 0,
"format": fc.cf(status),
})
c = idx[spec.name_header]
ws.conditional_format(3, c, last, c, {
"type": "formula",
"criteria": f"=${_colletter(last_col + 1)}4=\"Y\"",
"format": fc.cf("워치"),
})
for name in ("최근 7일", "최근 30일", "누적 유효등록"):
if name not in idx:
continue
c = idx[name]
ws.conditional_format(3, c, last, c, {
"type": "data_bar", "bar_color": PALETTE["ACCENT"], "bar_solid": True,
})
c = idx[spec.partner_header]
ws.conditional_format(3, c, last, c, {
"type": "cell", "criteria": ">=", "value": 5,
"format": fc._wb.add_format({"bold": True}),
})
c = idx[spec.domestic_header]
ws.conditional_format(3, c, last, c, {
"type": "text", "criteria": "containing",
"value": spec.domestic_values[0], "format": fc.cf("신규"),
})
if "전일 대비" in idx:
c = idx["전일 대비"]
ws.conditional_format(3, c, last, c, {
"type": "icon_set", "icon_style": "3_arrows_gray", "icons_only": False,
"icons": [{"criteria": ">=", "type": "number", "value": 1},
{"criteria": ">=", "type": "number", "value": 0}],
})
if "최근 등록일" in idx:
c = idx["최근 등록일"]
ws.conditional_format(3, c, last, c, {
"type": "formula",
"criteria": f'=AND(${_colletter(c)}4<>"",REPORT_DATE-${_colletter(c)}4>90)',
"format": fc.cf("없음"),
})
if "워치리스트" in idx:
c = idx["워치리스트"]
ws.conditional_format(3, c, last, c, {
"type": "text", "criteria": "containing", "value": "★",
"format": fc.cf("워치"),
})
def _colletter(col0: int) -> str:
"""0-index 열 번호 → 엑셀 열 문자."""
s = ""
n = col0 + 1
while n:
n, rem = divmod(n - 1, 26)
s = chr(65 + rem) + s
return s
def _as_row(r, spec, idx):
row = [None] * len(idx)
row[idx["순위"]] = r.rank
row[idx[spec.name_header]] = r.name
if spec.has_company_key:
row[idx["업체키"]] = r.key
row[idx["오늘 신규"]] = r.today_new
row[idx["오늘 변경"]] = r.today_changed
row[idx["오늘 취하"]] = r.today_withdrawn
row[idx["오늘 합계"]] = r.today_new + r.today_changed + r.today_withdrawn
if "최근 7일" in idx:
row[idx["최근 7일"]] = r.last7
row[idx["최근 30일"]] = r.last30
row[idx["누적 유효등록"]] = r.cumulative
row[idx[spec.partner_header]] = r.partner_count
row[idx[spec.country_header]] = r.country_count
row[idx["주요 제조국"]] = r.main_country
row[idx[spec.domestic_header]] = r.domestic
if "최근 등록일" in idx:
row[idx["최근 등록일"]] = r.last_date
if "워치리스트" in idx:
row[idx["워치리스트"]] = "★" if r.watch_flag == "Y" else "-"
if "전일 대비" in idx:
row[idx["전일 대비"]] = r.delta
row[idx["30일 추이"]] = ""
row[idx["상세"]] = "보기"
return row
8.11 src/dmf_crawler/report/sheets/s05_watchlist.py
"""05_워치리스트 — 읽기 전용 미러. active 항목이 0건이면 호출되지 않는다."""
from __future__ import annotations
from ..theme import NUMFMT, PALETTE, SHEET_NAMES, STATUS, qsheet
from ..widgets import apply_col_widths, write_sheet_header, write_table_header
HEADER_A = ("번호", "대상유형", "키워드", "매칭방식", "우선순위", "활성", "메모")
HEADER_B = ("오늘 매칭", "최근 7일", "최근 30일", "누적 매칭", "마지막 매칭일", "상세")
HEADER_C = ("키워드", "상태", "등록번호", "성분명", "업체명", "제조국가", "발급일자")
WIDTHS = [6, 12, 30, 12, 8, 6, 26, 1.5, 9, 9, 9, 9, 12, 8] # A..N
VALID = {
1: ["성분명", "업체명", "제조소명", "제조국가", "등록번호"],
3: ["부분일치", "정확일치", "정규식"],
4: [1, 2, 3],
5: ["Y", "N"],
}
def build(ctx) -> None:
ws, fc, rd = ctx.sheets["watchlist"], ctx.fc, ctx.rd
write_sheet_header(ws, fc, title="워치리스트 (읽기 전용)",
dashboard=SHEET_NAMES["dashboard"],
report_date=f"{rd.meta.report_date:%Y-%m-%d}",
generated_at=rd.meta.finished_at[-8:-3], last_col=13)
apply_col_widths(ws, WIDTHS)
write_table_header(ws, fc, 2, HEADER_A + ("",) + HEADER_B, {
2: "이 값이 성분명·업체명 등에 포함되면 히트로 센다",
3: "정규식 모드는 파이썬 re 문법을 따른다",
})
f_text = fc.cell_text()
f_ctr = fc.get(align="center", valign="vcenter", num_format=NUMFMT["TEXT"],
bottom=7, bottom_color=PALETTE["RULE"])
f_num = fc.cell_num("INT_DASH")
f_cum = fc.cell_num("INT")
f_date = fc.cell_date()
f_link = fc.link_internal()
ch = qsheet(SHEET_NAMES["changes"])
for i, w in enumerate(rd.watch):
wr = 3 + i
ws.set_row(wr, 16)
ws.write_number(wr, 0, w.no, f_num)
ws.write_string(wr, 1, w.target_type, f_ctr)
ws.write_string(wr, 2, w.keyword, f_text)
ws.write_string(wr, 3, w.match_mode, f_ctr)
ws.write_number(wr, 4, w.priority, f_num)
ws.write_string(wr, 5, w.active, f_ctr)
ws.write_string(wr, 6, w.memo or "-", f_text)
ws.write_formula(
wr, 8,
f'=IF($C{wr + 1}="",0,COUNTIF({ch}!$O$4:$O$100000,"*"&$C{wr + 1}&"*"))',
f_num, w.today)
ws.write_number(wr, 9, w.last7, f_num)
ws.write_number(wr, 10, w.last30, f_num)
ws.write_number(wr, 11, w.cumulative, f_cum)
if w.last_hit is None:
ws.write_string(wr, 12, "-", f_ctr)
else:
ws.write_datetime(wr, 12, w.last_hit, f_date)
ws.write_url(wr, 13, f"internal:{ch}!A3", f_link, string="보기")
last = 3 + len(rd.watch) - 1
for col, values in VALID.items():
ws.data_validation(3, col, last, col,
{"validate": "list", "source": values})
_conditional_ab(ws, fc, last)
# ── 블록 C: 매칭 상세표 ──────────────────────────────────────
cstart = last + 3
ws.write(cstart - 1, 0, "오늘 매칭 상세",
fc.get(font_size=12, bold=True, valign="vcenter"))
write_table_header(ws, fc, cstart, HEADER_C, height=24)
for i, d in enumerate(rd.watch_detail):
wr = cstart + 1 + i
ws.set_row(wr, 16)
ws.write_string(wr, 0, d.keyword, f_text)
ws.write_string(wr, 1, d.status, f_ctr)
ws.write_string(wr, 2, d.permit_no, f_text)
ws.write_string(wr, 3, d.ingredient_name or "-", f_text)
ws.write_string(wr, 4, d.applicant or "-", f_text)
ws.write_string(wr, 5, d.countries or "-", f_ctr)
if d.permit_date is None:
ws.write_string(wr, 6, "-", f_ctr)
else:
ws.write_datetime(wr, 6, d.permit_date, f_date)
cend = cstart + max(1, len(rd.watch_detail))
for label in ("신규", "변경", "취하"):
ws.conditional_format(cstart + 1, 0, cend, 6, {
"type": "formula",
"criteria": f'=$B{cstart + 2}="{label}"',
"format": fc.cf(label),
})
ws.freeze_panes(3, 3)
ws.protect("", {"objects": True, "scenarios": True,
"select_locked_cells": True, "select_unlocked_cells": True,
"sort": True, "autofilter": True})
ws.set_landscape()
ws.set_paper(9)
ws.fit_to_pages(1, 0)
ws.set_margins(0.4, 0.4, 0.6, 0.6)
ws.repeat_rows(0, 2)
ws.print_area(0, 0, cend, 13)
def _conditional_ab(ws, fc, last: int) -> None:
ws.conditional_format(3, 8, last, 8, { # CF-05-01
"type": "cell", "criteria": ">", "value": 0, "format": fc.cf("워치"),
})
ws.conditional_format(3, 12, last, 12, { # CF-05-02
"type": "formula", "criteria": "=$M4=REPORT_DATE", "format": fc.cf("워치"),
})
ws.conditional_format(3, 2, last, 2, { # CF-05-03
"type": "blanks", "format": fc.cf("취하"),
})
for c in (9, 10): # CF-05-04
ws.conditional_format(3, c, last, c, {
"type": "data_bar", "bar_color": PALETTE["WATCH"], "bar_solid": True,
})
8.12 src/dmf_crawler/report/sheets/s06_trend.py
"""06_추이 — 시각화 원본. 가장 먼저 만들어진다."""
from __future__ import annotations
from ..theme import NUMFMT, PALETTE, SHEET_NAMES, TABLE_STYLE
from ..widgets import apply_col_widths, write_sheet_header, write_table_header
HEADER = ("일자", "신규", "변경", "취하", "합계", "워치 히트",
"누적 유효등록", "수집 건수", "실행 상태")
WIDTHS = [12, 9, 9, 9, 9, 9, 12, 10, 10]
BAR_COLORS = {1: "#009E73", 2: "#E69F00", 3: "#D55E00"}
def build(ctx) -> None:
ws, fc, rd = ctx.sheets["trend"], ctx.fc, ctx.rd
write_sheet_header(ws, fc, title="일자별 추이",
dashboard=SHEET_NAMES["dashboard"],
report_date=f"{rd.meta.report_date:%Y-%m-%d}",
generated_at=rd.meta.finished_at[-8:-3], last_col=8)
apply_col_widths(ws, WIDTHS)
write_table_header(ws, fc, 2, HEADER, {
6: "그날 스냅샷의 유효 등록 총건수",
7: "그날 API 에서 실제로 수집한 행 수. 급감하면 붉게 표시된다",
})
f_num = fc.cell_num("INT_DASH")
f_cum = fc.cell_num("INT")
f_date = fc.cell_date()
f_ctr = fc.get(align="center", valign="vcenter", num_format=NUMFMT["TEXT"],
bottom=7, bottom_color=PALETTE["RULE"])
for i, t in enumerate(rd.trend):
wr = 3 + i
ws.set_row(wr, 16)
ws.write_datetime(wr, 0, t.day, f_date)
ws.write_number(wr, 1, t.new, f_num)
ws.write_number(wr, 2, t.changed, f_num)
ws.write_number(wr, 3, t.withdrawn, f_num)
ws.write_formula(wr, 4, f"=SUM($B{wr + 1}:$D{wr + 1})", f_num, t.total)
ws.write_number(wr, 5, t.watch_hits, f_num)
ws.write_number(wr, 6, t.active_total, f_cum)
ws.write_number(wr, 7, t.fetched_rows, f_cum)
ws.write_string(wr, 8, t.run_status, f_ctr)
last = 3 + len(rd.trend) - 1
ctx.trend_first_row = 4
ctx.trend_last_row = last + 1
ctx.spark_first_row = max(4, ctx.trend_last_row - 29)
ws.add_table(2, 0, last, 8, {
"name": "T_TREND", "style": TABLE_STYLE,
"banded_rows": True, "autofilter": True,
"columns": [{"header": h} for h in HEADER],
})
_conditional(ws, fc, last)
_hidden_blocks(ctx, ws, fc, rd)
ws.freeze_panes(3, 1)
ws.set_landscape()
ws.set_paper(9)
ws.fit_to_pages(1, 0)
ws.set_margins(0.4, 0.4, 0.6, 0.6)
ws.repeat_rows(0, 2)
ws.print_area(0, 0, last, 8)
def _conditional(ws, fc, last: int) -> None:
ws.conditional_format(3, 0, last, 0, { # CF-06-01
"type": "formula", "criteria": "=$A4=REPORT_DATE", "format": fc.cf("정보"),
})
for c, color in BAR_COLORS.items(): # CF-06-02
ws.conditional_format(3, c, last, c, {
"type": "data_bar", "bar_color": color, "bar_solid": True,
})
ws.conditional_format(3, 5, last, 5, {
"type": "cell", "criteria": ">", "value": 0, "format": fc.cf("워치"),
})
ws.conditional_format(4, 7, last, 7, { # CF-06-03
"type": "formula", "criteria": "=AND($H4<>\"\",$H4<$H3*0.95)",
"format": fc.cf("취하"),
})
for value, status in (("SUCCESS", "신규"), ("PARTIAL", "변경"),
("FAILED", "취하")): # CF-06-04
ws.conditional_format(3, 8, last, 8, {
"type": "text", "criteria": "containing", "value": value,
"format": fc.cf(status),
})
def _hidden_blocks(ctx, ws, fc, rd) -> None:
"""차트2·스파크라인 원본. 열 폭 0 + hidden."""
plain = fc.get()
num = fc.get(num_format=NUMFMT["INT"])
# AA(26)/AB(27): 제조국 Top 8 — 차트2 원본
ws.set_column(26, 27, 12, None, {"hidden": 1})
ws.write(2, 26, "제조국", plain)
ws.write(2, 27, "건수", plain)
for i, (country, n) in enumerate(rd.top_countries):
ws.write_string(3 + i, 26, country, plain)
ws.write_number(3 + i, 27, n, num)
# AD(29)~BH(59): 성분별 30일 시계열
ws.set_column(29, 59, 6, None, {"hidden": 1})
ws.write(2, 29, "성분명", plain)
for i, d in enumerate(rd.trend_dates[-30:]):
ws.write_datetime(2, 30 + i, d, fc.cell_date())
for r, agg in enumerate(rd.ingredients):
ws.write_string(3 + r, 29, agg.name, plain)
for c, v in enumerate(agg.series30[:30]):
ws.write_number(3 + r, 30 + c, v, num)
# BJ(61)~CN(91): 업체별 30일 시계열
ws.set_column(61, 91, 6, None, {"hidden": 1})
ws.write(2, 61, "업체명", plain)
for i, d in enumerate(rd.trend_dates[-30:]):
ws.write_datetime(2, 62 + i, d, fc.cell_date())
for r, agg in enumerate(rd.companies):
ws.write_string(3 + r, 61, agg.name, plain)
for c, v in enumerate(agg.series30[:30]):
ws.write_number(3 + r, 62 + c, v, num)
8.13 src/dmf_crawler/report/sheets/s99_meta.py
"""99_메타 — 수집 메타·소스·무결성 게이트·AI 상태·코드북·면책."""
from __future__ import annotations
from ..theme import NUMFMT, PALETTE, SHEET_NAMES
from ..widgets import apply_col_widths, write_sheet_header
WIDTHS = [28, 46, 16, 16, 30]
CODEBOOK = (
("상태", "신규", "어제 스냅샷에 없던 등록번호가 오늘 나타남"),
("상태", "변경", "양쪽에 있으나 비교 대상 6필드 중 하나 이상이 달라짐"),
("상태", "취하", "어제 있었으나 오늘 스냅샷에서 사라짐"),
("등록번호", "20121228-168-I-169-04", "발급일자8-성분일련-시행군-접수일련-성분내일련"),
("실행 상태", "SUCCESS", "전 단계 성공"),
("실행 상태", "PARTIAL", "무결성 게이트 차단 또는 선택 단계 실패. 수치는 직전 성공 기준"),
("실행 상태", "FAILED", "필수 단계 실패. 리포트가 갱신되지 않았을 수 있다"),
)
def build(ctx) -> None:
ws, fc, rd = ctx.sheets["meta"], ctx.fc, ctx.rd
m, ai = rd.meta, rd.ai
write_sheet_header(ws, fc, title="메타·로그",
dashboard=SHEET_NAMES["dashboard"],
report_date=f"{m.report_date:%Y-%m-%d}",
generated_at=m.finished_at[-8:-3], last_col=4)
apply_col_widths(ws, WIDTHS)
k = fc.get(bold=True, font_color=PALETTE["MUTED"], valign="vcenter")
v = fc.get(valign="vcenter", num_format=NUMFMT["TEXT"])
vd = fc.get(valign="vcenter", num_format=NUMFMT["DATE"])
vn = fc.get(valign="vcenter", num_format=NUMFMT["INT"])
link = fc.link_external()
hdr = fc.header(wrap=False)
# ── 블록 1: 실행 메타 (A3:B18) ──────────────────────────────
ws.write_string(2, 0, "리포트 기준일", k)
ws.write_datetime(2, 1, m.report_date, vd) # ← REPORT_DATE 원본
pairs = (
("수집 시작 시각", m.started_at), ("수집 종료 시각", m.finished_at),
("실행 ID", m.run_id), ("실행 상태", m.run_status),
("실행 트리거", m.trigger), ("실행 호스트", m.host),
("파이썬 버전", m.python_version), ("XlsxWriter 버전", m.xlsxwriter_version),
("직전 성공 실행", m.prev_run_id or "-"),
)
for i, (key, val) in enumerate(pairs):
ws.write_string(3 + i, 0, key, k)
ws.write_string(3 + i, 1, str(val), v)
files = (("실행 로그 경로", m.log_dir), ("이벤트 로그 경로", m.events_path),
("원문 아카이브", m.archive_dir), ("DB 백업", m.backup_path),
("스냅샷 해시", m.snapshot_hash), ("직전 리포트", m.prev_report_path))
for i, (key, val) in enumerate(files):
r = 12 + i
ws.write_string(r, 0, key, k)
if val and key != "스냅샷 해시":
ws.write_url(r, 1, f"external:{val}", link, string=val)
else:
ws.write_string(r, 1, val or "-", v)
# ── 블록 2: 소스 (A21:E22) ───────────────────────────────────
ws.write_row(20, 0, ("소스명", "URL", "HTTP", "수집행수", "비고"), hdr)
ws.write_string(21, 0, m.source_name, v)
ws.write_url(21, 1, m.source_url, link, string=m.source_url)
ws.write_string(21, 2, m.http_summary, v)
ws.write_number(21, 3, m.fetched_rows, vn)
ws.write_string(21, 4, "공공데이터포털 15057075", v)
# ── 블록 3: 무결성 게이트 (A26:E32) ──────────────────────────
ws.write_row(25, 0, ("게이트", "기대", "실제", "판정", "상세"), hdr)
for i, g in enumerate(rd.gates):
r = 26 + i
ws.write_string(r, 0, g.name, k)
ws.write_string(r, 1, g.expected, v)
ws.write_string(r, 2, g.actual, v)
ws.write_formula(r, 3, f"=IF($C{r + 1}=$B{r + 1},\"OK\",\"불일치\")",
v, g.verdict)
ws.write_string(r, 4, g.detail, v)
tail = 26 + len(rd.gates)
ws.write_string(tail, 0, "종합", k)
ws.write_formula(tail, 3,
f'=IF(COUNTIF($D$27:$D${tail},"불일치")=0,"OK","불일치")',
v, "OK" if all(g.verdict == "OK" for g in rd.gates) else "불일치")
for value, status in (("OK", "신규"), ("불일치", "취하")): # CF-99-01
ws.conditional_format(26, 3, tail, 3, {
"type": "text", "criteria": "containing", "value": value,
"format": fc.cf(status),
})
# ── 블록 4: AI 계층 상태 ─────────────────────────────────────
base = tail + 3
ws.write_string(base - 1, 0, "AI 계층 (agy)",
fc.get(font_size=12, bold=True, valign="vcenter"))
ai_pairs = (
("agy 사용 여부", ai.used), ("모델", ai.model or "-"),
("노력 수준", ai.effort or "-"),
("입력 토큰", f"{ai.input_tokens:,}"),
("출력 토큰", f"{ai.output_tokens:,}"),
("소요 시간(초)", f"{ai.duration_s:.1f}"),
("오늘 누적 토큰", f"{ai.tokens_today:,} / {ai.token_cap:,}"),
("헤드라인", ai.headline or "(AI 요약 없음)"),
("요약", ai.summary or "(AI 요약 없음)"),
("리스크 메모", ai.risk_note or "-"),
)
for i, (key, val) in enumerate(ai_pairs):
ws.write_string(base + i, 0, key, k)
ws.write_string(base + i, 1, str(val), v)
envr = base + len(ai_pairs)
ws.write_string(envr, 0, "봉투 원문 경로", k)
if ai.envelope_path:
ws.write_url(envr, 1, f"external:{ai.envelope_path}", link,
string=ai.envelope_path)
else:
ws.write_string(envr, 1, "-", v)
ws.conditional_format(base, 1, base, 1, { # CF-99-02
"type": "text", "criteria": "containing", "value": "실패",
"format": fc.cf("취하"),
})
# ── 블록 5: 코드북 ───────────────────────────────────────────
cb = envr + 3
ws.write_row(cb, 0, ("구분", "코드", "설명"), hdr)
for i, (g, code, desc) in enumerate(CODEBOOK):
ws.write_string(cb + 1 + i, 0, g, v)
ws.write_string(cb + 1 + i, 1, code, v)
ws.write_string(cb + 1 + i, 2, desc, v)
# ── 블록 6: 출처·면책 ────────────────────────────────────────
foot = cb + len(CODEBOOK) + 3
ws.merge_range(
foot, 0, foot, 4,
"출처: 공공데이터포털 「의약품 원료의약품 등록 정보」(서비스 15057075), "
"식품의약품안전처. 이용약관에 따라 출처를 표시하고 정중한 호출 간격을 지킨다.",
fc.footnote())
ws.merge_range(
foot + 1, 0, foot + 1, 4,
"면책: 본 리포트는 공개 데이터를 자동 수집·가공한 참고 자료이며, "
"원본과 차이가 있을 수 있고 법적 효력이 없다. "
"규제 판단은 반드시 원문 공고를 확인할 것.",
fc.footnote())
ws.freeze_panes(3, 0)
ws.set_portrait()
ws.set_paper(9)
ws.fit_to_pages(1, 0)
ws.set_margins(0.5, 0.5, 0.6, 0.6)
ws.print_area(0, 0, foot + 1, 4)
9. 인쇄·공유 설정
9.1 시트별 인쇄 설정표
| 시트 | 방향 | 용지 | 맞춤 | 제목 반복 | 인쇄 영역 | 눈금선 인쇄 | 가운데 |
|---|---|---|---|---|---|---|---|
00_대시보드 |
가로 | A4(set_paper(9)) |
fit_to_pages(1, 1) |
없음 | A1:S47 |
숨김(hide_gridlines(2)) |
가로 가운데 |
01_오늘변경분 |
가로 | A4 | fit_to_pages(1, 0) |
repeat_rows(0, 2) |
A1:S{last} |
기본 | 아니오 |
02_전체현황 |
가로 | A4 | fit_to_pages(1, 0) |
repeat_rows(0, 2) |
A1:O{last} (히든 열 제외) |
기본 | 아니오 |
03_성분별 |
가로 | A4 | fit_to_pages(1, 0) |
repeat_rows(0, 2) |
A1:P{total_row} |
기본 | 아니오 |
04_업체별 |
가로 | A4 | fit_to_pages(1, 0) |
repeat_rows(0, 2) |
A1:Q{total_row} |
기본 | 아니오 |
05_워치리스트 |
가로 | A4 | fit_to_pages(1, 0) |
repeat_rows(0, 2) |
A1:N{cend} |
기본 | 아니오 |
06_추이 |
가로 | A4 | fit_to_pages(1, 0) |
repeat_rows(0, 2) |
A1:I{last} |
기본 | 아니오 |
99_메타 |
세로 | A4 | fit_to_pages(1, 0) |
없음 | A1:E{foot+1} |
기본 | 아니오 |
fit_to_pages(1, 0)= 너비 1페이지, 높이 무제한. 원장이 수백 페이지가 되어도 열이 잘리지 않는다.99_메타만 세로다. 5열짜리 목록이라 가로로 하면 여백만 커진다.- 인쇄 영역에서 히든 열은 제외한다. 숨겨진 열은 인쇄되지 않지만, 인쇄 영역 끝을 히든 열에 두면 마지막 페이지에 빈 페이지가 생기는 경우가 있다.
9.2 여백·머리글·바닥글
ws.set_margins(left=0.4, right=0.4, top=0.6, bottom=0.6) # 인치
ws.set_header("") # 머리글 없음
ws.set_footer('&L&"맑은 고딕"&8{시트 이름}'
'&C&"맑은 고딕"&8&P / &N'
'&R&"맑은 고딕"&8&D')
| 코드 | 의미 |
|---|---|
&L / &C / &R |
좌 / 가운데 / 우 구역 |
&"맑은 고딕" |
폰트 지정 |
&8 |
8pt |
&P / &N |
현재 페이지 / 전체 페이지 |
&D / &T |
인쇄 날짜 / 시각 |
머리글을 비우는 이유: 1행에 이미 제목과 기준일이 있고, repeat_rows(0, 2) 로 매 페이지 반복되므로 머리글이 중복된다.
9.3 화면 공유용 설정
| 항목 | 값 | 이유 |
|---|---|---|
| 최초 열림 창 크기 | wb.set_size(1500, 900) |
노트북 화면에서 대시보드가 잘리지 않게 |
| 최초 활성 시트 | 00_대시보드 |
파일을 여는 사람이 먼저 볼 것 |
| 최초 활성 셀 | 각 시트 A1 (기본) |
스크롤 위치 초기화 |
| 확대 | 대시보드 90%, 나머지 100% | |
| 문서 속성 | set_properties(...) 로 제목·주제·작성자·키워드 채움 |
파일 탐색기 미리보기와 검색에 잡힘 |
| 시트 보호 | 05_워치리스트 만 |
읽기 전용 미러임을 물리적으로 표시 |
| 통합문서 보호 | 걸지 않는다 | 사용자가 시트를 복사·추가해 자기 분석을 할 수 있어야 한다 |
| 비밀번호 | 없음 | 자동 배치가 만드는 파일이라 비밀번호 관리 주체가 없다 |
9.4 PDF 변환
내장 변환은 제공하지 않는다. LibreOffice headless 는 아키텍처 부록에서 기각됐고, Excel COM 자동화는 Excel 설치와 대화형 세션을 요구해 배치(S4U)에서 동작하지 않는다. 사용자가 Excel 에서 파일 → 내보내기 → PDF 를 누르면 위 인쇄 설정이 그대로 적용된다. 인쇄 설정을 정확히 잡아 두는 것이 곧 PDF 품질 보증이다.
10. 파일명·보관 규약
10.1 파일명
| 종류 | 패턴 | 예 | 설정키 |
|---|---|---|---|
| 일자별 정본 | DMF_리포트_{date}.xlsx (date = YYYY-MM-DD) |
DMF_리포트_2026-09-02.xlsx |
report.filename_pattern |
| 최신본 고정 링크 | DMF_리포트_최신.xlsx |
좌동 | report.latest_link_name (빈 문자열이면 생성 안 함) |
| 잠김 폴백 | DMF_리포트_{date}_{HHMMSS}.xlsx |
DMF_리포트_2026-09-02_060241.xlsx |
자동 |
| 생성 중 임시 | .~dmf_XXXXXX.xlsx (tempfile.mkstemp) |
— | 자동, 반드시 같은 디렉터리 |
| 최신본 임시 | .~dmflatest_XXXXXX.xlsx |
— | 자동 |
날짜는 report_date 이지 실행 시각이 아니다. 06:00 실행이 자정을 넘겨 07:00 에 끝나도 파일명은 그날 날짜다. general.timezone(Asia/Seoul) 기준으로 판정한다.
파일명에 한글을 쓰는 이유: 사용자가 탐색기에서 바로 찾는 것이 이 프로젝트의 배포 방식이다. Windows 로 배포처가 한정되므로 인코딩 문제가 없다. 다만 이메일 첨부나 웹 업로드 시 깨질 수 있음을 README 에 적는다.
10.2 "최신본" 을 심볼릭 링크로 만들지 않는 이유
| 방식 | 문제 |
|---|---|
심볼릭 링크(os.symlink) |
Windows 에서 관리자 권한 또는 개발자 모드 필요. 배치는 일반 사용자 계정(S4U)으로 돈다 |
하드링크(os.link) |
원본을 보존기간 정리로 지우면 최신본만 남아, 어느 날짜인지 알 수 없게 된다 |
바로가기(.lnk) |
xlsx 가 아니라 파일 형식이 달라진다. 더블클릭 동작이 환경마다 다르다 |
| 원자적 복사(채택) | 디스크를 2배 쓰지만 파일 하나가 수 MB 수준이라 무시할 수 있다. 완전히 독립적이고 안전하다 |
refresh_latest_link() 는 최신본이 열려 있으면 조용히 실패하고 False 를 반환한다. 리포트 본체는 이미 저장됐으므로 실패시켜서는 안 된다. 경고만 ReportOutcome.warnings 에 남긴다.
10.3 보관 폴더 구조
reports\
├── DMF_리포트_최신.xlsx ← 항상 최신 성공본의 복사
├── DMF_리포트_2026-09-02.xlsx
├── DMF_리포트_2026-09-01.xlsx
├── DMF_리포트_2026-08-31.xlsx
└── … ← report.retain_days(기본 365) 초과분 삭제
하위 폴더를 만들지 않는다. 아키텍처 §2 의 확정 트리가 reports\ 를 평면으로 정의했고, 1년치 365개는 탐색기가 문제없이 다룬다. 날짜가 파일명에 있어 이름순 정렬이 곧 시간순이다.
10.4 보존 정리 규칙
finalize 스테이지에서 로그 정리와 함께 수행한다.
| 대상 | 보존 | 설정키 |
|---|---|---|
reports\DMF_리포트_YYYY-MM-DD.xlsx |
report.retain_days = 365일 |
report.retain_days |
reports\DMF_리포트_최신.xlsx |
삭제하지 않음 | — |
폴백 파일(_HHMMSS.xlsx) |
7일. 정규 파일이 같은 날짜로 존재하면 즉시 삭제 후보 | 하드코딩 |
잔여 임시 파일(.~dmf_*.xlsx) |
24시간 초과분 무조건 삭제 | 하드코딩 |
data\raw\YYYY-MM-DD\ |
source.archive_retain_days = 180일 |
별도 |
logs\run_*\ |
logging.retain_days = 90일 |
별도 |
def prune_reports(out_dir: Path, retain_days: int, today: date) -> list[Path]:
"""보존 기간 초과 리포트와 잔여 임시 파일을 지운다. 지운 경로 목록 반환."""
removed: list[Path] = []
cutoff = today - timedelta(days=retain_days)
for p in out_dir.glob("DMF_리포트_*.xlsx"):
if p.name == "DMF_리포트_최신.xlsx":
continue
m = re.match(r"DMF_리포트_(\d{4}-\d{2}-\d{2})", p.stem)
if not m:
continue
d = date.fromisoformat(m.group(1))
# 폴백 파일: 정규 파일이 있으면 7일, 없으면 정규와 동일 취급
is_fallback = re.search(r"_\d{6}$", p.stem) is not None
limit = today - timedelta(days=7) if is_fallback else cutoff
if d < limit:
p.unlink(missing_ok=True)
removed.append(p)
now = time.time()
for p in out_dir.glob(".~dmf*.xlsx"):
if now - p.stat().st_mtime > 86400:
p.unlink(missing_ok=True)
removed.append(p)
return removed
10.5 백업과의 관계
리포트는 재생성 가능한 파생물이다. SQLite 정본(data\dmf.sqlite3)과 원문 아카이브(data\raw\)가 살아 있으면 python -m dmf_crawler report-only --run-id <id> 로 언제든 다시 만들 수 있다. 따라서 backup\ 에는 리포트를 넣지 않는다. 백업 대상은 DB 하나다.
부록. 미해결 / 실측 필요
A. 실측이 필요한 것
- nedrug 성분명 검색 URL(
CCBAC03/getItem?...&itemName=)이 실제로 동작하는지. 실패 시 목록 URL 폴백으로 전환. - XlsxWriter
define_name()이 한글 정의 이름을 통과시키는지. 통과하면REPORT_DATE옆에기준일같은 한글 별칭을 추가할 수 있다(현재는 ASCII 만 사용). add_table의total_value옵션이 설치 버전에서 실제로 캐시값을 기록하는지. 기록하지 않으면 합계행을 표 밖의 일반 행으로 옮긴다.write_url로 만든external:로컬 경로 링크가 한글 경로(DMF_리포트_최신.xlsx)에서 정상 동작하는지. Excel 보안 설정에 따라 차단될 수 있다.- 조건부서식에서 정의된 이름(
REPORT_DATE)을 참조하는 수식(CF-02-02,CF-05-02,CF-06-01)이 Excel 에서 정상 평가되는지. 실패하면 기준일 문자열을 리터럴로 굽는다. - 스파크라인 30 포인트가
column타입에서 96px 폭(2열 병합)에 시각적으로 읽히는지. 안 읽히면 14 포인트로 줄인다. records테이블 행수가 수만 건일 때02_전체현황생성 시간과 파일 크기. 30초를 넘으면 Excel 표를 포기하고autofilter로 대체(constant_memory를 켤 수 있게).- 맑은 고딕 10pt 기준 열 폭 계수 1.8 의 실제 적합도. 첫 산출물을 열어 눈으로 확인하고 1.7~1.9 사이에서 보정.
SQL_TOP_COUNTRY의json_each트릭이 설치된 SQLite 버전에서 동작하는지(JSON1 확장 필요). 안 되면 파이썬에서 콤마 분해로 집계.06_추이히든 블록의 열 인덱스(AD=29, BJ=61)가 성분·업체 30일 시계열 폭과 충돌하지 않는지. Top N 이 커지면 블록 시작 열을 재배치.
B. 상위 문서 확정 후 갱신할 것
docs/design/02-data-model.md확정 시 §8.4 의 잠정 스키마·SQL 을 실제 DDL 에 맞춰 교체.records.status의 값 도메인(유효/취하)이 데이터 모델에서 확정되면CF-02-03·SQL_*_CUM의 리터럴을 맞춘다.watchlist테이블의 편집 UI 를docs/design/04-onboarding-wizard.md에서 확정. 현재는 "설정 GUI 에서 편집" 이라고만 정해 두었고 화면이 없다.enrichment스키마(headline/summary/risk_note/note/importance)가prompts/daily_briefing.schema.json과 일치하는지 M3 에서 교차 확인.
C. 의도적으로 채택하지 않은 것 (기록)
스냅샷·로그시트 — 리서치 07 제안. SQLiteevents가 정본이라 xlsx 중복.99_메타의 경로 포인터로 대체.- 상태
연차보고/사전등록— 원천 API 에 등록구분 필드가 없다. nedrug 화면 크롤링을 추가하면 그때 재검토. - 피벗테이블 — 파이썬에서 생성 불가.
03/04의 정적 집계표 + 차트로 대체(리서치 07 §10 의 방식 A). - 워치리스트 사용자 편집 왕복 — openpyxl 의존을 만들어야 해서 기각. SSOT 를 SQLite 로 옮겨 해결.
- 도형(shape) 기반 KPI 카드 — XlsxWriter 로 도형 그룹을 만들 수 없다. 셀 병합 + 테두리 + 배경으로 동일 효과.
- 신호등 아이콘셋 — CVD 위험(red/green + 동일 형태).
3_arrows_gray만 사용. constant_memory모드 — Excel 표와 배타적. 현재 규모에서 불필요.- PDF 자동 변환 — LibreOffice·Excel COM 둘 다 기각. 인쇄 설정으로 대체.
- Okabe-Ito Yellow(
#F0E442) — 흰 배경 대비 1.32. 팔레트에는 있으나 사용 금지.