DB_JOB/_스크립트/_buyeo_rejudge_NO.py

# -*- coding: utf-8 -*-
"""부여군 N(저작물 유형) + O/P/Q(공공누리) 전수 재판정.
사용자 검수(3~37행)를 정답으로 검증 후 38행~ 적용.

핵심(부여 buyeo 템플릿 특성):
- KOGL 마크는 게시판 '상세글'의 `div.gnuri_layer > .codeView01 > img[src=/_module/gnuri/images/img_opentypeNN.png]`
  + kogl.or.kr/info/licenseTypeN.do 링크. **본문(#txt) 바깥**이라 기존 phase234가 놓침.
- 상세 URL = `?mode=V&no=...`.
- 게시판은 글마다 유형 다를 수 있음 → O = **첫 마크 글의 유형**(사용자 컨벤션: 01030502 글1=1유형 → O=1유형).
- N 이미지: 장식/공통UI(file_icon·see_btn·btn_page·/skin/·/common/·ico/btn/bg) + KOGL 마크 제외.
  지도/PDF iframe = 이미지. 게시판 M=0 = 없음. 실제 업로드 사진만 이미지.

사용: python -X utf8 _buyeo_rejudge_NO.py --validate   # 3~37 사용자값과 비교
      python -X utf8 _buyeo_rejudge_NO.py --write       # 38행~ 기입(백업)
"""
import sys, warnings, re, time
from urllib.parse import urljoin
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests, openpyxl
from bs4 import BeautifulSoup

warnings.filterwarnings('ignore')
S = requests.Session(); S.headers.update({'User-Agent': 'Mozilla/5.0'})
XLSX = r'D:\01.프로젝트\DB수집\작업파일\광역_사이트맵\충청남도\7.부여군\충청남도_부여군.xlsx'
DOMAIN = 'buyeo.go.kr'
REF_MAX = 37   # 3~37 = 사용자 검수 정답

KOGL_IMG = re.compile(r'(?:new_)?img_open(?:type|code)(\d{1,2})\.(?:png|jpe?g|gif)', re.I)
KOGL_LINK = re.compile(r'licenseType(\d)', re.I)
DECO = re.compile(
    r'/site/common/img/|move\.png|no[-_]?img|blank\.|spacer\.|'
    r'/img/(?:icon|ico|bul|bullet|arrow|btn|bg|tit|h\d)|'
    r'ico_|btn_|bul_|bg_|_bg\.|icon_|'
    r'file_icon|_icon\.|icon\.gif|_btn\.|see_btn|flag\.|webaccess|'
    r'/skin/|/images/(?:kr/)?common(?:_new)?/|/common/img/|/_module/gnuri/|'
    r'btn_page|page_(?:next|prev|first|last)', re.I)
MAPPDF = re.compile(r'pdf|viewer\.html|/map|kakao|daum|=map|naver.*map|google.*map', re.I)
VIDEO = re.compile(r'youtube\.com|youtu\.be|/embed/|\.mp4|\.webm|vimeo', re.I)
DETAIL = re.compile(r'mode=V&no=', re.I)


def get(url):
    last = None
    for _ in range(3):
        try:
            r = S.get(url, verify=False, timeout=20)
            return BeautifulSoup(r.content, 'html.parser')
        except Exception as e:
            last = e
            time.sleep(1.2)
    raise last


def body(soup):
    return soup.select_one('#txt') or soup.select_one('#contents') or soup.select_one('main') or soup


def real_img(node):
    for im in node.find_all('img'):
        src = im.get('src') or ''
        if not src or KOGL_IMG.search(src) or DECO.search(src):
            continue
        return True
    return False


def media_flags(node):
    img = real_img(node)
    vid = bool(node.find('video'))
    for ifr in node.find_all('iframe'):
        s = ifr.get('src') or ''
        if VIDEO.search(s):
            vid = True
        elif MAPPDF.search(s):
            img = True
    for a in node.find_all('a'):
        if VIDEO.search(a.get('href') or ''):
            vid = True
    return img, vid


def kogl_from(html):
    """전체 HTML에서 KOGL 이미지유형·링크유형 추출."""
    imgs = [int(x) for x in KOGL_IMG.findall(html) if 1 <= int(x) <= 4]
    links = [int(x) for x in KOGL_LINK.findall(html) if 1 <= int(x) <= 4]
    return imgs, links


def _menu_cd(u):
    m = re.search(r'menu_dvs_cd=([^&]+)', u or '')
    return m.group(1) if m else None


def detail_links(soup, base):
    """같은 게시판(_prog/_board + 동일 menu_dvs_cd) 상세글만. 타 게시판 stray 링크 배제."""
    if '/_prog/_board/' not in base:
        return []                         # 게시판 템플릿이 아니면 상세 없음(scate 등)
    base_menu = _menu_cd(base)
    out = []
    for a in body(soup).find_all('a', href=True):
        h = a['href']
        if not DETAIL.search(h):
            continue
        full = urljoin(base, h)
        if base_menu and _menu_cd(full) and _menu_cd(full) != base_menu:
            continue                      # 다른 게시판 글 → 제외
        out.append(full)
    seen = set(); res = []
    for u in out:
        if u not in seen:
            seen.add(u); res.append(u)
    return res[:5]


def decide_O(img_types, link_types):
    """KOGL 규칙: 이미지유형 우선. 링크만이면 licenseType, 1·2·3·4 전부=범례→미부착."""
    if img_types:
        t = img_types[0]
        mism = bool(link_types) and (link_types[0] != t)
        return f'{t}유형', mism
    if link_types:
        if {1, 2, 3, 4}.issubset(set(link_types)):
            return '미부착', False
        return f'{link_types[0]}유형', False
    return '미부착', False


def judge(k, L, M):
    """반환 dict: N, O, P, Q, S(mismatch)."""
    try:
        soup = get(k)
    except Exception as e:
        return {'err': f'fetch:{str(e)[:25]}'}
    bd = body(soup); full = str(soup)
    txt = len(bd.get_text(strip=True)) >= 1
    is_board_url = '/_prog/_board/' in k        # 실제 게시판 URL만(.html·서브사이트는 게시판 아님)

    has_img, has_vid = media_flags(bd)
    # O: 페이지/리스트 자체 마크 우선
    img_t, link_t = kogl_from(full)
    o_loc = '게시판' if L == '게시판' else '페이지'
    o_imgs, o_links = list(img_t), list(link_t)

    # '빈 게시판→없음'은 진짜 게시판(_prog/_board)에 글이 0개일 때만.
    # L=게시판 M=0 이라도 .html 페이지·서브사이트는 본문내용으로 판정(원본 L/M 오기 보정).
    dls = detail_links(soup, k) if (L == '게시판') else []
    board0 = (L == '게시판' and is_board_url and (M in (0, '0', None)) and not dls)

    if L == '게시판' and not board0:
        # N: 상위 5글 본문에서 실제 이미지/영상 누적(이미지 콘텐츠 회수율↑)
        for idx, du in enumerate(dls):
            try:
                ds = get(du); db = body(ds); df = str(ds)
            except Exception:
                continue
            di, dv = media_flags(db)
            if di: has_img = True
            if dv: has_vid = True
            # O: 리스트 마크 없으면 '첫(대표) 글'만으로 판정
            #    (사용자 컨벤션: 01030502 글1=1유형→1유형 / 업무추진비 글1무·글5만3유형→미부착)
            if idx == 0 and not o_imgs and not o_links:
                dimg, dlink = kogl_from(df)
                if dimg or dlink:
                    o_imgs, o_links, o_loc = dimg, dlink, '게시물'
            if has_img and has_vid:
                break

    # N
    if board0:
        N = '없음'
    else:
        parts = []
        if txt: parts.append('어문')
        if has_img: parts.append('이미지')
        if has_vid: parts.append('영상')
        N = ','.join(parts) if parts else '없음'

    O, mism = decide_O(o_imgs, o_links)
    P = (o_loc if O != '미부착' else None)
    Q = ('Y' if (O != '미부착' and o_links) else None)
    return {'N': N, 'O': O, 'P': P, 'Q': Q, 'mismatch': mism}


def main():
    mode = '--validate' if '--validate' in sys.argv else ('--write' if '--write' in sys.argv else '--dry')
    wb = openpyxl.load_workbook(XLSX); ws = wb.active

    rows = []
    for r in range(3, ws.max_row + 1):
        k = ws.cell(r, 11).value; L = ws.cell(r, 12).value; M = ws.cell(r, 13).value
        if not (isinstance(k, str) and DOMAIN in k):
            continue
        if L == '사이트':
            continue
        if mode == '--validate' and r > REF_MAX:
            continue
        if mode == '--write' and r <= REF_MAX:
            continue
        rows.append((r, k, L, M))
    print(f'mode={mode} | 대상 {len(rows)}행')

    res = {}
    t0 = time.time()
    with ThreadPoolExecutor(max_workers=3) as ex:
        futs = {ex.submit(judge, k, L, M): (r, k, L, M) for r, k, L, M in rows}
        done = 0
        for fut in as_completed(futs):
            r, k, L, M = futs[fut]
            res[r] = fut.result()
            done += 1
            if done % 40 == 0:
                print(f'  {done}/{len(rows)} ({time.time()-t0:.0f}s)')
    print(f'크롤링 완료 ({time.time()-t0:.0f}s)')

    if mode == '--validate':
        miss = 0
        for r, k, L, M in rows:
            d = res[r]
            if d.get('err'):
                print(f'  행{r} ERR {d["err"]}'); continue
            curN = ws.cell(r, 14).value; curO = ws.cell(r, 15).value
            curP = ws.cell(r, 16).value; curQ = ws.cell(r, 17).value
            dn = (d['N'] or '') != (curN or '')
            do = (d['O'] or '') != (curO or '')
            dp = (d['P'] or '') != (curP or '')
            dq = (d['Q'] or '') != (curQ or '')
            if dn or do or dp or dq:
                miss += 1
                g = ws.cell(r, 7).value or ws.cell(r, 6).value
                print(f'  ✗행{r} {str(g)[:16]:16}| N:{curN}→{d["N"]} {"≠" if dn else "="} | '
                      f'O:{curO}→{d["O"]}{"≠" if do else "="} P:{curP}→{d["P"]}{"≠" if dp else ""} Q:{curQ}→{d["Q"]}{"≠" if dq else ""}')
        print(f'\n불일치 {miss}/{len(rows)}행 (0이면 로직이 사용자 검수와 100% 일치)')
        return

    # write (38행~)
    if mode == '--write':
        import shutil
        bk = XLSX.replace('.xlsx', '_backup_NO재판정전.xlsx'); shutil.copy(XLSX, bk)
        print('백업:', bk)
        chg = 0
        for r, k, L, M in rows:
            d = res[r]
            if d.get('err'):
                print(f'  행{r} ERR {d["err"]}'); continue
            old = (ws.cell(r,14).value, ws.cell(r,15).value, ws.cell(r,16).value, ws.cell(r,17).value)
            ws.cell(r, 14).value = d['N']
            ws.cell(r, 15).value = d['O']
            ws.cell(r, 16).value = d['P']
            ws.cell(r, 17).value = d['Q']
            if d['mismatch']:
                cur = (ws.cell(r,19).value or '').strip()
                ws.cell(r,19).value = '링크주소 오기' if not cur else cur+' / 링크주소 오기'
            new = (d['N'], d['O'], d['P'], d['Q'])
            if old != new:
                chg += 1
                g = ws.cell(r,7).value or ws.cell(r,6).value
                print(f'  행{r} {str(g)[:16]:16}| N:{old[0]}→{new[0]} O:{old[1]}→{new[1]} P:{old[2]}→{new[2]} Q:{old[3]}→{new[3]}')
        wb.save(XLSX)
        print(f'\n저장: {XLSX} ({chg}행 변경)')
        return

    print('(--validate 또는 --write 지정)')


if __name__ == '__main__':
    main()