Leets-Official · sky-0131 · Jul 17, 2026 · Jul 17, 2026 · Jul 17, 2026 · Jul 17, 2026
@@ -43,6 +43,9 @@ dependencies {
 	testCompileOnly 'org.projectlombok:lombok'
 	testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
 	testAnnotationProcessor 'org.projectlombok:lombok'
+
+	// JSONL 파일 파싱을 위해 사용한 ObjectMapper용 Jackson 라이브러리
+	implementation 'org.springframework.boot:spring-boot-starter-web'
 }
 
 tasks.named('test') {

@@ -0,0 +1,54 @@
+"""개발 세부직군 태그를 offset 페이징으로 훑어 후보 id 풀을 구성하고 N개 샤드로 분할.
+
+출력:
+  <out>/pool.json            : 전체 후보 id (순서보존 dedup)
+  <out>/shard_{k}.json       : k번째 샤드 id 목록 (멀티에이전트용)
+
+사용:
+  python build_pool.py --shards 10 --per-tag-max 400 --out ../out/shards
+"""
+from __future__ import annotations
+import argparse, json, os, sys, time
+from wanted import list_job_ids, DEV_TAG_IDS
+
+
+def build_pool(per_tag_max, page=100, delay=0.4):
+    seen, pool = set(), []
+    for tid in DEV_TAG_IDS:
+        got = 0
+        for off in range(0, per_tag_max, page):
+            try:
+                ids = list_job_ids(limit=page, offset=off, tag_type_id=tid)
+            except Exception as e:
+                print(f"[tag {tid}] off {off} 실패: {e}", file=sys.stderr)
+                break
+            if not ids:
+                break
+            new = 0
+            for j in ids:
+                if j not in seen:
+                    seen.add(j); pool.append(j); new += 1
+            got += len(ids)
+            time.sleep(delay)
+            if len(ids) < page:
+                break
+        print(f"[tag {tid}] 누적 풀 {len(pool)}", file=sys.stderr)
+    return pool
+
+
+if __name__ == "__main__":
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--shards", type=int, default=10)
+    ap.add_argument("--per-tag-max", type=int, default=400)
+    ap.add_argument("--out", default="../out/shards")
+    a = ap.parse_args()
+    os.makedirs(a.out, exist_ok=True)
+    pool = build_pool(a.per_tag_max)
+    json.dump(pool, open(os.path.join(a.out, "pool.json"), "w"))
+    # 라운드로빈 분할(인접 id 가 같은 샤드에 몰리지 않게)
+    shards = [[] for _ in range(a.shards)]
+    for i, jid in enumerate(pool):
+        shards[i % a.shards].append(jid)
+    for k, sh in enumerate(shards):
+        json.dump(sh, open(os.path.join(a.out, f"shard_{k}.json"), "w"))
+    print(f"\n풀 {len(pool)}건 → {a.shards}개 샤드 (각 ~{len(pool)//a.shards}건) → {a.out}/")
@@ -0,0 +1,61 @@
+"""공통 HTTP/파싱 헬퍼 (표준 라이브러리만 사용 — 외부 의존성 없음)."""
+from __future__ import annotations
+import gzip
+import json
+import re
+import urllib.parse
+import urllib.request
+
+UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+      "(KHTML, like Gecko) Chrome/120.0 Safari/537.36")
+
+
+def http_get(url, referer=None, timeout=30):
+    """브라우저 UA로 GET. gzip 응답도 처리. 본문 문자열 반환."""
+    req = urllib.request.Request(url, headers={
+        "User-Agent": UA,
+        "Accept-Language": "ko-KR,ko;q=0.9,en;q=0.8",
+    })
+    if referer:
+        req.add_header("Referer", referer)
+    with urllib.request.urlopen(req, timeout=timeout) as r:
+        data = r.read()
+        if r.headers.get("Content-Encoding") == "gzip":
+            data = gzip.decompress(data)
+    return data.decode("utf-8", "replace")
+
+
+def http_json(url, referer=None, timeout=30):
+    return json.loads(http_get(url, referer, timeout))
+
+
+def quote(s):
+    return urllib.parse.quote(str(s))
+
+
+def rsc_blob(html):
+    """잡코리아 Next.js(App Router) RSC: self.__next_f.push 조각들을 하나의 문자열로 결합."""
+    out = []
+    for p in re.findall(r"self\.__next_f\.push\((\[.*?\])\)", html, re.S):
+        try:
+            for el in json.loads(p):
+                if isinstance(el, str):
+                    out.append(el)
+        except Exception:
+            out.append(p)
+    return "".join(out)
+
+
+_PAREN = re.compile(r"\([^)]*\)")  # 영문 별칭 등 괄호군: 넵튠(Neptune)→넵튠, 크몽(kmong)→크몽
+_DROP = re.compile(r"㈜|주식회사|유한회사|inc\.?|co\.,?\s*ltd\.?|corp\.?|ltd\.?", re.I)
+
+
+def norm_company(name):
+    """기업명 매칭/조인 키: 괄호별칭·법인 표기·공백 제거 + 소문자.
+    원티드 '크몽(kmong)' ↔ 잡코리아 '㈜크몽' 같은 표기 차를 흡수한다."""
+    if not name:
+        return ""
+    n = _PAREN.sub("", name)   # (Neptune)/(kmong)/(주) 등 괄호군 제거
+    n = _DROP.sub("", n)       # ㈜/주식회사/Inc/Co.,Ltd 등 법인 표기 제거
+    n = re.sub(r"\s+", "", n)
+    return n.lower()
@@ -0,0 +1,126 @@
+"""잡코리아(enrichment) 기업 메타 추출기.
+
+원티드는 기업 메타가 약하다(업종 정도). 잡코리아 공고 RSC의 CORP_INFO 객체에는
+사원수·기업규모·업종·상장여부·주소가 정형으로 들어있다(실측 검증).
+
+경로(기업명만 알면 됨):
+  1) 잡코리아 통합검색 /Search/?stext={기업명}  → 해당 기업 공고 Gno 추출
+  2) /Recruit/GI_Read/{Gno} 의 RSC → CORP_INFO 회사객체 파싱
+
+한계:
+- 해당 기업이 잡코리아에 공고가 있어야 함(없으면 not_found → 원티드 industry_name으로 fallback).
+- 검색 첫 결과를 쓰므로 동명이인 가능 → 정규화 이름 일치(name_match)로 검증.
+"""
+from __future__ import annotations
+import json
+import re
+import sys
+
+from common import http_get, rsc_blob, norm_company, quote
+
+
+def search_gnos(company_name, limit=4):
+    """기업명으로 잡코리아 검색 → 후보 공고 Gno 들(순서 보존 dedup, 상위 limit개).
+    첫 결과가 무관한 회사일 수 있어, enrich 에서 이름 일치하는 후보를 고른다."""
+    html = http_get(f"https://www.jobkorea.co.kr/Search/?stext={quote(company_name)}",
+                    referer="https://www.jobkorea.co.kr/")
+    raw = re.findall(r"/Recruit/GI_Read/([0-9]+)", html)
+    if not raw:
+        raw = re.findall(r"GI_Read/([0-9]+)", rsc_blob(html))
+    out = []
+    for g in raw:                      # 순서 보존 dedup
+        if g not in out:
+            out.append(g)
+        if len(out) >= limit:
+            break
+    return out
+
+
+def _posting_company_name(html, blob):
+    """공고의 회사명(신뢰): JSON-LD hiringOrganization.name. HTML LD 우선, RSC fallback."""
+    for m in re.finditer(r'<script[^>]*application/ld\+json[^>]*>(.*?)</script>', html, re.S):
+        try:
+            j = json.loads(m.group(1))
+        except Exception:
+            continue
+        for it in (j if isinstance(j, list) else [j]):
+            if isinstance(it, dict) and it.get("@type") == "JobPosting":
+                org = it.get("hiringOrganization") or {}
+                if org.get("name"):
+                    return org["name"]
+    # fallback: RSC(이스케이프된 JSON) 안의 hiringOrganization name
+    m = re.search(r'hiringOrganization\\?"\s*:\s*\{[^}]*?name\\?"\s*:\s*\\?"([^"\\]{1,60})', blob)
+    return m.group(1) if m else None
+
+
+def fetch_corp_info(gno):
+    """공고 RSC의 CORP_INFO 회사객체에서 메타 추출. employeeCount 를 앵커로 사용."""
+    html = http_get(f"https://www.jobkorea.co.kr/Recruit/GI_Read/{gno}?sc=729&sn=103",
+                    referer="https://www.jobkorea.co.kr/")
+    blob = rsc_blob(html)
+    i = blob.find('"employeeCount"')
+    if i < 0:
+        return None
+    seg = blob[max(0, i - 500):i + 700]
+
+    def s(key):
+        m = re.search(r'"' + key + r'"\s*:\s*"([^"]{0,80})"', seg)
+        return m.group(1) if m else None
+
+    def n(key):
+        m = re.search(r'"' + key + r'"\s*:\s*([0-9]+)', seg)
+        return int(m.group(1)) if m else None
+
+    addr = None
+    am = re.search(r'"address"\s*:\s*\{[^}]*?"address"\s*:\s*"([^"]{0,80})"'
+                   r'(?:[^}]*?"addressDetail"\s*:\s*"([^"]{0,80})")?', seg)
+    if am:
+        addr = " ".join(p for p in [am.group(1), am.group(2)] if p).strip()
+
+    return {
+        "jobkorea_gno_ref": int(gno),
+        "company_name_jk": _posting_company_name(html, blob),  # JSON-LD 기준 (신뢰)
+        "employee_count": n("employeeCount"),
+        "company_type": s("companyTypeName"),   # 대기업/중견기업/중소기업
+        "industry": s("industryName"),
+        "stock_status": s("stockStatusName"),   # 코스피/코스닥/-
+        "hq_address": addr,
+    }
+
+
+def enrich(company_name, max_candidates=4):
+    """기업명 → 잡코리아 메타 dict. enrichment_status 포함.
+
+    상위 후보 Gno 들을 훑어 **이름이 일치하는 회사만** 채택한다. 일치가 없으면
+    틀린 회사의 사원수·규모를 저장하지 않는다(무결성). status:
+      enriched      — 이름 일치, 메타 채움
+      name_mismatch — 후보는 있었으나 이름 일치 없음(타사 메타 폐기, 감사용 rejected_name 만 남김)
+      not_found     — 검색 결과 없음
+    """
+    target = norm_company(company_name)
+    gnos = search_gnos(company_name, limit=max_candidates)
+    if not gnos:
+        return {"enrichment_status": "not_found"}
+
+    rejected = None
+    for gno in gnos:
+        meta = fetch_corp_info(gno)
+        if not meta:
+            continue
+        cand = meta.get("company_name_jk") or ""
+        if norm_company(cand) == target and target:
+            meta["name_match"] = True
+            meta["enrichment_status"] = "enriched"
+            meta["raw_jobkorea"] = {k: meta[k] for k in
+                                    ("employee_count", "company_type", "industry",
+                                     "stock_status", "hq_address", "company_name_jk")}
+            return meta
+        rejected = rejected or cand  # 첫 후보명만 감사용으로 기록
+    # 어떤 후보도 이름이 안 맞음 → 타사 메타는 버린다(저장 금지)
+    return {"enrichment_status": "name_mismatch", "name_match": False,
+            "rejected_name": rejected}
+
+
+if __name__ == "__main__":
+    name = sys.argv[1] if len(sys.argv) > 1 else "메리츠화재"
+    print(json.dumps(enrich(name), ensure_ascii=False, indent=2))
@@ -0,0 +1,54 @@
+"""샤드 결과 병합 + dedup → 최종 out/job_postings.jsonl, out/companies.jsonl + 통계.
+
+  공고 : external_id 기준 dedup
+  기업 : normalized_name 기준 dedup (enriched 를 우선 채택)
+
+사용:
+  python merge_shards.py --shards-dir ../out/shards --out ../out
+"""
+from __future__ import annotations
+import argparse, glob, json, os
+from collections import Counter
+
+RANK = {"enriched": 3, "name_mismatch": 2, "not_found": 1, None: 0}
+
+
+def main(shards_dir, out_dir):
+    posts, comps = {}, {}
+    for fp in sorted(glob.glob(os.path.join(shards_dir, "postings_*.jsonl"))):
+        for line in open(fp, encoding="utf-8"):
+            p = json.loads(line)
+            posts.setdefault(p["external_id"], p)            # 첫 등장 유지
+    for fp in sorted(glob.glob(os.path.join(shards_dir, "companies_*.jsonl"))):
+        for line in open(fp, encoding="utf-8"):
+            c = json.loads(line)
+            k = c["normalized_name"]
+            cur = comps.get(k)
+            # 더 좋은 보강상태를 우선
+            if cur is None or RANK.get(c.get("enrichment_status"), 0) > RANK.get(cur.get("enrichment_status"), 0):
+                comps[k] = c
+
+    os.makedirs(out_dir, exist_ok=True)
+    with open(os.path.join(out_dir, "job_postings.jsonl"), "w", encoding="utf-8") as f:
+        for p in posts.values():
+            f.write(json.dumps(p, ensure_ascii=False) + "\n")
+    with open(os.path.join(out_dir, "companies.jsonl"), "w", encoding="utf-8") as f:
+        for c in comps.values():
+            f.write(json.dumps(c, ensure_ascii=False) + "\n")
+
+    np = len(posts); nc = len(comps)
+    st = Counter(c.get("enrichment_status") for c in comps.values())
+    sk = sum(1 for p in posts.values() if p.get("skill_tags"))
+    th = sum(1 for p in posts.values() if p.get("thumbnail_url"))
+    print(f"=== 병합 완료 → {out_dir}/ ===")
+    print(f"공고 {np}건  (skill {sk}={100*sk//np}% / thumbnail {th}={100*th//np}%)")
+    print(f"기업 {nc}곳  보강: " + " ".join(f"{k}={v}" for k, v in st.most_common()))
+    print(f"  enriched 비율: {100*st.get('enriched',0)//nc}%")
+
+
+if __name__ == "__main__":
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--shards-dir", default="../out/shards")
+    ap.add_argument("--out", default="../out")
+    a = ap.parse_args()
+    main(a.shards_dir, a.out)
@@ -0,0 +1,90 @@
+"""오케스트레이션: 원티드 목록 → 공고 수집 → (기업별) 잡코리아 메타 보강 → JSONL 출력.
+
+출력:
+  companies.jsonl     : 기업 1행씩 (잡코리아 메타 보강 포함)
+  job_postings.jsonl  : 공고 1행씩 (company_normalized_name 으로 companies 와 조인)
+
+DB 적재: schema.sql 로 테이블 생성 후, JSONL 을 COPY/INSERT (가이드 참고).
+
+사용:
+  python run_pipeline.py --target 30 --per-tag 20 --delay 1.5 --out ../out
+"""
+from __future__ import annotations
+import argparse
+import json
+import os
+import sys
+import time
+
+from wanted import list_dev_job_ids, fetch_posting
+from jobkorea_company import enrich
+
+
+def run(target, per_tag, delay, out_dir):
+    os.makedirs(out_dir, exist_ok=True)
+    companies = {}          # normalized_name -> 기업 레코드
+    postings = []
+
+    # 목록 단계에서 개발 세부직군 태그로만 발견(사전 fetch 낭비 제거).
+    # target 의 3배까지 후보 확보(태그 누수분 + 중복 여유).
+    pool = list_dev_job_ids(per_tag=per_tag, max_total=target * 3 if target else None)
+    print(f"[discover] 개발 후보 {len(pool)}건 → 목표 {target or '전체'}건 추출", file=sys.stderr)
+
+    for jid in pool:
+        if target and len(postings) >= target:
+            break
+        try:
+            p = fetch_posting(jid)
+        except Exception as e:
+            print(f"[wanted] {jid} skip: {e}", file=sys.stderr)
+            continue
+        # 태그가 깨끗하지만 누수 가능 → 개발 직군 최종 가드(이제 저렴)
+        if p.get("category_parent") != "개발":
+            continue
+        postings.append(p)
+        print(f"  + [{len(postings)}] {jid}: {p['position']} @ {p['company']['name']} "
+              f"(skill {len(p['skill_tags'])})", file=sys.stderr)
+
+        cn = p["company"]["normalized_name"]
+        if cn and cn not in companies:
+            comp = dict(p["company"])
+            try:
+                meta = enrich(p["company"]["name"])   # 잡코리아 보강
+            except Exception as e:
+                meta = {"enrichment_status": "not_found", "error": str(e)}
+            # 원티드 기본 + 잡코리아 메타 병합 (잡코리아 industry 가 있으면 우선)
+            comp["industry"] = meta.get("industry") or comp.get("industry_name")
+            for k in ("jobkorea_gno_ref", "employee_count", "company_type",
+                      "stock_status", "hq_address", "name_match",
+                      "enrichment_status", "raw_jobkorea"):
+                if k in meta:
+                    comp[k] = meta[k]
+            comp.setdefault("enrichment_status", "not_found")
+            companies[cn] = comp
+            st = comp.get("enrichment_status")
+            print(f"    └ 기업메타[{st}]: {comp.get('company_type')}/"
+                  f"{comp.get('employee_count')}명/{comp.get('stock_status')}", file=sys.stderr)
+            time.sleep(delay)
+        time.sleep(delay)
+
+    with open(os.path.join(out_dir, "companies.jsonl"), "w", encoding="utf-8") as f:
+        for c in companies.values():
+            f.write(json.dumps(c, ensure_ascii=False) + "\n")
+    with open(os.path.join(out_dir, "job_postings.jsonl"), "w", encoding="utf-8") as f:
+        for p in postings:
+            p["company_normalized_name"] = p["company"]["normalized_name"]
+            f.write(json.dumps(p, ensure_ascii=False) + "\n")
+
+    enriched = sum(1 for c in companies.values() if c.get("enrichment_status") == "enriched")
+    print(f"\n완료: 공고 {len(postings)}건 / 기업 {len(companies)}곳 "
+          f"(잡코리아 보강 성공 {enriched}곳) → {out_dir}/")
+
+
+if __name__ == "__main__":
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--target", type=int, default=30, help="추출할 개발 공고 목표 수 (0=후보 전체)")
+    ap.add_argument("--per-tag", type=int, default=20, help="세부직군 태그당 목록 수")
+    ap.add_argument("--delay", type=float, default=1.5)
+    ap.add_argument("--out", default="../out")
+    a = ap.parse_args()
+    run(a.target, a.per_tag, a.delay, a.out)