Merge pull request #137 from Pseudo-Lab/feat/crawling

한국어 공고 크롤링 deadline 추가

Author: minglet

preprocess/korean_jd_crawling.py

@@ -1,4 +1,4 @@
-import json
+import csv
 from re import T
 import time
 import os
@@ -18,55 +18,54 @@ class TestCrawlingWanted:
     def __init__(self):
         self.endpoint = "https://www.wanted.co.kr"
         self.job_parent_category = 518
-        self.job_category_id = 10110  # 소프트웨어 엔지니어
-        self.job_category_name = "소프트웨어 엔지니어"
-        # self.job_category_id2name = {
-            # 10110: "소프트웨어 엔지니어",
-            # 873: "웹 개발자",
-            # 872: "서버 개발자",
-            # 669: "프론트엔드 개발자",
+        self.job_category_id2name = {
+            10110: "소프트웨어 엔지니어",
+            873: "웹 개발자",
+            872: "서버 개발자",
+            669: "프론트엔드 개발자",
             # 660: "자바 개발자",
             # 900: "C,C++ 개발자",
-            # 899: "파이썬 개발자",
-            # 1634: "머신러닝 엔지니어",
-            # 674: "DevOps / 시스템 관리자",
+            899: "파이썬 개발자",
+            1634: "머신러닝 엔지니어",
+            674: "DevOps / 시스템 관리자",
             # 665: "시스템,네트워크 관리자",
-            # 655: "데이터 엔지니어",
+            655: "데이터 엔지니어",
             # 895: "Node.js 개발자",
-            # 677: "안드로이드 개발자",
-            # 678: "iOS 개발자",
+            677: "안드로이드 개발자",
+            678: "iOS 개발자",
             # 658: "임베디드 개발자",
             # 877: "개발 매니저",
-            # 1024: "데이터 사이언티스트",
+            1024: "데이터 사이언티스트",
             # 1026: "기술지원",
-            # 676: "QA,테스트 엔지니어",
+            676: "QA,테스트 엔지니어",
             # 672: "하드웨어 엔지니어",
             # 1025: "빅데이터 엔지니어",
-            # 671: "보안 엔지니어",
+            671: "보안 엔지니어",
             # 876: "프로덕트 매니저",
             # 10111: "크로스플랫폼 앱 개발자",
             # 1027: "블록체인 플랫폼 엔지니어",
             # 10231: "DBA",
             # 893: "PHP 개발자",
             # 661: ".NET 개발자",
-            # 896: "영상,음성 엔지니어",
+            896: "영상,음성 엔지니어",
             # 10230: "ERP전문가",
             # 939: "웹 퍼블리셔",
             # 898: "그래픽스 엔지니어",
-            # 795: "CTO,Chief Technology Officer",
+            795: "CTO,Chief Technology Officer",
             # 10112: "VR 엔지니어",
             # 1022: "BI 엔지니어",
             # 894: "루비온레일즈 개발자",
             # 793: "CIO,Chief Information Officer"
-        # }
+        }
 
         self.tag2field_map = {
             "포지션 상세": "description",
             "주요업무": "main_work", 
             "자격요건": "qualification",
             "우대사항": "preferences",
             "혜택 및 복지": "welfare",
-            "기술스택 ・ 툴": "tech_list"
+            "기술스택 ・ 툴": "tech_list",
+            "마감일": "deadline"
         }
 
         # Chrome 드라이버 설정
@@ -163,21 +162,19 @@ def crawl_job_detail(self, position_url):
         # 회사 정보
         company_info = job_header.find("div", class_="JobHeader_JobHeader__Tools__lyxqQ")
         if company_info:
-            result['company_name'] = company_info.text.strip()
+            raw_company_info = company_info.text.strip()
+            result['company_name_raw'] = raw_company_info  # 원본 정보 보존
+            
+            # 회사 정보 파싱
+            parsed_info = self.parse_company_info(raw_company_info)
+            result['company_name'] = parsed_info['company_name']
+            result['location'] = parsed_info['location']
+            result['experience_requirement'] = parsed_info['experience']
+            
             company_link = company_info.find("a")
             if company_link:
                 result['company_id'] = company_link.get("href", "")
 
-        # 태그 정보
-        tags_div = job_header.find("div", class_="Tags_tagsClass__mvehZ")
-        if tags_div:
-            tag_list = tags_div.find_all("span")
-            result['tag_name'] = [tag.text.lstrip("#").strip() for tag in tag_list]
-            result['tag_id'] = [tag.get("href", "") for tag in tag_list]
-        else:
-            result['tag_name'] = []
-            result['tag_id'] = []
-        
         # 상세 내용
         job_body = soup.find("section", class_="JobContent_descriptionWrapper__RMlfm")
         if job_body:
@@ -186,13 +183,15 @@ def crawl_job_detail(self, position_url):
             for elem in job_body.find_all(["h3", "h2", "p", "li", "div"]):
                 if elem.name in ["h2", "h3"]:
                     title = elem.text.strip()
+                    
                     if title in self.tag2field_map:
                         current_field = self.tag2field_map[title]
                         result[current_field] = ""
                     else:
                         current_field = None
                 elif current_field:
                     text = elem.get_text(" ", strip=True)
+                    
                     if text and len(h3_text) > 0:
                         result[current_field] = h3_text
                         h3_text = text
@@ -202,12 +201,73 @@ def crawl_job_detail(self, position_url):
                 elif current_field is None:
                     text = elem.get_text(" ", strip=True)
                     h3_text += text
+                    
+        
+        deadline = soup.find("article", class_="JobDueTime_JobDueTime__yvhtg")
+        if deadline:
+            result['deadline'] = deadline.find("span").text.strip()
 
-        # 기술 스택 정제
-        if "tech_list" in result and result["tech_list"]:
-            result["tech_list"] = [t.strip() for t in result["tech_list"].split("・") if t.strip()]
+        # 태그 정보
+        tags_div = soup.find("ul", class_="CompanyTags_CompanyTags__list__XmzkW")
+        if tags_div:
+            tag_list = tags_div.find_all("li")
+            # 각 li에서 키워드 텍스트만 추출
+            keywords = []
+            for li in tag_list:
+                keyword_span = li.find("span", class_="wds-nkj4w6")
+                if keyword_span:
+                    keywords.append(keyword_span.text.strip())
+            
+            result['tag_name'] = keywords
         else:
-            result["tech_list"] = []
+            result['tag_name'] = []
+        
+        return result
+    
+    def parse_company_info(self, company_info_text):
+        """
+        회사 정보 텍스트를 파싱하여 회사명, 지역, 조건을 분리
+        예: "퓨쳐스콜레∙서울 성동구∙경력 5년 이상" -> 
+        {
+            "company_name": "퓨쳐스콜레",
+            "location": "서울 성동구", 
+            "experience": "경력 5년 이상"
+        }
+        """
+        if not company_info_text:
+            return {
+                "company_name": "",
+                "location": "",
+                "experience": ""
+            }
+        
+        # ∙ 또는 · 문자로 분리
+        parts = company_info_text.replace('·', '∙').split('∙')
+        parts = [part.strip() for part in parts if part.strip()]
+        
+        result = {
+            "company_name": "",
+            "location": "",
+            "experience": ""
+        }
+        
+        if len(parts) >= 1:
+            result["company_name"] = parts[0]
+        
+        if len(parts) >= 2:
+            # 두 번째 부분이 지역인지 확인 (시/도 이름이 포함되어 있는지)
+            location_keywords = ['서울', '부산', '대구', '인천', '광주', '대전', '울산', '세종', 
+                               '경기', '강원', '충북', '충남', '전북', '전남', '경북', '경남', '제주']
+            if any(keyword in parts[1] for keyword in location_keywords):
+                result["location"] = parts[1]
+                if len(parts) >= 3:
+                    result["experience"] = parts[2]
+            else:
+                # 지역이 아니면 경력 조건으로 간주
+                result["experience"] = parts[1]
+        
+        if len(parts) >= 3 and not result["experience"]:
+            result["experience"] = parts[2]
 
         return result
 
@@ -216,54 +276,75 @@ def run_test_crawling(self, limit=5):
         print("=== 원티드 테스트 크롤링 시작 ===")
 
         try:
-            # 1. URL 수집
-            position_urls = self.get_test_job_urls(limit)
-            if not position_urls:
-                print("수집할 URL이 없습니다.")
-                return
-            
-            # 2. 상세 정보 크롤링
-            results = []
-            for i, url in enumerate(position_urls, 1):
-                print(f"\n[{i}/{len(position_urls)}] 처리 중...")
-                result = self.crawl_job_detail(url)
-                if result:
-                    results.append(result)
-                    print(f"✅ 성공: {result.get('title', 'Unknown')}")
-                else:
-                    print("❌ 실패")
+            final_results = []
+            for self.job_category_id, self.job_category_name in self.job_category_id2name.items():
 
-                # 요청 간격 (서버 부하 방지)
-                time.sleep(1)
+                position_urls = self.get_test_job_urls(limit)
+                if not position_urls:
+                    print("수집할 URL이 없습니다.")
+                    return
+                
+                results = []
+                for i, url in enumerate(position_urls, 1):
+                    print(f"\n[{i}/{len(position_urls)}] 처리 중...")
+                    result = self.crawl_job_detail(url)
+                    if result:
+                        results.append(result)
+                        print(f"✅ 성공: {result.get('title', 'Unknown')}")
+                    else:
+                        print("❌ 실패")                    
+                    # 요청 간격 (서버 부하 방지)
+                    time.sleep(1)
+                
+                final_results.append(results)
+
+            # 3. 결과 저장 (CSV 형식)
+            output_file = "crawling_results.csv"
 
-            # 3. 결과 저장
-            output_file = "test_crawling_results.json"
-            with open(output_file, 'w', encoding='utf-8') as f:
-                json.dump(results, f, ensure_ascii=False, indent=2)
+            # 중첩된 리스트를 평면화
+            flattened_results = []
+            for category_results in final_results:
+                flattened_results.extend(category_results)
+            
+            if flattened_results:
+                # CSV 헤더 정의 (첫 번째 결과의 키를 기준으로)
+                fieldnames = list(flattened_results[0].keys())
+                
+                with open(output_file, 'w', encoding='utf-8', newline='') as f:
+                    writer = csv.DictWriter(f, fieldnames=fieldnames)
+                    writer.writeheader()
+                    
+                    for result in flattened_results:
+                        # 리스트 형태의 값들을 문자열로 변환 (예: tag_name)
+                        row = {}
+                        for key, value in result.items():
+                            if isinstance(value, list):
+                                row[key] = ', '.join(map(str, value))
+                            else:
+                                row[key] = value
+                        writer.writerow(row)
+            else:
+                print("저장할 데이터가 없습니다.")
+                return final_results
 
             print(f"\n=== 크롤링 완료 ===")
-            print(f"총 {len(results)}개 항목 수집")
+            print(f"총 {len(flattened_results)}개 항목 수집")
             print(f"결과 파일: {output_file}")
 
             # 간단한 결과 미리보기
             print("\n=== 결과 미리보기 ===")
-            for i, result in enumerate(results, 1):
+            for i, result in enumerate(flattened_results[:5], 1):  # 처음 5개만 미리보기
                 print(f"{i}. {result.get('company_name', 'Unknown')} - {result.get('title', 'Unknown')}")
                 print(f"   URL: {result.get('url', '')}")
                 print(f"   태그: {', '.join(result.get('tag_name', []))}")
                 print()
 
-            return results
+            return final_results
 
         finally:
             self.driver.quit()
             print("브라우저 종료")
 
 if __name__ == "__main__":
     crawler = TestCrawlingWanted()
-    results = crawler.run_test_crawling(limit=5)
-    
-    # # view_files.py로 결과 확인하기
-    # if results:
-    #     print("\n결과를 HTML로 보려면:")
-    #     print("python view_files.py test_crawling_results.json")
+    results = crawler.run_test_crawling(limit=7)