Snowflake ML JobsでPDFを一括画像化し、Cortex マルチモーダルで文字起こしをする

CREATE OR REPLACE STAGE test_stage
  ENCRYPTION = (TYPE = 'snowflake_sse');


CREATE OR REPLACE STAGE payload_stage
  ENCRYPTION = (TYPE = 'snowflake_sse');

from snowflake.ml.jobs import remote
import fitz, os, traceback

compute_pool = "MY_COMPUTE_POOL"

@remote(
    compute_pool,
    stage_name="payload_stage",          
    pip_requirements=['PyMuPDF'],        
    external_access_integrations=['PYPI_ACCESS_INTEGRATION'])
def convert_pdfs_to_images(stage_name, file_name, temp_input="tmp/input/", temp_output="tmp/output/"):
    """
    指定されたPDFファイルを画像に変換し、出力ディレクトリに保存します。

    Parameters:
    - stage_name (str): ステージの名前（例: '@pdf/'）
    - file_list (list): 変換するPDFファイル名のリスト
    - temp_input (str): 入力ファイルを保存する一時ディレクトリのパス
    - temp_output (str): 出力画像を保存する一時ディレクトリのパス

    Returns:
    - dict: 処理結果を含む辞書。例:
        {
            'success': True,
            'output_files': ['CreditConsumptionTable_page_1.png', 'Invoice2025_page_1.png'],
            'errors': []
        }
    """
    
    session = get_active_session()

    
    os.makedirs(temp_input, exist_ok=True)
    os.makedirs(temp_output, exist_ok=True)
    file_list = [file_name]
    output_files = []
    errors = []

    for file_name in file_list:
        try:
            
            session.file.get(os.path.join(stage_name, file_name), temp_input)

            
            base_name = os.path.splitext(file_name)[0]

            
            doc = fitz.open(os.path.join(temp_input, file_name))
            for i, page in enumerate(doc):
                pix = page.get_pixmap()
                output_filename = f"{base_name}-page_{i + 1}.png"
                output_path = os.path.join(temp_output, output_filename)
                pix.save(output_path)
                session.file.put(output_path, os.path.join(stage_name, 'output_image'), auto_compress=False)
                output_files.append(output_filename)

        except Exception as e:
            error_message = f"エラーが発生しました: {file_name}\n{traceback.format_exc()}"
            print(error_message)
            errors.append({'file': file_name, 'error': str(e)})

    success = len(errors) == 0
    return {
        'success': success,
        'output_files': output_files,
        'errors': errors
    }

stage = '@test_stage'
df = session.sql(
    "SELECT * FROM directory(@test_stage) WHERE relative_path LIKE ('%.pdf')"
).to_pandas()

for f in df['RELATIVE_PATH']:
    job = convert_pdfs_to_images(stage, f)

job.id 
job.status 
job.get_logs() 

from snowflake.ml.jobs import list_jobs
print(list_jobs())

SELECT SNOWFLAKE.CORTEX.COMPLETE(
  'pixtral-large',
  '入力された画像内にあるテキスト情報を、JSON形式で抽出してください',
  TO_FILE('@test_stage', 'consumptiontable-page_1.png')
) AS result;

Table 1(b) Snowflake Credit Table for Virtual Warehouse Services, Snowpark Optimized Warehouses

| Resource Constraints | XS | S | M | L | Credits Per Hour | XL | 2XL | 3XL | 4XL | 5XL | 6XL |
|----------------------|----|---|---|---|------------------|----|-----|-----|-----|-----|-----|
| MEMORY: 1x, vCPU: 1 | 1.00 | 2.00 | 4.00 | 8.00 | 16.00 | 32.00 | 64.00 | 128.00 | 256.00 | - | - |
| MEMORY: 16X, vCPU: 1 | 1.00 | 2.00 | 4.00 | 8.00 | 17.60 | 35.20 | 70.40 | 140.80 | 281.60 | 384.00 | 768.00 |
| MEMORY: 16X, vCPU: 2 | - | - | 6.00 | 12.00 | 24.00 | 48.00 | 96.00 | 192.00 | 384.00 | - | - |
| MEMORY: 64X, vCPU: 8 | - | - | 6.25 | 12.50 | 25.00 | 50.00 | 100.00 | 200.00 | - | - | - |
| MEMORY: 64X, vCPU: 16 | - | - | - | 16.00 | 32.00 | 64.00 | 128.00 | 256.00 | - | - | - |

WITH pages AS (
  SELECT relative_path
  FROM directory(@test_stage)
  WHERE relative_path ILIKE '%-page_%.png'
)
SELECT relative_path,
       SNOWFLAKE.CORTEX.COMPLETE(
          'pixtral-large',
          '入力された画像内にあるテキスト情報を、JSON形式で抽出してください',
          TO_FILE('@test_stage', relative_path)
       ) AS result
FROM pages;

CREATE OR REPLACE TABLE extract_text_raw (
  relative_path STRING,
  extracted_json VARIANT,
  processed_at TIMESTAMP_TZ DEFAULT CURRENT_TIMESTAMP()
);

INSERT INTO extract_text_raw
WITH pages AS (
  SELECT relative_path
  FROM directory(@test_stage)
  WHERE relative_path ILIKE '%-page_%.png'
)
SELECT relative_path,
       SNOWFLAKE.CORTEX.COMPLETE(
          'pixtral-large',
          '入力された画像内にあるテキスト情報を、JSON形式で抽出してください',
          TO_FILE('@test_stage', relative_path)
       ) AS result
FROM pages;