1. 업무 요구사항 및 목표

더존 ERP의 5개 테이블을 Feishu 다차원 표(Base)로 동기화합니다.

주요 요구사항은 다음과 같습니다.

• 기존 계약 120건 최초 일괄 등록
• 매일 새벽 배치로 마일스톤 달성률 수정
• 계약·지출·정산·지급·마일스톤 간 관계 유지
• 담당자 A는 정산서와 지급전표만 열람
• 중복 등록 방지 및 재실행 가능한 배치
• 추가·수정·삭제 작업의 감사 로그와 오류 재처리
• ERP와 Feishu 간 데이터 정합성 보장

제공된 cli_a1b2c3d4e5f6g7h8i9 값은 코드에 직접 넣지 않아야 합니다. 실제 운영 전 해당 값은 폐기 또는 교체하고, App ID와 App Secret은 Secret Manager에 보관하십시오.

---

2. 권장 데이터 구조

Feishu Base 구성

하나의 Base에 다음 5개 테이블을 구성합니다.

ERP 테이블	Feishu 테이블	주요 키
cost_contract	계약	계약번호
expense	지출결의	결의번호
settlement_sheet	정산서	정산번호
payment_slip	지급전표	전표번호
milestone_node	마일스톤	계약번호 + 마일스톤명

모든 테이블에는 다음 시스템 필드를 추가하는 것을 권장합니다.

• ERP_ID: ERP 원본 식별자
• 계약번호: 관계 조회용
• 동기화일시
• 동기화상태
• 원본수정일시
• 삭제여부
• 데이터해시

ERP_ID는 Feishu에서 중복을 방지하기 위한 고유 키로 사용합니다.

관계 설계

두 가지 방식이 있습니다.

권장 방식: ERP_ID 기반 비정규화 + Feishu 링크 필드

각 테이블에 계약번호를 일반 텍스트 필드로 저장하고, 필요하면 Feishu의 링크 레코드 필드를 추가합니다.

장점:

• 대량 배치가 단순함
• ERP 원본과 대조하기 쉬움
• 링크 레코드 생성 실패가 전체 동기화를 중단시키지 않음

대안: Feishu 링크 레코드 중심 구조

계약 테이블의 레코드 ID를 기준으로 지출결의, 정산서, 지급전표, 마일스톤을 링크합니다.

장점:

• Feishu 화면에서 계약별 하위 데이터를 쉽게 탐색 가능

단점:

• 계약 레코드 ID를 먼저 조회해야 함
• 관계 생성과 변경 처리가 복잡함

실무에서는 계약번호를 항상 보존하고, 사용자 화면 개선이 필요한 경우에만 링크 필드를 추가하는 구조가 안정적입니다.

---

3. 필드 매핑

계약

ERP 필드	Feishu 필드	타입
계약번호	계약번호	Text
거래처	거래처	Text
계약금액	계약금액	Number
계약일	계약일	Date
내부 키	ERP_ID	Text

지출결의

ERP 필드	Feishu 필드	타입
결의번호	결의번호	Text
계약번호FK	계약번호	Text 또는 Link
지출항목	지출항목	Text
금액	금액	Number

정산서

ERP 필드	Feishu 필드	타입
정산번호	정산번호	Text
계약번호FK	계약번호	Text 또는 Link
정산금액	정산금액	Number
정산일	정산일	Date

지급전표

ERP 필드	Feishu 필드	타입
전표번호	전표번호	Text
정산번호FK	정산번호	Text 또는 Link
지급액	지급액	Number
지급일	지급일	Date

마일스톤

ERP 필드	Feishu 필드	타입
계약번호FK	계약번호	Text 또는 Link
마일스톤명	마일스톤명	Text
달성률	달성률	Number
완료예정일	완료예정일	Date
복합 키	ERP_ID	Text

마일스톤은 계약번호 + 마일스톤명을 복합 키로 사용합니다.

예:

ERP_ID = C20260001#착공

---

4. 기술 아키텍처

더존 ERP
   │
   │ 추출: API, DB View, CSV 또는 SFTP
   ▼
ERP Adapter
   │
   ├─ 데이터 정규화
   ├─ 필수값 및 금액 검증
   ├─ 데이터 해시 생성
   └─ 변경분 추출
   ▼
Sync Engine
   │
   ├─ ERP_ID 조회
   ├─ add / modify / delete 분류
   ├─ 500건 단위 배치
   ├─ 재시도 및 Rate Limit 처리
   └─ 동기화 로그 저장
   ▼
Feishu Bitable API
   │
   ├─ 계약
   ├─ 지출결의
   ├─ 정산서
   ├─ 지급전표
   └─ 마일스톤

추가로 다음 저장소를 둡니다.

• sync_job: 배치 실행 기록
• sync_item: 레코드별 성공·실패 결과
• erp_feishu_mapping: ERP_ID와 Feishu record_id 매핑
• dead_letter: 반복 실패 데이터

---

5. 배치 처리 전략

최초 등록

기존 계약 120건은 다음 순서로 등록합니다.

1. 계약 등록
2. 지출결의 등록
3. 정산서 등록
4. 지급전표 등록
5. 마일스톤 등록
6. ERP_ID와 Feishu record_id 저장
7. 건수 및 금액 검증

Feishu Bitable API의 배치 크기는 API 제한에 맞춰 사용해야 하며, 일반적으로 500건 이하 단위로 나누는 것이 안전합니다.

매일 새벽 마일스톤 수정

1. ERP에서 마일스톤 데이터 조회
2. 계약번호 + 마일스톤명으로 ERP_ID 생성
3. Feishu에서 기존 ERP_ID 조회
4. 달성률 또는 완료예정일 변경 시에만 수정
5. 변경 건만 batch_update
6. 실패 건은 재처리 큐에 저장

마일스톤 달성률이 변경되지 않은 경우 API를 호출하지 않아야 합니다.

---

6. Python 구현 예시

환경 변수

export FEISHU_APP_ID="cli_xxxxxxxxx"
export FEISHU_APP_SECRET="vault에서_조회"
export FEISHU_APP_TOKEN="bascnxxxxxxxxx"

export FEISHU_CONTRACT_TABLE="tbl_contract"
export FEISHU_MILESTONE_TABLE="tbl_milestone"

App Secret은 .env 파일이나 소스 코드에 저장하지 말고 Vault, AWS Secrets Manager, GCP Secret Manager 등의 비밀 저장소에서 주입합니다.

Feishu 클라이언트

import os
import time
import requests
from typing import Iterable

BASE_URL = "https://open.feishu.cn/open-apis"

class FeishuClient:
    def __init__(self):
        self.app_id = os.environ["FEISHU_APP_ID"]
        self.app_secret = os.environ["FEISHU_APP_SECRET"]
        self.app_token = os.environ["FEISHU_APP_TOKEN"]
        self.token = self._get_tenant_token()

    def _get_tenant_token(self) -> str:
        response = requests.post(
            f"{BASE_URL}/auth/v3/tenant_access_token/internal",
            json={
                "app_id": self.app_id,
                "app_secret": self.app_secret,
            },
            timeout=30,
        )
        response.raise_for_status()

        data = response.json()
        if data.get("code") != 0:
            raise RuntimeError(f"토큰 발급 실패: {data}")

        return data["tenant_access_token"]

    def _request(self, method: str, path: str, payload: dict):
        headers = {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json",
        }

        for attempt in range(5):
            response = requests.request(
                method,
                f"{BASE_URL}{path}",
                headers=headers,
                json=payload,
                timeout=60,
            )

            if response.status_code == 429:
                wait_seconds = min(2 ** attempt, 30)
                time.sleep(wait_seconds)
                continue

            response.raise_for_status()
            result = response.json()

            if result.get("code") != 0:
                raise RuntimeError(result)

            return result

        raise RuntimeError("Feishu API 재시도 횟수를 초과했습니다.")

    def batch_create(self, table_id: str, records: list[dict]):
        return self._request(
            "POST",
            f"/bitable/v1/apps/{self.app_token}/tables/"
            f"{table_id}/records/batch_create",
            {"records": [{"fields": x} for x in records]},
        )

    def batch_update(self, table_id: str, records: list[dict]):
        return self._request(
            "POST",
            f"/bitable/v1/apps/{self.app_token}/tables/"
            f"{table_id}/records/batch_update",
            {
                "records": [
                    {
                        "record_id": x["record_id"],
                        "fields": x["fields"],
                    }
                    for x in records
                ]
            },
        )

    def batch_delete(self, table_id: str, record_ids: list[str]):
        return self._request(
            "POST",
            f"/bitable/v1/apps/{self.app_token}/tables/"
            f"{table_id}/records/batch_delete",
            {"records": record_ids},
        )

운영 시에는 토큰 만료 시간에 맞춘 캐시, 네트워크 오류 재시도, API 오류 코드별 분류를 추가해야 합니다.

---

7. ERP 데이터 변환

from datetime import datetime, timezone

def date_to_timestamp(value: str | None) -> int | None:
    if not value:
        return None

    dt = datetime.strptime(value, "%Y-%m-%d")
    dt = dt.replace(tzinfo=timezone.utc)
    return int(dt.timestamp() * 1000)


def contract_to_feishu(row: dict) -> dict:
    return {
        "ERP_ID": str(row["계약번호"]),
        "계약번호": str(row["계약번호"]),
        "거래처": row["거래처"],
        "계약금액": float(row["계약금액"]),
        "계약일": date_to_timestamp(row["계약일"]),
        "동기화상태": "정상",
    }


def milestone_to_feishu(row: dict) -> dict:
    contract_no = str(row["계약번호FK"])
    name = str(row["마일스톤명"])

    return {
        "ERP_ID": f"{contract_no}#{name}",
        "계약번호": contract_no,
        "마일스톤명": name,
        "달성률": float(row["달성률"]),
        "완료예정일": date_to_timestamp(row["완료예정일"]),
        "동기화상태": "정상",
    }

금액은 부동소수점 오차가 문제가 될 수 있으므로, 원천 ERP가 소수점 금액을 제공한다면 Decimal을 사용하는 것이 좋습니다.

---

8. 최초 120건 등록

def chunks(items: list[dict], size: int = 500):
    for i in range(0, len(items), size):
        yield items[i:i + size]


def initial_contract_load(client: FeishuClient, table_id: str, erp_rows: list[dict]):
    records = [contract_to_feishu(row) for row in erp_rows]

    if len(records) != 120:
        raise ValueError(f"예상 계약 건수는 120건이지만 {len(records)}건입니다.")

    for batch in chunks(records):
        client.batch_create(table_id, batch)

최초 등록 전에는 다음을 검사해야 합니다.

• 계약번호 중복 여부
• 계약금액이 음수인지 여부
• 계약일 형식
• 필수 거래처 누락 여부
• Feishu 테이블 필드명과 타입 일치 여부

---

9. 마일스톤 변경분 수정

실제로는 먼저 Feishu의 기존 레코드를 ERP_ID 기준으로 조회해야 합니다. 조회 결과를 다음과 같이 구성한다고 가정합니다.

# {
#     "C20260001#착공": "recxxxxxxxx",
#     "C20260001#준공": "recyyyyyyyy",
# }
existing_record_map = {}

수정 코드는 다음과 같습니다.

def update_milestones(
    client: FeishuClient,
    table_id: str,
    erp_rows: list[dict],
    existing_record_map: dict[str, str],
):
    updates = []
    creates = []

    for row in erp_rows:
        fields = milestone_to_feishu(row)
        erp_id = fields["ERP_ID"]
        record_id = existing_record_map.get(erp_id)

        if record_id:
            updates.append({
                "record_id": record_id,
                "fields": fields,
            })
        else:
            creates.append(fields)

    for batch in chunks(updates):
        client.batch_update(table_id, batch)

    for batch in chunks(creates):
        client.batch_create(table_id, batch)

더 효율적으로 운영하려면 ERP에서 최종수정일시를 받고, 이전 동기화 시점 이후 변경된 마일스톤만 전달받아야 합니다.

---

10. 삭제 처리

물리 삭제보다 다음 방식이 권장됩니다.

1. ERP에서 삭제된 레코드 식별
2. Feishu의 삭제여부를 Y로 수정
3. 화면 필터에서 삭제 데이터를 제외
4. 법적·감사 보존 기간 이후에만 물리 삭제

정말 물리 삭제가 필요한 경우:

def delete_records(
    client: FeishuClient,
    table_id: str,
    record_ids: list[str],
):
    for batch in chunks(record_ids):
        client.batch_delete(table_id, batch)

삭제는 반드시 다음 조건을 모두 만족할 때만 실행해야 합니다.

• ERP 삭제 목록이 확정됨
• 삭제 대상 건수가 예상 범위 내임
• 동기화 작업자가 승인함
• 삭제 전 백업 또는 감사 로그가 존재함

---

11. 담당자 A 권한 설계

담당자 A에게 전체 Base 접근 권한을 주고 특정 뷰만 공유하는 방식은 안전하지 않습니다. 뷰는 보안 경계로 간주하면 안 됩니다.

권장 구조는 다음과 같습니다.

• 계약, 지출결의, 마일스톤 테이블: 담당자 A 접근 차단
• 정산서, 지급전표 테이블: 담당자 A에게 읽기 전용 권한 부여
• 담당자 A에게 편집 권한 부여 금지
• API App 계정은 동기화에 필요한 테이블만 접근
• 관리자만 Base 구조와 권한 변경 가능
• 정산서와 지급전표에 민감정보가 있다면 별도 Base로 분리

가장 강한 보안 구조는 다음과 같습니다.

Base 1: 계약 / 지출결의 / 마일스톤
        일반 운영 담당자

Base 2: 정산서 / 지급전표
        담당자 A 읽기 전용
        재무 관리자 편집 가능

Feishu 관리 콘솔에서 테이블·필드·레코드 권한 기능을 사용할 수 있는 경우에도, 실제 테넌트 정책과 라이선스에서 해당 기능을 지원하는지 확인해야 합니다.

---

12. 보안 및 데이터 무결성

필수 보안 조치는 다음과 같습니다.

• App Secret을 소스 코드, 로그, Git에 저장하지 않음
• 제공된 인증값은 운영 전 교체
• HTTPS만 사용
• API 호출 로그에서 토큰과 개인정보 마스킹
• ERP_ID를 통한 멱등성 보장
• 배치별 job_id 생성
• 요청·응답 원문은 민감정보 제거 후 저장
• Feishu App 권한을 최소 범위로 설정
• 동기화 계정과 일반 사용자 계정 분리
• 실패 데이터는 Dead Letter Queue에 보관
• 배치 실행 전후 건수와 금액 합계 비교

예를 들어 다음 검증을 수행할 수 있습니다.

def validate_amounts(erp_rows: list[dict], feishu_rows: list[dict]):
    erp_total = sum(float(row["금액"]) for row in erp_rows)
    feishu_total = sum(float(row["금액"]) for row in feishu_rows)

    if round(erp_total, 2) != round(feishu_total, 2):
        raise ValueError(
            f"금액 합계 불일치: ERP={erp_total}, Feishu={feishu_total}"
        )

---

13. 운영 및 모니터링

배치마다 다음 지표를 기록합니다.

• 전체 대상 건수
• 추가 건수
• 수정 건수
• 삭제 건수
• 성공 건수
• 실패 건수
• 처리 시간
• ERP와 Feishu 금액 합계
• API 응답 오류 코드
• 마지막 성공 동기화 시각

권장 배치 정책:

매일 02:00  ERP 변경분 추출
매일 02:10  데이터 검증
매일 02:15  Feishu 추가·수정
매일 02:30  건수·금액 대조
매일 02:35  결과 알림

실패 시에는 전체 배치를 롤백하기보다, 성공 건은 유지하고 실패 건만 재처리하는 방식이 적합합니다. 단, 계약 생성과 하위 데이터 생성처럼 순서 의존성이 있는 작업은 단계별 완료 상태를 저장해야 합니다.

---

최종 권장안

• ERP_ID를 모든 테이블의 멱등성 키로 사용
• 계약번호를 모든 하위 테이블에 보존
• 최초 120건은 계약부터 순차 등록
• 매일 마일스톤 변경분만 수정
• 삭제는 기본적으로 논리 삭제
• 담당자 A는 정산서·지급전표 전용 읽기 권한만 부여
• App Secret은 별도 Vault에 보관하고 현재 노출된 인증값은 교체
• 배치별 로그, 재시도, 금액 대조, 실패 큐를 필수 구성
• Feishu API 호출 전 실제 테이블 ID, 필드 타입, 테넌트별 API 한도를 검증