台灣個資法(PIPA)CRM 顧客標籤與分眾合規設計指南

1# tag_registry.yaml — 由 CI 驗證,任何新標籤必須先進這份檔案
2- tag: purchased_category_skincare
3  tier: L1
4  purpose_codes: ["069", "090"]
5  markets: [TW, HK, SG]
6  retention_days: 1095
7  marketing_use: allowed_with_optout
8  owner: crm-ops
9
10- tag: browsed_maternity_products
11  tier: L2
12  purpose_codes: ["040"]
13  markets: [TW]
14  sensitivity_flag: health_inference   # 觸發人工覆核
15  retention_days: 180
16  marketing_use: requires_explicit_consent
17  owner: growth
18
19- tag: allergy_profile
20  tier: L3
21  purpose_codes: []
22  markets: []
23  marketing_use: prohibited
24  storage: isolated_vault_only

1# pre-deploy check
2python scripts/validate_tags.py \
3  --registry tag_registry.yaml \
4  --source dbt/models/marts/customer_tags.sql \
5  --fail-on unregistered,L3_in_crm
6
7# 預期輸出
8# ✓ 142 tags validated
9# ✗ ERROR: tag 'health_condition_diabetes' matches L3 pattern, not permitted in CRM layer
10# exit code 1

1-- Snowflake / BigQuery 皆適用
2CREATE TABLE consent_events (
3  event_id        STRING NOT NULL,
4  subject_id      STRING NOT NULL,      -- 內部 customer id,勿用身分證字號
5  market          STRING NOT NULL,      -- TW / HK / SG / AU
6  purpose_code    STRING NOT NULL,      -- 對應特定目的代號,例如 040
7  channel         STRING NOT NULL,      -- email / sms / line / push / call
8  status          STRING NOT NULL,      -- granted / withdrawn / expired
9  captured_at     TIMESTAMP NOT NULL,
10  source          STRING NOT NULL,      -- web_signup_v4 / pos_tablet / line_oa
11  notice_version  STRING NOT NULL,      -- 當時展示的告知文字版本
12  evidence_uri    STRING,               -- 畫面截圖或表單快照存放位置
13  ip_hash         STRING,
14  ingested_at     TIMESTAMP DEFAULT CURRENT_TIMESTAMP()
15);

1CREATE OR REPLACE VIEW consent_current AS
2SELECT subject_id, market, purpose_code, channel,
3       status, captured_at, notice_version
4FROM (
5  SELECT *, ROW_NUMBER() OVER (
6    PARTITION BY subject_id, market, purpose_code, channel
7    ORDER BY captured_at DESC
8  ) AS rn
9  FROM consent_events
10)
11WHERE rn = 1;

1// 前端埋點:同意狀態隨事件走
2analytics.identify(userId, {
3  market: 'TW',
4  crm_tier: 'gold'
5}, {
6  consent: {
7    categoryPreferences: {
8      Functional: true,
9      Marketing: true,
10      Advertising: false
11    }
12  },
13  // 自訂欄位:台灣個資法特定目的
14  tw_pdpa: {
15    purpose_codes: ['069', '090', '040'],
16    notice_version: 'tw-privacy-2026-01',
17    optout_channel_url: 'https://example.com/tw/unsubscribe'
18  }
19});

1SELECT Id, ContactPointEmail.EmailAddress, DataUsePurpose.Name,
2       PrivacyConsentStatus, EffectiveFrom, CaptureSource
3FROM ContactPointConsent
4WHERE ContactPointEmail.ParentId = :individualId
5  AND DataUsePurpose.Name = 'TW_Marketing_040'
6  AND PrivacyConsentStatus = 'OptIn'

1trigger BlockNonConsentedSend on MarketingSendRequest__c (before insert) {
2    Set<Id> individualIds = new Set<Id>();
3    for (MarketingSendRequest__c r : Trigger.new) individualIds.add(r.Individual__c);
4
5    Map<Id, ContactPointConsent> consented = new Map<Id, ContactPointConsent>();
6    for (ContactPointConsent c : [
7        SELECT Id, ContactPointEmail.ParentId
8        FROM ContactPointConsent
9        WHERE ContactPointEmail.ParentId IN :individualIds
10          AND DataUsePurpose.Name = 'TW_Marketing_040'
11          AND PrivacyConsentStatus = 'OptIn'
12    ]) {
13        consented.put(c.ContactPointEmail.ParentId, c);
14    }
15
16    for (MarketingSendRequest__c r : Trigger.new) {
17        if (r.Market__c == 'TW' && !consented.containsKey(r.Individual__c)) {
18            r.addError('TW PDPA: 缺少特定目的 040 之有效同意,寄送已阻擋。');
19        }
20    }
21}

1-- 這是唯一允許行銷團隊查詢的 audience 來源
2CREATE OR REPLACE SECURE VIEW audience_tw_email_marketing AS
3SELECT
4  c.subject_id,
5  c.email_hash,
6  t.tag,
7  t.tag_tier,
8  cc.notice_version,
9  cc.captured_at AS consent_captured_at
10FROM customers c
11JOIN customer_tags t   ON t.subject_id = c.subject_id
12JOIN consent_current cc
13  ON cc.subject_id  = c.subject_id
14 AND cc.market      = 'TW'
15 AND cc.purpose_code = '040'
16 AND cc.channel      = 'email'
17 AND cc.status       = 'granted'
18JOIN tag_registry r ON r.tag = t.tag
19WHERE c.market = 'TW'
20  AND r.tier IN ('L1','L2')
21  AND r.marketing_use <> 'prohibited'
22  AND t.updated_at >= DATEADD(day, -r.retention_days, CURRENT_DATE());

1# models/marts/schema.yml
2models:
3  - name: audience_tw_email_marketing
4    tests:
5      - dbt_utils.expression_is_true:
6          expression: "tag_tier != 'L3'"
7      - dbt_utils.expression_is_true:
8          expression: "consent_captured_at is not null"
9    columns:
10      - name: subject_id
11        tests: [unique_combination_of_columns, not_null]

1$ dbt test --select audience_tw_email_marketing
211:42:03  Running 4 tests
311:42:07  PASS expression_is_true_tag_tier ......... [PASS in 3.1s]
411:42:09  PASS expression_is_true_consent_captured . [PASS in 1.8s]
511:42:09  Done. PASS=4 WARN=0 ERROR=0

1# FastAPI 範例:統一 opt-out 端點,所有通路共用
2@app.post("/consent/withdraw")
3async def withdraw(payload: WithdrawRequest):
4    event = {
5        "event_id": str(uuid4()),
6        "subject_id": resolve_subject(payload.token),
7        "market": payload.market,
8        "purpose_code": "040",
9        "channel": payload.channel,      # email / sms / line / push / call
10        "status": "withdrawn",
11        "captured_at": datetime.now(timezone.utc),
12        "source": payload.source,
13        "notice_version": payload.notice_version,
14    }
15    await warehouse.insert("consent_events", event)
16    # 即時推回 CDP,不等 batch
17    await segment.track(
18        user_id=event["subject_id"],
19        event="Marketing Consent Withdrawn",
20        properties={"channel": payload.channel, "market": payload.market},
21    )
22    return {"status": "ok", "effective": "immediate"}

1-- BigQuery:建立區域鎖定的資料集
2CREATE SCHEMA `proj.crm_tw`
3OPTIONS (location = 'asia-east1', default_table_expiration_days = 1095);
4
5-- 分析層只收假名化輸出
6CREATE TABLE `proj.analytics_global.tw_tags_pseudonymised` AS
7SELECT TO_HEX(SHA256(CONCAT(subject_id, @pepper))) AS pid,
8       tag, tag_tier, updated_at
9FROM `proj.crm_tw.customer_tags`;

1-- 每日執行:清除逾期標籤
2DELETE FROM customer_tags ct
3USING tag_registry r
4WHERE ct.tag = r.tag
5  AND ct.updated_at < CURRENT_DATE() - r.retention_days;
6
7-- 休眠客戶:超過 36 個月無互動且無有效契約關係
8UPDATE customers
9SET status = 'pending_erasure', flagged_at = CURRENT_TIMESTAMP()
10WHERE last_interaction_at < DATEADD(month, -36, CURRENT_DATE())
11  AND active_contract = FALSE
12  AND market = 'TW';

1ALLOWED_TAGS = load_registry(tier_in=["L1", "L2"], market="TW",
2                             exclude_flags=["health_inference", "religion_inference"])
3
4response = client.chat.completions.create(
5    model="gpt-4.1-mini",
6    messages=[{
7        "role": "system",
8        "content": (
9            "你只能從下列封閉清單中選擇標籤,不得自行創造。"
10            "若對話涉及健康、醫療、宗教、政治或性生活,回傳 {\"tags\": [], \"blocked\": true}。\n"
11            f"允許清單:{json.dumps(ALLOWED_TAGS, ensure_ascii=False)}"
12        )
13    }, {"role": "user", "content": redact_pii(transcript)}],
14    response_format={"type": "json_schema", "json_schema": TAG_SCHEMA}
15)
16
17# 輸出後再過一次白名單,不信任模型自律
18tags = [t for t in json.loads(response.choices[0].message.content)["tags"]
19        if t in ALLOWED_TAGS]