Aislo/resources/knowledge/original/_pipeline/collect_cost_sources.py

# -*- coding: utf-8 -*-
"""원가 기초단가 원천 수집 스크립트 (미결 No.17 / 수집지침 W8 예정)

수집 대상 → 저장 위치 (original/원가계산/):
  1. 건설업 노임   : 대한건설협회 건설임금 게시판     → 노임단가_건설업_대한건설협회/
  2. 제조업 노임   : 중소기업중앙회 조사·통계 게시판   → 노임단가_제조업_중소기업중앙회/
  3. 건설공사 표준품셈 : 한국표준품셈정보원(kseis) 자료실 → 건설공사_표준품셈/

  4. 자재단가 스냅샷 : 나라장터 가격정보현황서비스 API → 자재단가/ (CSV+JSON, 수집일 파일명)
  5. 환율 스냅샷    : 한국은행 ECOS API 731Y001      → 환율_한국은행ECOS/ (CSV, 연초~수집일 일별)
  6. 유가 스냅샷    : 오피넷 API                      → 유가_오피넷/ (JSON, 수집일 — 전국평균·시도별 두 벌)

사용법:
  python collect_cost_sources.py docs      # 1~3 문서 다운로드 (게시판 최신글 자동 탐색은 미구현 —
                                           # 반기마다 아래 URL 상수를 최신 게시글로 갱신 후 실행)
  python collect_cost_sources.py values    # 4~6 값형 API 원본 스냅샷 저장 (키 = knowledge/.secrets.local.md)

주의: 문서 URL(게시글 seq·uuid)은 반기마다 바뀐다. 갱신 절차는 knowledge/01_수집지침.md W8 예정.
값형(4~6)은 원본 스냅샷을 폴더에 보존하고, 프로그램용 데이터셋은 여기서 별도 추출해 구성한다
(2026-08-14 사용자 결정). 표준시장단가는 수집 제외 (100억 미만 공사 미적용·품셈 방식과 별도 트랙).
"""

import csv
import json
import os
import re
import ssl
import sys
import urllib.request
from datetime import date
from pathlib import Path

BASE = Path(__file__).resolve().parent.parent / "원가계산"
SECRETS = Path(__file__).resolve().parent.parent.parent / ".secrets.local.md"
UA = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}

# ── 반기 갱신 대상 URL (2026 상반기/2026년판 기준) ──
DOC_SOURCES = [
    # (폴더, 파일명, URL, referer)
    (
        "노임단가_건설업_대한건설협회",
        "2026상반기_건설업_임금실태조사_대한건설협회.pdf",
        "https://www.cak.or.kr/download.do?uuid=a0e92670-b6b7-4764-a15f-40c34febeaa0.pdf",
        "https://www.cak.or.kr/lay1/S1T16C41/sublink.do",
    ),
    (
        "노임단가_제조업_중소기업중앙회",
        "2026상반기_중소제조업_직종별_임금조사_중소기업중앙회.pdf",
        None,  # kbiz는 view 페이지에서 download.do 링크 추출 필요 — VIEW_URL 사용
        "https://www.kbiz.or.kr/ko/contents/bbs/view.do?mnSeq=325&seq=163372",
    ),
    (
        "건설공사_표준품셈",
        "2026년_건설공사_표준품셈.pdf",
        "https://www.kseis.co.kr/bbs/data/dataFileDown.do?bbs_seq=64699193400142&file_no=1",
        "https://www.kseis.co.kr/bbs/data/dataDetail.do?bbs_seq=64699193400142&pgno=1",
    ),
    (
        "건설공사_표준품셈",
        "2026년_건설공사_표준품셈_개정사항.pdf",
        "https://www.kseis.co.kr/bbs/data/dataFileDown.do?bbs_seq=64699193400142&file_no=2",
        "https://www.kseis.co.kr/bbs/data/dataDetail.do?bbs_seq=64699193400142&pgno=1",
    ),
]

G2B_OPS = [
    ("getPriceInfoListFcltyCmmnMtrilEngrk", "토목"),
    ("getPriceInfoListFcltyCmmnMtrilBildng", "건축"),
    ("getPriceInfoListFcltyCmmnMtrilMchnEqp", "기계설비"),
    ("getPriceInfoListFcltyCmmnMtrilElctyIrmc", "전기정보통신"),
]

_ctx = ssl.create_default_context()
_ctx.check_hostname = False
_ctx.verify_mode = ssl.CERT_NONE  # 일부 기관 인증서 체인 불량 — 무결성은 매직바이트로 검사


def fetch(url, referer=None, timeout=120):
    h = dict(UA)
    if referer:
        h["Referer"] = referer
    req = urllib.request.Request(url, headers=h)
    with urllib.request.urlopen(req, timeout=timeout, context=_ctx) as r:
        return r.read()


def read_key(pattern, name):
    m = re.search(pattern, SECRETS.read_text(encoding="utf-8"))
    if not m:
        sys.exit(f"{name} 키를 .secrets.local.md에서 찾지 못함")
    return m.group(1)


def read_g2b_key():
    return read_key(r"인증키\(Encoding형[^`]*`([^`]+)`", "공공데이터포털")


def collect_docs():
    for folder, fname, url, referer in DOC_SOURCES:
        dest = BASE / folder / fname
        if url is None:  # kbiz: view 페이지에서 download 링크 추출
            html = fetch(referer).decode("utf-8", "replace")
            links = re.findall(r'(/[^"\']*download\.do\?[^"\']+)', html)
            if not links:
                print(f"SKIP {fname}: 다운로드 링크 못 찾음")
                continue
            url = "https://www.kbiz.or.kr" + links[0].replace("&", "&")
        data = fetch(url, referer)
        if (
            not data[:4] == b"%PDF"
            and not data[:4] == b"PK\x03\x04"
            and not data[:8].startswith(b"\xd0\xcf\x11\xe0")
        ):
            print(f"FAIL {fname}: PDF/HWP 아님 ({data[:8].hex()}) — URL 갱신 필요")
            continue
        dest.write_bytes(data)
        print(f"saved {folder}/{fname} ({len(data):,} B)")


def collect_values():
    """값형 3종 원본 스냅샷: 자재단가(CSV+JSON)·환율(CSV)·유가(JSON)."""
    today = date.today().isoformat()
    collect_snapshot(BASE / "자재단가")

    # 환율: 연초~오늘 일별, 5통화
    ecos = read_key(r"ECOS[^`]*\n- 인증키: `([^`]+)`", "ECOS")
    year = today[:4]
    rows = []
    for code, name in [
        ("0000001", "원/미국달러"),
        ("0000002", "원/일본엔100"),
        ("0000003", "원/유로"),
        ("0000012", "원/영국파운드"),
        ("0000053", "원/위안"),
    ]:
        url = (
            f"https://ecos.bok.or.kr/api/StatisticSearch/{ecos}/json/kr/1/400/"
            f"731Y001/D/{year}0101/{today.replace('-', '')}/{code}"
        )
        j = json.loads(fetch(url, timeout=60).decode("utf-8", "replace"))
        for r in j.get("StatisticSearch", {}).get("row", []):
            rows.append(
                {"통화": name, "항목코드": code, "일자": r["TIME"], "환율": r["DATA_VALUE"]}
            )
    out = BASE / "환율_한국은행ECOS" / f"환율_일별_{year}0101_{today}.csv"
    out.parent.mkdir(exist_ok=True)
    with open(out, "w", newline="", encoding="utf-8-sig") as f:
        w = csv.DictWriter(f, fieldnames=["통화", "항목코드", "일자", "환율"])
        w.writeheader()
        w.writerows(rows)
    print(f"saved {out.name} ({len(rows):,} rows)")

    # 유가: 오늘 전국 평균 + 유종별 최근 7일
    opinet = read_key(r"오피넷[^`]*`([^`]+)`", "오피넷")
    data = {"수집일": today}
    j = json.loads(
        fetch(f"https://www.opinet.co.kr/api/avgAllPrice.do?out=json&code={opinet}", timeout=60)
    )
    data["전국평균"] = j.get("RESULT", {}).get("OIL", [])
    data["최근7일"] = {}
    for prod in ["B027", "D047"]:  # 휘발유, 자동차용경유
        j = json.loads(
            fetch(
                f"https://www.opinet.co.kr/api/avgRecentPrice.do?out=json&code={opinet}&prodcd={prod}",
                timeout=60,
            )
        )
        data["최근7일"][prod] = j.get("RESULT", {}).get("OIL", [])
    out = BASE / "유가_오피넷" / f"유가_전국평균_{today}.json"
    out.parent.mkdir(exist_ok=True)
    out.write_text(json.dumps(data, ensure_ascii=False, indent=1), encoding="utf-8")
    print(f"saved {out.name}")

    # 유가: 시도별 (품셈 8-1-7 5호 「유류가격은 해당지역의 가격」 — 지역 단가의 원천)
    sido = {"수집일": today, "시도별": {}}
    for prod in ["B027", "D047", "C004"]:  # 휘발유, 자동차용경유, 실내등유
        j = json.loads(
            fetch(
                f"https://www.opinet.co.kr/api/avgSidoPrice.do?out=json&code={opinet}&prodcd={prod}",
                timeout=60,
            )
        )
        sido["시도별"][prod] = j.get("RESULT", {}).get("OIL", [])
    out = BASE / "유가_오피넷" / f"유가_시도별_{today}.json"
    out.write_text(json.dumps(sido, ensure_ascii=False, indent=1), encoding="utf-8")
    print(f"saved {out.name}")


def collect_snapshot(outdir=None):
    key = read_g2b_key()
    today = date.today().isoformat()
    rows, fieldnames = [], []
    for op, label in G2B_OPS:
        page, got, total = 1, 0, 1
        while got < total:
            url = (
                f"http://apis.data.go.kr/1230000/ao/PriceInfoService/{op}"
                f"?serviceKey={key}&pageNo={page}&numOfRows=999&type=json"
            )
            j = json.loads(fetch(url, timeout=60).decode("utf-8", "replace"))
            body = j.get("response", {}).get("body", {})
            total = int(body.get("totalCount") or 0)
            items = body.get("items", [])
            if isinstance(items, dict):
                items = items.get("item", [])
            if not items:
                break
            for it in items:
                it["분야"] = label
                rows.append(it)
                for k in it:
                    if k not in fieldnames:
                        fieldnames.append(k)
            got += len(items)
            page += 1
        print(f"{label}: {got}/{total}")
    outp = Path(outdir or BASE / "자재단가")
    outp.mkdir(exist_ok=True)
    out = outp / f"나라장터_시설공통자재_{today}.csv"
    with open(out, "w", newline="", encoding="utf-8-sig") as f:
        w = csv.DictWriter(f, fieldnames=fieldnames)
        w.writeheader()
        w.writerows(rows)
    (outp / f"나라장터_시설공통자재_{today}.json").write_text(
        json.dumps(rows, ensure_ascii=False), encoding="utf-8"
    )
    print(f"saved {out.name} ({len(rows):,} rows, +json)")


if __name__ == "__main__":
    mode = sys.argv[1] if len(sys.argv) > 1 else "docs"
    if mode == "docs":
        collect_docs()
    elif mode == "values":
        collect_values()
    elif mode == "snapshot":
        collect_snapshot(sys.argv[2] if len(sys.argv) > 2 else None)
    else:
        sys.exit("mode: docs | values | snapshot [출력폴더]")