【Dify×Salesforce×Claude】PDF転記作業を卒業！PDF解析からSalesforce登録まで自動化してみた

import json

def main(name: str, selfPromotion: str, skill: str, workExperience: str) -> dict:
    inner_dict = {
        "name": name,
        "selfPromotion": selfPromotion,
        "skill": skill,
        "workExperience": workExperience
    }
    return {
        "result": json.dumps(inner_dict, ensure_ascii=False)
    }

import requests
import json
import os

# --- メイン関数 ---
def main(name: str, selfPromotion: str, skill: str, workExperience: str, CLIENT_ID: str, CLIENT_SECRET: str, TOKEN_URL: str) -> dict:

    sobject_api_name = "Lead" # SObject名を指定

    last_name_val = name
    company_val = name 
    selfPromotion_val = selfPromotion
    status_val = "Open - Not Contacted"
    skill_val = skill
    workExperience_val = workExperience

    # Salesforceアクセストークン取得時に、main関数の引数として受け取った接続情報を使用
    access_token, instance_url, token_error = _get_salesforce_access_token(
        CLIENT_ID, 
        CLIENT_SECRET, 
        TOKEN_URL
    )
    if token_error:
        current_result = {
            "status": "error",
            "message": f"Salesforceアクセストークン取得失敗: {token_error}",
            "salesforce_response": None
        }
        return {"result": json.dumps(current_result, ensure_ascii=False)}

    if not access_token or not instance_url: # instance_urlもチェック
        current_result = {
            "status": "error",
            "message": "SalesforceアクセストークンまたはインスタンスURLが取得できませんでした。",
            "salesforce_response": None
        }
        return {"result": json.dumps(current_result, ensure_ascii=False)}
        
    record_payload = {
        "LastName": last_name_val,
        "Company": company_val,
        "SelfPromotion__c": selfPromotion_val,
        "Skill__c": skill_val,
        "WorkExperience__c": workExperience_val,
    }
    if sobject_api_name == "Lead":
        record_payload["Status"] = status_val
    
    sf_response, creation_error = _create_salesforce_record(
        access_token, instance_url, sobject_api_name, record_payload
    )

    current_result_dict = {}
    if creation_error:
        current_result_dict = {
            "status": "error",
            "message": f"Salesforceへのレコード作成失敗 ({sobject_api_name}): {creation_error}",
            "salesforce_response": sf_response
        }
    elif sf_response and sf_response.get("success"):
        record_id = sf_response.get("id")
        record_url = None
        if record_id:
            # ユーザー指定のURL形式で構築
            # 注意: ドメイン部分は固定です。instance_url から動的に取得する方がより堅牢ですが、
            #       ユーザーの指定に合わせています。
            record_url = f"https://cunning-otter-muyevc-dev-ed.trailblaze.lightning.force.com/lightning/r/{sobject_api_name}/{record_id}/view"
        
        current_result_dict = {
            "status": "success",
            "message": f"Salesforceにレコードが正常に作成されました (SObject: {sobject_api_name})。",
            "salesforce_response": sf_response,
            "record_url": record_url # 生成したURLを追加
        }
    else:
        current_result_dict = {
            "status": "error",
            "message": f"Salesforceでのレコード作成処理中にエラーが発生しました (SObject: {sobject_api_name})。",
            "salesforce_response": sf_response if sf_response else {"errors": ["Unknown error during record creation."]}
        }
    
    return {"result": json.dumps(current_result_dict, ensure_ascii=False)}

# --- ヘルパー関数 ---
def _get_salesforce_access_token(CLIENT_ID: str, CLIENT_SECRET: str, TOKEN_URL: str) -> tuple[str | None, str | None, str | None]:
    payload = {
        'grant_type': 'client_credentials',
        'client_id': CLIENT_ID,
        'client_secret': CLIENT_SECRET
    }
    headers = {'Content-Type': 'application/x-www-form-urlencoded'}
    
    try:
        response = requests.post(TOKEN_URL, headers=headers, data=payload, timeout=10)
        response.raise_for_status()
        token_data = response.json()
        access_token = token_data.get('access_token')
        instance_url = token_data.get('instance_url')
        if not access_token or not instance_url:
            return None, None, f"エラー: アクセストークンまたはインスタンスURLをレスポンスから取得できませんでした。 レスポンス: {token_data}"
        return access_token, instance_url, None
    except requests.exceptions.HTTPError as http_err:
        return None, None, f"HTTPエラーが発生しました: {http_err} - レスポンスボディ: {getattr(http_err.response, 'text', 'N/A')}"
    except requests.exceptions.RequestException as req_err:
        return None, None, f"リクエストエラーが発生しました: {req_err}"
    except json.JSONDecodeError:
        return None, None, f"レスポンスのJSONデコードに失敗しました。レスポンス: {getattr(response, 'text', 'N/A')}"

def _create_salesforce_record(access_token: str, instance_url: str, sobject_name: str, record_data: dict) -> tuple[dict | None, str | None]:
    api_endpoint = f"{instance_url}/services/data/v59.0/sobjects/{sobject_name}/"
    headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json'}
    try:
        response = requests.post(api_endpoint, headers=headers, json=record_data, timeout=10)
        response.raise_for_status()
        return response.json(), None
    except requests.exceptions.HTTPError as http_err:
        error_msg = f"HTTPエラーが発生しました: {http_err} - レスポンスボディ: {getattr(http_err.response, 'text', 'N/A')}"
        try:
            error_details = http_err.response.json()
            error_msg += f" Salesforceからのエラー詳細: {json.dumps(error_details, ensure_ascii=False)}"
        except json.JSONDecodeError:
            pass
        return None, error_msg
    except requests.exceptions.RequestException as req_err:
        return None, f"リクエストエラーが発生しました: {req_err}"