DB_JOB/_스크립트/_공공기관_navfix.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

113 lines
4.3 KiB
Python

# -*- coding: utf-8 -*-
"""공공기관 N 영상/오디오 거짓양성 교정.
원인: phase234 detect_media가 str(body)에 '.mp4'/'.mp3'만 있어도 영상/오디오로 오판(템플릿 배경영상/플레이어 스크립트).
교정: 영상=youtube/vimeo iframe·링크 또는 <video>실소스 / 오디오=<audio>실소스·미디어링크 만 인정. 게시판은 상세 표본도 확인.
대상: N에 '영상' 또는 '오디오' 포함 & L!=사이트 행. 거짓이면 N에서 제거.
사용: python _공공기관_navfix.py [기관명 ...]
"""
import sys, os, re, json, importlib.util, warnings
import openpyxl
warnings.filterwarnings('ignore')
spec = importlib.util.spec_from_file_location('p234', r'D:\01.프로젝트\DB수집\_스크립트\_공공기관_phase234.py')
P = importlib.util.module_from_spec(spec); spec.loader.exec_module(P)
OUTDIR = r'D:\01.프로젝트\DB수집\공공기관'
# 실제 콘텐츠 영상만: watch/embed/youtu.be ID/vimeo 숫자. ⛔채널·계정·SNS링크(/channel//user//@//c/) 제외
VID = re.compile(r'(youtube\.com/watch\?|youtube\.com/embed/|youtu\.be/[\w-]{6,}|player\.vimeo\.com/video|vimeo\.com/\d{5,})', re.I)
AUD_EXT = re.compile(r'\.(mp3|wav|m4a|ogg|flac)(\?|$|["\'&])', re.I)
def real_video(body):
for ifr in body.find_all('iframe'):
if VID.search(ifr.get('src', '') or ''):
return True
for a in body.find_all('a', href=True):
if VID.search(a['href']):
return True
for v in body.find_all('video'):
src = v.get('src') or (v.find('source').get('src') if v.find('source') else '')
if src and not src.lower().endswith(('.gif', '.png', '.jpg')):
return True
return False
def real_audio(body):
for au in body.find_all('audio'):
if au.get('src') or au.find('source'):
return True
for a in body.find_all('a', href=True):
if AUD_EXT.search(a['href']):
return True
for s in body.find_all('source'):
if AUD_EXT.search(s.get('src', '') or ''):
return True
return False
def rebuild_N(N, vid, aud):
parts = []
for p in (N or '').split(','):
if p == '영상' and not vid:
continue
if p == '오디오' and not aud:
continue
if p:
parts.append(p)
return ','.join(parts) if parts else '없음'
def run(name, sess):
xlsx = os.path.join(OUTDIR, f'{name}.xlsx')
wb = openpyxl.load_workbook(xlsx)
ws = wb.active
chg = 0
for r in range(3, ws.max_row + 1):
if ws.cell(r, 2).value is None:
break
L = ws.cell(r, 12).value
N = ws.cell(r, 14).value or ''
if L == '사이트' or ('영상' not in N and '오디오' not in N):
continue
url = ws.cell(r, 11).value
if not url or not isinstance(url, str) or not url.startswith('http'):
continue
soup, _ = P.fetch(sess, url)
if not soup:
continue
body = P.get_body(soup, P.BODY_SEL)
vid = real_video(body); aud = real_audio(body)
if L == '게시판' and (('영상' in N and not vid) or ('오디오' in N and not aud)):
for du in P.extract_detail_urls(body, url, limit=4):
ds, _ = P.fetch(sess, du)
if not ds:
continue
db = P.get_body(ds, P.BODY_SEL)
vid = vid or real_video(db); aud = aud or real_audio(db)
if vid and aud:
break
newN = rebuild_N(N, vid, aud)
if newN != N:
ws.cell(r, 14).value = newN; chg += 1
wb.save(xlsx)
print(f'[{name}] 영상/오디오 거짓제거 {chg}')
return chg
def main():
probe = {r['name']: r for r in json.load(open(r'D:\01.프로젝트\DB수집\_스크립트\_공공기관_probe.json', 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]
sess = P.make_session()
tot = 0
for p in order:
try:
tot += run(p['name'], sess)
except Exception as e:
print(f"[{p['name']}] 실패: {e}")
print(f'=== 영상/오디오 교정 합계 {tot}행 ===')
if __name__ == '__main__':
main()