DB_JOB/_스크립트/_재검수_searchform.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.0 KiB
Python

# -*- coding: utf-8 -*-
"""검색폼형 페이지 → 게시판(M=0) 재정정 (사용자룰 2026-06-04).
본문에 <select> 필터 1개 이상 + 검색/조회/찾기 버튼이 있으면 '검색폼형' = 게시판(게시물0).
대상: audit에서 페이지→게시판으로 잡혔다가 억제돼 아직 '페이지'인 행만 재검(검색신호 보유분).
텍스트 절차/안내 페이지(select 없음)는 페이지 유지.
사용: python -X utf8 _재검수_searchform.py <기관...> [--write]
"""
import sys, os, re, json, shutil, importlib.util, time
from concurrent.futures import ThreadPoolExecutor, as_completed
import openpyxl
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']
SEARCHBTN = re.compile(r'검색|조회|찾기')
def load():
sites,mods={},{}
for f in MODULES:
sp=importlib.util.spec_from_file_location(f[:-3],os.path.join(HERE,f));m=importlib.util.module_from_spec(sp);sp.loader.exec_module(m)
for k,v in m.SITES.items(): sites[k]=v;mods[k]=m
return sites,mods
def mkfetch(M,cfg):
if hasattr(M,'make_session'):
s=M.make_session(weak_ssl=cfg.get('weak_ssl',False)); return lambda u:M.fetch(s,u)
return lambda u:M.fetch(u)
# 누출방지: 본문이 '제대로 격리'된 콘텐츠 컨테이너일 때만 select 신뢰
CONTENT_OK = {'#main-contents','#content','#contents','.contents','#txt',
'#sub_content','.content','#cms','.cms_content','.sub_content'}
def is_searchform(M,url,bsel,fetch):
soup=None
for k in range(3):
soup=fetch(url)
if soup is not None: break
time.sleep(0.4*(k+1))
if soup is None: return None
matched=None
for sel in bsel:
if soup.select_one(sel): matched=sel; break
if matched not in CONTENT_OK: # whole-soup/#container/main 등 broad → select 신뢰불가, 보류
return False
b=soup.select_one(matched)
selects=b.find_all('select')
btn=bool(SEARCHBTN.search(' '.join((x.get_text() or '')+' '+(x.get('value') or '')
for x in b.find_all(['button','input','a']))))
return len(selects)>=1 and btn
def run(name,cfg,M,write):
jp=os.path.join(TEMP,f'_audit_{name}.json')
if not os.path.exists(jp): print(f'{name}: audit없음'); return
d=json.load(open(jp,encoding='utf-8'))
wb=openpyxl.load_workbook(cfg['xlsx']); ws=wb.active
cand=[x for x in d['L_diffs'] if x['oldL']=='페이지' and x['newL']=='게시판'
and ws.cell(x['r'],12).value=='페이지']
fetch=mkfetch(M,cfg); bsel=cfg['body_sel']
res={}
with ThreadPoolExecutor(max_workers=6) as ex:
futs={ex.submit(is_searchform,M,x['url'],bsel,fetch):x for x in cand}
for f in as_completed(futs):
x=futs[f]
try: res[x['r']]=(f.result(),x)
except: res[x['r']]=(None,x)
flip=[(r,x) for r,(v,x) in res.items() if v is True]
keep=[(r,x) for r,(v,x) in res.items() if v is False]
fail=[(r,x) for r,(v,x) in res.items() if v is None]
if write and flip:
bak=cfg['xlsx'].replace('.xlsx','_backup_검색폼게시판전.xlsx')
if not os.path.exists(bak): shutil.copy(cfg['xlsx'],bak)
for r,x in flip:
ws.cell(r,12).value='게시판'; ws.cell(r,13).value=0
wb.save(cfg['xlsx'])
print('%s: 억제후보%d → 검색폼게시판%d 페이지유지%d 실패%d %s'%(name,len(cand),len(flip),len(keep),len(fail),'[적용]' if write else '[DRY]'))
for r,x in sorted(flip)[:15]:
print(f' +게시판0 r{r} {(x["label"] or "")[:22]} {x["url"][:62]}')
if keep:
print(' (페이지유지 예:)')
for r,x in sorted(keep)[:5]:
print(f' r{r} {(x["label"] or "")[:22]} {x["url"][:55]}')
def main():
args=sys.argv[1:]; write='--write' in args
names=[a for a in args if not a.startswith('--')]
sites,mods=load()
for n in names:
if n in sites: run(n,sites[n],mods[n],write)
else: print(f'{n}: 없음')
if __name__=='__main__': main()