import os import re import asyncio import json import traceback import sys import threading import queue import pymysql from datetime import datetime, timedelta from playwright.async_api import async_playwright from dotenv import load_dotenv from sql_queries import CrawlerQueries load_dotenv(override=True) # 글로벌 중단 제어용 이벤트 crawl_stop_event = threading.Event() def get_db_connection(): """MySQL 데이터베이스(TEST) 연결을 반환""" return pymysql.connect( host=os.getenv('DB_HOST', 'localhost'), port=int(os.getenv('DB_PORT', 3306)), user=os.getenv('DB_USER', 'root'), password=os.getenv('DB_PASSWORD', '45278434'), database='PM_proto_test', # 테스트용 DB 고정 charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor ) def clean_date_string(date_str): """원본 crawler_service.py와 동일한 날짜 정리 로직""" if not date_str: return "" match = re.search(r'(\d{2})[./-](\d{2})[./-](\d{2})', date_str) if match: return f"20{match.group(1)}.{match.group(2)}.{match.group(3)}" return date_str[:10].replace("-", ".") def parse_log_id(log_id): """원본 crawler_service.py와 동일한 로그 ID 파싱 로직""" if not log_id or "_" not in log_id: return log_id try: parts = log_id.split('_') if len(parts) >= 4: date_part = clean_date_string(parts[1]) activity = parts[3].strip() activity = re.sub(r'\(.*?\)', '', activity).strip() return f"{date_part}, {activity}" except: pass return log_id def crawler_thread_worker(msg_queue, user_id, password): crawl_stop_event.clear() if sys.platform == 'win32': asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy()) loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) async def run(): async with async_playwright() as p: browser = None try: msg_queue.put(json.dumps({'type': 'log', 'message': '[TEST] 원본 수집 방식 복구 및 추론 엔진 가동...'})) browser = await p.chromium.launch(headless=True, args=[ "--no-sandbox", "--disable-dev-shm-usage", "--disable-blink-features=AutomationControlled" ]) context = await browser.new_context( viewport={'width': 1600, 'height': 900}, user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36" ) captured_data = {"tree": None, "_is_root_archive": False, "project_list": [], "last_project_data": None} async def global_interceptor(response): url = response.url try: if "getAllList" in url: data = await response.json() captured_data["project_list"] = data.get("data", []) elif "getTreeObject" in url: # [복구] 원본과 100% 동일한 루트 판정 로직 is_root = False if "params[resourcePath]=" in url: path_val = url.split("params[resourcePath]=")[1].split("&")[0] if path_val in ["%2F", "/"]: is_root = True if is_root: captured_data["tree"] = await response.json() captured_data["_is_root_archive"] = True elif "getData" in url and "overview" in url: captured_data["last_project_data"] = await response.json() except: pass context.on("response", global_interceptor) page = await context.new_page() await page.goto("https://overseas.projectmastercloud.com/dashboard", wait_until="domcontentloaded") if await page.locator("#login-by-id").is_visible(timeout=10000): await page.click("#login-by-id"); await page.fill("#user_id", user_id); await page.fill("#user_pw", password); await page.click("#login-btn") await page.wait_for_selector("h4.list__contents_aria_group_body_list_item_label", timeout=60000) await asyncio.sleep(2) project_names = list(dict.fromkeys([n.strip() for n in await page.locator("h4.list__contents_aria_group_body_list_item_label").all_inner_texts() if n.strip()])) count = len(project_names) for i, project_name in enumerate(project_names): if crawl_stop_event.is_set(): break msg_queue.put(json.dumps({'type': 'log', 'message': f'[TEST] [{i+1}/{count}] {project_name} 수집'})) p_match = next((p for p in captured_data["project_list"] if p.get('project_nm') == project_name or p.get('short_nm', '').strip() == project_name), None) current_p_id = p_match.get('project_id') if p_match else None try: # 1. 프로젝트 진입 target_el = page.locator(f"h4.list__contents_aria_group_body_list_item_label:has-text('{project_name}')").first await target_el.scroll_into_view_if_needed() box = await target_el.bounding_box() if box: await page.mouse.click(box['x'] + 5, box['y'] + 5) else: await target_el.click(force=True) await page.wait_for_selector("text=활동로그", timeout=30000) # 2. [복구] 최신 파일 수 실측 (원본의 수동 Fetch 방식 그대로) captured_data["tree"] = None; captured_data["_is_root_archive"] = False await page.evaluate("""() => { const baseUrl = window.location.origin + window.location.pathname.split('/').slice(0, 2).join('/'); fetch(`${baseUrl}/archive/getTreeObject?params[storageType]=CLOUD¶ms[resourcePath]=/`); }""") for _ in range(30): if captured_data["_is_root_archive"]: break await asyncio.sleep(0.5) actual_count = 0 if captured_data["tree"]: tree_data = captured_data["tree"] if isinstance(tree_data, list) and len(tree_data) > 0: tree_data = tree_data[0] if isinstance(tree_data, dict): tree = tree_data.get('currentTreeObject', tree_data) if isinstance(tree, dict): # 원본 파일 수 합산 로직 total = len(tree.get("file", {})) folders = tree.get("folder", {}) if isinstance(folders, dict): for f in folders.values(): total += int(f.get("filesCount", 0)) actual_count = total # 3. 활동로그 전수 수집 (하이브리드 방식: 최상단 우선 확보 + 전수 스크롤) all_logs = [] await page.get_by_text("활동로그").first.click() if await page.wait_for_selector("article.archive-modal", timeout=10000): # 날짜 필터 적용 (2020-01-01) inputs = await page.locator("article.archive-modal input").all() for inp in inputs: if (await inp.get_attribute("type")) == "date": await inp.fill("2020-01-01"); break apply_btn = page.locator("article.archive-modal").get_by_text("적용").first if await apply_btn.is_visible(): await apply_btn.click() # [핵심] 첫 번째 로그가 나타날 때까지 명시적 대기 (최대 10초) try: await page.wait_for_selector("article.archive-modal div[id*='_']", timeout=10000) except: pass await asyncio.sleep(2) # (1) 최상단 로그 즉시 확보 (안전장치) first_log_el = await page.locator("article.archive-modal div[id*='_']").first.get_attribute("id") if first_log_el: first_log_text = parse_log_id(first_log_el) if ", " in first_log_text: d, a = first_log_text.split(", ", 1) all_logs.append({'date': d, 'activity': a}) # (2) 전수 수집을 위한 무한 스크롤 및 지정된 클래스 내 ID 수집 last_count = len(all_logs) for _ in range(20): # 스크롤 수행 (사용자가 지정한 log-body 클래스 기준) await page.evaluate("""() => { const body = document.querySelector('.log-item-wrap.log-body.scrollbar.scroll-container') || document.querySelector('article.archive-modal .modal-body') || document.querySelector('article.archive-modal'); if (body) body.scrollTop = body.scrollHeight; }""") await asyncio.sleep(1.5) # 사용자 지정 클래스 내의 모든 div ID 수집 # .log-item-wrap.log-body.scrollbar.scroll-container 내부의 div들을 타겟팅 selector = ".log-item-wrap.log-body.scrollbar.scroll-container div" current_elements = await page.locator(selector).all() # 만약 지정된 클래스로 검색되지 않을 경우 기존 div[id*='_']를 백업으로 사용 if not current_elements: current_elements = await page.locator("article.archive-modal div[id*='_']").all() seen_ids = {f"{log['date']}, {log['activity']}" for log in all_logs} for el in current_elements: log_id = await el.get_attribute("id") if not log_id: continue log_text = parse_log_id(log_id) if ", " in log_text and log_text not in seen_ids: d, a = log_text.split(", ", 1) all_logs.append({'date': d, 'activity': a}) seen_ids.add(log_text) if len(all_logs) == last_count: break last_count = len(all_logs) if not all_logs: msg_queue.put(json.dumps({'type': 'log', 'message': f' - [주의] {project_name}: 수집된 로그가 없습니다.'})) await page.keyboard.press("Escape") # 4. 파일 수 추론 (전수 보존 모드) history_map = {} curr_calc_count = actual_count if all_logs: # 오늘 날짜 강제 주입 대신 수집된 로그의 실제 날짜 사용 for log in all_logs: d_db = log['date'].replace(".", "-") act = log['activity'] if d_db not in history_map: history_map[d_db] = {"log": act, "count": curr_calc_count} if "업로드" in act: curr_calc_count -= 1 elif "삭제" in act: curr_calc_count += 1 if curr_calc_count < 0: curr_calc_count = 0 history_map[d_db]["count"] = curr_calc_count else: # 로그가 전혀 없을 경우에만 기본값 생성 today_str = datetime.now().strftime("%Y-%m-%d") history_map[today_str] = {"log": "기존 상태 유지 (활동 없음)", "count": actual_count} # 5. DB 저장 if current_p_id: with get_db_connection() as conn: with conn.cursor() as cursor: for date_key, data in history_map.items(): cursor.execute(CrawlerQueries.UPSERT_HISTORY_WITH_DATE, (current_p_id, date_key, f"{date_key.replace('-', '.')}, {data['log']}", data['count'])) conn.commit() msg_queue.put(json.dumps({'type': 'log', 'message': f' - [성공] 실측 {actual_count}개 기준 시계열 적재 완료'})) await page.goto("https://overseas.projectmastercloud.com/dashboard", wait_until="domcontentloaded") except Exception as e: msg_queue.put(json.dumps({'type': 'log', 'message': f' - {project_name} 에러: {str(e)}'})) await page.goto("https://overseas.projectmastercloud.com/dashboard") msg_queue.put(json.dumps({'type': 'done', 'data': []})) except Exception as e: msg_queue.put(json.dumps({'type': 'log', 'message': f'치명적 오류: {str(e)}'})) finally: if browser: await browser.close() msg_queue.put(None) loop.run_until_complete(run()) loop.close() async def run_crawler_service(): msg_queue = queue.Queue() thread = threading.Thread(target=crawler_thread_worker, args=(msg_queue, os.getenv("PM_USER_ID"), os.getenv("PM_PASSWORD"))) thread.start() while True: try: msg = await asyncio.to_thread(msg_queue.get, timeout=1.0) if msg is None: break yield f"data: {msg}\n\n" except queue.Empty: if not thread.is_alive(): break await asyncio.sleep(0.1) thread.join()