CLIP API

import json
import sys
import time
from collections import Counter
from pathlib import Path

import requests

BASE_URL = "https://clip.floaterlab.com"
API_KEY = "YOUR_API_KEY"
COVERAGES = "medical_in,hospitalization"
DOC_EXTS = {".pdf", ".jpg", ".jpeg", ".png", ".tif", ".tiff"}
# 서류 이름 표에 있는 종류만 판정에 보냅니다.
KNOWN_DOC_TYPES = {
    "보험금청구서", "신분증", "통장사본", "진단서", "입퇴원확인서",
    "진료비세부내역서", "진료비계산서·영수증", "처방전", "수술확인서",
    "사망진단서", "가족관계증명서", "병리조직검사 결과지", "위임장",
}
HEADERS = {"x-clip-key": API_KEY}
MAX_RETRIES = 2  # 5xx 응답은 최대 2번 다시 보냅니다
TIMEOUT = 1200  # 클라이언트 타임아웃 20분
POLL_SECONDS = 3
OUT_PATH = Path("insurance-result.json")


def send(method, path, make_kwargs):
    """요청을 보내고, 5xx 응답이면 최대 MAX_RETRIES 번 다시 보냅니다. 파일은 매번 새로 만듭니다."""
    for attempt in range(MAX_RETRIES + 1):
        res = requests.request(method, BASE_URL + path, headers=HEADERS, timeout=TIMEOUT, **make_kwargs())
        if res.status_code < 500 or attempt == MAX_RETRIES:
            return res
        print(f"  서버 오류 {res.status_code}, 재시도 {attempt + 1}/{MAX_RETRIES}")
        time.sleep(2)


def poll(path, job_id, label):
    """pending 이 아닐 때까지 3초 간격으로 조회합니다. done 이나 error 응답은 작업을 삭제하므로 바로 돌려줍니다."""
    last_note = None
    while True:
        time.sleep(POLL_SECONDS)
        res = requests.get(BASE_URL + path, params={"job": job_id}, timeout=60)
        try:
            body = res.json()
        except ValueError:
            body = {"status": "error", "error": f"HTTP {res.status_code}"}
        if body.get("status") != "pending":
            return body
        if body.get("note") and body["note"] != last_note:
            last_note = body["note"]
            print(f"  {label}: {last_note}")


def judge(judge_inputs):
    """판정을 요청하고 결과를 기다립니다. 실패하면 RuntimeError 를 던집니다."""
    pairs = [("coverages", COVERAGES)] + [("docTypes", doc_type) for _, _, doc_type in judge_inputs]
    res = send("POST", "/api/insurance/judge", lambda: {
        "data": pairs,
        "files": [("files", (name, content)) for name, content, _ in judge_inputs],
    })
    job = res.json() if res.ok else {}
    if not job.get("jobId"):
        raise RuntimeError(f"판정 요청 실패: {job.get('error') or res.status_code}")
    body = poll("/api/insurance/judge", job["jobId"], "판정")
    if body.get("status") != "done":
        raise RuntimeError(f"판정 실패: {body.get('error') or body.get('status')}")
    return body["results"]


def main():
    folder = Path(sys.argv[1] if len(sys.argv) > 1 else "claim-docs")
    paths = sorted(p for p in folder.iterdir() if p.suffix.lower() in DOC_EXTS)
    if not paths:
        sys.exit(f"{folder} 폴더에 서류 파일이 없습니다.")

    # 1) 서류를 이름 순서대로 올리고 서류 종류와 핵심정보를 받습니다.
    records = []
    judge_inputs = []  # (파일 이름, 파일 내용, 서류 종류). 판정 요청에 이 순서로 보냅니다.
    for path in paths:
        rec = {"file": path.name, "status": "error", "type": None, "keys": [], "error": None}
        records.append(rec)
        print(f"[분류] {path.name}")
        try:
            data = path.read_bytes()
            res = send("POST", "/api/insurance/upload", lambda: {"files": {"file": (path.name, data)}})
            up = res.json() if res.ok else {}
            if not up.get("fileId"):
                raise RuntimeError(up.get("error") or f"업로드 실패 ({res.status_code})")

            file_id = up["fileId"]
            res = send("POST", "/api/insurance/process", lambda: {"json": {"fileId": file_id, "page": 1, "mode": "full"}})
            job = res.json() if res.ok else {}
            if not job.get("jobId"):
                raise RuntimeError(job.get("error") or f"처리 요청 실패 ({res.status_code})")

            body = poll("/api/insurance/process", job["jobId"], path.name)
            if body.get("status") != "done":
                raise RuntimeError(body.get("error") or body.get("status"))
            result = body["result"]
            rec["status"] = "done"
            rec["type"] = result["type"]
            rec["keys"] = [{"key": k["key"], "value": k.get("value", "")} for k in result.get("keys", [])]
            if result["type"] in KNOWN_DOC_TYPES:
                judge_inputs.append((path.name, data, result["type"]))
        except (RuntimeError, ValueError, requests.RequestException) as e:
            rec["error"] = str(e)
        print(f"  -> {rec['type'] or '실패: ' + str(rec['error'])}")

    # 2) 인식된 서류 종류로만 담보별 AI 판정을 요청합니다.
    verdicts = []
    error = None
    print()
    if not judge_inputs:
        error = "판정에 쓸 수 있는 서류가 없습니다."
    else:
        print(f"[판정] 담보 {COVERAGES} · 서류 {len(judge_inputs)}건")
        try:
            verdicts = judge(judge_inputs)
        except (RuntimeError, ValueError, requests.RequestException) as e:
            error = str(e)

    # 3) 판정이 실패해도 분류 결과는 남도록 먼저 저장합니다.
    OUT_PATH.write_text(
        json.dumps({"records": records, "verdicts": verdicts, "error": error}, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    if error:
        sys.exit(f"{error}\n분류 결과는 {OUT_PATH} 에 저장했습니다.")

    # 4) 요약을 출력합니다.
    counts = Counter(v["verdict"] for v in verdicts)
    print()
    print("== 요약 ==")
    print(f"서류 {len(records)}건 · 분류 {sum(r['status'] == 'done' for r in records)}건 · 실패 {sum(r['status'] == 'error' for r in records)}건")
    print("판정 " + " · ".join(f"{k} {counts.get(k, 0)}" for k in ("PASS", "FAIL", "REVIEW", "UNKNOWN")))
    print()
    print("== 담보별 판정 ==")
    for v in verdicts:
        print(f"{v['coverageId']:<18}{v['no']:<8}{v['verdict']:<9}{v['reason']}")
    print()
    print(f"결과 저장: {OUT_PATH}")


if __name__ == "__main__":
    main()

[분류] 01_진단서.pdf
  -> 진단서
[분류] 02_입퇴원확인서.pdf
  -> 입퇴원확인서
[분류] 03_통장사본.jpg
  -> 통장사본
[분류] 04_기타.pdf
  -> document

[판정] 담보 medical_in,hospitalization · 서류 3건
  판정: 서류 색인 중
  판정: 점검항목 판정 중

== 요약 ==
서류 4건 · 분류 4건 · 실패 0건
판정 PASS 1 · FAIL 0 · REVIEW 3 · UNKNOWN 1

== 담보별 판정 ==
medical_in        MI-4    REVIEW   진단서에서 진단명은 확인되었으나, 면책 해당 여부는 계약 정보가 있어야 판단할 수 있습니다.
medical_in        MI-5    PASS     입원 기간과 치료 내용이 기재되어 있어 기준을 충족합니다.
hospitalization   HP-2    REVIEW   입퇴원확인서의 입원 기간은 확인되었으나, 보장 개시 여부는 계약 정보가 필요합니다.
hospitalization   HP-3    REVIEW   진단서와 입퇴원확인서의 내용은 확인되었으나, 재입원 규정 적용 여부는 약관과 계약 정보로 최종 판단이 필요합니다.
hospitalization   HP-4    UNKNOWN  판정 응답을 해석하지 못했습니다.

결과 저장: insurance-result.json

import json
import sys
import time
from pathlib import Path

import requests

BASE_URL = "https://clip.floaterlab.com".rstrip("/")
API_KEY = "YOUR_API_KEY"
HEADERS = {"x-clip-key": API_KEY}
EXTENSIONS = {".pdf", ".png", ".jpg", ".jpeg", ".tif", ".tiff", ".webp"}
CONNECT_RETRIES = 2  # 연결 오류(응답을 받지 못함)일 때만 다시 보내는 횟수
POLL_SECONDS = 3     # 결과 조회 간격
MAX_POLLS = 600      # 결과 조회 최대 횟수(약 30분)
RESULTS_FILE = Path("results.json")


def call(method, path, timeout=60, retries=0, **kwargs):
    """연결 오류일 때만 retries 번 다시 보냅니다. 5xx 응답은 다시 보내지 않고 그대로 돌려줍니다."""
    for attempt in range(retries + 1):
        try:
            return requests.request(method, BASE_URL + path, headers=HEADERS, timeout=timeout, **kwargs)
        except requests.RequestException as e:
            if attempt == retries:
                raise RuntimeError(f"{method} {path} 연결 실패: {e}")
            time.sleep(2 ** attempt)


def upload(path):
    data = path.read_bytes()
    # 업로드는 다시 보내지 않습니다. 같은 파일이 두 번 올라갈 수 있기 때문입니다.
    res = call("POST", "/api/shipping/upload", timeout=300, files={"file": (path.name, data)})
    if res.status_code != 200:
        raise RuntimeError(f"업로드 실패 {res.status_code}: {res.text[:200]}")
    file_id = res.json().get("fileId")
    if not file_id:
        raise RuntimeError("업로드 응답에 fileId 가 없습니다.")
    return file_id


def count_pages(path):
    suffix = path.suffix.lower()
    try:
        if suffix == ".pdf":
            from pypdf import PdfReader
            return len(PdfReader(str(path)).pages)
        if suffix in {".tif", ".tiff"}:
            from PIL import Image
            with Image.open(path) as im:
                return getattr(im, "n_frames", 1)
    except ImportError:
        print(f"  [경고] {path.name}: 쪽 수를 세려면 PDF 는 pypdf, TIFF 는 Pillow 가 필요합니다. pip install pypdf pillow 를 실행하십시오. 이 파일은 1쪽으로 처리합니다.")
    return 1


def process_page(file_id, page):
    res = call("POST", "/api/shipping/process", json={"fileId": file_id, "page": page, "mode": "full"})
    if res.status_code != 200:
        raise RuntimeError(f"처리 요청 실패 {res.status_code}: {res.text[:200]}")
    job_id = res.json()["jobId"]
    for _ in range(MAX_POLLS):
        time.sleep(POLL_SECONDS)
        res = call("GET", "/api/shipping/process", timeout=60, retries=CONNECT_RETRIES, params={"job": job_id})
        try:
            body = res.json()
        except ValueError:
            raise RuntimeError(f"조회 응답을 읽지 못했습니다(HTTP {res.status_code}).")
        # 실패한 작업은 HTTP 상태가 5xx 여도 본문의 status 로 판단합니다.
        status = body.get("status")
        if status == "pending":
            continue
        if status == "done":
            return body["result"]
        if status == "error":
            raise RuntimeError(body.get("error", "알 수 없는 오류"))
        raise RuntimeError(f"작업 상태를 알 수 없습니다({status}). 이미 받은 결과이거나 서버가 재시작되었을 수 있습니다.")
    raise TimeoutError("결과가 제한 시간 안에 오지 않았습니다.")


def group_documents(pages):
    """쪽의 type 이 같고 연속인 쪽을 한 서류로 묶습니다."""
    docs = []
    for p in pages:
        last = docs[-1] if docs else None
        if last and last["documentType"] == p["type"] and last["pageEnd"] == p["page"] - 1:
            last["pageEnd"] = p["page"]
            last["_pages"].append(p)
        else:
            docs.append({"documentType": p["type"], "pageStart": p["page"], "pageEnd": p["page"], "_pages": [p]})
    result = []
    for d in docs:
        values = {}
        for p in d["_pages"]:
            for item in p.get("keys", []):
                if item.get("value") and item["key"] not in values:
                    values[item["key"]] = item["value"]
        result.append({"documentType": d["documentType"], "pageStart": d["pageStart"],
                       "pageEnd": d["pageEnd"], "keyValues": values})
    return result


def process_file(path):
    file_id = upload(path)
    page_count = count_pages(path)
    pages, errors = [], []
    for page in range(1, page_count + 1):
        try:
            pages.append(process_page(file_id, page))
        except Exception as e:
            errors.append({"page": page, "error": str(e)})
    return {
        "fileName": path.name,
        "fileId": file_id,
        "pageCount": page_count,
        "documents": group_documents(pages),
        "errors": errors,
    }


def main():
    if len(sys.argv) != 2:
        sys.exit("사용법: python review_folder.py <파일 폴더>")
    folder = Path(sys.argv[1])
    files = sorted(p for p in folder.iterdir() if p.is_file() and p.suffix.lower() in EXTENSIONS)

    summary = []
    for path in files:
        print(f"[처리] {path.name}")
        try:
            summary.append(process_file(path))
        except Exception as e:
            summary.append({"fileName": path.name, "error": str(e)})
        # 파일 하나를 끝낼 때마다 저장합니다. 중간에 멈춰도 끝난 파일의 결과는 남습니다.
        RESULTS_FILE.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")

    for item in summary:
        print()
        print(f"■ {item['fileName']}")
        if "error" in item:
            print(f"  실패: {item['error']}")
            continue
        for doc in item["documents"]:
            print(f"  - {doc['documentType']} (p.{doc['pageStart']}-{doc['pageEnd']})")
            for key, value in list(doc["keyValues"].items())[:5]:
                print(f"      {key}: {value}")
        for err in item["errors"]:
            print(f"  쪽 {err['page']} 실패: {err['error']}")

    print()
    print(f"결과를 {RESULTS_FILE} 에 저장했습니다.")


if __name__ == "__main__":
    main()

$ python review_folder.py ./shipping-docs
[처리] 001-invoice.pdf
[처리] 002-bl.pdf

■ 001-invoice.pdf
  - Commercial Invoice (p.1-2)
      Invoice No.: INV-2026-0412
      Amount: USD 48,200.00

■ 002-bl.pdf
  - Bill of Lading (p.1-1)
      B/L No.: HDMU0412xxxx

결과를 results.json 에 저장했습니다.

{
  "037": "1. 예, 전 직원에 대하여 성희롱 예방교육을 실시하고 있습니다.",
  "052": "3. 해당사항 없음 (법적 측정 대상이 아닌 업종 및 기업)"
}

{
  "fileName": "2025_실태표(작성본).xlsx",
  "documentId": "",
  "matches": [],
  "unmet": [],
  "candidateCount": 0,
  "capped": false,
  "summary": "실태표 양식 문서입니다. 증빙으로 판정하지 않으며, 실태표 값은 declared 필드로 보내십시오."
}

import csv
import io
import json
import sys
import time
from pathlib import Path

import requests

BASE_URL = "https://clip.floaterlab.com"
API_KEY = "YOUR_API_KEY"
EVAL_YEAR = 2025
RETRY_WAITS = [2, 4]  # 재시도 전 대기(초), 최대 2회
EXTS = {".pdf", ".png", ".jpg", ".jpeg", ".tif", ".tiff", ".webp",
        ".xlsx", ".xls", ".docx", ".doc", ".pptx", ".ppt", ".hwp", ".hwpx"}

# 지표번호 -> 문항 번호 (실태표 값을 보내는 문항)
DECLARED_ITEMS = {
    'E-005-001-001': '005',
    'E-005-002-001': '006',
    'E-005-003-001': '007',
    'E-006-002-001': '008',
    'E-007-002-001': '009',
    'E-007-003-001': '011',
    'E-008-001-001': '012',
    'E-008-001-002': '013',
    'E-008-002-001': '014',
    'E-009-003-001': '018',
    'S-002-002-001': '020',
    'S-002-002-002': '022',
    'S-002-005-003': '035',
    'S-002-007-001': '037',
    'S-002-007-002': '038',
    'S-002-007-003': '039',
    'S-002-007-005': '041',
    'S-002-007-008': '043',
    'S-003-003-002': '046',
    'S-004-003-001': '051',
    'S-005-001-001': '052',
    'S-005-002-001': '053',
    'S-005-003-001': '054',
    'S-006-002-001': '057',
    'S-006-005-001': '058',
    'S-010-001-001': '061',
    'G-001-001-002': '063',
    'SF-001-002-001': '071',
    'SF-001-002-002': '072',
    'SF-001-003-001': '073',
    'SF-002-001-001': '074',
    'SF-002-001-002': '075',
    'SF-002-001-003': '076',
    'SF-002-001-004': '077',
    'SF-002-002-001': '078',
    'SF-002-002-002': '079',
    'SF-003-002-002': '082',
    'SF-004-001-002': '085',
    'SF-005-002-001': '090',
    'SF-005-002-002': '091',
}


def read_csv(path):
    raw = Path(path).read_bytes()
    try:
        text = raw.decode("utf-8-sig")
    except UnicodeDecodeError:
        text = raw.decode("cp949")  # 한국어 Excel 이 저장한 CSV
    return list(csv.DictReader(io.StringIO(text)))


def to_declared(csv_path):
    """실태표 CSV(수동항목 양식)를 declared 객체로 바꾼다. 받아들이지 않은 행은 알려 준다."""
    declared = {}
    for row in read_csv(csv_path):
        code = (row.get("지표번호") or "").strip()
        no = DECLARED_ITEMS.get(code)
        if not no:
            continue
        num = (row.get("결과번호") or "").strip()
        text = (row.get("결과문구") or "").strip()
        if not num:
            continue  # 답하지 않은 문항
        if not num.isdigit() or int(num) < 1:
            print(f"  건너뜀: {code} 결과번호 {num!r} (숫자 하나가 아님)")
            continue
        n = int(num)
        declared[no] = f"{n}. {text}" if text else str(n)  # 결과문구가 없으면 번호만 보낸다
    return declared


def check_document(session, path, declared):
    """문서 1건을 /api/esg/check 로 보낸다. 5xx·네트워크 오류만 2회까지 다시 보낸다."""
    data = {"evalYear": str(EVAL_YEAR)}
    if declared:
        data["declared"] = json.dumps(declared, ensure_ascii=False)
    last = ""
    for attempt in range(len(RETRY_WAITS) + 1):
        try:
            with open(path, "rb") as fp:
                res = session.post(f"{BASE_URL}/api/esg/check",
                                   files={"file": (path.name, fp)},
                                   data=data, timeout=3600)
        except requests.RequestException as e:
            last = str(e)
        else:
            if res.status_code < 500:
                if not res.ok:  # 4xx 는 재시도하지 않는다
                    raise RuntimeError(f"HTTP {res.status_code} {res.text[:200]}")
                try:
                    return res.json()
                except ValueError:
                    raise RuntimeError("응답을 JSON 으로 읽지 못했습니다")
            last = f"HTTP {res.status_code}"
        if attempt < len(RETRY_WAITS):
            time.sleep(RETRY_WAITS[attempt])
    raise RuntimeError(last)


def group_by_item(results):
    """문서별 판정(matches)을 문항 번호 기준으로 모은다."""
    items = {}
    for res in results:
        for m in res.get("matches", []):
            item = items.setdefault(m["id"], {
                "id": m["id"], "group": m["group"], "question": m["question"], "documents": []})
            item["documents"].append({
                "fileName": res["fileName"],
                "levelText": m.get("levelText", ""),
                "reason": m.get("reason", ""),
            })
    return [items[k] for k in sorted(items)]


def main():
    if len(sys.argv) != 3:
        sys.exit("사용법: python esg_check.py 증빙폴더 실태표.csv")
    folder, csv_path = Path(sys.argv[1]), sys.argv[2]
    declared = to_declared(csv_path)
    print(f"실태표 반영: {len(declared)}개 문항")

    session = requests.Session()
    session.headers["x-clip-key"] = API_KEY
    files = sorted(p for p in folder.iterdir()
                   if p.suffix.lower() in EXTS and "실태표" not in p.name)

    results, failed = [], []
    for i, path in enumerate(files, 1):
        print(f"[{i}/{len(files)}] {path.name} 판정 중")
        try:
            results.append(check_document(session, path, declared))
        except RuntimeError as e:
            failed.append({"fileName": path.name, "error": str(e)})
            print(f"  실패: {e}")

    items = group_by_item(results)
    out = {
        "evalYear": EVAL_YEAR,
        "documents": [r["fileName"] for r in results],
        "failed": failed,
        "items": items,
    }
    Path("esg_result.json").write_text(json.dumps(out, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"판정 완료 {len(results)}건, 실패 {len(failed)}건")
    print(f"충족 근거가 있는 문항 {len(items)}개: {', '.join(i['id'] for i in items)}")
    print("결과 저장: esg_result.json")


if __name__ == "__main__":
    main()

실태표 반영: 12개 문항
[1/3] 2025_성희롱예방교육이수현황.xlsx 판정 중
[2/3] 2025_임금대장.xlsx 판정 중
[3/3] 2025_환경정책서.pdf 판정 중
판정 완료 3건, 실패 0건
충족 근거가 있는 문항 3개: 001, 033, 037
결과 저장: esg_result.json

import json
import sys
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

import requests

BASE_URL = "https://clip.floaterlab.com"
API_KEY = "YOUR_API_KEY"
URL = BASE_URL + "/api/construct/item-check"
MAX_WORKERS = 3  # 동시에 보내는 파일 수


def send(item, path):
    """파일 1건을 문항 1개에 대해 보낸다. 5xx 와 네트워크 오류는 2회까지 재시도한다."""
    for attempt in range(3):
        try:
            with open(path, "rb") as f:
                res = requests.post(
                    URL,
                    headers={"x-clip-key": API_KEY},
                    files={"file": (path.name, f)},
                    data={"item": item},
                    timeout=1200,
                )
        except requests.RequestException as e:
            if attempt == 2:
                return {"error": str(e)}
            time.sleep(2 * (attempt + 1))
            continue
        if res.status_code >= 500 and attempt < 2:
            time.sleep(2 * (attempt + 1))
            continue
        if not res.ok:
            return {"error": f"HTTP {res.status_code}: {res.text[:200]}"}
        return res.json()
    return {"error": "재시도 횟수를 넘었습니다"}


def judge(item, path):
    body = send(item, path)
    if "error" in body:
        return {"file": path.name, "status": "오류", "reason": body["error"], "pages": []}
    match = body.get("match") or {}
    reason = match.get("reason", "")
    pages = sorted({s["page"] for s in match.get("sources", [])})
    if body.get("passed"):
        status = "충족"
    elif not reason:
        status = "확인필요"  # 판정 응답이 비어 있어 다시 확인해야 한다
    else:
        status = "미충족"
    return {"file": path.name, "status": status, "reason": reason, "pages": pages}


def is_item(name):
    parts = name.split("-")
    return len(parts) == 2 and all(p.isdigit() for p in parts)


def collect(root):
    pairs = []
    for folder in sorted(root.iterdir()):
        if not folder.is_dir():
            continue
        if not is_item(folder.name):
            print(f"건너뜀: {folder.name} (문항번호 폴더가 아닙니다)")
            continue
        for path in sorted(folder.iterdir()):
            if path.is_file():
                pairs.append((folder.name, path))
    return pairs


def item_key(item):
    return tuple(int(x) for x in item.split("-"))


def main():
    root = Path(sys.argv[1] if len(sys.argv) > 1 else "서류")
    out = Path(sys.argv[2] if len(sys.argv) > 2 else "문항별결과.json")
    pairs = collect(root)
    if not pairs:
        sys.exit("점검할 서류가 없습니다. 문항번호 폴더 아래에 파일을 넣어 주세요.")

    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
        judged = list(pool.map(lambda p: (p[0], judge(*p)), pairs))

    files_by_item = {}
    for item, rec in judged:
        files_by_item.setdefault(item, []).append(rec)

    summary = {}
    for item, recs in files_by_item.items():
        statuses = [r["status"] for r in recs]
        if "충족" in statuses:
            final = "충족"  # 한 문항에 서류가 여러 개면 하나라도 충족이면 충족
        elif "확인필요" in statuses:
            final = "확인필요"
        elif "미충족" in statuses:
            final = "미충족"
        else:
            final = "오류"
        summary[item] = {"status": final, "files": recs}

    items = sorted(summary, key=item_key)
    print(f"문항별 점검 결과 (서류 {len(pairs)}건)")
    for item in items:
        s = summary[item]
        names = ", ".join(r["file"] for r in s["files"])
        print(f"  {item:<6} {s['status']:<6} {names}")
    counts = {k: sum(1 for s in summary.values() if s["status"] == k) for k in ("충족", "미충족", "확인필요", "오류")}
    print(f"충족 {counts['충족']}개 · 미충족 {counts['미충족']}개 · 확인필요 {counts['확인필요']}개 · 오류 {counts['오류']}개")

    out.write_text(json.dumps({i: summary[i] for i in items}, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"저장: {out}")


if __name__ == "__main__":
    main()

문항별 점검 결과 (서류 5건)
  1-1    충족     안전보건방침.pdf
  2-1    미충족    성과측정표.xlsx
  10-1   충족     선임계.pdf, 관리자선임보완.hwp
  10-2   확인필요   보건관리자선임.pdf
충족 2개 · 미충족 1개 · 확인필요 1개 · 오류 0개
저장: 문항별결과.json

import time
import requests

BASE = 'https://clip.floaterlab.com'
HEADERS = {'x-clip-key': 'YOUR_API_KEY'}


def upload(path):
    with open(path, 'rb') as f:
        r = requests.post(BASE + '/api/extract/upload', headers=HEADERS, files={'file': f}, timeout=600)
    r.raise_for_status()
    return r.json()['fileId']


def run_page(file_id, page):
    r = requests.post(
        BASE + '/api/extract/process',
        headers=HEADERS,
        json={'fileId': file_id, 'page': page, 'mode': 'full'},
        timeout=30,
    )
    r.raise_for_status()
    job_id = r.json()['jobId']

    deadline = time.time() + 60 * 60
    while time.time() < deadline:
        time.sleep(3)
        body = requests.get(BASE + '/api/extract/process', params={'job': job_id}, timeout=30).json()
        if body.get('status') == 'done':
            return body['result']
        if body.get('status') in ('error', 'unknown'):
            raise RuntimeError(body.get('error', 'job not found'))
    raise TimeoutError('page %d did not finish in time' % page)


file_id = upload('document.pdf')
page_count = 3  # 직접 계산한 페이지 수
results = [run_page(file_id, p) for p in range(1, page_count + 1)]

{
  "page": 1,
  "type": "Letter of Credit",
  "keys": [
    {
      "key": "L/C No.",
      "value": "M079Y2606NU00178",
      "reference": {
        "1": [
          23,
          64
        ]
      },
      "confidence": 0.9
    },
    {
      "key": "Issue Date",
      "value": "260617",
      "processed": "2026-06-17",
      "reference": {
        "1": [
          69
        ]
      },
      "confidence": 0.93
    },
    {
      "key": "Amount",
      "value": "USD 80,816.94",
      "processed": {
        "currency": "USD",
        "amount": 80816.94
      },
      "reference": {
        "1": [
          116,
          117
        ]
      },
      "confidence": 0.93
    }
  ],
  "words": {
    "23": [
      0.1558,
      0.1044,
      0.2906,
      0.1141
    ],
    "64": [
      0.0969,
      0.3406,
      0.2333,
      0.3497
    ],
    "69": [
      0.096,
      0.3719,
      0.1493,
      0.3811
    ],
    "116": [
      0.2018,
      0.5898,
      0.2413,
      0.6018
    ],
    "117": [
      0.2494,
      0.5898,
      0.3398,
      0.6018
    ]
  }
}

def confidence_band(score):
    if not isinstance(score, (int, float)):
        return None  # 구간 없음: 집계에서 제외
    if score >= 0.85:
        return '양호'
    if score >= 0.70:
        return '보통'
    if score >= 0.50:
        return '주의'
    return '위험'


def needs_review_count(keys):
    return sum(1 for k in keys if confidence_band(k.get('confidence')) in ('주의', '위험'))

import csv
import glob
import json
import os
import sys
import time

import requests

BASE = 'https://clip.floaterlab.com'
HEADERS = {'x-clip-key': 'YOUR_API_KEY'}
EXTENSIONS = ('.pdf', '.png', '.jpg', '.jpeg', '.webp', '.tif', '.tiff')
MAX_RETRIES = 2  # 결과 조회(GET)가 5xx 응답이나 연결 오류를 받으면 최대 2번 다시 보냅니다
POLL_INTERVAL = 3  # 결과 조회 간격(초)
POLL_TIMEOUT = 30 * 60  # 쪽 하나를 기다리는 최대 시간(초)


def page_count(path):
    """PDF 는 쪽수를, TIFF 는 장수를 셉니다. 그 밖의 이미지는 1쪽입니다."""
    lower = path.lower()
    if lower.endswith('.pdf'):
        from pypdf import PdfReader
        return len(PdfReader(path).pages)
    if lower.endswith(('.tif', '.tiff')):
        from PIL import Image
        with Image.open(path) as im:
            return getattr(im, 'n_frames', 1)
    return 1


def call(method, path, retry=False, timeout=60, **kwargs):
    """요청을 보내고 (상태 코드, 본문) 을 돌려줍니다. retry=True 일 때만 5xx 나 연결 오류를 다시 보냅니다."""
    attempts = MAX_RETRIES + 1 if retry else 1
    for attempt in range(attempts):
        try:
            r = requests.request(method, BASE + path, headers=HEADERS, timeout=timeout, **kwargs)
            if r.status_code < 500 or attempt == attempts - 1:
                try:
                    body = r.json()
                except ValueError:
                    body = {}
                return r.status_code, body
        except requests.ConnectionError as e:
            if attempt == attempts - 1:
                raise RuntimeError(f'{method} {path} 연결 오류: {e}')
        time.sleep(5 * (attempt + 1))


def send(method, path, timeout=60, **kwargs):
    """업로드·처리 요청(POST)용. 다시 보내지 않습니다 — 같은 파일이 또 올라가거나 작업이 두 번 돌 수 있습니다.
    4xx·5xx 는 서버가 준 error 문구를 그대로 오류 메시지로 씁니다."""
    status, body = call(method, path, retry=False, timeout=timeout, **kwargs)
    if status >= 400:
        raise RuntimeError(f"{method} {path} 실패 (HTTP {status}): {body.get('error') or body}")
    return body


def upload(path):
    with open(path, 'rb') as f:
        data = f.read()
    body = send('POST', '/api/extract/upload', timeout=600, files={'file': (os.path.basename(path), data)})
    return body['fileId']


def submit_page(file_id, page):
    body = send('POST', '/api/extract/process', json={'fileId': file_id, 'page': page, 'mode': 'full'}, timeout=30)
    return body['jobId']


def wait_result(job_id):
    """작업이 끝날 때까지 조회합니다. done 과 error 는 한 번 읽으면 서버에서 지워지므로 받은 값을 바로 씁니다."""
    deadline = time.time() + POLL_TIMEOUT
    while time.time() < deadline:
        time.sleep(POLL_INTERVAL)
        try:
            _, body = call('GET', '/api/extract/process', retry=True, timeout=30, params={'job': job_id})
        except RuntimeError:
            continue  # 연결 오류가 계속되면 다음 조회에서 다시 봅니다
        state = body.get('status')
        if state == 'done':
            return body['result']
        if state == 'error':
            raise RuntimeError(body.get('error') or '처리 중 오류가 났습니다')
        if state == 'unknown':
            raise RuntimeError('결과를 찾을 수 없습니다(서버가 재시작됐을 수 있습니다). 이 쪽의 처리 요청을 다시 보내세요.')
    raise TimeoutError('제한 시간 안에 끝나지 않았습니다')


def extract_file(path):
    """파일의 쪽마다 분류·추출을 요청합니다. 쪽 하나가 실패해도 나머지 쪽은 계속 처리합니다."""
    file_id = upload(path)
    pages, errors = [], []
    for page in range(1, page_count(path) + 1):
        try:
            pages.append(wait_result(submit_page(file_id, page)))
        except Exception as e:
            errors.append({'page': page, 'error': str(e)})
    return pages, errors


def group_documents(pages):
    """연속된 같은 유형의 쪽을 한 문서로 묶습니다. 서버는 문서 경계를 알려주지 않습니다."""
    docs = []
    for pr in sorted(pages, key=lambda x: x['page']):
        dtype = pr.get('type') or 'document'  # 빈 유형은 미분류(document)로 묶습니다
        last = docs[-1] if docs else None
        if last and last['documentType'] == dtype and last['pageEnd'] == pr['page'] - 1:
            last['pageEnd'] = pr['page']
            last['keys'].extend(pr.get('keys', []))
        else:
            docs.append({'documentType': dtype, 'pageStart': pr['page'], 'pageEnd': pr['page'],
                         'keys': list(pr.get('keys', []))})
    return docs


def band(score):
    """신뢰도 점수를 네 구간으로 나눕니다. 서버 값이 아니라 이 스크립트가 매기는 구간입니다."""
    if not isinstance(score, (int, float)):
        return ''
    if score >= 0.85:
        return '양호'
    if score >= 0.70:
        return '보통'
    if score >= 0.50:
        return '주의'
    return '위험'


def main():
    if len(sys.argv) != 2:
        sys.exit('사용법: python extract_batch.py 서류폴더')
    paths = sorted(p for p in glob.glob(os.path.join(sys.argv[1], '*')) if p.lower().endswith(EXTENSIONS))
    if not paths:
        sys.exit('처리할 파일이 없습니다')

    report, rows = [], []
    for path in paths:
        name = os.path.basename(path)
        try:
            pages, errors = extract_file(path)
        except Exception as e:
            report.append({'fileName': name, 'documents': [], 'errors': [{'page': 0, 'error': str(e)}]})
            continue
        docs = group_documents(pages)
        report.append({'fileName': name, 'documents': docs, 'errors': errors})
        for i, doc in enumerate(docs, 1):
            for k in doc['keys'] or [None]:
                score = k.get('confidence') if k else None
                rows.append({
                    '파일명': name,
                    '문서 번호': i,
                    '문서 유형': doc['documentType'],
                    '시작 쪽': doc['pageStart'],
                    '끝 쪽': doc['pageEnd'],
                    '키': k['key'] if k else '',
                    '값': k['value'] if k else '',
                    '신뢰도': '' if score is None else score,
                    '구간': band(score),
                    '확인 필요': 'Y' if band(score) in ('주의', '위험') else '',
                })

    with open('extract-result.json', 'w', encoding='utf-8') as f:
        json.dump(report, f, ensure_ascii=False, indent=2)
    if rows:
        with open('extract-keys.csv', 'w', newline='', encoding='utf-8-sig') as f:
            writer = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
            writer.writeheader()
            writer.writerows(rows)

    docs_total = sum(len(r['documents']) for r in report)
    review = sum(1 for row in rows if row['확인 필요'])
    failed = sum(len(r['errors']) for r in report)
    print(f'파일 {len(report)}개 · 문서 {docs_total}건 · CSV 행 {len(rows)}건')
    for r in report:
        print(f"- {r['fileName']}")
        for d in r['documents']:
            pages = f"p.{d['pageStart']}" + (f"-{d['pageEnd']}" if d['pageEnd'] > d['pageStart'] else '')
            print(f"    {d['documentType']} ({pages}) · 키 {len(d['keys'])}건")
        for e in r['errors']:
            where = f"p.{e['page']}" if e['page'] else '파일'
            print(f"    오류 {where}: {e['error']}")
    print(f'확인 필요 {review}건 · 실패 {failed}건')
    print('저장: extract-result.json, extract-keys.csv')


if __name__ == '__main__':
    main()

파일 1개 · 문서 3건 · CSV 행 12건
- 거래처_제출서류.pdf
    Letter of Credit (p.1-2) · 키 6건
    Bill of Lading (p.3) · 키 5건
    document (p.4) · 키 0건
    오류 p.5: Processing failed: timeout
확인 필요 2건 · 실패 1건
저장: extract-result.json, extract-keys.csv