"""모든 사이트 URL 실접속 검증 — breadcrumb/title로 D~G 매칭 확인. 사이트별 path 추출 로직: - 인천광역시: title에서 '>' split, '|' 앞까지 — "홈>인천소식>새소식>..." - 전라남도: title을 '|'로 split, 첫 토큰(사이트명) 제거 - 고용노동부: .location 또는 .path 텍스트 (홈 으로 이동 X Y Z) - 과학기술정보통신부: .location 등 + 또는 h1/h2 fallback (봇 차단 가능) - 교육부: 동일 - 보건복지부: title을 '<'로 split + 역순 ("X < Y < Z" → [Z, Y, X]) - 성평등가족부: h2 + breadcrumb - 외교부: title의 '|' 앞 부분 + h2 병렬 처리(ThreadPoolExecutor)로 속도 확보. 결과는 row 번호별로 저장. """ import sys, io, json, re, time from concurrent.futures import ThreadPoolExecutor, as_completed sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') import requests, urllib3 urllib3.disable_warnings() HEADERS = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36', 'Accept': 'text/html,application/xhtml+xml,*/*;q=0.9', 'Accept-Language': 'ko-KR,ko;q=0.9', 'Accept-Encoding': 'gzip, deflate', } # 세션은 thread별로 def make_session(): s = requests.Session() s.headers.update(HEADERS) return s from bs4 import BeautifulSoup def extract_path(site, html, url): """사이트별 path 추출. (path_tokens, title, h2) 반환.""" soup = BeautifulSoup(html, 'html.parser') title = (soup.title.get_text(' ', strip=True) if soup.title else '') h1 = (soup.select_one('h1').get_text(' ', strip=True) if soup.select_one('h1') else '') h2 = (soup.select_one('h2').get_text(' ', strip=True) if soup.select_one('h2') else '') path = [] if site == '인천광역시': # 1순위: .content-location-inner — "Home 인천소식 새소식" el = soup.select_one('.content-location-inner, .content-location, .location-inner') if el: # 자식 a/span/li 우선 items = [] for inner in ['a', 'li', 'span']: items = [e.get_text(' ', strip=True) for e in el.select(inner) if e.get_text(strip=True)] if items: break if not items: items = el.get_text(' ', strip=True).split() path = [x for x in items if x.lower() not in ['home', '홈', '으로 이동']] if not path: # 폴백: title의 ">" 구분 t = title if '|' in t: t = t.split('|')[0] toks = [x.strip() for x in t.split('>') if x.strip()] path = toks elif site == '전라남도': # title: "전라남도청 | 참여와 소통 | 도민의 소리 | 우리동네 숨은 선행방" toks = [x.strip() for x in title.split('|') if x.strip()] path = toks[1:] if toks else [] # 첫 토큰(사이트명) 제거 elif site == '고용노동부': for sel in ['.location', '.path', '#location']: el = soup.select_one(sel) if el: # li 단위로 추출 시도 for inner in ['li', 'a', 'span']: items = [e.get_text(' ', strip=True) for e in el.select(inner) if e.get_text(strip=True)] if items: items = [re.sub(r'\s+', ' ', x).strip() for x in items] items = [x for x in items if x and not re.search(r'홈|Home|으로 이동|인쇄|새창|new window|즐겨찾기', x, re.IGNORECASE)] if items: path = items break break elif site == '보건복지부': # title: "정보목록 검색(14.1월~현재) < 정보목록 < 정보공개 < 정보공개 : 힘이 되는 평생 친구, 보건복지부" # ':' 앞만 사용, 역순 t = title.split(':')[0] if ':' in title else title toks = [x.strip() for x in t.split('<') if x.strip()] path = list(reversed(toks)) # ["정보공개", "정보공개", "정보목록", "정보목록 검색..."] # .location 셀렉터도 시도 el = soup.select_one('.location') if el: txt = el.get_text(' ', strip=True) # "홈 정보공개 정보공개 정보목록 정보목록 검색..." for inner in ['li', 'a', 'span']: items = [e.get_text(' ', strip=True) for e in el.select(inner) if e.get_text(strip=True)] if items: items = [x for x in items if x not in ['홈', 'Home']] if items: path = items break elif site == '성평등가족부': # title="성평등가족부" 뿐 / breadcrumb 없음 # h2 첫 = 대분류, h2 두번째 = 페이지명 h2s = [e.get_text(' ', strip=True) for e in soup.select('h2') if e.get_text(strip=True)] h2s = [x for x in h2s if x not in ['통합검색', '검색']] if h2s: path = h2s[:2] # 첫번째=대분류, 두번째=페이지명만 추출 가능 (제한적) elif site == '외교부': # breadcrumb 없음. title="X | 외교부" → 페이지명만 toks = [x.strip() for x in title.split('|') if x.strip()] toks = [x for x in toks if '외교부' not in x and x not in ['홈', 'Home']] # h2도 같이 if h2 and h2 not in toks and '외교부' not in h2: toks.append(h2) path = list(dict.fromkeys(toks)) # 중복 제거 (순서 유지) elif site == '교육부': # title: "교육부 > 국민참여·민원 > 참여·소통 > 전자공청회" toks = [x.strip() for x in title.split('>') if x.strip()] path = [x for x in toks if x not in ['교육부', '대한민국 교육부', 'MOE']] # .location 셀렉터 시도 (백업) if not path: el = soup.select_one('.location, .breadcrumb, .nav_path, .path') if el: items = [e.get_text(' ', strip=True) for e in el.select('a, li, span') if e.get_text(strip=True)] items = [x for x in items if x not in ['홈', 'Home']] if items: path = items elif site == '과학기술정보통신부': # 1순위: .sub_location — "홈 국민참여 적극행정 제도소개 적극행정·소극행정 정의" el = soup.select_one('.sub_location, .sub-location, .location, .breadcrumb, .path, .now_loc') if el: items = [] for inner in ['a', 'li', 'span']: items = [e.get_text(' ', strip=True) for e in el.select(inner) if e.get_text(strip=True)] if items: break items = [x for x in items if x not in ['홈', 'Home']] if items: path = items if not path: toks = [x.strip() for x in re.split(r'[>|]', title) if x.strip()] path = [x for x in toks if '과학기술' not in x and 'MSIT' not in x] # 공통 후처리 — '홈' 제거 path = [p for p in path if p.strip() not in ['홈', 'Home', 'HOME', '']] return path, title, h2 def fetch_one(site, row, url): sess = make_session() try: r = sess.get(url, verify=False, timeout=15, allow_redirects=True) if r.status_code == 200 and len(r.text) > 500: path, title, h2 = extract_path(site, r.text, url) return {'row': row, 'site': site, 'url': url, 'http': r.status_code, 'path': path, 'title': title[:200], 'h2': h2[:100], 'final_url': r.url} else: return {'row': row, 'site': site, 'url': url, 'http': r.status_code, 'path': [], 'title': '', 'h2': '', 'final_url': r.url} except Exception as e: return {'row': row, 'site': site, 'url': url, 'http': 0, 'path': [], 'title': '', 'h2': '', 'error': str(e)[:100]} def main(sites_to_run=None, limit=None): with open(r'D:\01.프로젝트\DB수집\2주차\검토_리포트\excel_data.json', encoding='utf-8') as f: excel = json.load(f) # 처리할 작업 목록 jobs = [] for site, rows in excel.items(): if sites_to_run and site not in sites_to_run: continue for r in rows: if not r['K'] or not isinstance(r['K'], str) or not r['K'].startswith('http'): continue if r['S'] == '외부링크': continue jobs.append((site, r['row'], r['K'])) if limit: jobs = jobs[:limit] print(f'총 작업: {len(jobs)}건') results = {} t0 = time.time() with ThreadPoolExecutor(max_workers=10) as ex: futs = {ex.submit(fetch_one, s, r, u): (s, r, u) for s, r, u in jobs} done = 0 for fut in as_completed(futs): res = fut.result() results[res['row']] = res done += 1 if done % 100 == 0: elapsed = time.time() - t0 print(f' {done}/{len(jobs)} ({elapsed:.0f}s)') out = r'D:\01.프로젝트\DB수집\2주차\검토_리포트\verify_results.json' with open(out, 'w', encoding='utf-8') as f: json.dump(results, f, ensure_ascii=False, indent=2) print(f'\n저장: {out}') print(f'총 소요: {time.time()-t0:.0f}s') if __name__ == '__main__': sites = sys.argv[1].split(',') if len(sys.argv) > 1 and sys.argv[1] != 'all' else None limit = int(sys.argv[2]) if len(sys.argv) > 2 else None main(sites_to_run=sites, limit=limit)