df16c98366
공공기관2 작업 중. _temp 몽타주(재생성가능)는 제외. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
322 lines
12 KiB
Python
322 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""공공기관 Phase 2~4: L(게시판형태)·M(수량)·N(저작물유형)·O/P/Q(공공누리).
|
|
|
|
시·군용 _chungnam_phase234_all.py 로직 재사용 + 범용화(넓은 본문셀렉터·다양한 상세패턴·오디오·이미지노이즈필터).
|
|
출력: 공공기관\{기관}.xlsx 의 L~Q 채움 (역순 실행 권장).
|
|
사용: python _공공기관_phase234.py [기관명 ...] (없으면 번호 역순 전체)
|
|
"""
|
|
import sys, os, re, json, time, warnings
|
|
from urllib.parse import urljoin, urlparse
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
import openpyxl
|
|
|
|
warnings.filterwarnings('ignore')
|
|
UA = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
|
|
H = {'User-Agent': UA}
|
|
OUTDIR = r'D:\01.프로젝트\DB수집\공공기관'
|
|
PROBE = r'D:\01.프로젝트\DB수집\_스크립트\_공공기관_probe.json'
|
|
|
|
TOTAL_PAT = re.compile(r'총\s*(?:게시물|건수)?\s*[:\-]?\s*(\d[\d,]*)\s*(?:건|개|page|페이지|item)', re.I)
|
|
TOTAL_PAT2 = re.compile(r'(?:전체|총|total)\s*[:\-]?\s*(\d[\d,]*)', re.I)
|
|
KOGL_IMG_PAT = re.compile(r'(?:new_)?img_open(?:type|code)(\d{1,2})\.(?:png|jpe?g|gif)', re.I)
|
|
KOGL_LINK_PAT = re.compile(r'kogl\.or\.kr/info/licenseType(\d)', re.I)
|
|
YOUTUBE_PAT = re.compile(r'(?:youtube\.com|youtu\.be|vimeo\.com)', re.I)
|
|
VIDEO_EXT = re.compile(r'\.(mp4|webm|mov|avi|m3u8)(?:\?|$)', re.I)
|
|
AUDIO_EXT = re.compile(r'\.(mp3|wav|m4a|ogg|flac)(?:\?|$|["\'&])', re.I)
|
|
IMG_NOISE = re.compile(r'(ico[_\-/]|/icon|logo|btn|bul[_\-]|bg[_\-]|banner|sns|blank|spacer|loading|arrow|/dot|line[_\.]|top_|foot|header|common|btn_|_icon|symbol|copyright|qr_|movie_ico|no_img|noimage|share|facebook|insta|youtube_ic|blog|twitter|naver|kakao)', re.I)
|
|
DETAIL_PAT = re.compile(r'(mode=V|view\.do|/view|read\.do|/read|detail\.do|/detail|bbsView|nttId=|articleNo=|boardSeq=|bbtSn=|idx=|seq=|bIdx=|board_no=|wr_id=|dataSid=|ntceSn=|brdId=|bcIdx=|page_idx=|=view)', re.I)
|
|
|
|
BODY_SEL = [
|
|
'#content', '#contents', '#contentsArea', '.contentsArea', '.content', '.contents',
|
|
'#sub_content', '.sub_content', '.sub_contents', '#subContent', '.subContent',
|
|
'#container .content', '.board_view', '.bbs_view', '.view_con', '.view_cont',
|
|
'.board', '.bbs', '#bbs', '.sub_cont', '#cont', '.cont_area', '#content_area',
|
|
'main', '#main', 'article', '.board_list', '.bbs_list',
|
|
]
|
|
|
|
|
|
def make_session(weak=False):
|
|
s = requests.Session()
|
|
s.headers.update(H)
|
|
return s
|
|
|
|
|
|
def fetch(session, url, timeout=12):
|
|
try:
|
|
r = session.get(url, timeout=timeout, verify=False, allow_redirects=True)
|
|
meta = re.search(rb'charset=["\']?\s*([\w-]+)', r.content[:4096], re.I)
|
|
r.encoding = meta.group(1).decode(errors='ignore') if meta else r.apparent_encoding
|
|
if r.status_code == 200:
|
|
return BeautifulSoup(r.text, 'html.parser'), r.url
|
|
except Exception:
|
|
pass
|
|
return None, None
|
|
|
|
|
|
def get_body(soup, selectors):
|
|
for sel in selectors:
|
|
try:
|
|
el = soup.select_one(sel)
|
|
if el and len(el.get_text(strip=True)) > 20:
|
|
return el
|
|
except Exception:
|
|
pass
|
|
return soup
|
|
|
|
|
|
def detect_form(body):
|
|
has_paging = bool(body.select('.pagination, .paging, nav.paging, .page_nav, .paginate, .pgwrap, .board_paging, .num_wrap'))
|
|
text_inputs = [i for i in body.find_all('input')
|
|
if (i.get('type') or 'text').lower() in ('text', 'search')]
|
|
has_search = len(text_inputs) >= 1
|
|
has_listtable = bool(body.select('table.board_list, table.bbs_list, ul.board_list, .board_list tbody tr, .bbs_list li'))
|
|
txt = body.get_text(' ', strip=True)
|
|
m = TOTAL_PAT.search(txt) or TOTAL_PAT2.search(txt)
|
|
total = None
|
|
if m:
|
|
digits = m.group(1).replace(',', '')
|
|
if digits.isdigit():
|
|
total = int(digits)
|
|
is_board = has_paging or has_listtable or (total is not None and (has_search or has_paging or has_listtable))
|
|
if is_board:
|
|
return '게시판', total if total is not None else 0
|
|
if has_search and (has_paging or has_listtable):
|
|
return '게시판', total if total is not None else 0
|
|
return '페이지', 1
|
|
|
|
|
|
def extract_detail_urls(body, base_url, limit=6):
|
|
urls, seen = [], set()
|
|
for a in body.find_all('a', href=True):
|
|
h = a['href']
|
|
if not h or h.startswith('#') or h.lower().startswith('javascript:'):
|
|
continue
|
|
if DETAIL_PAT.search(h):
|
|
full = urljoin(base_url, h)
|
|
if full not in seen:
|
|
seen.add(full); urls.append(full)
|
|
if len(urls) >= limit:
|
|
break
|
|
return urls
|
|
|
|
|
|
def detect_media(body):
|
|
has_text = len(body.get_text(strip=True)) > 30
|
|
has_image = False
|
|
for img in body.find_all('img'):
|
|
src = img.get('src') or img.get('data-src') or ''
|
|
if not src or KOGL_IMG_PAT.search(src) or IMG_NOISE.search(src):
|
|
continue
|
|
w = img.get('width', '')
|
|
try:
|
|
if w and int(re.sub(r'\D', '', w) or 0) and int(re.sub(r'\D', '', w)) < 60:
|
|
continue
|
|
except Exception:
|
|
pass
|
|
has_image = True
|
|
break
|
|
has_video = bool(body.find_all('video'))
|
|
if not has_video:
|
|
for ifr in body.find_all('iframe'):
|
|
if YOUTUBE_PAT.search(ifr.get('src', '')):
|
|
has_video = True; break
|
|
if not has_video:
|
|
for a in body.find_all('a', href=True):
|
|
if YOUTUBE_PAT.search(a['href']):
|
|
has_video = True; break
|
|
if not has_video and VIDEO_EXT.search(str(body)):
|
|
has_video = True
|
|
has_audio = bool(body.find_all('audio')) or bool(AUDIO_EXT.search(str(body)))
|
|
return has_image, has_video, has_audio, has_text
|
|
|
|
|
|
def n_string(has_text, has_image, has_video, has_audio):
|
|
parts = []
|
|
if has_text:
|
|
parts.append('어문')
|
|
if has_image:
|
|
parts.append('이미지')
|
|
if has_video:
|
|
parts.append('영상')
|
|
if has_audio:
|
|
parts.append('오디오')
|
|
return ','.join(parts) if parts else '없음'
|
|
|
|
|
|
def img_has_valid_anchor(img):
|
|
p = img.parent
|
|
while p is not None:
|
|
if p.name == 'a':
|
|
href = p.get('href', '')
|
|
return bool(href and not href.startswith('#') and not href.lower().startswith('javascript:'))
|
|
p = p.parent
|
|
return False
|
|
|
|
|
|
def detect_kogl(body):
|
|
types, q_y = set(), False
|
|
for a in body.find_all('a', href=True):
|
|
m = KOGL_LINK_PAT.search(a['href'])
|
|
if m:
|
|
types.add(int(m.group(1))); q_y = True
|
|
for img in body.find_all('img'):
|
|
m = KOGL_IMG_PAT.search(img.get('src', ''))
|
|
if m and 1 <= int(m.group(1)) <= 4:
|
|
types.add(int(m.group(1)))
|
|
if img_has_valid_anchor(img):
|
|
q_y = True
|
|
for el in body.find_all(style=True):
|
|
m = KOGL_IMG_PAT.search(el.get('style', ''))
|
|
if m and 1 <= int(m.group(1)) <= 4:
|
|
types.add(int(m.group(1)))
|
|
# 1·2·3·4 전부 = 범례페이지 = 미부착
|
|
if types == {1, 2, 3, 4}:
|
|
return set(), None
|
|
if not types:
|
|
return set(), None
|
|
return types, ('Y' if q_y else 'N')
|
|
|
|
|
|
def process_row(session, url, body_selectors):
|
|
out = {'L': '', 'M': '', 'N': '', 'O': '', 'P': '', 'Q': '', 'note': ''}
|
|
soup, final = fetch(session, url)
|
|
if soup is None:
|
|
out['note'] = '접근 실패'
|
|
return out
|
|
body = get_body(soup, body_selectors)
|
|
form, count = detect_form(body)
|
|
out['L'] = form
|
|
out['M'] = count if form == '게시판' else 1
|
|
has_img, has_vid, has_aud, has_txt = detect_media(body)
|
|
types_main, q_main = detect_kogl(body)
|
|
P = '게시판' if types_main else ''
|
|
types_all = set(types_main)
|
|
q_flags = [q_main] if q_main else []
|
|
if form == '게시판':
|
|
for du in extract_detail_urls(body, final or url, limit=6):
|
|
d_soup, _ = fetch(session, du, timeout=10)
|
|
if not d_soup:
|
|
continue
|
|
d_body = get_body(d_soup, body_selectors)
|
|
di, dv, da, dt = detect_media(d_body)
|
|
has_img |= di; has_vid |= dv; has_aud |= da; has_txt |= dt
|
|
dt_types, dt_q = detect_kogl(d_body)
|
|
if dt_types and not types_main and not P:
|
|
P = '게시물'
|
|
types_all |= dt_types
|
|
if dt_q:
|
|
q_flags.append(dt_q)
|
|
out['N'] = n_string(has_txt, has_img, has_vid, has_aud)
|
|
if not types_all:
|
|
out['O'] = '미부착'
|
|
else:
|
|
out['O'] = ','.join(f'{n}유형' for n in sorted(types_all))
|
|
out['P'] = P if P else '게시판'
|
|
out['Q'] = 'Y' if 'Y' in q_flags else 'N'
|
|
return out
|
|
|
|
|
|
def run_site(name, num, body_selectors, workers=8):
|
|
xlsx = os.path.join(OUTDIR, f'{name}.xlsx')
|
|
if not os.path.exists(xlsx):
|
|
print(f'[{name}] 파일 없음 — 스킵'); return None
|
|
wb = openpyxl.load_workbook(xlsx)
|
|
ws = wb.active
|
|
START = 3
|
|
END = START - 1
|
|
for r in range(START, ws.max_row + 1):
|
|
if ws.cell(r, 2).value is None:
|
|
break
|
|
END = r
|
|
if END < START:
|
|
print(f'[{name}] 데이터행 없음'); return None
|
|
# 이미 처리됨(이어하기): L열 채워진 비율 ≥90%면 스킵
|
|
filled = sum(1 for r in range(START, END + 1) if ws.cell(r, 12).value)
|
|
if '--force' not in sys.argv and filled >= (END - START + 1) * 0.9:
|
|
print(f'[{num}.{name}] 이미 처리됨({filled}/{END-START+1}) — 스킵')
|
|
return {'name': name, 'forms': {'(skip)': filled}, 'attach': 0, 'rows': END - START + 1}
|
|
tasks = []
|
|
for r in range(START, END + 1):
|
|
url = ws.cell(r, 11).value
|
|
is_ext = (ws.cell(r, 19).value == '외부링크')
|
|
tasks.append((r, url, is_ext))
|
|
n_ext = sum(1 for t in tasks if t[2])
|
|
print(f'\n[{num}.{name}] {len(tasks)}행 (외부 {n_ext}) 처리…')
|
|
t0 = time.time()
|
|
session = make_session()
|
|
results = {}
|
|
|
|
def worker(task):
|
|
row, url, is_ext = task
|
|
if is_ext:
|
|
return row, {'L': '사이트', 'M': '', 'N': '', 'O': '', 'P': '', 'Q': '', 'note': ''}
|
|
if not url or not isinstance(url, str) or not url.startswith('http'):
|
|
return row, {'L': '', 'M': '', 'N': '', 'O': '', 'P': '', 'Q': '', 'note': 'URL 없음'}
|
|
return row, process_row(session, url, body_selectors)
|
|
|
|
with ThreadPoolExecutor(max_workers=workers) as ex:
|
|
futs = [ex.submit(worker, t) for t in tasks]
|
|
done = 0
|
|
for fut in as_completed(futs):
|
|
row, res = fut.result()
|
|
results[row] = res
|
|
done += 1
|
|
if done % 50 == 0 or done == len(tasks):
|
|
print(f' {done}/{len(tasks)} ({time.time()-t0:.0f}s)')
|
|
|
|
for r in range(START, END + 1):
|
|
res = results.get(r)
|
|
if not res:
|
|
continue
|
|
if res.get('L'):
|
|
ws.cell(r, 12).value = res['L']
|
|
if res.get('M') != '':
|
|
ws.cell(r, 13).value = res['M']
|
|
if res.get('N'):
|
|
ws.cell(r, 14).value = res['N']
|
|
if res.get('O'):
|
|
ws.cell(r, 15).value = res['O']
|
|
if res.get('P'):
|
|
ws.cell(r, 16).value = res['P']
|
|
if res.get('Q'):
|
|
ws.cell(r, 17).value = res['Q']
|
|
if res.get('note') and not ws.cell(r, 19).value:
|
|
ws.cell(r, 19).value = res['note']
|
|
wb.save(xlsx)
|
|
forms, attach = {}, 0
|
|
for res in results.values():
|
|
forms[res.get('L', '')] = forms.get(res.get('L', ''), 0) + 1
|
|
if res.get('O') and res.get('O') != '미부착':
|
|
attach += 1
|
|
fs = ' '.join(f'{k}{v}' for k, v in forms.items() if k)
|
|
print(f' ✓ [{name}] {fs} | 공공누리부착 {attach} ({time.time()-t0:.0f}s)')
|
|
return {'name': name, 'forms': forms, 'attach': attach, 'rows': len(tasks)}
|
|
|
|
|
|
def main():
|
|
probe = {r['name']: r for r in json.load(open(PROBE, encoding='utf-8'))}
|
|
order = sorted(probe.values(), key=lambda x: -int(x['num'])) # 번호 역순
|
|
only = sys.argv[1:]
|
|
if only:
|
|
order = [p for p in order if p['name'] in only or str(p['num']) in only]
|
|
summ = []
|
|
for p in order:
|
|
try:
|
|
r = run_site(p['name'], int(p['num']), BODY_SEL)
|
|
if r:
|
|
summ.append(r)
|
|
except Exception as e:
|
|
import traceback
|
|
print(f"[{p['name']}] 실패: {e}")
|
|
traceback.print_exc()
|
|
print('\n=== 요약 ===')
|
|
for s in summ:
|
|
fs = ' '.join(f'{k}{v}' for k, v in s['forms'].items() if k)
|
|
print(f" {s['name']}: {fs} | 부착{s['attach']} /{s['rows']}행")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|