업무상 질병 판정서에서 항목 추출

extract_disease_cases_detail_v4.rb

require 'csv'
require 'json'
require 'net/http'
require 'set'
require 'time'

OLLAMA_HOST  = ENV.fetch('OLLAMA_HOST',  'localhost')
OLLAMA_PORT  = ENV.fetch('OLLAMA_PORT',  11434).to_i
OLLAMA_MODEL = ENV.fetch('OLLAMA_MODEL', 'gemma4:latest')
OLLAMA_PATH  = '/api/chat'

MAX_RETRIES  = ENV.fetch('MAX_RETRIES', 3).to_i
COOLDOWN     = ENV.fetch('COOLDOWN', 0.2).to_f
LIMIT        = ENV.fetch('LIMIT', 0).to_i

OUTPUT_PATH  = ENV.fetch('OUTPUT_PATH', 'test_v4.csv')
ERROR_LOG    = ENV.fetch('ERROR_LOG',   'test_errors_v4.log')

MAIN_HEADERS = %w[
  case_no
  성별
  발병_당시_연령
  신장
  체중
  BMI_지수
  우세손
  흡연_여부
  총_흡연량
  음주_여부
  주당_음주_횟수
  기저질환_보유
  과거_산재_처리_이력
  직종명
  고용_형태
  근무_형태
  현_직종_총_종사기간
  1주_평균_근무시간
  1일_평균_근무시간
  부담_신체_부위
  주요_부적절한_자세
  중량물_취급_여부
  취급_물품_최대_무게
  1일_취급_총_누적_중량
  기타_유해요인_노출
  최초_신청_상병명
  최종_의학적_확인_상병명
  의학영상_MRI_등_주요_소견
  상병명_변경_인정_여부
  업무관련성_평가
  업무부담_가중요인_노출
  최종_산재_인정_여부
  판단_주요_근거
]

COLUMN_PROMPTS = {
  'medical_records' => {
    'fields' => %w[성별 발병_당시_연령 신장 체중 BMI_지수 우세손 흡연_여부 총_흡연량 음주_여부 주당_음주_횟수 기저질환_보유 과거_산재_처리_이력 최초_신청_상병명 최종_의학적_확인_상병명 의학영상_MRI_등_주요_소견 상병명_변경_인정_여부],
    'instruction' => '환자 진료기록. 환자의 기본 신체 정보, 생활습관, 기저질환, 상병 및 진단 정보를 추출.',
    'schema' => <<~SCHEMA,
      - "성별": "남" 또는 "여"
      - "발병_당시_연령": 숫자 (세)
      - "신장": 숫자 (cm)
      - "체중": 숫자 (kg)
      - "BMI_지수": 숫자
      - "우세손": "오른손", "왼손", "양손" 중 하나
      - "흡연_여부": "현재_흡연", "과거_흡연", "비흡연" 중 하나
      - "총_흡연량": 숫자 (갑년, 없으면 null)
      - "음주_여부": "음주" 또는 "비음주"
      - "주당_음주_횟수": 숫자 (없으면 null)
      - "기저질환_보유": 질병 이름 배열 (없으면 null)
      - "과거_산재_처리_이력": "Y" 또는 "N"
      - "최초_신청_상병명": 질병 이름 배열 (없으면 null)
      - "최종_의학적_확인_상병명": 최초와 동일한 목록 (없으면 null)
      - "의학영상_MRI_등_주요_소견": 의학적 의견 배열(없으면 null)
      - "상병명_변경_인정_여부": "Y" 또는 "N"
    SCHEMA
    'example' => '{"성별": "남", "발병_당시_연령": 52, "신장": 170, "체중": 75, "BMI_지수": 25.95, "우세손": "오른손", "흡연_여부": "비흡연", "총_흡연량": null, "음주_여부": "비음주", "주당_음주_횟수": null, "기저질환_보유": ["고혈압"], "과거_산재_처리_이력": "N", "최초_신청_상병명": ["뇌출혈"], "최종_의학적_확인_상병명": ["뇌출혈"], "의학영상_MRI_등_주요_소견": ["특이소견_없음"], "상병명_변경_인정_여부": "N"}'
  },
  'recognized_facts' => {
    'fields' => %w[직종명 고용_형태 근무_형태 현_직종_총_종사기간 1주_평균_근무시간 1일_평균_근무시간 부담_신체_부위 주요_부적절한_자세 중량물_취급_여부 취급_물품_최대_무게 1일_취급_총_누적_중량 기타_유해요인_노출],
    'instruction' => '산업재해 위원회 인정한 사실. 근로자의 직업력, 근무 조건, 작업 내용, 신체부담 및 유해요인 정보를 추출.',
    'schema' => <<~SCHEMA,
      - "직종명": 직업 종류 이름
      - "고용_형태": 일하는 형태
      - "근무_형태": 교대 또는 근무 시간 등에 대한 형태
      - "현_직종_총_종사기간": 숫자 (개월)
      - "1주_평균_근무시간": 숫자 (시간)
      - "1일_평균_근무시간": 숫자 (시간)
      - "부담_신체_부위": 신체 부위 배열
      - "주요_부적절한_자세": 자세 배열
      - "중량물_취급_여부": "Y" 또는 "N"
      - "취급_물품_최대_무게": 숫자 (kg)
      - "1일_취급_총_누적_중량": 숫자 (kg)
      - "기타_유해요인_노출": 그 밖의 원인이 되는 상황 배열 (없으면 null)
    SCHEMA
    'example' => '{"직종명": "환경미화원", "고용_형태": "일용직", "근무_형태": "고정주간근무", "현_직종_총_종사기간": 36, "1주_평균_근무시간": 40, "1일_평균_근무시간": 8, "부담_신체_부위": null, "주요_부적절한_자세": null, "중량물_취급_여부": "Y", "취급_물품_최대_무게": 20, "1일_취급_총_누적_중량": 500, "기타_유해요인_노출": null}'
  },
  'committee_decision' => {
    'fields' => %w[업무관련성_평가 업무부담_가중요인_노출 최종_산재_인정_여부 판단_주요_근거],
    'instruction' => '산업재해 보상 위원회의 판정 결정문. 업무관련성 평가, 최종 인정 여부, 판단 근거를 추출.',
    'schema' => <<~SCHEMA,
      - "업무관련성_평가": "매우_높음", "높음", "보통", "낮음", "매우_낮음", "미흡" 중 하나
      - "업무부담_가중요인_노출": 더 부담이 되는 원인 배열 (없으면 null)
      - "최종_산재_인정_여부": "인정", "불인정", "일부인정", "변경인정" 중 하나
      - "판단_주요_근거": 근거 배열 (없으면 null)
    SCHEMA
    'example' => '{"업무관련성_평가": "낮음", "업무부담_가중요인_노출": ["해당없음"], "최종_산재_인정_여부": "불인정", "판단_주요_근거": ["개인_기저질환_악화", "객관적_직업력_부족"]}'
  }
}

NORMALIZATION_RULES = {
  '과거_산재_처리_이력' => { /있음|yes|y/i => 'Y', /없음|no|n/i => 'N' },
  '중량물_취급_여부'   => { /있음|yes|y/i => 'Y', /없음|no|n/i => 'N' },
  '업무관련성_평가'   => { /매우_?높음/ => '매우_높음', /높음/ => '높음', /보통/ => '보통', /매우_?낮음/ => '매우_낮음', /낮음/ => '낮음', /미흡/ => '미흡' },
  '최종_산재_인정_여부' => { /불인정|불승인|미인정|인정.*되지/ => '불인정', /일부인정|일부/ => '일부인정', /인정|승인/ => '인정', /변경/ => '변경인정' }
}

class Extractor
  def run
    input_file_name = "disease_cases_details.csv"
    output_file_name = "test_v4.csv"
    check_ollama!
    ensure_headers(output_file_name, MAIN_HEADERS)

    done  = completed_case_nos(output_file_name)
    total = [ File.foreach(input_file_name).count - 1, 1 ].max
    warn "Resuming: #{done.size}/#{total} already done." if done.size > 0

    total_done = done.size
    processed = 0

    CSV.foreach(input_file_name, headers: true) do |row|
      next if done.include?(row['case_no'])
      break if LIMIT > 0 && processed >= LIMIT

      case_no = row['case_no']
      data = {}

      http = Net::HTTP.new("localhost", 11434)
      http.read_timeout = 300
      http.open_timeout = 30
      http.start

      %w[medical_records recognized_facts committee_decision].each do |col|
        text = row[col]
        next if text.nil? || text.strip.empty?

        result = extract_column(text, col, case_no, http)
        data.merge!(result)
        sleep(COOLDOWN)
      end

      CSV.open(output_file_name, 'a') do |csv|
        csv << MAIN_HEADERS.map do |h|
          if h == 'case_no'
            case_no
          else
            case data[h]
            when nil then ''
            when Array then data[h].join('|')
            else data[h].to_s
            end
          end
        end
      end

      processed += 1
      pct = ((done.size + processed) * 100.0 / total).round(1)
      warn "[#{Time.now.strftime("%Y-%m-%d %H:%M:%S")}] #{processed} / #{LIMIT} (#{(processed / LIMIT.to_f * 100.0).round(1)}% #{pct}% total)"
      http&.finish if http&.started?
      sleep(100)
    ensure
      http&.finish if http&.started?
    end

    total_done += processed
    warn "Done. #{processed} new cases. Total output: #{total_done}."
    warn "Errors → #{ERROR_LOG}" if File.exist?(ERROR_LOG) && File.size(ERROR_LOG) > 0
  end

  private

  def check_ollama!
    http = Net::HTTP.new(OLLAMA_HOST, OLLAMA_PORT)
    http.open_timeout = 5
    http.get('/api/tags')
    warn "Ollama OK (#{OLLAMA_HOST}:#{OLLAMA_PORT}, model: #{OLLAMA_MODEL})"
  rescue => e
    abort "Ollama not reachable at #{OLLAMA_HOST}:#{OLLAMA_PORT} — #{e.message}\nStart with: ollama serve"
  end

  def call_api(prompt, http)
    body = JSON.dump(
      model:    OLLAMA_MODEL,
      messages: [ { role: 'user', content: prompt } ],
      stream:   false
    )
    req      = Net::HTTP::Post.new(OLLAMA_PATH, 'Content-Type' => 'application/json')
    req.body = body
    res      = http.request(req)

    unless res.is_a?(Net::HTTPSuccess)
      raise "Ollama HTTP #{res.code}: #{res.body[0, 200]}"
    end

    parsed = JSON.parse(res.body)
    parsed.dig('message', 'content') ||
      raise("Unexpected response shape: #{res.body[0, 200]}")
  end

  def extract_column(text, col, case_no, http)
    col_prompt= COLUMN_PROMPTS[col]
    prompt = <<~PROMPT
      #{col_prompt['instruction']}

      [규칙]
      1. 텍스트에서 명시적으로 확인되지 않거나 추정할 수 없는 경우, 반드시 null 사용.
      2. 배열 타입은 JSON 배열 []로 출력. 값이 없으면 null 사용.
      3. 아래에 명시된 필드만 추출. 다른 필드는 추가 금지.
      4. JSON 형식으로만 응답하. 추가 설명 없이 순수 JSON만 출력.

      [추출할 필드]
      #{col_prompt['schema']}
      [JSON 예시]
      #{col_prompt['example']}

      [텍스트]
      #{text}
    PROMPT

    retries = 0

    begin
      response = call_api(prompt, http)
      data = extract_json(response, case_no)
      if data
        fields = COLUMN_PROMPTS[col]['fields']
        data   = data.select { |k, _| fields.include?(k) }
        data   = normalize(data, col)
        return data
      end
    rescue Net::ReadTimeout, Net::OpenTimeout => e
      warn "Timeout #{case_no}/#{col} (try #{retries}): #{e.message}"
      log_error(case_no, 'TIMEOUT', "#{col}: #{e.message}")
      sleep(15 * (retries + 1)) if (retries += 1) < MAX_RETRIES
      retry if retries < MAX_RETRIES
    rescue => e
      warn "Error #{case_no}/#{col} (try #{retries}): #{e.message}"
      log_error(case_no, 'ERROR', "#{col}: #{e.message}")
      sleep(5) if (retries += 1) < MAX_RETRIES
      retry if retries < MAX_RETRIES
    end

    {}
  end

  def extract_json(response, case_no = nil)
    json_str = if response =~ /```(?:json)?\s*(.*?)\s*```/m
      $1.strip
    else
      s = response.index('{')
      e = response.rindex('}')
      return nil if s.nil? || e.nil?
      response[s..e]
    end
    JSON.parse(json_str)
  rescue => e
    warn "JSON parse error (#{case_no}): #{e.message}"
    log_error(case_no, 'JSON_PARSE_ERROR', e.message) if case_no
    nil
  end

  def normalize(data, col)
    COLUMN_PROMPTS[col]['fields'].each do |field|
      next unless data.key?(field)
      val = data[field]
      next if val.nil?

      if (rules = NORMALIZATION_RULES[field])
        if val.is_a?(String)
          matched = rules.find { |re, _| val.match?(re) }
          data[field] = matched ? matched[1] : nil
        elsif val.is_a?(Array)
          data[field] = val.map do |item|
            next item unless item.is_a?(String)
            matched = rules.find { |re, _| item.match?(re) }
            matched ? matched[1] : item
          end.uniq
        end
      end

      if %w[현_직종_총_종사기간 1주_평균_근무시간 1일_평균_근무시간 취급_물품_최대_무게 1일_취급_총_누적_중량 총_흡연량 주당_음주_횟수 발병_당시_연령 신장 체중 BMI_지수].include?(field)
        if val.is_a?(String)
          data[field] = val.match?(/^\d+(\.\d+)?$/) ? (val.include?('.') ? val.to_f : val.to_i) : nil
        end
      end
    end
    data
  end

  def completed_case_nos(filename)
    return Set.new unless File.exist?(filename) && File.size(filename) > 0
    completed = Set.new
    CSV.foreach(filename, headers: true) { completed << it["case_no"] if it["case_no"] && it["case_no"].length > 0 }
    completed
  end

  def ensure_headers(path, headers)
    return if File.exist?(path) && File.size(path) > 0
    CSV.open(path, 'w') { |csv| csv << headers }
  end

  def log_error(case_no, type, msg)
    File.open(ERROR_LOG, 'a') { |f| f.puts "#{Time.now.iso8601}\t#{case_no}\t#{type}\t#{msg.gsub(/[\t\n]/, ' ')}" }
  end

  def count_total_cases(path)
    n = File.foreach(path).count - 1
    n < 1 ? 1 : n
  end
end

Extractor.new.run