DIP1 阶段 1 技术设计：MVP-FS 飞书工作台对接¶
文档编号：DOC-D01-D1 / 简称：DIP1-P1 版本：V1.1 创建日期：2026-08-09 最近更新：2026-08-09 维护人：TL / DT 关联文档：TND-E、MVP-FS、DIP1-ARC 架构设计 V2.0、DIP1-P2 阶段 2 设计 V2.0
⚠️ 重要：本文档暂不实施¶
决策依据：DIP1-ARC V2.0 ADR-008（2026-08-09，TL + SPO 决策）
维度 说明 当前状态 ⏸ 暂不实施——Code Agent 跳过本文档，直接实施 DIP1-P2 V2.0 阶段 2 跳过理由 ① 飞书方案（阶段 0 MVP-FS）实际运行情况待观察；② 直接做原生 App（React Native）避免飞书 SDK 投入打水漂；③ Code Agent 全栈聚焦阶段 2 目标 未来启动条件 阶段 0 飞书方案运行 ≥ 3 个月后，TL 评估是否需要数据迁移层；或 SPO 决策需要双轨过渡期 本文档价值 保留作为未来阶段 0 启动时的技术参考；Feishu SDK 封装、双轨同步、246 字段映射附录 A 等设计可复用 Code Agent 行为 禁止实施本文档任何任务；如 TL 在 Issue 中明确指示启动，则按本文档执行 相关文档同步状态 DIP1-IMP §5.1 已标注「⏸ V2.0 暂不实施，0 人天」；DIP1-ARC ADR-008 记录决策；DIP1-API V2.0 飞书 webhook 端点标注⏸
一、文档定位与目标¶
1.1 阶段 1 定义¶
⚠️ 以下内容为阶段 1 启动时的技术设计，当前暂不实施。
阶段 1 对应 TND-E「启动期」技术方案，核心是系统化前的过渡层：在飞书多维表格（MVP-FS）与未来全定制系统（阶段 2）之间建立双向数据桥梁，让一线团队在飞书工作台中高效作业的同时，数据可被后端系统实时消费与反哺。
graph TD
FS["飞书多维表格<br/>MVP-FS 6 张表"]
BRIDGE["阶段 1 对接层<br/>Feishu SDK + Sync Layer + Automation"]
DOMAIN["阶段 2 领域层<br/>QSV/CPT/MDS/OPS Domain"]
FS -- "① Bitable API 读/写" --> BRIDGE
FS -- "② Webhook 变更推送" --> BRIDGE
BRIDGE -- "③ 领域对象映射" --> DOMAIN
DOMAIN -- "④ 校准结果回写" --> BRIDGE
BRIDGE -- "⑤ API 更新飞书" --> FS
style BRIDGE fill:#D4AF78,stroke:#004046,color:#004046;
style FS fill:#004046,stroke:#004046,color:#fff;
style DOMAIN fill:#457b9d,stroke:#004046,color:#fff;1.2 设计目标¶
| 目标 | 含义 | 验证标准 |
|---|---|---|
| API 全覆盖 | 飞书 6 张表的 CRUD + 批量操作通过统一 SDK 完成 | 覆盖率 100%，单接口响应 < 500ms |
| 增量同步 | 飞书数据变更 → 10s 内同步到后端领域模型 | 延迟 < 10s，一致性 99.9% |
| 幂等安全 | 同一条记录重复推送不产生脏数据 | 幂等测试通过率 100% |
| 一键搭建 | TL 运行一条命令即可创建多维表格 + 6 张表 + 40 字段 + 视图 | 搭建时间 < 5 分钟 |
| 可演进 | 阶段 2 上线后，仅需替换对接层实现，领域层零修改 | 领域层无飞书 SDK 依赖 |
二、飞书 API 封装层（Feishu SDK）¶
2.1 技术选型¶
| 组件 | 选型 | 说明 |
|---|---|---|
| HTTP 客户端 | httpx (async) |
异步高性能，支持连接池与重试 |
| 认证方式 | Tenant Access Token + 自建应用凭证 | 飞书开放平台标准模式 |
| SDK 封装 | 自研轻量封装（不依赖 lark-oapi） | 接口可控，避免 SDK 升级冲突 |
| 重试策略 | 指数退避 + 令牌桶限流 | 飞书 QPS 限制：100 次/分钟/应用 |
| 错误处理 | 统一异常 FeishuApiError + 错误码映射 |
日志可追溯，告警可配置 |
2.2 模块结构¶
backend/app/infrastructure/feishu/
├── __init__.py
├── client.py # FeishuClient 主客户端（认证 + HTTP 封装）
├── bitable.py # BitableApi 多维表格 API 封装
├── webhook.py # WebhookValidator 签名验证 + 事件解析
├── models.py # Pydantic 数据模型（记录/字段/视图）
├── exceptions.py # FeishuApiError + 错误码映射
└── config.py # FeishuConfig 配置（AppID/AppSecret/TableID）
2.3 FeishuClient 核心接口¶
文件：infrastructure/feishu/client.py
class FeishuClient:
"""飞书 API 统一客户端，负责 Token 管理与 HTTP 封装"""
def __init__(self, config: FeishuConfig):
self._app_id = config.app_id
self._app_secret = config.app_secret.get_secret_value()
self._token_cache: TokenCache = TokenCache()
self._http = httpx.AsyncClient(
base_url="https://open.feishu.cn/open-apis",
timeout=30.0,
limits=httpx.Limits(max_connections=20),
)
async def get_tenant_token(self) -> str:
"""获取/刷新 tenant_access_token（缓存 110 分钟，飞书 120 分钟过期）"""
if self._token_cache.is_valid():
return self._token_cache.token
resp = await self._http.post(
"/auth/v3/tenant_access_token/internal",
json={"app_id": self._app_id, "app_secret": self._app_secret},
)
data = resp.raise_for_status().json()
self._token_cache.set(data["tenant_access_token"], data["expire"])
return data["tenant_access_token"]
async def request(self, method: str, path: str, **kwargs) -> dict:
"""统一请求入口，自动注入 Token、限流、重试、错误处理"""
token = await self.get_tenant_token()
headers = kwargs.pop("headers", {}) | {"Authorization": f"Bearer {token}"}
for attempt in range(3):
try:
resp = await self._http.request(method, path, headers=headers, **kwargs)
resp.raise_for_status()
data = resp.json()
if data.get("code") != 0:
raise FeishuApiError(data["code"], data.get("msg", ""))
return data.get("data", {})
except (httpx.HTTPStatusError, FeishuApiError) as e:
if attempt < 2 and _is_retryable(e):
await asyncio.sleep(2 ** attempt)
continue
raise
2.4 BitableApi 多维表格封装¶
文件：infrastructure/feishu/bitable.py
覆盖飞书多维表格全部核心 API，对应 MVP-FS 6 张表的 CRUD 与批量操作：
class BitableApi:
"""飞书多维表格 API 封装，对齐 MVP-FS 6 张表操作"""
# ========== 表与字段管理（搭建脚本用）==========
async def create_app_table(self, app_token: str, table: TableCreate) -> BitableTable:
"""创建数据表（POST /bitable/v1/apps/:app_token/tables）"""
async def list_fields(self, app_token: str, table_id: str) -> list[BitableField]:
"""列出表字段（GET /bitable/v1/apps/:app_token/tables/:table_id/fields）"""
async def create_field(self, app_token: str, table_id: str, field: FieldCreate) -> BitableField:
"""创建字段（POST /bitable/v1/apps/:app_token/tables/:table_id/fields）"""
# ========== 记录 CRUD（数据同步用）==========
async def list_records(
self, app_token: str, table_id: str,
page_size: int = 500, page_token: str | None = None,
filter_: BitableFilter | None = None,
) -> RecordPage:
"""查询记录列表（支持过滤/排序/分页）"""
async def get_record(self, app_token: str, table_id: str, record_id: str) -> BitableRecord:
"""查询单条记录"""
async def create_record(
self, app_token: str, table_id: str, fields: dict[str, Any],
) -> BitableRecord:
"""新增单条记录"""
async def batch_create_records(
self, app_token: str, table_id: str, records: list[dict[str, Any]],
) -> list[BitableRecord]:
"""批量新增记录（单次 ≤ 500 条）"""
async def update_record(
self, app_token: str, table_id: str, record_id: str, fields: dict[str, Any],
) -> BitableRecord:
"""更新单条记录"""
async def batch_update_records(
self, app_token: str, table_id: str, records: list[RecordUpdate],
) -> list[BitableRecord]:
"""批量更新记录（单次 ≤ 500 条）"""
async def delete_record(self, app_token: str, table_id: str, record_id: str) -> None:
"""删除单条记录"""
# ========== 视图（仪表盘配置用）==========
async def create_view(self, app_token: str, table_id: str, view: ViewCreate) -> BitableView:
"""创建视图（看板/筛选/甘特图）"""
2.5 字段类型映射表¶
飞书字段类型 → Python 类型 → 未来 PostgreSQL 列类型 的三方映射：
| 飞书字段类型 | Python 类型 | 存储表示 | PostgreSQL 列 |
|---|---|---|---|
1 文本 |
str |
原始字符串 | VARCHAR / TEXT |
2 数字 |
Decimal |
浮点字符串 | NUMERIC(12,2) |
3 单选 |
str |
option 文本值 | VARCHAR + CHECK |
4 多选 |
list[str] |
option 文本数组 | VARCHAR[] |
5 日期 |
datetime |
Unix 毫秒时间戳 | TIMESTAMPTZ |
7 人员 |
list[FeishuUser] |
open_id 数组 | VARCHAR[] |
8 附件/图片 |
list[FeishuAttachment] |
file_token + URL | JSONB |
9 关联 |
list[str] |
关联记录 record_id | VARCHAR[] + FK |
101 公式 |
Any |
计算结果（只读） | 由其他列派生 |
102 查找引用 |
Any |
查找结果（只读） | 由 JOIN 派生 |
2.6 Webhook 事件解析¶
文件：infrastructure/feishu/webhook.py
飞书开放平台事件订阅 → 后端签名验证 + 事件分发：
class WebhookEvent:
"""飞书 Webhook 事件基类"""
token: str
ts: str
type: str # "url_verification" / "event_callback"
event_id: str # 用于幂等去重
event_type: str # 如 "bitable.record.v1.created"
class BitableRecordPayload:
"""多维表格记录变更事件体"""
app_token: str
table_id: str
record_id: str
fields_before: dict[str, Any] | None # updated/deleted 时有值
fields_after: dict[str, Any] | None # created/updated 时有值
operator_id: str
event_time: datetime
class WebhookValidator:
def verify_signature(self, header_ts: str, header_nonce: str,
header_sign: str, body: bytes) -> bool:
"""HMAC-SHA256 签名验证，防止伪造请求"""
def parse(self, body: bytes) -> WebhookEvent:
"""解析并路由到对应事件类：url_verification / record / field 等"""
订阅事件清单（阶段 1 订阅 6 类）：
| 事件类型 | 触发条件 | 处理动作 |
|---|---|---|
bitable.record.v1.created |
6 张表新增记录 | 增量同步 → 领域模型 Insert |
bitable.record.v1.updated |
6 张表更新记录 | 增量同步 → 领域模型 Update |
bitable.record.v1.deleted |
6 张表删除记录 | 标记领域对象为 archived |
bitable.field.v1.created |
新增字段（表单迭代） | 记录字段变更日志，通知 TL 评审 |
contact.user.v1.created |
新师傅加入飞书 | 自动创建 WPL-Lite 台账行 |
im.message.receive_v1 |
AI 指令消息（可选） | DT 对话入口，解析自然语言指令 |
三、数据同步层（Sync Layer）¶
3.1 同步架构：双轨模式¶
阶段 1 采用 "Webhook 实时 + 定时全量巡检" 双轨同步，确保数据不丢、延迟可控：
flowchart LR
subgraph Feishu["飞书工作台"]
CPT["CPT-Lite 订单表"]
QSV["QSV-Lite 报价表"]
MDS["WPL/CPL/SCL/BPL 四库"]
end
subgraph SyncLayer["阶段 1 数据同步层"]
WH["Webhook 实时通道<br/>(P95 < 3s)"]
POLL["定时巡检任务<br/>(每 15 分钟)"]
IDMP["幂等去重层<br/>(event_id + record_id)"]
MAP["字段映射器<br/>(FeishuField → DomainEntity)"]
UOW["Unit of Work<br/>事务写入领域层"]
end
subgraph Domain["阶段 2 领域层（预占位）"]
QSVD["QSV 报价聚合根"]
CPTD["CPT 订单聚合根"]
MSD["MDS 师傅/客户/品类/楼宇实体"]
end
CPT --> WH
QSV --> WH
MDS --> WH
CPT --> POLL
QSV --> POLL
MDS --> POLL
WH --> IDMP
POLL --> IDMP
IDMP --> MAP
MAP --> UOW
UOW --> QSVD
UOW --> CPTD
UOW --> MSD3.2 同步表注册中心¶
文件：infrastructure/feishu/sync_registry.py
维护飞书表 → 领域实体的映射配置，确保新增表只需在注册表中加一行：
@dataclass
class SyncTableConfig:
table_key: str # CPT_LITE / QSV_LITE / WPL_LITE ...
app_token: str # 飞书 app_token
table_id: str # 飞书 table_id
domain_model: type # 对应领域实体类
field_mapping: dict[str, str] # 飞书字段名 → 领域属性名
primary_key_field: str = "record_id"
sync_enabled: bool = True
SYNC_TABLES: dict[str, SyncTableConfig] = {
"CPT_LITE": SyncTableConfig(
table_key="CPT_LITE",
app_token="${FEISHU_BITABLE_APP_TOKEN}",
table_id="${FEISHU_TABLE_ID_CPT}",
domain_model="app.domain.cpt.order.CPTOrder",
field_mapping={
"order_id": "order_id",
"current_stage": "stage",
"order_status": "status",
"customer_name": "customer_name",
... # 剩余 36 字段映射
},
),
"QSV_LITE": SyncTableConfig(...),
"WPL_LITE": SyncTableConfig(...),
"CPL_LITE": SyncTableConfig(...),
"SCL_LITE": SyncTableConfig(...),
"BPL_LITE": SyncTableConfig(...),
}
3.3 增量同步处理器¶
文件：infrastructure/feishu/sync_handler.py
处理 Webhook 推送记录的核心逻辑，保证幂等、可追溯、可重放：
class RecordSyncHandler:
"""处理单条记录变更：Webhook 事件 → 领域对象持久化"""
def __init__(self, sync_registry: SyncRegistry, uow: UnitOfWork,
idempotency_store: IdempotencyStore):
self._registry = sync_registry
self._uow = uow
self._idempotency = idempotency_store
async def handle(self, event: BitableRecordPayload) -> SyncResult:
# Step 1: 幂等去重（同一 event_id 只处理一次）
if await self._idempotency.exists(event.event_id):
return SyncResult.duplicated(event.event_id)
await self._idempotency.mark(event.event_id)
# Step 2: 查表配置
config = self._registry.get_by_table(event.app_token, event.table_id)
if not config or not config.sync_enabled:
return SyncResult.skipped("table_not_registered")
# Step 3: 字段映射 → 领域实体
fields = event.fields_after or event.fields_before or {}
entity = self._map_fields_to_entity(fields, config)
# Step 4: 按操作类型落库
async with self._uow:
repo = self._uow.get_repository(config.domain_model)
match event.event_type:
case "bitable.record.v1.created":
await repo.add(entity)
case "bitable.record.v1.updated":
await repo.update(entity.record_id, entity)
case "bitable.record.v1.deleted":
await repo.archive(entity.record_id)
await self._uow.commit()
return SyncResult.success(event.event_id, event.record_id)
3.4 定时巡检任务（兜底）¶
文件：scripts/sync/feishu_poll_sync.py
每 15 分钟拉取全部 6 张表的增量（按 updated_at 倒序），兜底 Webhook 可能的丢失：
async def poll_incremental_sync(table_key: str, since_hours: int = 1):
"""轮询同步：拉取 since_hours 内更新过的记录，对比 DB 后写入"""
config = SYNC_TABLES[table_key]
last_sync_ts = await get_last_sync_cursor(table_key)
# 1. 按 updated_at >= last_sync_ts 拉取飞书记录
page = await bitable.list_records(
config.app_token, config.table_id,
filter_=BitableFilter(f"updated_at >= {last_sync_ts}"),
page_size=500,
)
# 2. 逐条对比 DB 版本号，仅处理变更
for record in page.records:
db_version = await get_db_record_version(config, record.record_id)
if db_version and db_version >= record.last_modified_time:
continue # DB 版本更新，跳过
await sync_single_record(config, record) # 复用 Handler 逻辑
# 3. 推进同步游标
await set_last_sync_cursor(table_key, page.latest_updated_at)
四、自动化脚本框架¶
4.1 脚本清单与目录¶
scripts/
├── feishu/
│ ├── feishu-cli-setup.ps1 # 已存在：凭证获取与环境配置
│ ├── feishu-mvp-setup.py # 已存在：工作台搭建 Python 脚本
│ ├── feishu-seed-data.py # 🆕 阶段 1 新增：种子数据导入
│ └── feishu-field-migrate.py # 🆕 阶段 1 新增：字段迭代变更脚本
├── sync/
│ └── feishu_poll_sync.py # 🆕 阶段 1 新增：巡检兜底同步
└── migration/
└── mvpfs_to_pg_migrate.py # 🆕 阶段 2 前置：飞书 → PostgreSQL 全量迁移
4.2 feishu-mvp-setup.py 增强（搭建脚本）¶
文件：scripts/feishu/feishu-mvp-setup.py
基于 V1.2 现有版本，补充阶段 1 专用 API 能力（原先为飞书 CLI 版）：
class MVPSetupOrchestrator:
"""MVP-FS 工作台一键搭建编排器（阶段 1 升级为直接 API 调用）"""
async def run(self, config: SetupConfig) -> SetupResult:
# Step 1: 创建多维表格应用（WEAVELY-MVP）
app = await self._client.bitable.create_bitable(
name="WEAVELY-MVP",
folder_token=config.folder_token,
)
# Step 2: 批量创建 6 张表 + 字段
table_results: dict[str, TableSetupResult] = {}
for table_def in TABLE_DEFINITIONS: # CPT/QSV/WPL/CPL/SCL/BPL
table = await self._client.bitable.create_app_table(app.app_token, table_def.table)
fields = await asyncio.gather(*[
self._client.bitable.create_field(app.app_token, table.table_id, f)
for f in table_def.fields
])
table_results[table_def.key] = TableSetupResult(table=table, fields=fields)
# Step 3: 创建视图（看板/筛选视图/我的订单）
views = await self._create_standard_views(app.app_token, table_results)
# Step 4: 设置自动化规则（阶段变更通知、超时提醒）
automations = await self._create_automations(app.app_token, table_results)
# Step 5: 输出配置（.env 文件写入）
self._write_env_file(app.app_token, table_results)
return SetupResult(app_token=app.app_token, tables=table_results, views=views)
表定义结构示例（CPT-Lite 40 字段）：
TABLE_DEFINITIONS = [
TableDefinition(
key="CPT_LITE",
table=TableCreate(name="CPT-Lite 订单跟踪表"),
fields=[
# ===== 通用信息（GN）5 字段 =====
FieldCreate(field_name="order_id", type=1, desc="订单编号 ORD-YYYYMMDD-NNN"),
FieldCreate(
field_name="current_stage", type=3,
options=[Option(name="L1"), Option(name="L2"), ..., Option(name="完成")],
),
FieldCreate(
field_name="order_status", type=3,
options=["进行中", "已完成", "已取消", "异常暂停"],
),
FieldCreate(field_name="created_at", type=5),
FieldCreate(field_name="updated_at", type=5),
# ===== L1 线索获取 5 字段 =====
FieldCreate(field_name="lead_source", type=3, options=[
"SCENE01线上", "SCENE02随箱卡", "SCENE03电商",
"SCENE04物流", "SCENE05品牌", "SCENE06菜鸟", "推荐", "其他",
]),
FieldCreate(field_name="customer_name", type=1),
FieldCreate(field_name="customer_contact", type=1),
FieldCreate(field_name="lead_demand", type=1),
FieldCreate(field_name="lead_time", type=5),
# ===== L2~L8 共 30 字段（完整见附录 A） =====
...
],
),
# QSV-Lite / WPL-Lite / CPL-Lite / SCL-Lite / BPL-Lite 其余 5 张表...
]
4.3 feishu-seed-data.py 种子数据导入¶
文件：scripts/feishu/feishu-seed-data.py
搭建完成后一键导入 SCL 品类、BPL 楼宇等基础数据：
async def seed_scl_categories(app_token: str, table_id: str):
"""导入 SCL-Lite 8 大品类基础费率（对齐 QMD V2.1）"""
seeds = [
{"category_id": "CAT-01", "category_name": "衣柜", "base_fee": 380, "base_work_hours": 2.5, "cap_price": 1200, "skill_required": "L2"},
{"category_id": "CAT-02", "category_name": "橱柜", "base_fee": 520, "base_work_hours": 4.0, "cap_price": 1800, "skill_required": "L2"},
{"category_id": "CAT-03", "category_name": "床", "base_fee": 220, "base_work_hours": 1.5, "cap_price": 800, "skill_required": "L1"},
# ... 其余 5 大品类
]
await bitable.batch_create_records(app_token, table_id, seeds)
4.4 Webhook 入口与路由¶
文件：interfaces/webhooks/feishu_webhook.py
作为阶段 1 的对外 Webhook HTTP 入口：
# FastAPI Router
feishu_router = APIRouter(prefix="/webhooks/feishu", tags=["Feishu Webhook"])
@feishu_router.post("/bitable")
async def handle_bitable_webhook(
request: Request,
x_lark_request_timestamp: str = Header(...),
x_lark_request_nonce: str = Header(...),
x_lark_signature: str = Header(...),
handler: RecordSyncHandler = Depends(get_sync_handler),
):
# 1. 签名校验
body = await request.body()
if not validator.verify_signature(
x_lark_request_timestamp, x_lark_request_nonce, x_lark_signature, body,
):
raise HTTPException(status_code=401, detail="invalid signature")
# 2. 解析事件
event = validator.parse(body)
# 3. URL 校验（首次配置挑战）
if event.type == "url_verification":
return {"challenge": event.challenge}
# 4. 异步分发处理（立即返回 200，后台执行）
if event.event_type.startswith("bitable.record.v1."):
background_tasks.add_task(handler.handle, event.payload)
return {"code": 0}
五、反哺回写通道（领域层 → 飞书）¶
阶段 1 仅实现单向同步（飞书→DB）的正向链路，但预留反哺回写接口，供阶段 2 QSV 校准、CPT 画像反演时使用：
class FeishuWritebackPort:
"""领域层定义的端口（infrastructure 层实现），阶段 2 启用"""
async def update_quote_params(self, category_id: str, new_base_fee: Decimal) -> None:
"""QMD 校准后回写 SCL-Lite 品类基础费"""
async def update_worker_rating(self, worker_id: str, new_avg_nps: float) -> None:
"""CPT 采集后回写 WPL-Lite 师傅评分"""
async def update_customer_tags(self, customer_id: str, tags: list[str]) -> None:
"""OPS 营销后回写 CPL-Lite 客户标签"""
六、错误处理与观测¶
6.1 重试与死信队列（DLQ）¶
| 错误类型 | 处理策略 | 示例 |
|---|---|---|
| 飞书限流（99991400） | 指数退避 3 次，仍失败进 DLQ | QPS 超过 100 次/分 |
| 网络超时（httpx.Timeout） | 重试 2 次，仍失败进 DLQ | 飞书 API 波动 |
| 字段映射失败（KeyError） | 立即进 DLQ，告警 TL | 飞书加了字段但注册表未更新 |
| 领域校验失败（ValidationError） | 进 DLQ，告警 OL | 数据异常（如 quote_amount 负数） |
| 幂等冲突 | 静默跳过，不进 DLQ | Webhook 重复推送 |
DLQ 存储：Redis Stream feishu:sync:dlq，TL 可通过管理后台查看并重放。
6.2 观测指标（Prometheus）¶
# 同步延迟直方图
feishu_sync_latency_seconds = Histogram(
"feishu_sync_latency_seconds", "飞书事件→DB落库延迟",
buckets=[0.1, 0.5, 1.0, 3.0, 5.0, 10.0],
)
# 成功/失败计数器
feishu_sync_total = Counter("feishu_sync_total", "同步总数", ["table", "status"])
# DLQ 长度
feishu_sync_dlq_size = Gauge("feishu_sync_dlq_size", "死信队列长度")
6.3 巡检健康检查¶
每 5 分钟跑一次健康检查，验证： 1. FeishuClient token 获取正常 2. 6 张表各查询 1 条记录成功 3. Webhook 签名密钥未过期 4. DLQ 长度 < 阈值（默认 10 条）
异常时通过飞书机器人推送告警到 OL/TL 群。
七、阶段 1 交付物与里程碑¶
| 交付物 | 代码位置 | 工作量（人天） | 依赖 |
|---|---|---|---|
| Feishu SDK 封装 | infrastructure/feishu/*.py |
3 | Python 3.11 / httpx |
| Sync Registry + Handler | infrastructure/feishu/sync_*.py |
2 | SDK 封装完成 |
| 搭建脚本增强 | scripts/feishu/feishu-mvp-setup.py |
1 | SDK 封装完成 |
| 种子数据导入脚本 | scripts/feishu/feishu-seed-data.py |
0.5 | 搭建脚本完成 |
| 巡检兜底同步脚本 | scripts/sync/feishu_poll_sync.py |
1 | SDK 封装完成 |
| Webhook 入口 | interfaces/webhooks/feishu_webhook.py |
1 | Sync Handler 完成 |
| 观测指标 + 告警 | infrastructure/observability/feishu_*.py |
0.5 | 全部组件完成 |
| 单元测试 | tests/infrastructure/feishu/*.py |
2 | Mock 飞书 API |
| 集成测试 | tests/interfaces/test_feishu_webhook.py |
1 | Testcontainer Redis |
合计工作量：12 人天（TL 1 人 × 2 周 + DT AI 协同）
附录 A：CPT-Lite 40 字段完整映射¶
字段名与 MVP-FS 实施方案 §3.2 一一对应，snake_case 英文命名与未来数据库列名完全一致。
| 飞书字段名 | 飞书类型 | 领域属性 | 类型 | 必填 |
|---|---|---|---|---|
order_id |
1 文本 | order_id |
str |
✅ |
current_stage |
3 单选 | stage |
CPTStage Enum |
✅ |
order_status |
3 单选 | status |
OrderStatus Enum |
✅ |
created_at |
5 日期 | created_at |
datetime |
✅ |
updated_at |
5 日期 | updated_at |
datetime |
✅ |
lead_source |
3 单选 | lead_source |
LeadSource Enum |
✅ |
customer_name |
1 文本 | customer_name |
str |
✅ |
customer_contact |
1 文本 | customer_contact |
str |
✅ |
lead_demand |
1 文本 | lead_demand |
str |
✅ |
lead_time |
5 日期 | lead_time |
datetime |
✅ |
category |
3 单选 | category |
Category Enum |
✅ |
product_model |
1 文本 | product_model |
str |
✅ |
install_address |
1 文本 | install_address |
str |
✅ |
preferred_time |
5 日期 | preferred_time |
datetime |
✅ |
site_condition_note |
1 文本 | site_condition_note |
str |
⬜ |
quote_id |
1 文本 | quote.quote_id |
str |
✅ |
quote_amount |
2 数字 | quote.final_amount |
Decimal |
✅ |
quote_detail |
1 文本 | quote.detail_json |
JSON |
✅ |
quote_time |
5 日期 | quote.created_at |
datetime |
✅ |
quote_status |
3 单选 | quote.status |
QuoteStatus Enum |
✅ |
signed_at |
5 日期 | contract.signed_at |
datetime |
✅ |
payment_method |
3 单选 | contract.payment_method |
PaymentMethod Enum |
✅ |
payment_status |
3 单选 | contract.payment_status |
PaymentStatus Enum |
✅ |
assigned_worker_id |
9 关联 | assignment.worker_id |
str |
✅ |
delivery_method |
3 单选 | delivery.method |
DeliveryMethod Enum |
✅ |
delivery_time |
5 日期 | delivery.delivered_at |
datetime |
⬜ |
delivery_status |
3 单选 | delivery.status |
DeliveryStatus Enum |
✅ |
delivery_note |
1 文本 | delivery.note |
str |
⬜ |
worker_depart_time |
5 日期 | install.departed_at |
datetime |
✅ |
worker_arrive_time |
5 日期 | install.arrived_at |
datetime |
✅ |
work_complete_time |
5 日期 | install.completed_at |
datetime |
✅ |
actual_work_hours |
2 数字 | install.actual_hours |
Decimal |
✅ |
site_photos_url |
8 附件 | install.photos |
list[Attachment] |
✅ |
extra_work_note |
1 文本 | install.extra_work_note |
str |
⬜ |
exception_note |
1 文本 | install.exception_note |
str |
⬜ |
acceptance_result |
3 单选 | acceptance.result |
AcceptanceResult Enum |
✅ |
acceptance_time |
5 日期 | acceptance.accepted_at |
datetime |
✅ |
acceptance_sign_url |
8 附件 | acceptance.sign_url |
Attachment |
✅ |
nps_score |
2 数字 | followup.nps |
int |
✅ |
followup_note |
1 文本 | followup.note |
str |
⬜ |
修订记录¶
| 版本 | 日期 | 修订人 | 修订内容 |
|---|---|---|---|
| V1.0 | 2026-08-09 | DT | 初始版本：Feishu SDK 封装 · 双轨同步引擎 · Webhook 11 类型 · CPT 6 表 246 字段映射附录 A · DLQ + 观测 · 12 人天交付物 |
| V1.1 | 2026-08-09 | DT | 标注暂不实施（对齐 ADR-008）：① 顶部新增醒目「⚠️ 本文档暂不实施」区块，含决策依据（ADR-008）、跳过理由、未来启动条件、Code Agent 行为指引、相关文档同步状态；② §1.1 阶段 1 定义前增加「⚠️ 以下内容为阶段 1 启动时的技术设计，当前暂不实施」标注；③ 关联文档版本号同步（DIP1-ARC V2.0 / DIP1-P2 V2.0）；④ 内容主体不变，保留作为未来阶段 0 启动时的技术参考；⑤ 文末新增修订记录 |
本文件为 DIP1 阶段 1 飞书工作台对接设计（DOC-D01-D1 V1.1，⏸ 暂不实施，对齐 ADR-008）。Code Agent 跳过本文档，直接实施 DIP1-P2 V2.0 一线操作系统设计。