df16c98366
공공기관2 작업 중. _temp 몽타주(재생성가능)는 제외. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
393 lines
15 KiB
Python
393 lines
15 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""공공기관 Phase 1: 사이트맵/메가메뉴 → 메뉴트리 D~K (Playwright 렌더 + 범용 중첩리스트 추출).
|
|
|
|
출력: D:\\01.프로젝트\\DB수집\\공공기관\\{기관}.xlsx (평면배치)
|
|
시트명: {번호}_{기관}
|
|
사용: python _공공기관_phase1.py [기관명 ...] (인자 없으면 OVERRIDES 정의된 전체)
|
|
"""
|
|
import sys, io, os, re, shutil, json, warnings
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
|
|
from copy import copy
|
|
from urllib.parse import urljoin, urlparse
|
|
import openpyxl
|
|
from bs4 import BeautifulSoup
|
|
from openpyxl.styles import Alignment, Font
|
|
|
|
warnings.filterwarnings('ignore')
|
|
TEMPLATE = r'D:\01.프로젝트\DB수집\자료_취합_예시.xlsx'
|
|
OUTDIR = r'D:\01.프로젝트\DB수집\공공기관'
|
|
PROBE = r'D:\01.프로젝트\DB수집\_스크립트\_공공기관_probe.json'
|
|
UA = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
|
|
|
|
# 기관별 오버라이드: sitemap URL + 컨테이너 셀렉터(선택) + 렌더 대기(wait). probe json을 기본값으로.
|
|
OVERRIDES = {
|
|
'국립생태원': {'sitemap': 'https://www.nie.re.kr/nie/main/main.do', 'wait': 5000},
|
|
'국토안전관리원': {'use_home': True, 'sel': '.all-menu', 'wait': 4500},
|
|
'국립중앙의료원': {'use_home': True, 'sel': '#siteMap', 'wait': 4500},
|
|
'건설근로자공제회': {'use_home': True, 'wait': 4500},
|
|
'국민연금공단': {'sitemap': 'https://www.nps.or.kr/main.do', 'sel': '.allmenu-container', 'wait': 4000},
|
|
'국민건강보험공단': {'sitemap': 'https://www.nhis.or.kr/nhis/index.do', 'sel': 'nav.head-gnb', 'wait': 4000},
|
|
}
|
|
|
|
JS_PATS = [
|
|
# goMenuPage('MENUID','/realpath',...) 형: 2번째 인자가 실제 URL
|
|
re.compile(r"""go\w*(?:Menu|Page|Move)\w*\s*\(\s*['"][^'"]*['"]\s*,\s*['"]([^'"]+)['"]""", re.I),
|
|
re.compile(r"""go(?:Menu|SubMenu|Page|Link|Url|View|Move)\s*\(\s*['"]([^'"]+)['"]""", re.I),
|
|
re.compile(r"""(?:location\.href|window\.location(?:\.href)?)\s*=\s*(?:encodeURI\()?\s*['"]([^'"]+)['"]"""),
|
|
re.compile(r"""window\.open\s*\(\s*['"]([^'"]+)['"]"""),
|
|
re.compile(r"""(?:fn_?\w*|aLink|movePage|menuMove)\s*\(\s*['"]([^'"]+)['"]""", re.I),
|
|
]
|
|
|
|
|
|
def js_href(a):
|
|
s = (a.get('href', '') or '') + ' ' + (a.get('onclick', '') or '')
|
|
for p in JS_PATS:
|
|
mm = p.search(s)
|
|
if mm:
|
|
u = mm.group(1).strip()
|
|
if u.startswith(('/', 'http', './', '?', '../')):
|
|
return u
|
|
return ''
|
|
|
|
|
|
def clean(s):
|
|
return re.sub(r'\s+', ' ', (s or '')).strip().replace('\xa0', '')
|
|
|
|
|
|
def render(url, wait=2500):
|
|
from playwright.sync_api import sync_playwright
|
|
with sync_playwright() as p:
|
|
b = p.chromium.launch()
|
|
pg = b.new_page(user_agent=UA, viewport={'width': 1440, 'height': 2400})
|
|
try:
|
|
pg.goto(url, timeout=30000, wait_until='domcontentloaded')
|
|
except Exception:
|
|
pass
|
|
pg.wait_for_timeout(wait)
|
|
html = pg.content()
|
|
final = pg.url
|
|
b.close()
|
|
return final, html
|
|
|
|
|
|
SITEMAP_HINTS = ('sitemap', 'site_map', 'sitemapwrap', 'site-map', 'allmenu', 'all_menu', 'all-menu',
|
|
'menu_all', 'menuall', 'totalmenu', 'total_menu', 'totmenu', 'full_menu', 'fullmenu',
|
|
'gnb_all', 'gnball')
|
|
|
|
|
|
def pick_container(soup, prefer=None):
|
|
if prefer:
|
|
el = soup.select_one(prefer)
|
|
if el and len(el.find_all('a')) >= 8:
|
|
return el
|
|
cands = []
|
|
for d in soup.find_all(['div', 'section', 'main', 'nav', 'ul']):
|
|
cls = ' '.join(d.get('class', [])).lower() + ' ' + (d.get('id', '') or '').lower()
|
|
clsn = cls.replace(' ', '').replace('-', '').replace('_', '')
|
|
if any(x in cls for x in ['footer', 'aside']):
|
|
continue
|
|
ac = len(d.find_all('a'))
|
|
ul = len(d.find_all('ul'))
|
|
if ac < 12:
|
|
continue
|
|
score = ac + ul * 2
|
|
# 사이트맵/전체메뉴 컨테이너 강력 우대
|
|
if any(h in clsn for h in SITEMAP_HINTS):
|
|
score += 1000
|
|
# 전역 헤더/상단 gnb는 (사이트맵 페이지에선) 감점
|
|
if 'header' in cls or ('gnb' in clsn and not any(h in clsn for h in SITEMAP_HINTS)):
|
|
score -= 200
|
|
cands.append((score, ac, d))
|
|
if not cands:
|
|
return None
|
|
cands.sort(key=lambda x: x[0], reverse=True)
|
|
return cands[0][2]
|
|
|
|
|
|
NOISE_LABELS = {'메뉴없음', '메뉴 없음', '홈', 'home', 'home으로', '홈으로', 'eng', 'english', '로그인',
|
|
'login', '회원가입', '검색', 'search', '바로가기', '본문바로가기', '닫기', 'close', '전체메뉴',
|
|
'사이트맵', 'sitemap'}
|
|
|
|
|
|
def node_depth(el, container):
|
|
"""container 내부에서 el 위의 ul/ol/dl 조상 개수 = 중첩 깊이."""
|
|
d = 0
|
|
p = el.parent
|
|
while p is not None and p is not container:
|
|
if getattr(p, 'name', None) in ('ul', 'ol', 'dl'):
|
|
d += 1
|
|
p = p.parent
|
|
return d
|
|
|
|
|
|
def parse_table_sitemap(table):
|
|
"""테이블형 사이트맵: tr > th(대분류, 빈칸=이어받기) + th(중분류 a) + td(소분류 a들)."""
|
|
rows = []
|
|
curD = ''
|
|
for tr in table.find_all('tr'):
|
|
ths = tr.find_all('th', recursive=False)
|
|
if ths:
|
|
d_txt = clean(ths[0].get_text())
|
|
# 첫 th에 a가 없고 span/텍스트면 대분류 라벨(빈칸이면 이어받기)
|
|
if d_txt and not (len(ths) == 1 and ths[0].find('a')):
|
|
# 첫 th가 곧 중분류 a 단독인 경우는 제외 위해 a 유무 확인
|
|
if not ths[0].find('a') or len(ths) >= 2:
|
|
if not ths[0].find('a'):
|
|
curD = d_txt
|
|
E = ''; Eh = ''
|
|
if len(ths) >= 2:
|
|
ea = ths[1].find('a')
|
|
if ea:
|
|
E = clean(ea.get_text()); Eh = js_href(ea) or (ea.get('href') or '')
|
|
elif len(ths) == 1 and ths[0].find('a'):
|
|
ea = ths[0].find('a'); E = clean(ea.get_text()); Eh = js_href(ea) or (ea.get('href') or '')
|
|
td = tr.find('td')
|
|
leaves = td.find_all('a') if td else []
|
|
if leaves:
|
|
for la in leaves:
|
|
lab = clean(la.get_text())
|
|
if not lab:
|
|
continue
|
|
rows.append({'path': [curD, E, lab], 'href': js_href(la) or (la.get('href') or '')})
|
|
elif E:
|
|
rows.append({'path': [curD, E], 'href': Eh})
|
|
return [r for r in rows if r['path'] and any(r['path'])]
|
|
|
|
|
|
def extract_rows(container):
|
|
"""앵커 중심 깊이추출: 각 a/heading의 ul조상 수로 컬럼 결정(정규화). 범용. 테이블형 자동전환."""
|
|
tbls = container.find_all('table')
|
|
if tbls and sum(len(t.find_all('th')) for t in tbls) >= 4 and len(container.find_all('ul')) <= 1:
|
|
rows = []
|
|
for t in tbls:
|
|
rows += parse_table_sitemap(t)
|
|
if rows:
|
|
return rows
|
|
nodes = [] # [depth, label, href]
|
|
for el in container.find_all(['a', 'button', 'h2', 'h3', 'h4', 'h5', 'h6', 'dt']):
|
|
name = el.name
|
|
if name == 'a':
|
|
href = (el.get('href') or '').strip()
|
|
if href.startswith('#') or href.lower().startswith('javascript:') or not href:
|
|
href = js_href(el)
|
|
label = clean(el.get_text())
|
|
elif name == 'button':
|
|
cls = ' '.join(el.get('class', [])).lower()
|
|
if not (el.get('aria-expanded') is not None or el.get('aria-haspopup')
|
|
or any(x in cls for x in ('depth', 'trigger', 'gnb', '1d', '2d', '3d', 'menu'))):
|
|
continue
|
|
href = ''
|
|
label = clean(el.get('data-dir') or el.get('title') or el.get_text())
|
|
else:
|
|
if el.find('a'): # heading 안의 a는 따로 잡힘 → 중복방지
|
|
continue
|
|
href = ''
|
|
label = clean(el.get_text())
|
|
if not label or label.lower() in NOISE_LABELS or len(label) > 60:
|
|
continue
|
|
nodes.append([node_depth(el, container), label, href])
|
|
# 연속 중복 제거
|
|
dedup = []
|
|
for n in nodes:
|
|
if dedup and dedup[-1] == n:
|
|
continue
|
|
dedup.append(n)
|
|
nodes = dedup
|
|
if not nodes:
|
|
return []
|
|
mind = min(n[0] for n in nodes)
|
|
cols = [min(n[0] - mind, 6) for n in nodes]
|
|
out = []
|
|
path = [''] * 7
|
|
for i, (depth, label, href) in enumerate(nodes):
|
|
c = cols[i]
|
|
path[c] = label
|
|
for k in range(c + 1, 7):
|
|
path[k] = ''
|
|
is_leaf = (i == len(nodes) - 1) or (cols[i + 1] <= c)
|
|
if href or is_leaf:
|
|
out.append({'path': path[:c + 1], 'href': href})
|
|
return out
|
|
|
|
|
|
def rows_to_dicts(raw):
|
|
out = []
|
|
for it in raw:
|
|
p = it['path']
|
|
href = it.get('href', '')
|
|
if href.startswith('#') or href.lower().startswith('javascript:'):
|
|
href = ''
|
|
row = {'D': '', 'E': '', 'F': '', 'G': '', 'H': '', 'I': '', 'J': '', 'href': href}
|
|
for i, lab in enumerate(p[:7]):
|
|
row['DEFGHIJ'[i]] = lab
|
|
out.append(row)
|
|
return out
|
|
|
|
|
|
def write_excel(name, num, base, raw_rows):
|
|
domain = urlparse(base).netloc
|
|
output = os.path.join(OUTDIR, f'{name}.xlsx')
|
|
|
|
def abs_url(href):
|
|
if not href:
|
|
return ''
|
|
href = href.strip()
|
|
if href.startswith(('javascript:', '#')):
|
|
return ''
|
|
if href.startswith(('http://', 'https://')):
|
|
return href
|
|
return urljoin(base + '/', href)
|
|
|
|
def is_external(url):
|
|
return url.startswith(('http://', 'https://')) and domain not in url
|
|
|
|
# 부모-자식 URL 중복 제거 (전 깊이) — 부모행 D~lc 동일 + lc+1 채워짐 + K동일
|
|
def colval(r, c):
|
|
return r.get(c, '') or ''
|
|
final = []
|
|
i = 0
|
|
removed = 0
|
|
cols = 'DEFGHIJ'
|
|
while i < len(raw_rows):
|
|
row = raw_rows[i]
|
|
# 마지막 채워진 컬럼 찾기
|
|
lc = -1
|
|
for ci, c in enumerate(cols):
|
|
if colval(row, c) != '':
|
|
lc = ci
|
|
dup = False
|
|
if i + 1 < len(raw_rows) and lc >= 0 and lc < 6:
|
|
nxt = raw_rows[i + 1]
|
|
same = all(colval(row, cols[k]) == colval(nxt, cols[k]) for k in range(lc + 1))
|
|
if (same and colval(nxt, cols[lc + 1]) != '' and colval(row, cols[lc + 1]) == ''
|
|
and row.get('href', '') == nxt.get('href', '')):
|
|
dup = True
|
|
if dup:
|
|
removed += 1
|
|
i += 1
|
|
continue
|
|
final.append(row)
|
|
i += 1
|
|
|
|
if not final:
|
|
print(f' [{name}] 행 0개 — 스킵')
|
|
return 0
|
|
shutil.copy(TEMPLATE, output)
|
|
wb = openpyxl.load_workbook(output)
|
|
ws = wb.active
|
|
ws.title = f'{num:02d}_{name}'
|
|
HEADER = {'B1:R1', 'S1:W1', 'Y1:AA1'}
|
|
for rng in [str(m) for m in ws.merged_cells.ranges if str(m) not in HEADER]:
|
|
ws.unmerge_cells(rng)
|
|
for row in ws.iter_rows(min_row=3, max_row=ws.max_row, min_col=1, max_col=ws.max_column):
|
|
for cell in row:
|
|
cell.value = None
|
|
START = 3
|
|
template_r = 3
|
|
cur_max = ws.max_row
|
|
for idx, item in enumerate(final, start=START):
|
|
if idx > cur_max:
|
|
for c in range(1, ws.max_column + 1):
|
|
srcc = ws.cell(template_r, c)
|
|
tgt = ws.cell(idx, c)
|
|
if srcc.has_style:
|
|
tgt.font = copy(srcc.font); tgt.fill = copy(srcc.fill)
|
|
tgt.border = copy(srcc.border); tgt.alignment = copy(srcc.alignment)
|
|
tgt.number_format = srcc.number_format; tgt.protection = copy(srcc.protection)
|
|
url = abs_url(item.get('href', ''))
|
|
ws.cell(idx, 2).value = idx - 2
|
|
ws.cell(idx, 3).value = name
|
|
for ci, c in enumerate('DEFGHIJ'):
|
|
ws.cell(idx, 4 + ci).value = item.get(c, '')
|
|
ws.cell(idx, 11).value = url
|
|
if is_external(url):
|
|
ws.cell(idx, 19).value = '외부링크'
|
|
END = START + len(final) - 1
|
|
center = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
|
left = Alignment(horizontal='left', vertical='center', wrap_text=False)
|
|
|
|
def merge_runs(col_letter, col_idx, group_cols=()):
|
|
runs = []
|
|
cur_val = ws.cell(START, col_idx).value
|
|
cur_grp = tuple(ws.cell(START, g).value for g in group_cols)
|
|
run_start = START
|
|
for r in range(START + 1, END + 1):
|
|
v = ws.cell(r, col_idx).value
|
|
g = tuple(ws.cell(r, gg).value for gg in group_cols)
|
|
if v == cur_val and g == cur_grp:
|
|
continue
|
|
if cur_val not in (None, '') and r - 1 > run_start:
|
|
runs.append((run_start, r - 1))
|
|
cur_val, cur_grp, run_start = v, g, r
|
|
if cur_val not in (None, '') and END > run_start:
|
|
runs.append((run_start, END))
|
|
for s, e in runs:
|
|
ws.merge_cells(f'{col_letter}{s}:{col_letter}{e}')
|
|
ws.cell(s, col_idx).alignment = center
|
|
return len(runs)
|
|
|
|
n_f = merge_runs('F', 6, group_cols=(4, 5))
|
|
n_e = merge_runs('E', 5, group_cols=(4,))
|
|
n_d = merge_runs('D', 4)
|
|
for r in range(START, END + 1):
|
|
for c in (4, 5, 6, 7):
|
|
if ws.cell(r, c).value is not None:
|
|
ws.cell(r, c).alignment = center
|
|
for r in range(1, END + 1):
|
|
ws.row_dimensions[r].height = 15
|
|
link_n = 0
|
|
for r in range(START, END + 1):
|
|
cell = ws.cell(r, 11)
|
|
u = cell.value
|
|
if u and isinstance(u, str) and u.startswith(('http://', 'https://')):
|
|
cell.hyperlink = u
|
|
old = cell.font
|
|
cell.font = Font(name=old.name or '맑은 고딕', size=old.size or 11,
|
|
bold=old.bold, italic=old.italic, color='0000FF', underline='single')
|
|
cell.alignment = left
|
|
link_n += 1
|
|
wb.save(output)
|
|
ext_n = sum(1 for r in range(START, END + 1) if ws.cell(r, 19).value == '외부링크')
|
|
print(f' [{name}] 원본{len(raw_rows)}→중복{removed}→{len(final)}행 | 병합D{n_d}E{n_e}F{n_f} | 외부{ext_n} K링크{link_n} → {output}')
|
|
return len(final)
|
|
|
|
|
|
def load_probe():
|
|
with open(PROBE, encoding='utf-8') as f:
|
|
return {r['name']: r for r in json.load(f)}
|
|
|
|
|
|
def main():
|
|
probe = load_probe()
|
|
targets = sys.argv[1:] if len(sys.argv) > 1 else list(probe.keys())
|
|
for name in targets:
|
|
p = probe.get(name)
|
|
if not p:
|
|
print(f'[{name}] probe 정보 없음 — 스킵'); continue
|
|
ov = OVERRIDES.get(name, {})
|
|
num = int(p['num'])
|
|
base = p['base']
|
|
if ov.get('use_home'):
|
|
sm = p['home']
|
|
else:
|
|
sm = ov.get('sitemap') or p.get('sitemap') or p['home']
|
|
wait = ov.get('wait', 3000)
|
|
print(f"\n=== {num}.{name} === render {sm}")
|
|
try:
|
|
final, html = render(sm, wait=wait)
|
|
soup = BeautifulSoup(html, 'html.parser')
|
|
cont = pick_container(soup, ov.get('sel'))
|
|
if not cont:
|
|
print(f' [{name}] 컨테이너 못찾음 (a없음)'); continue
|
|
raw = extract_rows(cont)
|
|
dicts = rows_to_dicts(raw)
|
|
write_excel(name, num, base, dicts)
|
|
except Exception as e:
|
|
import traceback
|
|
print(f' [{name}] 실패: {e}')
|
|
traceback.print_exc()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|