# -*- coding: utf-8 -*- """미검수 기관 재검수 audit (읽기전용) — 2026-06-03. 각 기관 엑셀의 K(URL) 행을 재접속해서: - L(게시판/페이지) : detect_form 재판정 → 기존 L과 diff - N(어문/이미지/영상/오디오) : 현행규칙(3-1a/3-3/3-4b, _redo_N_all의 media2) 재판정 → 기존 N과 diff 엑셀은 절대 수정하지 않음. diff만 JSON+MD로 출력. 검수완료 15곳은 제외(읽기만). 대상=미검수 27곳. 사용: python -X utf8 _재검수_audit.py [기관...] # 미지정 시 27곳 전체 """ import sys, os, io, re, importlib.util, warnings, json, time from concurrent.futures import ThreadPoolExecutor, as_completed import openpyxl warnings.filterwarnings('ignore') HERE = os.path.dirname(os.path.abspath(__file__)) OUT = os.path.join(HERE, '..', '_temp') MODULES = ['_chungnam_phase234_all.py', '_chungbuk_phase234_all.py', '_jeonbuk_phase234_all.py'] INSPECTED = {'계룡시','공주시','금산군','논산시','보령시','당진시','부여군','서천군','예산군','태안군', '고창군','김제시','남원시','군산시','무주군'} TARGETS = { # 미검수 27곳 '서산시','아산시','천안시','청양군','홍성군', '괴산군','단양군','보은군','영동군','옥천군','음성군','제천시','증평군','진천군','청주시','충주시', '부안군','순창군','완주군','익산시','임실군','장수군','전주시','정읍시','진안군', '서귀포시','제주시', } # ── N 이미지 분류 (벤더 공통, _redo_N_all과 동일) ────────── DECO = re.compile(r'/common/|move\.png|no[-_]?img|blank|spacer|/ico|/btn|bullet|arrow|/bg|icon|see_btn|/sample|mimetype|/file_|filedown|btn_dir', re.I) EXCLUDE = re.compile( r'한눈에|흐름도|절차도|처리절차|이용절차' r'|로고(?!송)|logo(?!song)|아이콘|배너|banner' r'|신문고|relation_item|tracer|headline' r'|카피라이트|copyright|copy_logo|popup|/pup/|wa_mk|웹접근성|품질인증', re.I) AUDIO = re.compile(r'\.(?:mp3|wav|m4a|ogg|flac)\b', re.I) MAPPDF = re.compile(r'pdf|viewer\.html|/map|kakao|daum.*map', re.I) def load_module(fname): spec = importlib.util.spec_from_file_location(fname[:-3], os.path.join(HERE, fname)) m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m) return m def real_imgs(M, body): out = [] for img in body.find_all('img'): src = img.get('src') or '' if not src or M.KOGL_IMG_PAT.search(src) or DECO.search(src): continue if EXCLUDE.search(src + ' ' + (img.get('alt') or '')): continue out.append(img) return out def media2(M, body): has_text = len(body.get_text(strip=True)) > 30 img = len(real_imgs(M, body)) > 0 vid = False for ifr in body.find_all('iframe'): s = ifr.get('src') or '' if M.YOUTUBE_PAT.search(s): vid = True elif MAPPDF.search(s): img = True if not vid and (body.find('video') or body.find('a', href=M.YOUTUBE_PAT) or M.VIDEO_EXT.search(str(body))): vid = True aud = bool(body.find('audio')) or bool(AUDIO.search(str(body))) return img, vid, aud, has_text def make_fetch(M, cfg): if hasattr(M, 'make_session'): sess = M.make_session(weak_ssl=cfg.get('weak_ssl', False)) return lambda u: M.fetch(sess, u) return lambda u: M.fetch(u) def n_join(txt, img, vid, aud): parts = [] if txt: parts.append('어문') if img: parts.append('이미지') if vid: parts.append('영상') if aud: parts.append('오디오') return ','.join(parts) if parts else '없음' def assess(M, url, oldL, body_sel, fetchfn): """반환: dict(newL, newM, newN) 또는 None(접근실패→기존유지).""" soup = None for k in range(3): soup = fetchfn(url) if soup is not None: break time.sleep(0.5 * (k + 1)) if soup is None: return None body = M.get_body(soup, body_sel) form, count = M.detect_form(body) # L 재판정 (기존 로직 그대로) img, vid, aud, txt = media2(M, body) if not (txt or img or vid or aud): # 빈 본문 — 신뢰불가. L은 보고하되 N은 기존유지 신호 return {'newL': form, 'newM': (count if form=='게시판' else 1), 'newN': None} if form == '게시판': detail_urls = M.extract_detail_urls(body, url, limit=5) data_rows = [tr for tr in body.select('table tbody tr, .board_list li, ul.bbs_list li') if tr.find('a')] empty_msg = bool(re.search(r'게시물이?\s*없|등록된\s*(?:게시물|자료)\s*가?\s*없|자료가\s*없', body.get_text(' ', strip=True))) if not detail_urls and not data_rows and (empty_msg or not txt): return {'newL': form, 'newM': 0, 'newN': '없음'} for du in detail_urls: ds = fetchfn(du) if not ds: continue db = M.get_body(ds, body_sel) i2, v2, a2, t2 = media2(M, db) img = img or i2; vid = vid or v2; aud = aud or a2; txt = txt or t2 return {'newL': form, 'newM': (count if form=='게시판' else 1), 'newN': n_join(txt, img, vid, aud)} def label_of(ws, r): vals = [ws.cell(r, c).value for c in range(4, 11)] # D~J vals = [str(v).strip() for v in vals if v not in (None, '')] return vals[-1] if vals else '' def do_inst(M, name, cfg): xlsx = cfg['xlsx'] if not os.path.exists(xlsx): return {'name': name, 'error': 'no_file'} body_sel = cfg['body_sel'] fetchfn = make_fetch(M, cfg) wb = openpyxl.load_workbook(xlsx, read_only=False) ws = wb.active jobs = [] for r in range(3, ws.max_row + 1): K = ws.cell(r, 11).value L = ws.cell(r, 12).value if not (isinstance(K, str) and K.startswith('http')): continue if L not in ('페이지', '게시판'): # 사이트/빈칸 스킵 continue jobs.append((r, K, L, ws.cell(r,2).value, ws.cell(r,13).value, ws.cell(r,14).value, label_of(ws, r))) results = {} with ThreadPoolExecutor(max_workers=6) as ex: futs = {ex.submit(assess, M, K, L, body_sel, fetchfn): (r,K,L,B,Mq,Nq,lab) for (r,K,L,B,Mq,Nq,lab) in jobs} for f in as_completed(futs): r,K,L,B,Mq,Nq,lab = futs[f] try: results[r] = (f.result(), K,L,B,Mq,Nq,lab) except Exception as e: results[r] = (None, K,L,B,Mq,Nq,lab) L_diffs, N_diffs, fails = [], [], 0 for r in sorted(results): res, K,L,B,Mq,Nq,lab = results[r] if res is None: fails += 1 continue if res['newL'] != L: L_diffs.append({'r':r,'B':B,'label':lab,'url':K,'oldL':L,'newL':res['newL'], 'oldM':Mq,'newM':res['newM']}) newN = res['newN'] if newN is not None and newN != (Nq or ''): N_diffs.append({'r':r,'B':B,'label':lab,'url':K,'oldN':Nq,'newN':newN,'L':L}) return {'name': name, 'rows': len(jobs), 'fails': fails, 'L_diff_n': len(L_diffs), 'N_diff_n': len(N_diffs), 'L_diffs': L_diffs, 'N_diffs': N_diffs} def main(): only = set(a for a in sys.argv[1:] if not a.startswith('--')) sel = (only & TARGETS) if only else TARGETS os.makedirs(OUT, exist_ok=True) grand = [] for fname in MODULES: M = load_module(fname) for name, cfg in M.SITES.items(): if name not in sel: continue t0 = time.time() res = do_inst(M, name, cfg) res['secs'] = round(time.time()-t0) grand.append(res) tag = res.get('error') or f"행{res['rows']} L변경{res['L_diff_n']} N변경{res['N_diff_n']} 실패{res['fails']} ({res['secs']}s)" print(f'{name:7} {tag}', flush=True) # 기관별 즉시 저장(중간 크래시 대비) with io.open(os.path.join(OUT, f'_audit_{name}.json'), 'w', encoding='utf-8') as fp: json.dump(res, fp, ensure_ascii=False, indent=1) with io.open(os.path.join(OUT, '_재검수_audit_all.json'), 'w', encoding='utf-8') as fp: json.dump(grand, fp, ensure_ascii=False, indent=1) print('='*55) print(f"총 {len(grand)}곳 | L변경합 {sum(g.get('L_diff_n',0) for g in grand)} | " f"N변경합 {sum(g.get('N_diff_n',0) for g in grand)} | " f"접근실패합 {sum(g.get('fails',0) for g in grand)}") if __name__ == '__main__': main()