DB_JOB/_스크립트/_재검수_apply.py
hehihoho3 df16c98366 백업: DB수집 전체 스냅샷 (공공기관2 정리 전)
공공기관2 작업 중. _temp 몽타주(재생성가능)는 제외.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-18 18:15:40 +09:00

92 lines
4.1 KiB
Python

# -*- coding: utf-8 -*-
"""재검수 audit diff를 엑셀에 반영(정정). 신뢰분류 필터 적용.
적용 규칙(보수적):
- L: 페이지→게시판 & newM>=3 → L=게시판, M=newM (실글수 확인된 실게시판)
- L: 게시판→페이지 & oldM<=1 → L=페이지, M=1 (빈게시판/오분류)
- N: 모든 newN 반영(현행 이미지규칙·노이즈제외 적용된 값)
- 억제(미적용·로그만): 페이지→게시판 M<3(검색창/총계미파싱)·게시판→페이지 oldM>1(일시오류 의심)
백업 *_backup_재검수정정전.xlsx. 행수·구조 무변경(셀값만).
사용: python -X utf8 _재검수_apply.py <기관...> [--dry]
"""
import sys, os, io, json, shutil, importlib.util, re
import openpyxl
# 게시판 승격은 URL이 게시판 엔드포인트일 때만(내러티브 '총 N개' 오파싱·예산금액 차단)
BOARD_URL = re.compile(
r'selectBbsNttList|/bbs/|BBSMSTR|board/list|list\.buan|selectBoardList'
r'|selectCntrct|selectPric|/list\.do|mode=list|nttList', re.I)
M_CAP = 200000 # 글수가 이보다 크면 금액/통계 오파싱 의심 → 미적용
HERE = os.path.dirname(os.path.abspath(__file__))
TEMP = os.path.join(HERE, '..', '_temp')
MODULES = ['_chungnam_phase234_all.py', '_chungbuk_phase234_all.py', '_jeonbuk_phase234_all.py']
def to_int(v):
try: return int(v)
except: return 0
def load_sites():
sites = {}
for f in MODULES:
spec = importlib.util.spec_from_file_location(f[:-3], os.path.join(HERE, f))
m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m)
sites.update(m.SITES)
return sites
def apply_inst(name, cfg, dry):
jp = os.path.join(TEMP, f'_audit_{name}.json')
if not os.path.exists(jp):
return f'{name}: audit json 없음'
d = json.load(open(jp, encoding='utf-8'))
xlsx = cfg['xlsx']
wb = openpyxl.load_workbook(xlsx); ws = wb.active
L_applied, L_skip, N_applied, mreview = [], [], [], []
for x in d.get('L_diffs', []):
r, o, n, nm, om = x['r'], x['oldL'], x['newL'], to_int(x['newM']), to_int(x['oldM'])
url = x.get('url', '')
if o == '페이지' and n == '게시판' and 3 <= nm <= M_CAP and BOARD_URL.search(url):
if not dry:
ws.cell(r, 12).value = '게시판'; ws.cell(r, 13).value = nm
L_applied.append((r, x['label'], f'페이지→게시판 M{om}{nm}'))
elif o == '게시판' and n == '페이지' and om <= 1:
if not dry:
ws.cell(r, 12).value = '페이지'; ws.cell(r, 13).value = 1
L_applied.append((r, x['label'], f'게시판→페이지 M{om}→1'))
else:
L_skip.append((r, x['label'], f'{o}{n} M{om}{nm}'))
if o == '페이지' and n == '게시판': # 실게시판 가능성(M미파싱/URL불명/금액의심)
mreview.append((r, x['label'], url, nm))
for x in d.get('N_diffs', []):
r, n = x['r'], x['newN']
if not dry:
ws.cell(r, 14).value = n
N_applied.append((r, x['label'], f"{x['oldN']}{n}"))
if not dry and (L_applied or N_applied):
bak = xlsx.replace('.xlsx', '_backup_재검수정정전.xlsx')
if not os.path.exists(bak):
shutil.copy(xlsx, bak)
wb.save(xlsx)
return {'name': name, 'L적용': len(L_applied), 'L억제': len(L_skip),
'N적용': len(N_applied), 'M검토필요': len(mreview),
'L_applied': L_applied, 'mreview': mreview}
def main():
args = sys.argv[1:]
dry = '--dry' in args
names = [a for a in args if not a.startswith('--')]
sites = load_sites()
for name in names:
if name not in sites:
print(f'{name}: SITES에 없음'); continue
res = apply_inst(name, sites[name], dry)
if isinstance(res, str):
print(res); continue
print(f"{res['name']:6} {'[DRY]' if dry else '[적용]'} L적용{res['L적용']} L억제{res['L억제']} N적용{res['N적용']} M검토{res['M검토필요']}")
for r, lab, msg in res['L_applied']:
print(f' +L r{r} {(lab or "")[:24]} {msg}')
if __name__ == '__main__':
main()