mirror of
https://github.com/Tencent/WeKnora.git
synced 2026-10-13 16:57:24 +08:00
feat(datasource/dingtalk): 摄取钉钉"上传文件"（docx/pptx/xlsx/pdf） (#3786)
* feat(datasource/dingtalk): 摄取钉钉"上传文件"（docx/pptx/xlsx/pdf）
钉钉连接器此前只导入原生 .adoc 文档：知识库里的上传文件既不会出现在选择器里，
也不会被同步，而且没有任何提示。
- 节点判定拆成 isOnlineDocument / isBinaryDocument / binaryContentType，isDocument()
变成"在线文档 ∪ 二进制文档"，上传文件因此可被枚举与摄取；表格（axls）等其它原生
类型仍不在其中，避免枚举出同步读不了的节点。
- 上传文件通过钉盘下载 API 逐字节传输，交给 WeKnora 自己的解析器：先用
/v2.0/doc/dentries/{id}/queryDentryId 把节点 UUID 换成数字 spaceId/dentryId
（存储 API 拒绝非数字 spaceId），再取预签名 URL 下载。
- 大小上限加在传输上而不是列表元数据上：node.size 与存储对象并不一致。
- 预签名 URL 短命且带版本，传输失败会用新签名重试。
- Validate 对上传文件"解析位置但不传 payload"，凭证校验不必为每个候选文件下载
数 MB；adoc 仍走 blocks API。
- 同步按类型分派：adoc 渲染 blocks，上传文件落原始字节并带上具体 MIME 类型与只含
一个扩展名的文件名；下载失败与读取失败一样是可重试的失败项，不推进游标。
本提交只含 C（上传文件摄取）。不含：校验循环容错（A）、跳过项日志与可见性（B）、
原生表格 axls（D）、多维表/手工引用/前端（E）。
* test(datasource/dingtalk): 上传文件用例改用合成文档名与占位 ID
- 夹具、断言与文件名注释里的文档名统一换成合成名 Employee-Handbook.docx，
用例语义不变：仍覆盖"名字已带扩展名时不重复追加"与 PK\x03\x04 字节断言。
- 去掉测试注释里"as verified against the live tenant"的表述，改为描述调用顺序。
- dentry 响应示例与用例中的数字 ID 换成明确占位符，不再出现可被误认的真实 ID。
* fix(datasource/dingtalk): 上传文件摄取改为数据源级开关，超限不再重试
评审意见（#3786）：上传文件是本次新增的摄取路径，存量数据源升级后无需用户显式启用
就会在下一次同步（含增量）开始下载并向量化所选范围内的 docx/pptx/xlsx/pdf，可能
增加存储与 embedding 成本；同时单次下载里超出体积上限被当成可重试错误，没有
Content-Length 的超大文件会重签名再下满 3 次。
- 新增 `Settings.include_uploaded_files`（默认 false，未设置/非法值一律按 false）。
读取走与本仓库语雀连接器同名的 settingBool(v, def)，并新增 documentSettings /
parseDocumentSettings 承载开关。
- 判定拆成"有没有读取路径"（isDocument）与"这个数据源收不收"（isIngestible）：
Validate 的探测、ListResources、全量同步与增量同步（共用 scanWorkspace）四处
全部改用 isIngestible，关闭时行为与 main 一致。
- skipReason 对"能读但没开开关"的上传文件给出可操作的原因：说明
include_uploaded_files 未启用，而不是伪装成类型不支持；媒体文件与真正没有
摄取路径的类型保持原有说法。
- readLimited 改为返回哨兵错误 errDocumentTooLarge（同时包装上限字节数），
fetchSigned 据此返回 retryable=false：体积是对象的属性，新签名只会拿到同样的
字节。普通读错误仍然 retryable=true，仍会重签名重试。
- 前端数据源编辑弹窗（同步策略步骤）加对应开关，未触碰时不写回 settings；5 个
语言包补齐文案。
- website-docs 钉钉一节写明支持范围、默认值、64 MiB 单文件上限与成本影响。
测试补 include_uploaded_files 的未设置/显式关闭/显式开启三种状态：校验、资源列表、
全量同步、增量同步四处各自断言，直接选择上传文件在关闭时仍是失败的资源；流式超限
用例断言签名查询与对象下载各只发生一次，另加"普通读错误仍重试"的用例。
---------
Co-authored-by: linus-liu-web <11640009+linus-liu-web@users.noreply.github.com>
Co-authored-by: linus-liu-web <linus-liu-web@users.noreply.github.com>
This commit is contained in:
linus-liu-web
co-authored by
linus-liu-web
linus-liu-web
linus-liu-web
linus-liu-web
parent
abc365f919
commit
8319d59dd0
@@ -7057,6 +7057,9 @@ export default {
|
||||
yuqueFolderModeHint: 'Files documents by their Yuque table-of-contents path. Note: folder moves made in the knowledge base afterwards are overwritten by the Yuque structure the next time that document syncs.',
|
||||
yuqueTOCOnly: 'Sync only documents visible in the Yuque TOC',
|
||||
yuqueTOCOnlyHint: 'Requires the "Mirror the Yuque TOC" layout. Documents already in the knowledge base are untouched — one the Yuque TOC does not list is simply no longer added, never deleted.',
|
||||
dingtalkIngestLabel: 'DingTalk ingestion scope',
|
||||
dingtalkIncludeUploadedFiles: 'Ingest uploaded files (docx / pptx / xlsx / pdf)',
|
||||
dingtalkIncludeUploadedFilesHint: 'Off by default. When on, Word / PowerPoint / Excel / PDF files uploaded to the selected scope are downloaded and parsed into the knowledge base, up to 64 MiB per file; their text counts towards storage and embedding cost. When off they are simply skipped — the sync log says the switch is not enabled, and copies already in the knowledge base are kept.',
|
||||
prereqBarText_dingtalk: 'First time? Click to see the DingTalk app setup guide',
|
||||
prereqStep1Brief_dingtalk: 'Create an internal enterprise app',
|
||||
prereqStep1Desc_dingtalk: 'Create an internal app in DingTalk Open Platform and copy its Client ID and Client Secret.',
|
||||
|
||||
@@ -7057,6 +7057,9 @@ export default {
|
||||
yuqueFolderModeHint: 'Yuqueの目次階層に従って文書を配置します。なお、その後ナレッジベース上で手動で移動したフォルダは、次回その文書を同期したときにYuqueの構成へ上書きされます。',
|
||||
yuqueTOCOnly: 'Yuqueの目次に表示される文書だけ同期',
|
||||
yuqueTOCOnlyHint: '「Yuqueの目次どおりに階層化」を選ぶと有効になります。すでにナレッジベースにある文書はそのまま残り、Yuqueの目次にない文書は新たに追加されなくなるだけで、削除はされません。',
|
||||
dingtalkIngestLabel: 'DingTalk の取り込み範囲',
|
||||
dingtalkIncludeUploadedFiles: 'アップロードファイル（docx / pptx / xlsx / pdf）を取り込む',
|
||||
dingtalkIncludeUploadedFilesHint: '既定ではオフです。オンにすると、選択範囲にアップロードされた Word / PowerPoint / Excel / PDF ファイルをダウンロードして解析し、ナレッジベースに取り込みます（1 ファイルあたり 64 MiB まで）。本文はストレージと埋め込みのコストに算入されます。オフの場合は単にスキップされ、同期ログにスイッチが無効である旨が記録されます。既にナレッジベースにあるコピーは削除されません。',
|
||||
prereqBarText_dingtalk: '初めての利用ですか？DingTalkアプリの設定ガイドを確認',
|
||||
prereqStep1Brief_dingtalk: '企業内部アプリを作成',
|
||||
prereqStep1Desc_dingtalk: 'DingTalk開発者プラットフォームで企業内部アプリを作成し、Client IDとClient Secretを取得します。',
|
||||
|
||||
@@ -1069,6 +1069,9 @@ export default {
|
||||
yuqueFolderModeHint: 'Yuque 목차 계층에 따라 문서를 배치합니다. 참고: 이후 지식베이스에서 직접 옮긴 폴더는 해당 문서가 다음에 동기화될 때 Yuque 구조로 덮어써집니다.',
|
||||
yuqueTOCOnly: 'Yuque 목차에 표시되는 문서만 동기화',
|
||||
yuqueTOCOnlyHint: '"Yuque 목차대로 계층 구성"을 선택해야 동작합니다. 이미 지식베이스에 있는 문서는 그대로 유지되며, Yuque 목차에 없는 문서는 새로 추가되지 않을 뿐 삭제되지 않습니다.',
|
||||
dingtalkIngestLabel: 'DingTalk 수집 범위',
|
||||
dingtalkIncludeUploadedFiles: '업로드 파일(docx / pptx / xlsx / pdf) 수집',
|
||||
dingtalkIncludeUploadedFilesHint: '기본값은 꺼짐입니다. 켜면 선택한 범위에 업로드된 Word / PowerPoint / Excel / PDF 파일을 내려받아 파싱해 지식베이스에 넣습니다(파일당 최대 64 MiB). 본문은 저장소와 임베딩 비용에 포함됩니다. 꺼져 있으면 해당 파일은 건너뛰고 동기화 로그에 스위치가 꺼져 있다고 기록되며, 이미 지식베이스에 있는 사본은 삭제되지 않습니다.',
|
||||
prereqBarText_dingtalk: '처음 사용하시나요? 클릭하여 DingTalk 앱 설정 가이드를 확인하세요',
|
||||
prereqStep1Brief_dingtalk: '기업 내부 앱 생성',
|
||||
prereqStep1Desc_dingtalk: 'DingTalk Open Platform에서 기업 내부 앱을 생성하고 Client ID와 Client Secret을 복사하세요.',
|
||||
|
||||
@@ -1045,6 +1045,9 @@ export default {
|
||||
yuqueFolderModeHint: 'Раскладывает документы по папкам согласно оглавлению Yuque. Обратите внимание: папки, перемещённые вручную в базе знаний, будут перезаписаны структурой Yuque при следующей синхронизации этого документа.',
|
||||
yuqueTOCOnly: 'Синхронизировать только видимые в оглавлении Yuque документы',
|
||||
yuqueTOCOnlyHint: 'Требуется режим «Как в оглавлении Yuque». Документы, уже добавленные в базу знаний, остаются нетронутыми — документ, которого нет в оглавлении Yuque, просто перестаёт добавляться, но не удаляется.',
|
||||
dingtalkIngestLabel: 'Область импорта DingTalk',
|
||||
dingtalkIncludeUploadedFiles: 'Импортировать загруженные файлы (docx / pptx / xlsx / pdf)',
|
||||
dingtalkIncludeUploadedFilesHint: 'По умолчанию выключено. Если включено, файлы Word / PowerPoint / Excel / PDF, загруженные в выбранную область, скачиваются и разбираются в базу знаний — до 64 МиБ на файл; их текст учитывается в расходах на хранение и эмбеддинги. Если выключено, они просто пропускаются: в журнале синхронизации указывается, что переключатель не включён, а уже загруженные копии сохраняются.',
|
||||
prereqBarText_dingtalk: 'Используете впервые? Откройте руководство по настройке приложения DingTalk',
|
||||
prereqStep1Brief_dingtalk: 'Создайте внутреннее корпоративное приложение',
|
||||
prereqStep1Desc_dingtalk: 'Создайте внутреннее приложение в DingTalk Open Platform и скопируйте Client ID и Client Secret.',
|
||||
|
||||
@@ -1047,6 +1047,9 @@ export default {
|
||||
yuqueFolderModeHint: '按语雀的目录层级放置文档。注意：之后在知识库里手动调整的目录，会在该文档下次同步时被覆盖回语雀的结构。',
|
||||
yuqueTOCOnly: '只同步语雀目录中可见的文档',
|
||||
yuqueTOCOnlyHint: '需要先选择「按语雀目录分层」。已经同步进知识库的文档不受影响——语雀中不在目录里的文档只是不再新增，不会被删除。',
|
||||
dingtalkIngestLabel: '钉钉摄取范围',
|
||||
dingtalkIncludeUploadedFiles: '摄取知识库中的上传文件（docx / pptx / xlsx / pdf）',
|
||||
dingtalkIncludeUploadedFilesHint: '默认关闭。开启后，所选范围内的 Word / PowerPoint / Excel / PDF 上传文件会被下载并解析入库，单个文件上限 64 MiB；这些文件的正文会计入存储与向量化成本。关闭时它们只是被跳过，同步日志会说明开关未开启，不会删除知识库中已有的副本。',
|
||||
prereqBarText_dingtalk: '首次使用？点击查看钉钉应用配置指引',
|
||||
prereqStep1Brief_dingtalk: '创建企业内部应用',
|
||||
prereqStep1Desc_dingtalk: '在钉钉开放平台创建企业内部应用，并获取 Client ID 和 Client Secret',
|
||||
|
||||
@@ -1047,6 +1047,9 @@ export default {
|
||||
yuqueFolderModeHint: '按語雀的目錄層級放置文件。注意：之後在知識庫裡手動調整的目錄，會在該文件下次同步時被覆蓋回語雀的結構。',
|
||||
yuqueTOCOnly: '只同步語雀目錄中可見的文件',
|
||||
yuqueTOCOnlyHint: '需要先選擇「按語雀目錄分層」。已經同步進知識庫的文件不受影響──語雀中不在目錄裡的文件只是不再新增，不會被刪除。',
|
||||
dingtalkIngestLabel: '釘釘攝取範圍',
|
||||
dingtalkIncludeUploadedFiles: '攝取知識庫中的上傳檔案（docx / pptx / xlsx / pdf）',
|
||||
dingtalkIncludeUploadedFilesHint: '預設關閉。開啟後，所選範圍內的 Word / PowerPoint / Excel / PDF 上傳檔案會被下載並解析入庫，單一檔案上限 64 MiB；這些檔案的正文會計入儲存與向量化成本。關閉時它們只會被跳過，同步日誌會說明開關未開啟，不會刪除知識庫中已有的副本。',
|
||||
prereqBarText_dingtalk: '首次使用？點擊查看釘釘應用設定指引',
|
||||
prereqStep1Brief_dingtalk: '建立企業內部應用',
|
||||
prereqStep1Desc_dingtalk: '在釘釘開放平台建立企業內部應用，並獲取 Client ID 和 Client Secret',
|
||||
|
||||
@@ -341,6 +341,38 @@ test('an existing Yuque data source on the TOC layout reports it', async () => {
|
||||
} finally { f.close() }
|
||||
})
|
||||
|
||||
test('a DingTalk data source reports the uploaded-file switch, defaulting to off', async () => {
|
||||
const off = await fixture({ type: 'dingtalk', settings: {} })
|
||||
try {
|
||||
// The connector treats a missing key as false, so an existing source that
|
||||
// predates this control must render as off — and stay untouched.
|
||||
assert.equal(off.vm.dingtalkIncludeUploadedFiles, false)
|
||||
assert.equal(off.vm.form.config.settings.include_uploaded_files, undefined)
|
||||
} finally { off.close() }
|
||||
|
||||
const on = await fixture({ type: 'dingtalk', settings: { include_uploaded_files: true } })
|
||||
try {
|
||||
assert.equal(on.vm.dingtalkIncludeUploadedFiles, true)
|
||||
} finally { on.close() }
|
||||
|
||||
const explicitOff = await fixture({ type: 'dingtalk', settings: { include_uploaded_files: false } })
|
||||
try {
|
||||
assert.equal(explicitOff.vm.dingtalkIncludeUploadedFiles, false)
|
||||
} finally { explicitOff.close() }
|
||||
})
|
||||
|
||||
test('toggling the DingTalk uploaded-file switch writes the settings key', async () => {
|
||||
const f = await fixture({ type: 'dingtalk', settings: {} })
|
||||
try {
|
||||
f.vm.dingtalkIncludeUploadedFiles = true
|
||||
assert.equal(f.vm.form.config.settings.include_uploaded_files, true)
|
||||
f.vm.dingtalkIncludeUploadedFiles = false
|
||||
assert.equal(f.vm.form.config.settings.include_uploaded_files, false)
|
||||
// The switch is scoped to DingTalk: no other connector's key is written.
|
||||
assert.equal(f.vm.form.config.settings.include_sheets, undefined)
|
||||
} finally { f.close() }
|
||||
})
|
||||
|
||||
test('Cloud hierarchy limitation stays visible after an empty space expansion', async () => {
|
||||
const f = await fixture()
|
||||
try {
|
||||
|
||||
@@ -227,6 +227,19 @@ const yuqueTOCOnly = computed({
|
||||
},
|
||||
})
|
||||
|
||||
// DingTalk ingestion of uploaded Office/PDF files. Like the Yuque switches the
|
||||
// key lives in the raw settings bag, and a data source created before this
|
||||
// control existed carries no key at all — the getter therefore reports the
|
||||
// connector default (off) instead of rendering the checkbox blank, and nothing
|
||||
// is written back until the user toggles it, so merely opening an existing
|
||||
// source never changes what it syncs.
|
||||
const dingtalkIncludeUploadedFiles = computed({
|
||||
get: () => form.value.config.settings?.include_uploaded_files === true,
|
||||
set: (on: boolean) => {
|
||||
form.value.config.settings = { ...form.value.config.settings, include_uploaded_files: on }
|
||||
},
|
||||
})
|
||||
|
||||
// Step 2: Resources
|
||||
const resources = ref<Resource[]>([])
|
||||
const loadingResources = ref(false)
|
||||
@@ -1991,6 +2004,17 @@ const drawerConfirmText = computed(() => {
|
||||
</div>
|
||||
</section>
|
||||
|
||||
<!-- DingTalk only: which extra node types the connector may ingest. -->
|
||||
<section v-if="form.type === 'dingtalk'" class="setting-drawer__section">
|
||||
<h4 class="setting-drawer__section-title">{{ t('datasource.dingtalkIngestLabel') }}</h4>
|
||||
<div class="form-item form-item--flat">
|
||||
<t-checkbox v-model="dingtalkIncludeUploadedFiles">
|
||||
{{ t('datasource.dingtalkIncludeUploadedFiles') }}
|
||||
</t-checkbox>
|
||||
</div>
|
||||
<p class="form-desc">{{ t('datasource.dingtalkIncludeUploadedFilesHint') }}</p>
|
||||
</section>
|
||||
|
||||
<!-- Yuque only: how synced documents are laid out, and what may be admitted. -->
|
||||
<section v-if="form.type === 'yuque'" class="setting-drawer__section">
|
||||
<h4 class="setting-drawer__section-title">{{ t('datasource.yuqueFolderModeLabel') }}</h4>
|
||||
|
||||
@@ -24,6 +24,15 @@ const (
|
||||
maxResponseBytes = 16 << 20
|
||||
maxPages = 1000
|
||||
maxAttempts = 3
|
||||
|
||||
// downloadTimeout is deliberately longer than apiTimeout: the JSON calls
|
||||
// are small, while an uploaded document can be several megabytes.
|
||||
downloadTimeout = 5 * time.Minute
|
||||
|
||||
// maxDocumentBytes bounds a single uploaded file transfer. DingTalk lists a
|
||||
// "size" per node that does not match the stored object, so the cap is
|
||||
// enforced on the transfer itself instead of on listing metadata.
|
||||
maxDocumentBytes = 64 << 20
|
||||
)
|
||||
|
||||
type config struct {
|
||||
@@ -84,16 +93,66 @@ type node struct {
|
||||
HasChildren bool `json:"hasChildren"`
|
||||
}
|
||||
|
||||
// binaryDocumentTypes maps the extension of an uploaded DingTalk file to the
|
||||
// MIME type WeKnora stores it under. Only formats WeKnora can parse are listed;
|
||||
// native ALIDOC files have their own online read APIs and are absent on
|
||||
// purpose, as are media files that would be pointless to index as text.
|
||||
var binaryDocumentTypes = map[string]string{
|
||||
"docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
||||
"pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation",
|
||||
"xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||||
"pdf": "application/pdf",
|
||||
}
|
||||
|
||||
func (n node) isFolder() bool {
|
||||
return strings.EqualFold(n.Type, "FOLDER")
|
||||
}
|
||||
|
||||
func (n node) isDocument() bool {
|
||||
// isOnlineDocument reports whether the node is a native DingTalk document whose
|
||||
// blocks can be read through the doc suites API.
|
||||
func (n node) isOnlineDocument() bool {
|
||||
return strings.EqualFold(n.Type, "FILE") &&
|
||||
strings.EqualFold(n.Category, "ALIDOC") &&
|
||||
strings.EqualFold(n.Extension, "adoc")
|
||||
}
|
||||
|
||||
// binaryContentType returns the MIME type of an uploaded file node. Uploaded
|
||||
// files are listed as DOCUMENT; ALIDOC covers DingTalk's own document family,
|
||||
// which is never downloaded as a file.
|
||||
func (n node) binaryContentType() (string, bool) {
|
||||
if !strings.EqualFold(n.Type, "FILE") || !strings.EqualFold(n.Category, "DOCUMENT") {
|
||||
return "", false
|
||||
}
|
||||
contentType, ok := binaryDocumentTypes[strings.ToLower(strings.TrimSpace(n.Extension))]
|
||||
return contentType, ok
|
||||
}
|
||||
|
||||
// isBinaryDocument reports whether the node is an uploaded file this connector
|
||||
// can download and hand to WeKnora's own parsers.
|
||||
func (n node) isBinaryDocument() bool {
|
||||
_, ok := n.binaryContentType()
|
||||
return ok
|
||||
}
|
||||
|
||||
// isDocument reports whether the connector has a read path for the node at all:
|
||||
// a native adoc document through the blocks API, or an uploaded file through
|
||||
// the storage download API. Whether a given data source takes it is
|
||||
// isIngestible's answer, because an upload is guarded by its own switch.
|
||||
func (n node) isDocument() bool {
|
||||
return n.isOnlineDocument() || n.isBinaryDocument()
|
||||
}
|
||||
|
||||
// isIngestible reports whether this data source ingests the node: a read path
|
||||
// must exist and the switch guarding it must be on. Validate, the resource
|
||||
// picker and both sync modes select nodes through this one predicate, so a
|
||||
// switch that is off is off in every one of them.
|
||||
func (n node) isIngestible(settings documentSettings) bool {
|
||||
if n.isOnlineDocument() {
|
||||
return true
|
||||
}
|
||||
return settings.IncludeUploadedFiles && n.isBinaryDocument()
|
||||
}
|
||||
|
||||
func (n node) title() string {
|
||||
if title := strings.TrimSpace(n.Name); title != "" {
|
||||
return title
|
||||
@@ -123,6 +182,10 @@ type dingTalkAPI interface {
|
||||
listNodes(context.Context, string) ([]node, error)
|
||||
listNodesPage(ctx context.Context, parentNodeID, pageToken string) ([]node, string, error)
|
||||
documentBlocks(context.Context, string) ([]json.RawMessage, error)
|
||||
// verifyDocumentDownload proves an uploaded file is downloadable without
|
||||
// transferring its body, so validation stays cheap for multi-megabyte files.
|
||||
verifyDocumentDownload(context.Context, string) error
|
||||
downloadDocument(context.Context, string) ([]byte, error)
|
||||
}
|
||||
|
||||
type client struct {
|
||||
@@ -133,6 +196,12 @@ type client struct {
|
||||
http *http.Client
|
||||
sleep func(context.Context, time.Duration) error
|
||||
|
||||
// fileHTTP transfers document bytes. It is separate from http so a large
|
||||
// download is not bound by the short control-plane timeout.
|
||||
fileHTTP *http.Client
|
||||
// downloadLimit overrides maxDocumentBytes; zero means the default.
|
||||
downloadLimit int64
|
||||
|
||||
token string
|
||||
tokenExpiry time.Time
|
||||
}
|
||||
@@ -144,10 +213,27 @@ func newClient(cfg *config) *client {
|
||||
appKey: cfg.ClientID,
|
||||
appSecret: cfg.ClientSecret,
|
||||
http: datasource.NewConnectorHTTPClient(apiTimeout),
|
||||
fileHTTP: datasource.NewConnectorHTTPClient(downloadTimeout),
|
||||
sleep: sleepContext,
|
||||
}
|
||||
}
|
||||
|
||||
// transferHTTP returns the client used for byte transfers.
|
||||
func (c *client) transferHTTP() *http.Client {
|
||||
if c.fileHTTP != nil {
|
||||
return c.fileHTTP
|
||||
}
|
||||
return c.http
|
||||
}
|
||||
|
||||
// maxDownloadBytes returns the effective per-document size cap.
|
||||
func (c *client) maxDownloadBytes() int64 {
|
||||
if c.downloadLimit > 0 {
|
||||
return c.downloadLimit
|
||||
}
|
||||
return maxDocumentBytes
|
||||
}
|
||||
|
||||
type accessTokenResponse struct {
|
||||
AccessToken string `json:"accessToken"`
|
||||
ExpireIn int64 `json:"expireIn"`
|
||||
@@ -408,6 +494,196 @@ func (c *client) documentBlocks(ctx context.Context, documentID string) ([]json.
|
||||
return nil, fmt.Errorf("DingTalk document block pagination exceeded %d pages", maxPages)
|
||||
}
|
||||
|
||||
// dentryRef identifies one file inside 钉盘. Wiki node IDs are UUIDs, but the
|
||||
// storage APIs accept only the numeric spaceId/dentryId pair, so every upload
|
||||
// is translated first. Response shape:
|
||||
//
|
||||
// {"dentryId":"<numeric>","spaceId":"<numeric>","dentryUuid":"<uuid>"}
|
||||
type dentryRef struct {
|
||||
DentryID string `json:"dentryId"`
|
||||
SpaceID string `json:"spaceId"`
|
||||
DentryUUID string `json:"dentryUuid"`
|
||||
}
|
||||
|
||||
// resolveDentry maps a wiki node UUID to its 钉盘 storage identifiers.
|
||||
func (c *client) resolveDentry(ctx context.Context, documentID string) (dentryRef, error) {
|
||||
query := url.Values{"operatorId": {c.operator}}
|
||||
path := "/v2.0/doc/dentries/" + url.PathEscape(documentID) +
|
||||
"/queryDentryId?" + query.Encode()
|
||||
|
||||
var ref dentryRef
|
||||
if err := c.doJSON(ctx, http.MethodGet, path, nil, true, &ref); err != nil {
|
||||
return dentryRef{}, fmt.Errorf("resolve DingTalk document location: %w", err)
|
||||
}
|
||||
ref.DentryID = strings.TrimSpace(ref.DentryID)
|
||||
ref.SpaceID = strings.TrimSpace(ref.SpaceID)
|
||||
if ref.DentryID == "" || ref.SpaceID == "" {
|
||||
return dentryRef{}, errors.New("DingTalk returned no storage location for the document")
|
||||
}
|
||||
return ref, nil
|
||||
}
|
||||
|
||||
// downloadTarget is a short-lived pre-signed URL for one document version.
|
||||
type downloadTarget struct {
|
||||
URL string
|
||||
Headers map[string]string
|
||||
}
|
||||
|
||||
type downloadInfoResponse struct {
|
||||
Protocol string `json:"protocol"`
|
||||
HeaderSignatureInfo *struct {
|
||||
Headers map[string]string `json:"headers"`
|
||||
ResourceURLs []string `json:"resourceUrls"`
|
||||
ExpirationSeconds int64 `json:"expirationSeconds"`
|
||||
} `json:"headerSignatureInfo"`
|
||||
}
|
||||
|
||||
// requestDownloadURL asks the storage API for a pre-signed download URL.
|
||||
// preferIntranet stays false so the URL is reachable from outside DingTalk's
|
||||
// own network.
|
||||
func (c *client) requestDownloadURL(ctx context.Context, ref dentryRef) (downloadTarget, error) {
|
||||
query := url.Values{"unionId": {c.operator}}
|
||||
path := "/v1.0/storage/spaces/" + url.PathEscape(ref.SpaceID) +
|
||||
"/dentries/" + url.PathEscape(ref.DentryID) + "/downloadInfos/query?" + query.Encode()
|
||||
body := map[string]interface{}{
|
||||
"option": map[string]interface{}{"version": 1, "preferIntranet": false},
|
||||
}
|
||||
|
||||
var response downloadInfoResponse
|
||||
if err := c.doJSON(ctx, http.MethodPost, path, body, true, &response); err != nil {
|
||||
return downloadTarget{}, fmt.Errorf("query DingTalk download URL: %w", err)
|
||||
}
|
||||
info := response.HeaderSignatureInfo
|
||||
if info == nil || len(info.ResourceURLs) == 0 {
|
||||
return downloadTarget{}, fmt.Errorf(
|
||||
"DingTalk returned no download URL for the document (protocol %q)",
|
||||
strings.TrimSpace(response.Protocol))
|
||||
}
|
||||
target := downloadTarget{
|
||||
URL: strings.TrimSpace(info.ResourceURLs[0]),
|
||||
Headers: info.Headers,
|
||||
}
|
||||
if target.URL == "" {
|
||||
return downloadTarget{}, errors.New("DingTalk returned an empty download URL")
|
||||
}
|
||||
return target, nil
|
||||
}
|
||||
|
||||
// verifyDocumentDownload proves an uploaded file is reachable end to end
|
||||
// without transferring its body: the node must resolve to a storage location
|
||||
// and yield a pre-signed URL. Validate uses this so credentials can be checked
|
||||
// without downloading several megabytes per attempt.
|
||||
func (c *client) verifyDocumentDownload(ctx context.Context, documentID string) error {
|
||||
ref, err := c.resolveDentry(ctx, documentID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = c.requestDownloadURL(ctx, ref)
|
||||
return err
|
||||
}
|
||||
|
||||
// downloadDocument fetches the raw bytes of an uploaded document. The pre-signed
|
||||
// URL is versioned and short lived, so it is re-requested whenever a transfer
|
||||
// attempt fails in a way a fresh signature could fix.
|
||||
func (c *client) downloadDocument(ctx context.Context, documentID string) ([]byte, error) {
|
||||
ref, err := c.resolveDentry(ctx, documentID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var lastErr error
|
||||
for attempt := 0; attempt < maxAttempts; attempt++ {
|
||||
if attempt > 0 {
|
||||
if err := c.wait(ctx, retryDelay(attempt-1)); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
target, err := c.requestDownloadURL(ctx, ref)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
data, retryable, err := c.fetchSigned(ctx, target)
|
||||
if err == nil {
|
||||
return data, nil
|
||||
}
|
||||
lastErr = err
|
||||
if !retryable || isContextError(err) {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return nil, lastErr
|
||||
}
|
||||
|
||||
// fetchSigned transfers a document body from a pre-signed URL. It reports
|
||||
// whether the failure is worth retrying with a freshly signed URL.
|
||||
func (c *client) fetchSigned(ctx context.Context, target downloadTarget) ([]byte, bool, error) {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, target.URL, nil)
|
||||
if err != nil {
|
||||
return nil, false, fmt.Errorf("create DingTalk download request: %w", err)
|
||||
}
|
||||
for name, value := range target.Headers {
|
||||
req.Header.Set(name, value)
|
||||
}
|
||||
// The signature covers the object as stored. Asking for an identity
|
||||
// encoding keeps a proxy from re-encoding the bytes into an unparseable
|
||||
// document.
|
||||
req.Header.Set("Accept-Encoding", "identity")
|
||||
|
||||
resp, err := c.transferHTTP().Do(req)
|
||||
if err != nil {
|
||||
return nil, true, fmt.Errorf("download DingTalk document: %w", redactRequestError(err))
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
|
||||
if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
|
||||
// Object-store failures are small XML bodies. Only a bounded prefix is
|
||||
// read, and decodeAPIError never echoes the body verbatim.
|
||||
detail, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
|
||||
retryable := isTransient(resp.StatusCode) ||
|
||||
resp.StatusCode == http.StatusUnauthorized ||
|
||||
resp.StatusCode == http.StatusForbidden ||
|
||||
resp.StatusCode == http.StatusNotFound
|
||||
return nil, retryable, fmt.Errorf(
|
||||
"download DingTalk document: %w", decodeAPIError(resp.StatusCode, detail))
|
||||
}
|
||||
|
||||
limit := c.maxDownloadBytes()
|
||||
if resp.ContentLength > limit {
|
||||
return nil, false, fmt.Errorf(
|
||||
"%w: %d bytes announced, limit is %d bytes",
|
||||
errDocumentTooLarge, resp.ContentLength, limit)
|
||||
}
|
||||
data, err := readLimited(resp.Body, limit)
|
||||
if err != nil {
|
||||
if errors.Is(err, errDocumentTooLarge) {
|
||||
return nil, false, fmt.Errorf("download DingTalk document: %w", err)
|
||||
}
|
||||
return nil, true, fmt.Errorf("read DingTalk document: %w", err)
|
||||
}
|
||||
return data, false, nil
|
||||
}
|
||||
|
||||
// errDocumentTooLarge marks a document whose transfer passed the configured
|
||||
// size cap. The size is a property of the stored object rather than a transient
|
||||
// failure — a freshly signed URL returns the very same bytes — so a caller must
|
||||
// not retry it.
|
||||
var errDocumentTooLarge = errors.New("document exceeds the download limit")
|
||||
|
||||
// readLimited reads at most limit bytes and fails when the body is longer, so
|
||||
// an over-sized document is reported instead of silently truncated. The failure
|
||||
// wraps errDocumentTooLarge so the streaming guard and the announced-length
|
||||
// guard report the same, recognisable error.
|
||||
func readLimited(body io.Reader, limit int64) ([]byte, error) {
|
||||
data, err := io.ReadAll(io.LimitReader(body, limit+1))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if int64(len(data)) > limit {
|
||||
return nil, fmt.Errorf("%w of %d bytes", errDocumentTooLarge, limit)
|
||||
}
|
||||
return data, nil
|
||||
}
|
||||
|
||||
type apiError struct {
|
||||
status int
|
||||
code string
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
@@ -258,6 +259,335 @@ func TestReadBodyRejectsOversizedResponses(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// The whole uploaded-file chain, in the order the provider is called:
|
||||
// node UUID → numeric 钉盘 ids → pre-signed URL → raw bytes.
|
||||
func TestClientDownloadsUploadedDocumentThroughStorageAPI(t *testing.T) {
|
||||
const payload = "PK\x03\x04word-bytes"
|
||||
var dentryQueries, downloadQueries, objectFetches int
|
||||
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
switch {
|
||||
case r.URL.Path == "/v1.0/oauth2/accessToken":
|
||||
_, _ = w.Write([]byte(`{"accessToken":"token","expireIn":7200}`))
|
||||
case strings.HasSuffix(r.URL.Path, "/queryDentryId"):
|
||||
dentryQueries++
|
||||
if r.Method != http.MethodGet ||
|
||||
r.URL.EscapedPath() != "/v2.0/doc/dentries/node%2Fid/queryDentryId" {
|
||||
t.Errorf("dentry request = %s %s", r.Method, r.URL.EscapedPath())
|
||||
}
|
||||
if r.URL.Query().Get("operatorId") != "union/user" ||
|
||||
r.Header.Get("x-acs-dingtalk-access-token") != "token" {
|
||||
t.Errorf("dentry request query = %#v, headers = %#v", r.URL.Query(), r.Header)
|
||||
}
|
||||
_, _ = w.Write([]byte(
|
||||
`{"dentryId":"100000000001","spaceId":"200000000002","dentryUuid":"node/id"}`,
|
||||
))
|
||||
case strings.HasSuffix(r.URL.Path, "/downloadInfos/query"):
|
||||
downloadQueries++
|
||||
if r.Method != http.MethodPost ||
|
||||
r.URL.Path != "/v1.0/storage/spaces/200000000002/dentries/100000000001/downloadInfos/query" {
|
||||
t.Errorf("download request = %s %s", r.Method, r.URL.Path)
|
||||
}
|
||||
if r.URL.Query().Get("unionId") != "union/user" {
|
||||
t.Errorf("download request query = %#v", r.URL.Query())
|
||||
}
|
||||
body, _ := io.ReadAll(r.Body)
|
||||
if !strings.Contains(string(body), `"preferIntranet":false`) ||
|
||||
!strings.Contains(string(body), `"version":1`) {
|
||||
t.Errorf("download request body = %s", body)
|
||||
}
|
||||
_, _ = w.Write([]byte(`{
|
||||
"protocol":"HEADER_SIGNATURE",
|
||||
"headerSignatureInfo":{
|
||||
"headers":{"Authorization":"OSS key:signature","x-oss-date":"Sun, 27 Sep 2026 03:34:30 GMT"},
|
||||
"resourceUrls":["http://` + r.Host + `/oss/object"],
|
||||
"expirationSeconds":900
|
||||
}
|
||||
}`))
|
||||
case r.URL.Path == "/oss/object":
|
||||
objectFetches++
|
||||
if r.Header.Get("Authorization") != "OSS key:signature" ||
|
||||
r.Header.Get("x-oss-date") != "Sun, 27 Sep 2026 03:34:30 GMT" {
|
||||
t.Errorf("signed request headers = %#v", r.Header)
|
||||
}
|
||||
if r.Header.Get("Accept-Encoding") != "identity" {
|
||||
t.Errorf("signed request encoding = %q", r.Header.Get("Accept-Encoding"))
|
||||
}
|
||||
_, _ = w.Write([]byte(payload))
|
||||
default:
|
||||
http.NotFound(w, r)
|
||||
}
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
client := testClient(server)
|
||||
data, err := client.downloadDocument(context.Background(), "node/id")
|
||||
if err != nil {
|
||||
t.Fatalf("downloadDocument() error = %v", err)
|
||||
}
|
||||
if string(data) != payload {
|
||||
t.Fatalf("downloadDocument() = %q, want %q", data, payload)
|
||||
}
|
||||
if dentryQueries != 1 || downloadQueries != 1 || objectFetches != 1 {
|
||||
t.Fatalf("requests dentry=%d download=%d object=%d",
|
||||
dentryQueries, downloadQueries, objectFetches)
|
||||
}
|
||||
|
||||
// Validation resolves the same location without transferring the payload.
|
||||
if err := client.verifyDocumentDownload(context.Background(), "node/id"); err != nil {
|
||||
t.Fatalf("verifyDocumentDownload() error = %v", err)
|
||||
}
|
||||
if downloadQueries != 2 || objectFetches != 1 {
|
||||
t.Fatalf("verify transferred bytes: download=%d object=%d", downloadQueries, objectFetches)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClientRejectsIncompleteDentryResponse(t *testing.T) {
|
||||
for _, body := range []string{
|
||||
`{"dentryId":"1"}`,
|
||||
`{"spaceId":"2"}`,
|
||||
`{"dentryId":" ","spaceId":"2"}`,
|
||||
`{}`,
|
||||
} {
|
||||
t.Run(body, func(t *testing.T) {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write([]byte(body))
|
||||
}))
|
||||
defer server.Close()
|
||||
c := testClient(server)
|
||||
c.token = "cached"
|
||||
c.tokenExpiry = time.Now().Add(time.Hour)
|
||||
if _, err := c.resolveDentry(context.Background(), "doc"); err == nil {
|
||||
t.Fatalf("resolveDentry() accepted %s", body)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestClientRejectsDownloadWithoutResourceURL(t *testing.T) {
|
||||
for _, body := range []string{
|
||||
`{}`,
|
||||
`{"protocol":"HEADER_SIGNATURE","headerSignatureInfo":null}`,
|
||||
`{"protocol":"HEADER_SIGNATURE","headerSignatureInfo":{"resourceUrls":[]}}`,
|
||||
`{"protocol":"SIGNATURE","headerSignatureInfo":{"resourceUrls":[" "]}}`,
|
||||
} {
|
||||
t.Run(body, func(t *testing.T) {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write([]byte(body))
|
||||
}))
|
||||
defer server.Close()
|
||||
c := testClient(server)
|
||||
c.token = "cached"
|
||||
c.tokenExpiry = time.Now().Add(time.Hour)
|
||||
_, err := c.requestDownloadURL(context.Background(), dentryRef{DentryID: "1", SpaceID: "2"})
|
||||
if err == nil {
|
||||
t.Fatalf("requestDownloadURL() accepted %s", body)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// An expired pre-signed URL must be re-signed rather than failing the document.
|
||||
func TestClientResignsDownloadURLAfterObjectStoreRefusal(t *testing.T) {
|
||||
var downloadQueries, objectFetches int
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
switch {
|
||||
case strings.HasSuffix(r.URL.Path, "/queryDentryId"):
|
||||
_, _ = w.Write([]byte(`{"dentryId":"1","spaceId":"2"}`))
|
||||
case strings.HasSuffix(r.URL.Path, "/downloadInfos/query"):
|
||||
downloadQueries++
|
||||
_, _ = w.Write([]byte(`{"protocol":"HEADER_SIGNATURE","headerSignatureInfo":{
|
||||
"headers":{"Authorization":"OSS key:sig` + string(rune('0'+downloadQueries)) + `"},
|
||||
"resourceUrls":["http://` + r.Host + `/oss/object"]}}`))
|
||||
case r.URL.Path == "/oss/object":
|
||||
objectFetches++
|
||||
if objectFetches == 1 {
|
||||
w.WriteHeader(http.StatusForbidden)
|
||||
_, _ = w.Write([]byte(`<Error><Code>SignatureDoesNotMatch</Code></Error>`))
|
||||
return
|
||||
}
|
||||
if r.Header.Get("Authorization") != "OSS key:sig2" {
|
||||
t.Errorf("retry reused the stale signature: %#v", r.Header)
|
||||
}
|
||||
_, _ = w.Write([]byte("bytes"))
|
||||
default:
|
||||
http.NotFound(w, r)
|
||||
}
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
c := testClient(server)
|
||||
c.token = "cached"
|
||||
c.tokenExpiry = time.Now().Add(time.Hour)
|
||||
data, err := c.downloadDocument(context.Background(), "doc")
|
||||
if err != nil {
|
||||
t.Fatalf("downloadDocument() error = %v", err)
|
||||
}
|
||||
if string(data) != "bytes" || downloadQueries != 2 || objectFetches != 2 {
|
||||
t.Fatalf("data = %q, download=%d object=%d", data, downloadQueries, objectFetches)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClientBoundsDocumentDownloads(t *testing.T) {
|
||||
t.Run("declared length", func(t *testing.T) {
|
||||
var downloadQueries, objectFetches int
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
switch {
|
||||
case strings.HasSuffix(r.URL.Path, "/queryDentryId"):
|
||||
_, _ = w.Write([]byte(`{"dentryId":"1","spaceId":"2"}`))
|
||||
case strings.HasSuffix(r.URL.Path, "/downloadInfos/query"):
|
||||
downloadQueries++
|
||||
_, _ = w.Write([]byte(`{"protocol":"HEADER_SIGNATURE","headerSignatureInfo":{
|
||||
"resourceUrls":["http://` + r.Host + `/oss/object"]}}`))
|
||||
default:
|
||||
objectFetches++
|
||||
w.Header().Set("Content-Length", strconv.Itoa(4096))
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
c := testClient(server)
|
||||
c.token = "cached"
|
||||
c.tokenExpiry = time.Now().Add(time.Hour)
|
||||
c.downloadLimit = 1024
|
||||
data, err := c.downloadDocument(context.Background(), "doc")
|
||||
if err == nil || !errors.Is(err, errDocumentTooLarge) ||
|
||||
!strings.Contains(err.Error(), "1024 bytes") {
|
||||
t.Fatalf("downloadDocument() = %q, %v; want a size refusal", data, err)
|
||||
}
|
||||
if downloadQueries != 1 || objectFetches != 1 {
|
||||
t.Fatalf("over-sized download retried: signature queries = %d, object fetches = %d",
|
||||
downloadQueries, objectFetches)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("streamed length", func(t *testing.T) {
|
||||
var downloadQueries, objectFetches int
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
switch {
|
||||
case strings.HasSuffix(r.URL.Path, "/queryDentryId"):
|
||||
_, _ = w.Write([]byte(`{"dentryId":"1","spaceId":"2"}`))
|
||||
case strings.HasSuffix(r.URL.Path, "/downloadInfos/query"):
|
||||
downloadQueries++
|
||||
_, _ = w.Write([]byte(`{"protocol":"HEADER_SIGNATURE","headerSignatureInfo":{
|
||||
"resourceUrls":["http://` + r.Host + `/oss/object"]}}`))
|
||||
default:
|
||||
objectFetches++
|
||||
// Flushing first drops Content-Length, so only the streaming
|
||||
// guard can stop an over-sized body.
|
||||
w.WriteHeader(http.StatusOK)
|
||||
if flusher, ok := w.(http.Flusher); ok {
|
||||
flusher.Flush()
|
||||
}
|
||||
_, _ = w.Write([]byte(strings.Repeat("x", 4096)))
|
||||
}
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
c := testClient(server)
|
||||
c.token = "cached"
|
||||
c.tokenExpiry = time.Now().Add(time.Hour)
|
||||
c.downloadLimit = 1024
|
||||
if _, err := c.downloadDocument(context.Background(), "doc"); err == nil ||
|
||||
!errors.Is(err, errDocumentTooLarge) ||
|
||||
!strings.Contains(err.Error(), "1024 bytes") {
|
||||
t.Fatalf("downloadDocument() error = %v, want a size refusal", err)
|
||||
}
|
||||
// A body with no Content-Length is only measurable while it is read, so
|
||||
// before the sentinel existed every attempt paid for a fresh signature
|
||||
// and another over-sized read before failing. The size is a property of
|
||||
// the document, not a transient fault: one attempt, one signature, one
|
||||
// transfer.
|
||||
if downloadQueries != 1 || objectFetches != 1 {
|
||||
t.Fatalf("over-sized download retried: signature queries = %d, object fetches = %d",
|
||||
downloadQueries, objectFetches)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// A read failure that is not a size refusal stays retryable: the transfer is
|
||||
// re-signed and re-attempted, and the document is still returned.
|
||||
func TestClientRetriesTransientReadFailuresWithAFreshSignature(t *testing.T) {
|
||||
var downloadQueries, objectFetches int
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
switch {
|
||||
case strings.HasSuffix(r.URL.Path, "/queryDentryId"):
|
||||
_, _ = w.Write([]byte(`{"dentryId":"1","spaceId":"2"}`))
|
||||
case strings.HasSuffix(r.URL.Path, "/downloadInfos/query"):
|
||||
downloadQueries++
|
||||
_, _ = w.Write([]byte(`{"protocol":"HEADER_SIGNATURE","headerSignatureInfo":{
|
||||
"headers":{"Authorization":"OSS key:sig` + strconv.Itoa(downloadQueries) + `"},
|
||||
"resourceUrls":["http://` + r.Host + `/oss/object"]}}`))
|
||||
default:
|
||||
http.NotFound(w, r)
|
||||
}
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
c := testClient(server)
|
||||
c.token = "cached"
|
||||
c.tokenExpiry = time.Now().Add(time.Hour)
|
||||
// The object store is served by a transport that truncates the first body
|
||||
// mid-stream, which surfaces as a plain read error rather than a size
|
||||
// refusal, and succeeds on the second attempt.
|
||||
c.fileHTTP = &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
|
||||
objectFetches++
|
||||
body := io.NopCloser(strings.NewReader("bytes"))
|
||||
if objectFetches == 1 {
|
||||
body = io.NopCloser(&failingReader{data: "byt"})
|
||||
}
|
||||
return &http.Response{
|
||||
StatusCode: http.StatusOK,
|
||||
Body: body,
|
||||
Header: make(http.Header),
|
||||
Request: req,
|
||||
}, nil
|
||||
})}
|
||||
|
||||
data, err := c.downloadDocument(context.Background(), "doc")
|
||||
if err != nil {
|
||||
t.Fatalf("downloadDocument() error = %v", err)
|
||||
}
|
||||
if string(data) != "bytes" || downloadQueries != 2 || objectFetches != 2 {
|
||||
t.Fatalf("data = %q, download=%d object=%d; want a re-signed retry",
|
||||
data, downloadQueries, objectFetches)
|
||||
}
|
||||
}
|
||||
|
||||
// failingReader yields its data and then fails, standing in for a connection
|
||||
// dropped in the middle of a transfer.
|
||||
type failingReader struct {
|
||||
data string
|
||||
read bool
|
||||
}
|
||||
|
||||
func (r *failingReader) Read(p []byte) (int, error) {
|
||||
if r.read {
|
||||
return 0, errors.New("connection reset by peer")
|
||||
}
|
||||
r.read = true
|
||||
return copy(p, r.data), nil
|
||||
}
|
||||
|
||||
func TestReadLimitedRejectsOverlongBody(t *testing.T) {
|
||||
if data, err := readLimited(strings.NewReader("1234"), 4); err != nil || string(data) != "1234" {
|
||||
t.Fatalf("readLimited() = %q, %v", data, err)
|
||||
}
|
||||
if _, err := readLimited(strings.NewReader("12345"), 4); err == nil ||
|
||||
!errors.Is(err, errDocumentTooLarge) ||
|
||||
!strings.Contains(err.Error(), "download limit") {
|
||||
t.Fatalf("readLimited() error = %v, want errDocumentTooLarge", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDocumentBlocksRejectsIncompleteResponse(t *testing.T) {
|
||||
for _, body := range []string{`{}`, `{"success":false}`, `{"success":true}`, `{"success":true,"result":null}`} {
|
||||
t.Run(body, func(t *testing.T) {
|
||||
|
||||
@@ -45,12 +45,18 @@ var unsupportedDocumentExtensions = map[string]string{
|
||||
// just listed will not be ingested. Both answers are deterministic: retrying
|
||||
// the sync can never turn such a node into a document, which is what separates
|
||||
// a skip from a failed read that must be retried.
|
||||
func skipReason(n node) string {
|
||||
func skipReason(n node, settings documentSettings) string {
|
||||
extension := strings.ToLower(strings.TrimSpace(n.Extension))
|
||||
_, knownMediaExtension := mediaExtensions[extension]
|
||||
if strings.EqualFold(n.Category, "VIDEO") || knownMediaExtension {
|
||||
return "video/media files are deliberately not downloaded by this connector"
|
||||
}
|
||||
// An uploaded file this connector could read, but this data source was
|
||||
// never told to. Naming the switch makes the skip actionable instead of
|
||||
// looking like a missing feature.
|
||||
if n.isBinaryDocument() && !settings.IncludeUploadedFiles {
|
||||
return "uploaded files are not ingested because include_uploaded_files is not enabled for this data source"
|
||||
}
|
||||
if label := unsupportedDocumentExtensions[extension]; label != "" {
|
||||
return label + " has no ingest path in this connector yet"
|
||||
}
|
||||
@@ -77,7 +83,8 @@ var (
|
||||
|
||||
type apiFactory func(*config) dingTalkAPI
|
||||
|
||||
// Connector imports native DingTalk documents through the Wiki and Blocks APIs.
|
||||
// Connector imports DingTalk wiki documents: native adoc documents through the
|
||||
// blocks API, and uploaded Office/PDF files through the 钉盘 download API.
|
||||
type Connector struct {
|
||||
newAPI apiFactory
|
||||
}
|
||||
@@ -118,6 +125,7 @@ func (c *Connector) Validate(ctx context.Context, dataSourceConfig *types.DataSo
|
||||
return err
|
||||
}
|
||||
api := c.api(cfg)
|
||||
settings := parseDocumentSettings(dataSourceConfig)
|
||||
workspaces, err := api.listWorkspaces(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("validate DingTalk data source: %w", err)
|
||||
@@ -136,7 +144,9 @@ func (c *Connector) Validate(ctx context.Context, dataSourceConfig *types.DataSo
|
||||
if rootNodeID == "" {
|
||||
continue
|
||||
}
|
||||
document, found, unexplored, err := firstValidateDocument(ctx, api, rootNodeID, &listings)
|
||||
document, found, unexplored, err := firstValidateDocument(
|
||||
ctx, api, rootNodeID, settings, &listings,
|
||||
)
|
||||
if err != nil {
|
||||
lastErr = fmt.Errorf("workspace %q: %w", item.Name, err)
|
||||
// A cancelled or timed-out request proves nothing about the
|
||||
@@ -201,8 +211,15 @@ const maxValidateListings = 20
|
||||
// unexplored reports that some nodes were never seen, because the budget ran
|
||||
// out or a folder could not be listed, so the walk cannot prove the workspace
|
||||
// holds no readable document.
|
||||
//
|
||||
// The walk selects nodes with the same settings-aware predicate the picker and
|
||||
// the sync use, so a type the data source does not ingest is never probed.
|
||||
func firstValidateDocument(
|
||||
ctx context.Context, api dingTalkAPI, rootNodeID string, listings *int,
|
||||
ctx context.Context,
|
||||
api dingTalkAPI,
|
||||
rootNodeID string,
|
||||
settings documentSettings,
|
||||
listings *int,
|
||||
) (document node, found, unexplored bool, err error) {
|
||||
queue := []string{rootNodeID}
|
||||
visited := map[string]struct{}{rootNodeID: {}}
|
||||
@@ -235,7 +252,7 @@ func firstValidateDocument(
|
||||
break
|
||||
}
|
||||
for _, child := range children {
|
||||
if child.isDocument() {
|
||||
if child.isIngestible(settings) {
|
||||
return child, true, unexplored, nil
|
||||
}
|
||||
if child.isFolder() || child.HasChildren {
|
||||
@@ -263,16 +280,23 @@ func firstValidateDocument(
|
||||
return node{}, false, unexplored, nil
|
||||
}
|
||||
|
||||
// verifyDocument proves one visible document is readable by calling the read
|
||||
// API that backs its ingest path — the blocks API for adoc today. The caller
|
||||
// only reaches this function for nodes isDocument accepts, and a native type
|
||||
// becomes a document, gains a sync read path and gains its probe here in the
|
||||
// same change. Types that are not ingestible yet are therefore not documents
|
||||
// yet: uploaded files and native spreadsheets join all three together when
|
||||
// their ingest paths land, instead of being probed here in advance.
|
||||
// verifyDocument proves one visible document is readable. Native documents are
|
||||
// probed through their own read API — the blocks API for adoc — while uploaded
|
||||
// files resolve their 钉盘 download location, which exercises the same
|
||||
// permissions and returns the same errors a real download would, without
|
||||
// transferring the payload.
|
||||
//
|
||||
// #3782 landed a narrower version of this probe (the blocks API only) with a
|
||||
// comment saying uploaded files and native spreadsheets join it "when their
|
||||
// ingest paths land". This is that follow-up: an uploaded file is not an adoc,
|
||||
// so probing it through documentBlocks would report a readable file as
|
||||
// unreadable and reject a working data source.
|
||||
func verifyDocument(ctx context.Context, api dingTalkAPI, document node) error {
|
||||
_, err := api.documentBlocks(ctx, document.ID)
|
||||
return err
|
||||
if document.isOnlineDocument() {
|
||||
_, err := api.documentBlocks(ctx, document.ID)
|
||||
return err
|
||||
}
|
||||
return api.verifyDocumentDownload(ctx, document.ID)
|
||||
}
|
||||
|
||||
// ListResources lazily lists selectable workspaces, folders and documents.
|
||||
@@ -286,6 +310,7 @@ func (c *Connector) ListResources(
|
||||
return nil, err
|
||||
}
|
||||
api := c.api(cfg)
|
||||
settings := parseDocumentSettings(dataSourceConfig)
|
||||
if strings.TrimSpace(parentID) == "" {
|
||||
workspaces, err := api.listWorkspaces(ctx)
|
||||
if err != nil {
|
||||
@@ -345,7 +370,7 @@ func (c *Connector) ListResources(
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
scopes, failures, err := resolveSyncScopes(ctx, api, workspaces, []string{parentID})
|
||||
scopes, failures, err := resolveSyncScopes(ctx, api, workspaces, []string{parentID}, settings)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -383,7 +408,7 @@ func (c *Connector) ListResources(
|
||||
}
|
||||
resources := make([]types.Resource, 0, len(children))
|
||||
for _, child := range children {
|
||||
if !child.isFolder() && !child.isDocument() {
|
||||
if !child.isFolder() && !child.isIngestible(settings) {
|
||||
continue
|
||||
}
|
||||
if child.WorkspaceID != "" && child.WorkspaceID != parentRef.WorkspaceID {
|
||||
@@ -599,11 +624,12 @@ func (c *Connector) sync(
|
||||
}
|
||||
|
||||
api := c.api(cfg)
|
||||
settings := parseDocumentSettings(dataSourceConfig)
|
||||
workspaces, err := api.listWorkspaces(ctx)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
scopes, failures, err := resolveSyncScopes(ctx, api, workspaces, selected)
|
||||
scopes, failures, err := resolveSyncScopes(ctx, api, workspaces, selected, settings)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
@@ -639,7 +665,7 @@ func (c *Connector) sync(
|
||||
oldRevisions = stored
|
||||
}
|
||||
}
|
||||
documents, skipped, err := scanScope(ctx, api, scope)
|
||||
documents, skipped, err := scanScope(ctx, api, scope, settings)
|
||||
if err != nil {
|
||||
if isContextError(err) {
|
||||
return nil, nil, err
|
||||
@@ -660,7 +686,7 @@ func (c *Connector) sync(
|
||||
// read below, which must stay retryable.
|
||||
for _, node := range skipped {
|
||||
logger.Infof(ctx, "[DingTalk] skip node %s (name=%q type=%s category=%s extension=%s): %s",
|
||||
node.ID, node.title(), node.Type, node.Category, node.Extension, skipReason(node))
|
||||
node.ID, node.title(), node.Type, node.Category, node.Extension, skipReason(node, settings))
|
||||
}
|
||||
|
||||
newRevisions := make(map[string]string, len(documents))
|
||||
@@ -681,7 +707,7 @@ func (c *Connector) sync(
|
||||
continue
|
||||
}
|
||||
|
||||
blocks, err := api.documentBlocks(ctx, document.ID)
|
||||
item, err := readDocument(ctx, api, scope.ResourceID, scope.Reference.WorkspaceID, document)
|
||||
if err != nil {
|
||||
if isContextError(err) {
|
||||
return nil, nil, err
|
||||
@@ -700,13 +726,7 @@ func (c *Connector) sync(
|
||||
}
|
||||
continue
|
||||
}
|
||||
rendered := renderDocument(document.title(), blocks)
|
||||
// The renderer stays context-free; the warning is emitted here,
|
||||
// where both the request context and the node identity are known.
|
||||
warnUnknownBlockTypes(ctx, document, rendered)
|
||||
items = append(items, fetchedDocument(
|
||||
scope.ResourceID, scope.Reference.WorkspaceID, document, rendered,
|
||||
))
|
||||
items = append(items, item)
|
||||
synced++
|
||||
newRevisions[document.ID] = revision
|
||||
}
|
||||
@@ -768,6 +788,7 @@ func resolveSyncScopes(
|
||||
api dingTalkAPI,
|
||||
workspaces []workspace,
|
||||
resourceIDs []string,
|
||||
settings documentSettings,
|
||||
) ([]syncScope, map[string]error, error) {
|
||||
byID := make(map[string]workspace, len(workspaces))
|
||||
for _, item := range workspaces {
|
||||
@@ -843,7 +864,7 @@ func resolveSyncScopes(
|
||||
return syncScope{
|
||||
ResourceID: canonicalID, Reference: ref, StartNodeID: selectedNode.ID,
|
||||
}, nil
|
||||
case selectedNode.isDocument():
|
||||
case selectedNode.isIngestible(settings):
|
||||
document := selectedNode
|
||||
return syncScope{
|
||||
ResourceID: canonicalID, Reference: ref, Document: &document,
|
||||
@@ -910,11 +931,16 @@ func childByID(children []node, nodeID string) (node, bool) {
|
||||
// nodes it saw but cannot ingest. The skipped nodes are returned rather than
|
||||
// discarded so the caller can report them: a full sync must never look clean
|
||||
// while silently dropping part of the tree.
|
||||
func scanScope(ctx context.Context, api dingTalkAPI, scope syncScope) ([]node, []node, error) {
|
||||
func scanScope(
|
||||
ctx context.Context,
|
||||
api dingTalkAPI,
|
||||
scope syncScope,
|
||||
settings documentSettings,
|
||||
) ([]node, []node, error) {
|
||||
if scope.Document != nil {
|
||||
return []node{*scope.Document}, nil, nil
|
||||
}
|
||||
return scanWorkspace(ctx, api, scope.Reference.WorkspaceID, scope.StartNodeID)
|
||||
return scanWorkspace(ctx, api, scope.Reference.WorkspaceID, scope.StartNodeID, settings)
|
||||
}
|
||||
|
||||
// scanWorkspace walks a workspace subtree breadth-first. Folders are traversal
|
||||
@@ -924,6 +950,7 @@ func scanWorkspace(
|
||||
api dingTalkAPI,
|
||||
workspaceID string,
|
||||
rootNodeID string,
|
||||
settings documentSettings,
|
||||
) ([]node, []node, error) {
|
||||
queue := []string{rootNodeID}
|
||||
visitedParents := make(map[string]struct{})
|
||||
@@ -961,7 +988,7 @@ func scanWorkspace(
|
||||
return nil, nil, fmt.Errorf("DingTalk workspace exceeds %d nodes", maxTraversalNodes)
|
||||
}
|
||||
switch {
|
||||
case child.isDocument():
|
||||
case child.isIngestible(settings):
|
||||
documents = append(documents, child)
|
||||
case child.isFolder():
|
||||
// Containers hold no content of their own, so they are never
|
||||
@@ -1049,39 +1076,143 @@ func warnUnknownBlockTypes(ctx context.Context, document node, rendered renderRe
|
||||
common.PipelineWarn(ctx, "DingTalkConnector", "unknown_block_types", unknown.fields())
|
||||
}
|
||||
|
||||
func fetchedDocument(
|
||||
// readDocument turns one supported wiki node into a FetchedItem. Native adoc
|
||||
// documents are rendered from their blocks, and uploaded files are transferred
|
||||
// byte-for-byte so WeKnora's own parsers can read them.
|
||||
func readDocument(
|
||||
ctx context.Context,
|
||||
api dingTalkAPI,
|
||||
sourceResourceID string,
|
||||
workspaceID string,
|
||||
document node,
|
||||
rendered renderResult,
|
||||
) types.FetchedItem {
|
||||
metadata := map[string]string{
|
||||
) (types.FetchedItem, error) {
|
||||
switch {
|
||||
case document.isOnlineDocument():
|
||||
blocks, err := api.documentBlocks(ctx, document.ID)
|
||||
if err != nil {
|
||||
return types.FetchedItem{}, err
|
||||
}
|
||||
rendered := renderDocument(document.title(), blocks)
|
||||
// The renderer stays context-free; the warning is emitted here, where
|
||||
// both the request context and the node identity are known. Uploaded
|
||||
// files take the default branch and have no rendering step to lose.
|
||||
warnUnknownBlockTypes(ctx, document, rendered)
|
||||
return fetchedDocument(sourceResourceID, workspaceID, document, rendered), nil
|
||||
default:
|
||||
data, err := api.downloadDocument(ctx, document.ID)
|
||||
if err != nil {
|
||||
return types.FetchedItem{}, err
|
||||
}
|
||||
return fetchedBinaryDocument(sourceResourceID, workspaceID, document, data), nil
|
||||
}
|
||||
}
|
||||
|
||||
// documentMetadata describes a wiki node the same way for rendered and
|
||||
// downloaded documents, so both stay searchable by their DingTalk identity.
|
||||
func documentMetadata(workspaceID string, document node) map[string]string {
|
||||
return map[string]string{
|
||||
"channel": types.ChannelDingtalk,
|
||||
"workspace_id": workspaceID,
|
||||
"node_id": document.ID,
|
||||
"category": document.Category,
|
||||
"extension": document.Extension,
|
||||
}
|
||||
}
|
||||
|
||||
// documentURL is the address a reader can open the node at in DingTalk.
|
||||
func documentURL(document node) string {
|
||||
if url := strings.TrimSpace(document.URL); url != "" {
|
||||
return url
|
||||
}
|
||||
return "https://alidocs.dingtalk.com/i/nodes/" + url.PathEscape(document.ID)
|
||||
}
|
||||
|
||||
func fetchedDocument(
|
||||
sourceResourceID string,
|
||||
workspaceID string,
|
||||
document node,
|
||||
rendered renderResult,
|
||||
) types.FetchedItem {
|
||||
metadata := documentMetadata(workspaceID, document)
|
||||
if len(rendered.UnknownTypes) > 0 {
|
||||
metadata["unknown_block_types"] = strings.Join(rendered.UnknownTypes, ",")
|
||||
}
|
||||
documentURL := strings.TrimSpace(document.URL)
|
||||
if documentURL == "" {
|
||||
documentURL = "https://alidocs.dingtalk.com/i/nodes/" + url.PathEscape(document.ID)
|
||||
}
|
||||
return types.FetchedItem{
|
||||
ExternalID: document.ID,
|
||||
Title: document.title(),
|
||||
Content: []byte(rendered.Markdown),
|
||||
ContentType: "text/markdown",
|
||||
FileName: sanitizeFilename(document.title()) + ".md",
|
||||
URL: documentURL,
|
||||
FileName: renderedDocumentFileName(document.title()),
|
||||
URL: documentURL(document),
|
||||
UpdatedAt: document.modifiedAt(),
|
||||
Metadata: metadata,
|
||||
SourceResourceID: sourceResourceID,
|
||||
}
|
||||
}
|
||||
|
||||
// renderedDocumentFileName names a document this connector renders as markdown.
|
||||
// The node's name is used as reported — a DingTalk name normally carries its
|
||||
// own extension — but ".md" is appended exactly once, so a name that already
|
||||
// ends in .md does not grow a second extension.
|
||||
func renderedDocumentFileName(title string) string {
|
||||
name := sanitizeFilename(title)
|
||||
if stem, trimmed := trimSuffixFold(name, ".md"); trimmed {
|
||||
name = stem
|
||||
}
|
||||
return name + ".md"
|
||||
}
|
||||
|
||||
// fetchedBinaryDocument wraps an uploaded file's raw bytes. The file name keeps
|
||||
// the original extension because that is what selects the parser during
|
||||
// ingestion; the MIME type records the same fact and is set to a concrete Office
|
||||
// type rather than the generic octet-stream the object store reports.
|
||||
func fetchedBinaryDocument(
|
||||
sourceResourceID string,
|
||||
workspaceID string,
|
||||
document node,
|
||||
data []byte,
|
||||
) types.FetchedItem {
|
||||
contentType, ok := document.binaryContentType()
|
||||
if !ok {
|
||||
contentType = "application/octet-stream"
|
||||
}
|
||||
return types.FetchedItem{
|
||||
ExternalID: document.ID,
|
||||
Title: document.title(),
|
||||
Content: data,
|
||||
ContentType: contentType,
|
||||
FileName: binaryDocumentFileName(document),
|
||||
URL: documentURL(document),
|
||||
UpdatedAt: document.modifiedAt(),
|
||||
Metadata: documentMetadata(workspaceID, document),
|
||||
SourceResourceID: sourceResourceID,
|
||||
}
|
||||
}
|
||||
|
||||
// binaryDocumentFileName derives the stored file name. DingTalk node names
|
||||
// normally already end in the extension, so it is replaced rather than
|
||||
// appended: "Employee-Handbook.docx" must not become
|
||||
// "Employee-Handbook.docx.docx".
|
||||
func binaryDocumentFileName(document node) string {
|
||||
extension := strings.ToLower(strings.TrimSpace(document.Extension))
|
||||
name := strings.TrimSpace(document.title())
|
||||
if extension == "" {
|
||||
return sanitizeFilename(name)
|
||||
}
|
||||
if stem, ok := trimSuffixFold(name, "."+extension); ok {
|
||||
name = stem
|
||||
}
|
||||
return sanitizeFilename(name) + "." + extension
|
||||
}
|
||||
|
||||
// trimSuffixFold removes suffix from value, comparing case-insensitively.
|
||||
func trimSuffixFold(value, suffix string) (string, bool) {
|
||||
if !strings.HasSuffix(strings.ToLower(value), strings.ToLower(suffix)) {
|
||||
return value, false
|
||||
}
|
||||
return value[:len(value)-len(suffix)], true
|
||||
}
|
||||
|
||||
func failedDocument(
|
||||
sourceResourceID string,
|
||||
workspaceID string,
|
||||
|
||||
@@ -14,12 +14,17 @@ import (
|
||||
)
|
||||
|
||||
type fakeAPI struct {
|
||||
workspaces []workspace
|
||||
nodes map[string][]node
|
||||
blocks map[string][]json.RawMessage
|
||||
nodeErrors map[string]error
|
||||
blockErrors map[string]error
|
||||
blockCalls map[string]int
|
||||
workspaces []workspace
|
||||
nodes map[string][]node
|
||||
blocks map[string][]json.RawMessage
|
||||
downloads map[string][]byte
|
||||
nodeErrors map[string]error
|
||||
blockErrors map[string]error
|
||||
verifyErrors map[string]error
|
||||
dlErrors map[string]error
|
||||
blockCalls map[string]int
|
||||
verifyCalls map[string]int
|
||||
dlCalls map[string]int
|
||||
}
|
||||
|
||||
func (f *fakeAPI) listWorkspaces(context.Context) ([]workspace, error) {
|
||||
@@ -50,6 +55,25 @@ func (f *fakeAPI) documentBlocks(_ context.Context, documentID string) ([]json.R
|
||||
return f.blocks[documentID], nil
|
||||
}
|
||||
|
||||
func (f *fakeAPI) verifyDocumentDownload(_ context.Context, documentID string) error {
|
||||
if f.verifyCalls == nil {
|
||||
f.verifyCalls = make(map[string]int)
|
||||
}
|
||||
f.verifyCalls[documentID]++
|
||||
return f.verifyErrors[documentID]
|
||||
}
|
||||
|
||||
func (f *fakeAPI) downloadDocument(_ context.Context, documentID string) ([]byte, error) {
|
||||
if f.dlCalls == nil {
|
||||
f.dlCalls = make(map[string]int)
|
||||
}
|
||||
f.dlCalls[documentID]++
|
||||
if err := f.dlErrors[documentID]; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return f.downloads[documentID], nil
|
||||
}
|
||||
|
||||
func testConnector(api dingTalkAPI) *Connector {
|
||||
return &Connector{newAPI: func(*config) dingTalkAPI { return api }}
|
||||
}
|
||||
@@ -66,6 +90,15 @@ func testConfig(resources ...string) *types.DataSourceConfig {
|
||||
}
|
||||
}
|
||||
|
||||
// testConfigWithUploadedFiles turns include_uploaded_files on. Cases that
|
||||
// exercise the upload ingest path have to opt in, because the switch defaults
|
||||
// to off and testConfig deliberately leaves the settings bag empty.
|
||||
func testConfigWithUploadedFiles(resources ...string) *types.DataSourceConfig {
|
||||
cfg := testConfig(resources...)
|
||||
cfg.Settings = map[string]interface{}{"include_uploaded_files": true}
|
||||
return cfg
|
||||
}
|
||||
|
||||
func rawJSON(value string) json.RawMessage {
|
||||
return json.RawMessage(value)
|
||||
}
|
||||
@@ -687,7 +720,7 @@ func TestScanScopeReportsSkippedNodesAndSingleDocumentScopeHasNone(t *testing.T)
|
||||
ResourceID: "resource",
|
||||
Reference: resourceReference{WorkspaceID: "space"},
|
||||
StartNodeID: "root",
|
||||
})
|
||||
}, documentSettings{})
|
||||
if err != nil {
|
||||
t.Fatalf("scanScope() error = %v", err)
|
||||
}
|
||||
@@ -713,33 +746,39 @@ func TestScanScopeReportsSkippedNodesAndSingleDocumentScopeHasNone(t *testing.T)
|
||||
ResourceID: "resource",
|
||||
Reference: resourceReference{WorkspaceID: "space"},
|
||||
Document: &document,
|
||||
})
|
||||
}, documentSettings{})
|
||||
if err != nil || len(documents) != 1 || documents[0].ID != "doc" || len(skipped) != 0 {
|
||||
t.Fatalf("single-document scanScope() = %#v, %#v, %v", documents, skipped, err)
|
||||
}
|
||||
}
|
||||
|
||||
// The reason has to say which kind of unsupported a node is: a video is skipped
|
||||
// on purpose, a native type has simply not been implemented.
|
||||
// on purpose, a native type has simply not been implemented, and an uploaded
|
||||
// file the connector could read is skipped only because its switch is off.
|
||||
func TestSkipReasonDistinguishesMediaFromUnimplementedTypes(t *testing.T) {
|
||||
enabled := documentSettings{IncludeUploadedFiles: true}
|
||||
for _, testCase := range []struct {
|
||||
label string
|
||||
node node
|
||||
want string
|
||||
label string
|
||||
node node
|
||||
settings documentSettings
|
||||
want string
|
||||
}{
|
||||
{
|
||||
"video category",
|
||||
node{Type: "FILE", Category: "VIDEO", Extension: "mp4"},
|
||||
documentSettings{},
|
||||
"video/media files are deliberately not downloaded by this connector",
|
||||
},
|
||||
{
|
||||
"video extension without a category",
|
||||
node{Type: "FILE", Category: "OTHER", Extension: "MOV"},
|
||||
documentSettings{},
|
||||
"video/media files are deliberately not downloaded by this connector",
|
||||
},
|
||||
{
|
||||
"audio extension",
|
||||
node{Type: "FILE", Category: "OTHER", Extension: "mp3"},
|
||||
documentSettings{},
|
||||
"video/media files are deliberately not downloaded by this connector",
|
||||
},
|
||||
{
|
||||
@@ -748,28 +787,602 @@ func TestSkipReasonDistinguishesMediaFromUnimplementedTypes(t *testing.T) {
|
||||
// regression guard on the map.
|
||||
"type without a dedicated label",
|
||||
node{Type: "FILE", Category: "ALIDOC", Extension: "axls"},
|
||||
documentSettings{},
|
||||
"no ingest path for this DingTalk node type in this connector yet",
|
||||
},
|
||||
{
|
||||
"multidimensional table",
|
||||
node{Type: "FILE", Category: "ALIDOC", Extension: "able"},
|
||||
documentSettings{},
|
||||
"DingTalk multi-dimensional table has no ingest path in this connector yet",
|
||||
},
|
||||
{
|
||||
"mind map",
|
||||
node{Type: "FILE", Category: "ALIDOC", Extension: "amind"},
|
||||
documentSettings{},
|
||||
"DingTalk mind map has no ingest path in this connector yet",
|
||||
},
|
||||
{
|
||||
"unknown type",
|
||||
node{Type: "FILE", Category: "OTHER", Extension: "bin"},
|
||||
documentSettings{},
|
||||
"no ingest path for this DingTalk node type in this connector yet",
|
||||
},
|
||||
{
|
||||
// The skip is not a missing feature: the file is readable, the data
|
||||
// source simply never opted in, and the log has to say so.
|
||||
"uploaded file with the switch off",
|
||||
node{Type: "FILE", Category: "DOCUMENT", Extension: "docx"},
|
||||
documentSettings{},
|
||||
"uploaded files are not ingested because include_uploaded_files is not enabled for this data source",
|
||||
},
|
||||
{
|
||||
"uploaded file with the switch explicitly off",
|
||||
node{Type: "FILE", Category: "DOCUMENT", Extension: "pdf"},
|
||||
documentSettings{IncludeUploadedFiles: false},
|
||||
"uploaded files are not ingested because include_uploaded_files is not enabled for this data source",
|
||||
},
|
||||
{
|
||||
// A node that is not an uploaded file keeps its own reason even
|
||||
// while the switch is on, so the switch never relabels a node it
|
||||
// does not govern.
|
||||
"unsupported type with the switch on",
|
||||
node{Type: "FILE", Category: "ALIDOC", Extension: "amind"},
|
||||
enabled,
|
||||
"DingTalk mind map has no ingest path in this connector yet",
|
||||
},
|
||||
{
|
||||
"media with the switch on",
|
||||
node{Type: "FILE", Category: "VIDEO", Extension: "mp4"},
|
||||
enabled,
|
||||
"video/media files are deliberately not downloaded by this connector",
|
||||
},
|
||||
} {
|
||||
t.Run(testCase.label, func(t *testing.T) {
|
||||
if got := skipReason(testCase.node); got != testCase.want {
|
||||
if got := skipReason(testCase.node, testCase.settings); got != testCase.want {
|
||||
t.Fatalf("skipReason() = %q, want %q", got, testCase.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// The switch is read from the settings bag the way Yuque reads its own, and a
|
||||
// missing or malformed value keeps uploaded files out.
|
||||
func TestParseDocumentSettingsReadsTheUploadedFilesSwitch(t *testing.T) {
|
||||
for _, testCase := range []struct {
|
||||
label string
|
||||
settings map[string]interface{}
|
||||
want bool
|
||||
}{
|
||||
{"unset", nil, false},
|
||||
{"empty bag", map[string]interface{}{}, false},
|
||||
{"explicit true", map[string]interface{}{"include_uploaded_files": true}, true},
|
||||
{"explicit false", map[string]interface{}{"include_uploaded_files": false}, false},
|
||||
{"string true", map[string]interface{}{"include_uploaded_files": "true"}, true},
|
||||
{"string ON", map[string]interface{}{"include_uploaded_files": " ON "}, true},
|
||||
{"string 1", map[string]interface{}{"include_uploaded_files": "1"}, true},
|
||||
{"string yes", map[string]interface{}{"include_uploaded_files": "yes"}, true},
|
||||
{"string off", map[string]interface{}{"include_uploaded_files": "off"}, false},
|
||||
{"string False", map[string]interface{}{"include_uploaded_files": "False"}, false},
|
||||
{"string 0", map[string]interface{}{"include_uploaded_files": "0"}, false},
|
||||
{"string no", map[string]interface{}{"include_uploaded_files": "no"}, false},
|
||||
{"garbage string", map[string]interface{}{"include_uploaded_files": "maybe"}, false},
|
||||
{"wrong type", map[string]interface{}{"include_uploaded_files": 1}, false},
|
||||
// A switch for another type must never turn this one on.
|
||||
{"unrelated key", map[string]interface{}{"include_sheets": true}, false},
|
||||
} {
|
||||
t.Run(testCase.label, func(t *testing.T) {
|
||||
cfg := testConfig()
|
||||
cfg.Settings = testCase.settings
|
||||
if got := parseDocumentSettings(cfg).IncludeUploadedFiles; got != testCase.want {
|
||||
t.Fatalf("IncludeUploadedFiles = %v, want %v", got, testCase.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
// A nil config cannot be parsed by the connector, but reading its settings
|
||||
// must not panic either.
|
||||
if got := parseDocumentSettings(nil).IncludeUploadedFiles; got {
|
||||
t.Fatalf("parseDocumentSettings(nil).IncludeUploadedFiles = true, want false")
|
||||
}
|
||||
}
|
||||
|
||||
// binaryNode is an uploaded file as the wiki API lists it: a FILE in the
|
||||
// DOCUMENT category carrying the original extension.
|
||||
func binaryNode(id, name, extension string) node {
|
||||
return node{
|
||||
ID: id, WorkspaceID: "space", Name: name, Type: "FILE",
|
||||
Category: "DOCUMENT", Extension: extension, ModifiedTime: "2026-07-25T08:00:00Z",
|
||||
}
|
||||
}
|
||||
|
||||
func uploadedDocumentsFixture() *fakeAPI {
|
||||
return &fakeAPI{
|
||||
workspaces: []workspace{{ID: "space", RootNodeID: "root"}},
|
||||
nodes: map[string][]node{
|
||||
"root": {
|
||||
{
|
||||
ID: "adoc", WorkspaceID: "space", Name: "Runbook.adoc", Type: "FILE",
|
||||
Category: "ALIDOC", Extension: "adoc", ModifiedTime: "r1",
|
||||
},
|
||||
binaryNode("docx", "Employee-Handbook.docx", "docx"),
|
||||
binaryNode("pptx", "Deck.pptx", "pptx"),
|
||||
binaryNode("xlsx", "Sheet.xlsx", "xlsx"),
|
||||
binaryNode("pdf", "Policy.pdf", "pdf"),
|
||||
{
|
||||
ID: "video", WorkspaceID: "space", Name: "Lesson.mp4", Type: "FILE",
|
||||
Category: "VIDEO", Extension: "mp4",
|
||||
},
|
||||
{
|
||||
ID: "aitable", WorkspaceID: "space", Name: "Table.able", Type: "FILE",
|
||||
Category: "ALIDOC", Extension: "able",
|
||||
},
|
||||
},
|
||||
},
|
||||
blocks: map[string][]json.RawMessage{
|
||||
"adoc": {rawJSON(`{"blockType":"paragraph","paragraph":{"text":"run"}}`)},
|
||||
},
|
||||
downloads: map[string][]byte{
|
||||
"docx": []byte("PK\x03\x04word-bytes"),
|
||||
"pptx": []byte("PK\x03\x04slides-bytes"),
|
||||
"xlsx": []byte("PK\x03\x04sheet-bytes"),
|
||||
"pdf": []byte("%PDF-1.7-policy-bytes"),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// Uploaded Office and PDF files must be downloaded byte-for-byte with a
|
||||
// concrete content type and a file name that keeps exactly one extension, while
|
||||
// adoc keeps going through the blocks API and unsupported nodes are left out.
|
||||
func TestFetchAllDownloadsUploadedDocuments(t *testing.T) {
|
||||
api := uploadedDocumentsFixture()
|
||||
items, err := testConnector(api).FetchAll(
|
||||
context.Background(), testConfigWithUploadedFiles("space"), []string{"space"},
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("FetchAll() error = %v", err)
|
||||
}
|
||||
if len(items) != 5 {
|
||||
t.Fatalf("FetchAll() returned %d items, want 5: %#v", len(items), items)
|
||||
}
|
||||
|
||||
byID := make(map[string]types.FetchedItem, len(items))
|
||||
for _, item := range items {
|
||||
byID[item.ExternalID] = item
|
||||
}
|
||||
for _, skipped := range []string{"video", "aitable"} {
|
||||
if _, exists := byID[skipped]; exists {
|
||||
t.Fatalf("unsupported node %q was synced: %#v", skipped, byID[skipped])
|
||||
}
|
||||
}
|
||||
if api.blockCalls["adoc"] != 1 || len(api.blockCalls) != 1 {
|
||||
t.Fatalf("block calls = %#v, want only the adoc document", api.blockCalls)
|
||||
}
|
||||
if len(api.dlCalls) != 4 {
|
||||
t.Fatalf("download calls = %#v, want one per uploaded document", api.dlCalls)
|
||||
}
|
||||
|
||||
want := []struct {
|
||||
id string
|
||||
contentType string
|
||||
fileName string
|
||||
content string
|
||||
}{
|
||||
{
|
||||
"docx",
|
||||
"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
||||
"Employee-Handbook.docx",
|
||||
"PK\x03\x04word-bytes",
|
||||
},
|
||||
{
|
||||
"pptx",
|
||||
"application/vnd.openxmlformats-officedocument.presentationml.presentation",
|
||||
"Deck.pptx",
|
||||
"PK\x03\x04slides-bytes",
|
||||
},
|
||||
{
|
||||
"xlsx",
|
||||
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||||
"Sheet.xlsx",
|
||||
"PK\x03\x04sheet-bytes",
|
||||
},
|
||||
{"pdf", "application/pdf", "Policy.pdf", "%PDF-1.7-policy-bytes"},
|
||||
}
|
||||
for _, expected := range want {
|
||||
item, exists := byID[expected.id]
|
||||
if !exists {
|
||||
t.Fatalf("uploaded document %q was not synced: %#v", expected.id, byID)
|
||||
}
|
||||
if string(item.Content) != expected.content || item.ContentType != expected.contentType ||
|
||||
item.FileName != expected.fileName {
|
||||
t.Fatalf("%s item = %#v (content_type %q, file_name %q)",
|
||||
expected.id, item, item.ContentType, item.FileName)
|
||||
}
|
||||
if item.SourceResourceID != "space" || item.Metadata["extension"] == "" ||
|
||||
item.Metadata["channel"] != types.ChannelDingtalk {
|
||||
t.Fatalf("%s metadata = %#v", expected.id, item.Metadata)
|
||||
}
|
||||
if item.URL == "" || item.UpdatedAt.IsZero() {
|
||||
t.Fatalf("%s lost its URL or timestamp: %#v", expected.id, item)
|
||||
}
|
||||
}
|
||||
if byID["adoc"].ContentType != "text/markdown" ||
|
||||
string(byID["adoc"].Content) != "# Runbook.adoc\n\nrun\n" {
|
||||
t.Fatalf("adoc item = %#v", byID["adoc"])
|
||||
}
|
||||
}
|
||||
|
||||
// Selecting an uploaded file directly must resolve to a single-document scope.
|
||||
func TestFetchAllHandlesUploadedDocumentSelection(t *testing.T) {
|
||||
api := uploadedDocumentsFixture()
|
||||
resourceID, err := encodeResourceReference(resourceReference{
|
||||
WorkspaceID: "space", NodeID: "docx",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
items, err := testConnector(api).FetchAll(
|
||||
context.Background(), testConfigWithUploadedFiles(resourceID), []string{resourceID},
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("FetchAll() error = %v", err)
|
||||
}
|
||||
if len(items) != 1 || items[0].ExternalID != "docx" ||
|
||||
string(items[0].Content) != "PK\x03\x04word-bytes" ||
|
||||
items[0].SourceResourceID != resourceID {
|
||||
t.Fatalf("FetchAll() = %#v", items)
|
||||
}
|
||||
if len(api.dlCalls) != 1 || api.dlCalls["docx"] != 1 {
|
||||
t.Fatalf("download calls = %#v", api.dlCalls)
|
||||
}
|
||||
}
|
||||
|
||||
// A failed download must surface as a retryable failure item and must not
|
||||
// advance the cursor, exactly like a failed block read.
|
||||
func TestIncrementalSyncRetriesUploadedDownloadFailures(t *testing.T) {
|
||||
api := uploadedDocumentsFixture()
|
||||
document := binaryNode("docx", "Employee-Handbook.docx", "docx")
|
||||
document.ModifiedTimestamp = 1_768_000_000_000
|
||||
api.nodes["root"] = []node{document}
|
||||
api.dlErrors = map[string]error{"docx": errors.New("unsupported file type")}
|
||||
|
||||
cursorMap, err := encodeCursor(&cursorState{
|
||||
Version: cursorVersion,
|
||||
Resources: map[string]map[string]string{"space": {"docx": "r1"}},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
items, next, syncErr := testConnector(api).FetchIncremental(
|
||||
context.Background(), testConfigWithUploadedFiles("space"), &types.SyncCursor{ConnectorCursor: cursorMap},
|
||||
)
|
||||
var partial *datasource.PartialFetchError
|
||||
if !errors.As(syncErr, &partial) {
|
||||
t.Fatalf("FetchIncremental() error = %v, want PartialFetchError", syncErr)
|
||||
}
|
||||
if len(items) != 1 || items[0].Metadata["error_reason_code"] != "dingtalk_document_failed" ||
|
||||
!strings.Contains(items[0].Metadata["error"], "unsupported file type") ||
|
||||
len(items[0].Content) != 0 {
|
||||
t.Fatalf("failure item = %#v", items)
|
||||
}
|
||||
state, err := decodeCursor(next)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if state.Resources["space"]["docx"] != "r1" {
|
||||
t.Fatalf("failed download advanced the cursor: %#v", state.Resources)
|
||||
}
|
||||
|
||||
// The same document succeeds once the provider recovers.
|
||||
delete(api.dlErrors, "docx")
|
||||
items, _, err = testConnector(api).FetchIncremental(
|
||||
context.Background(), testConfigWithUploadedFiles("space"), &types.SyncCursor{ConnectorCursor: cursorMap},
|
||||
)
|
||||
if err != nil || len(items) != 1 || string(items[0].Content) != "PK\x03\x04word-bytes" {
|
||||
t.Fatalf("retry result = %#v, %v", items, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Validate must prove an uploaded document is reachable without paying for the
|
||||
// transfer, and must still report a provider refusal.
|
||||
func TestValidateProbesUploadedDocumentsWithoutDownloading(t *testing.T) {
|
||||
api := &fakeAPI{
|
||||
workspaces: []workspace{{ID: "space", RootNodeID: "root", Name: "Space"}},
|
||||
nodes: map[string][]node{
|
||||
"root": {binaryNode("docx", "Employee-Handbook.docx", "docx")},
|
||||
},
|
||||
}
|
||||
|
||||
if err := testConnector(api).Validate(context.Background(), testConfigWithUploadedFiles()); err != nil {
|
||||
t.Fatalf("Validate() error = %v", err)
|
||||
}
|
||||
if api.verifyCalls["docx"] != 1 || len(api.dlCalls) != 0 {
|
||||
t.Fatalf("verify calls = %#v, download calls = %#v", api.verifyCalls, api.dlCalls)
|
||||
}
|
||||
|
||||
api.verifyErrors = map[string]error{"docx": errors.New("missing Storage.File.Read")}
|
||||
err := testConnector(api).Validate(context.Background(), testConfigWithUploadedFiles())
|
||||
if err == nil || !strings.Contains(err.Error(), "missing Storage.File.Read") {
|
||||
t.Fatalf("Validate() error = %v, want provider refusal", err)
|
||||
}
|
||||
}
|
||||
|
||||
// switchStates are the three states every switch test has to cover: the key is
|
||||
// absent from a data source created before the switch existed, explicitly
|
||||
// false, and explicitly true.
|
||||
func switchStates(t *testing.T, key string) []struct {
|
||||
label string
|
||||
settings map[string]interface{}
|
||||
on bool
|
||||
} {
|
||||
t.Helper()
|
||||
return []struct {
|
||||
label string
|
||||
settings map[string]interface{}
|
||||
on bool
|
||||
}{
|
||||
{key + " unset", nil, false},
|
||||
{key + " explicitly off", map[string]interface{}{key: false}, false},
|
||||
{key + " explicitly on", map[string]interface{}{key: true}, true},
|
||||
}
|
||||
}
|
||||
|
||||
// Validate must honour the switch: with it off an uploaded file is not a
|
||||
// document, so it is neither probed nor treated as the one readable document a
|
||||
// data source has to prove. With it on the same file is probed through the
|
||||
// download-location lookup, never through the body transfer.
|
||||
func TestValidateHonoursTheUploadedFilesSwitch(t *testing.T) {
|
||||
for _, testCase := range switchStates(t, "include_uploaded_files") {
|
||||
t.Run(testCase.label, func(t *testing.T) {
|
||||
api := &fakeAPI{
|
||||
workspaces: []workspace{{ID: "space", RootNodeID: "root", Name: "Space"}},
|
||||
nodes: map[string][]node{
|
||||
"root": {binaryNode("docx", "Employee-Handbook.docx", "docx")},
|
||||
},
|
||||
verifyErrors: map[string]error{"docx": errors.New("missing Storage.File.Read")},
|
||||
}
|
||||
cfg := testConfig()
|
||||
cfg.Settings = testCase.settings
|
||||
|
||||
err := testConnector(api).Validate(context.Background(), cfg)
|
||||
if testCase.on {
|
||||
if err == nil || !strings.Contains(err.Error(), "missing Storage.File.Read") {
|
||||
t.Fatalf("Validate() error = %v, want the probe to fail loudly", err)
|
||||
}
|
||||
if api.verifyCalls["docx"] != 1 {
|
||||
t.Fatalf("verify calls = %#v, want exactly one upload probe", api.verifyCalls)
|
||||
}
|
||||
} else {
|
||||
if err != nil {
|
||||
t.Fatalf("Validate() error = %v, want acceptance with nothing to probe", err)
|
||||
}
|
||||
if len(api.verifyCalls) != 0 {
|
||||
t.Fatalf("verify calls = %#v, want no upload probe while the switch is off", api.verifyCalls)
|
||||
}
|
||||
}
|
||||
if len(api.dlCalls) != 0 {
|
||||
t.Fatalf("download calls = %#v, want no body transfer during validation", api.dlCalls)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// The picker follows the same switch, so a node the sync would skip is never
|
||||
// offered for selection.
|
||||
func TestListResourcesHonoursTheUploadedFilesSwitch(t *testing.T) {
|
||||
for _, testCase := range switchStates(t, "include_uploaded_files") {
|
||||
t.Run(testCase.label, func(t *testing.T) {
|
||||
api := uploadedDocumentsFixture()
|
||||
rootID, err := encodeResourceReference(resourceReference{WorkspaceID: "space"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cfg := testConfig()
|
||||
cfg.Settings = testCase.settings
|
||||
|
||||
resources, err := testConnector(api).ListResources(context.Background(), cfg, rootID)
|
||||
if err != nil {
|
||||
t.Fatalf("ListResources() error = %v", err)
|
||||
}
|
||||
names := make(map[string]bool, len(resources))
|
||||
for _, resource := range resources {
|
||||
names[resource.Name] = true
|
||||
}
|
||||
if !names["Runbook.adoc"] {
|
||||
t.Fatalf("the adoc document is missing from %#v", names)
|
||||
}
|
||||
for _, uploaded := range []string{"Employee-Handbook.docx", "Deck.pptx", "Sheet.xlsx", "Policy.pdf"} {
|
||||
if names[uploaded] != testCase.on {
|
||||
t.Fatalf("resource %q listed = %v, want %v (%#v)",
|
||||
uploaded, names[uploaded], testCase.on, names)
|
||||
}
|
||||
}
|
||||
// A node with no ingest path is hidden either way.
|
||||
if names["Table.able"] || names["Lesson.mp4"] {
|
||||
t.Fatalf("unsupported nodes were listed: %#v", names)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Full sync and incremental sync share one scan, so the switch has to hold in
|
||||
// both. With it off, the pre-#3786 behaviour stands: the uploaded files are
|
||||
// reported as skipped with an actionable reason and never downloaded.
|
||||
func TestSyncHonoursTheUploadedFilesSwitch(t *testing.T) {
|
||||
for _, testCase := range switchStates(t, "include_uploaded_files") {
|
||||
t.Run(testCase.label, func(t *testing.T) {
|
||||
api := uploadedDocumentsFixture()
|
||||
cfg := testConfig("space")
|
||||
cfg.Settings = testCase.settings
|
||||
logs := captureLogs(t)
|
||||
|
||||
items, err := testConnector(api).FetchAll(context.Background(), cfg, []string{"space"})
|
||||
if err != nil {
|
||||
t.Fatalf("FetchAll() error = %v", err)
|
||||
}
|
||||
wantUploads := testCase.on
|
||||
for _, id := range []string{"docx", "pptx", "xlsx", "pdf"} {
|
||||
got := false
|
||||
for _, item := range items {
|
||||
if item.ExternalID == id {
|
||||
got = true
|
||||
}
|
||||
}
|
||||
if got != wantUploads {
|
||||
t.Fatalf("FetchAll() synced %q = %v, want %v: %#v", id, got, wantUploads, items)
|
||||
}
|
||||
}
|
||||
if hasAdoc := func() bool {
|
||||
for _, item := range items {
|
||||
if item.ExternalID == "adoc" {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}(); !hasAdoc {
|
||||
t.Fatalf("adoc document disappeared from %#v", items)
|
||||
}
|
||||
|
||||
downloads := len(api.dlCalls)
|
||||
if wantUploads && downloads != 4 {
|
||||
t.Fatalf("download calls = %#v, want one per uploaded document", api.dlCalls)
|
||||
}
|
||||
if !wantUploads {
|
||||
if downloads != 0 {
|
||||
t.Fatalf("download calls = %#v, want none while the switch is off", api.dlCalls)
|
||||
}
|
||||
wantLog := "uploaded files are not ingested because include_uploaded_files " +
|
||||
"is not enabled for this data source"
|
||||
if !strings.Contains(logs.String(), wantLog) {
|
||||
t.Fatalf("skip log is missing %q:\n%s", wantLog, logs.String())
|
||||
}
|
||||
}
|
||||
|
||||
// The incremental path walks the same scan, so it must agree.
|
||||
api.dlCalls = nil
|
||||
cursorMap, err := encodeCursor(&cursorState{
|
||||
Version: cursorVersion,
|
||||
Resources: map[string]map[string]string{"space": {}},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
items, _, err = testConnector(api).FetchIncremental(
|
||||
context.Background(), cfg, &types.SyncCursor{ConnectorCursor: cursorMap},
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("FetchIncremental() error = %v", err)
|
||||
}
|
||||
wantItems := 1
|
||||
if wantUploads {
|
||||
wantItems = 5
|
||||
}
|
||||
if len(items) != wantItems {
|
||||
t.Fatalf("FetchIncremental() returned %d items, want %d: %#v", len(items), wantItems, items)
|
||||
}
|
||||
if wantUploads && len(api.dlCalls) != 4 {
|
||||
t.Fatalf("incremental download calls = %#v, want one per uploaded document", api.dlCalls)
|
||||
}
|
||||
if !wantUploads && len(api.dlCalls) != 0 {
|
||||
t.Fatalf("incremental download calls = %#v, want none while the switch is off", api.dlCalls)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Selecting an uploaded file directly is a scope the sync cannot resolve while
|
||||
// the switch is off, exactly as it could not before the upload path existed.
|
||||
func TestSelectingAnUploadedFileNeedsTheSwitch(t *testing.T) {
|
||||
api := uploadedDocumentsFixture()
|
||||
resourceID, err := encodeResourceReference(resourceReference{WorkspaceID: "space", NodeID: "docx"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cfg := testConfig(resourceID)
|
||||
cfg.Settings = map[string]interface{}{"include_uploaded_files": false}
|
||||
|
||||
items, err := testConnector(api).FetchAll(context.Background(), cfg, []string{resourceID})
|
||||
var partial *datasource.PartialFetchError
|
||||
if !errors.As(err, &partial) {
|
||||
t.Fatalf("FetchAll() error = %v, want PartialFetchError", err)
|
||||
}
|
||||
if len(api.dlCalls) != 0 {
|
||||
t.Fatalf("download calls = %#v, want none while the switch is off", api.dlCalls)
|
||||
}
|
||||
if len(items) != 1 || items[0].Metadata["error_reason_code"] != "dingtalk_resource_failed" {
|
||||
t.Fatalf("FetchAll() = %#v, want one failed resource", items)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBinaryDocumentFileNameKeepsASingleExtension(t *testing.T) {
|
||||
for _, testCase := range []struct {
|
||||
name string
|
||||
extension string
|
||||
want string
|
||||
}{
|
||||
{"Employee-Handbook.docx", "docx", "Employee-Handbook.docx"},
|
||||
{"Deck", "pptx", "Deck.pptx"},
|
||||
{"Report.PDF", "pdf", "Report.pdf"},
|
||||
{"quarterly/2026:sheet.xlsx", "xlsx", "quarterly_2026_sheet.xlsx"},
|
||||
{"", "xlsx", "untitled.xlsx"},
|
||||
{".pdf", "pdf", "untitled.pdf"},
|
||||
} {
|
||||
t.Run(testCase.name+"."+testCase.extension, func(t *testing.T) {
|
||||
got := binaryDocumentFileName(node{Name: testCase.name, Extension: testCase.extension})
|
||||
if got != testCase.want {
|
||||
t.Fatalf("binaryDocumentFileName() = %q, want %q", got, testCase.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// isDocument answers whether a read path exists at all, and isIngestible
|
||||
// answers whether this data source takes the node: the blocks API for adoc
|
||||
// always, the 钉盘 download API for an uploaded file only once
|
||||
// include_uploaded_files is on. Native DingTalk types stay out either way, so
|
||||
// nothing is ever enumerated that the sync cannot read.
|
||||
func TestNodeDocumentClassification(t *testing.T) {
|
||||
for _, testCase := range []struct {
|
||||
label string
|
||||
node node
|
||||
document bool
|
||||
online bool
|
||||
binary bool
|
||||
}{
|
||||
{"adoc", node{Type: "FILE", Category: "ALIDOC", Extension: "adoc"}, true, true, false},
|
||||
{"axls", node{Type: "FILE", Category: "ALIDOC", Extension: "axls"}, false, false, false},
|
||||
{"docx", node{Type: "FILE", Category: "DOCUMENT", Extension: "docx"}, true, false, true},
|
||||
{"pptx", node{Type: "FILE", Category: "DOCUMENT", Extension: "pptx"}, true, false, true},
|
||||
{"xlsx", node{Type: "FILE", Category: "DOCUMENT", Extension: "xlsx"}, true, false, true},
|
||||
{"pdf", node{Type: "FILE", Category: "DOCUMENT", Extension: "pdf"}, true, false, true},
|
||||
{"uppercase", node{Type: "file", Category: "document", Extension: "PDF"}, true, false, true},
|
||||
{"mp4", node{Type: "FILE", Category: "VIDEO", Extension: "mp4"}, false, false, false},
|
||||
{"aitable", node{Type: "FILE", Category: "ALIDOC", Extension: "able"}, false, false, false},
|
||||
{"mindmap", node{Type: "FILE", Category: "ALIDOC", Extension: "amind"}, false, false, false},
|
||||
{"upload category", node{Type: "FILE", Category: "FILE", Extension: "pdf"}, false, false, false},
|
||||
{"other category", node{Type: "FILE", Category: "OTHER", Extension: "docx"}, false, false, false},
|
||||
{"folder", node{Type: "FOLDER", Extension: "docx"}, false, false, false},
|
||||
{"no extension", node{Type: "FILE", Category: "DOCUMENT"}, false, false, false},
|
||||
} {
|
||||
t.Run(testCase.label, func(t *testing.T) {
|
||||
if got := testCase.node.isDocument(); got != testCase.document {
|
||||
t.Fatalf("isDocument() = %v, want %v", got, testCase.document)
|
||||
}
|
||||
if got := testCase.node.isOnlineDocument(); got != testCase.online {
|
||||
t.Fatalf("isOnlineDocument() = %v, want %v", got, testCase.online)
|
||||
}
|
||||
if got := testCase.node.isBinaryDocument(); got != testCase.binary {
|
||||
t.Fatalf("isBinaryDocument() = %v, want %v", got, testCase.binary)
|
||||
}
|
||||
// Off by default: with every switch unset only the native adoc
|
||||
// path is admitted, and the switch must never admit a node that
|
||||
// has no read path at all.
|
||||
if got := testCase.node.isIngestible(documentSettings{}); got != testCase.online {
|
||||
t.Fatalf("isIngestible(off) = %v, want %v", got, testCase.online)
|
||||
}
|
||||
enabled := documentSettings{IncludeUploadedFiles: true}
|
||||
if got := testCase.node.isIngestible(enabled); got != testCase.document {
|
||||
t.Fatalf("isIngestible(on) = %v, want %v", got, testCase.document)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,53 @@
|
||||
package dingtalk
|
||||
|
||||
import (
|
||||
"strings"
|
||||
|
||||
"github.com/Tencent/WeKnora/internal/types"
|
||||
)
|
||||
|
||||
// settingBool interprets a data source settings value as a boolean, returning
|
||||
// def when the value is absent or of an unrecognised shape. Settings arrive
|
||||
// from JSON, so a hand-edited config may carry either a real bool or its string
|
||||
// form. The accepted spellings are the ones the Yuque connector's settingBool
|
||||
// accepts, so every connector reads its switches the same way.
|
||||
func settingBool(v interface{}, def bool) bool {
|
||||
switch t := v.(type) {
|
||||
case bool:
|
||||
return t
|
||||
case string:
|
||||
switch strings.ToLower(strings.TrimSpace(t)) {
|
||||
case "true", "1", "yes", "on":
|
||||
return true
|
||||
case "false", "0", "no", "off":
|
||||
return false
|
||||
}
|
||||
}
|
||||
return def
|
||||
}
|
||||
|
||||
// documentSettings holds the optional ingestion switches carried in
|
||||
// DataSourceConfig.Settings.
|
||||
//
|
||||
// Every switch defaults to false: a data source created before the switch
|
||||
// existed keeps the behaviour it had then until an operator turns the switch on
|
||||
// deliberately, so an upgrade never starts downloading or embedding content
|
||||
// nobody asked for.
|
||||
type documentSettings struct {
|
||||
// IncludeUploadedFiles admits uploaded Office/PDF files (FILE/DOCUMENT
|
||||
// nodes) that are downloaded from 钉盘 and parsed by WeKnora itself.
|
||||
IncludeUploadedFiles bool
|
||||
}
|
||||
|
||||
// parseDocumentSettings reads the settings bag of a data source. A missing or
|
||||
// malformed value leaves the switch off, so a hand-edited config can only ever
|
||||
// opt in to an ingest path, never alter one by accident.
|
||||
func parseDocumentSettings(dataSourceConfig *types.DataSourceConfig) documentSettings {
|
||||
var raw map[string]interface{}
|
||||
if dataSourceConfig != nil {
|
||||
raw = dataSourceConfig.Settings
|
||||
}
|
||||
return documentSettings{
|
||||
IncludeUploadedFiles: settingBool(raw["include_uploaded_files"], false),
|
||||
}
|
||||
}
|
||||
@@ -172,10 +172,13 @@
|
||||
#### 钉钉文档（`connector/dingtalk/`）
|
||||
|
||||
- **认证**：企业内部应用 Client ID、Client Secret 和有目标知识库访问权限的操作人 Union ID；开通 `Wiki.Workspace.Read`、`Wiki.Node.Read`、`Storage.File.Read` 后发布应用。
|
||||
- **范围**：选择知识库、文件夹或单篇 `ALIDOC/adoc` 在线文档，通过公开 Wiki / Blocks API 转为 Markdown。当前不导入钉钉表格或普通上传附件，也不依赖异步导出回调。
|
||||
- **范围**：选择知识库、文件夹或单篇 `ALIDOC/adoc` 在线文档，通过公开 Wiki / Blocks API 转为 Markdown。当前不导入钉钉表格，也不依赖异步导出回调。
|
||||
- **上传文件（默认关闭）**：`Settings.include_uploaded_files`（默认 `false`）开启后，所选范围内钉盘中的普通上传附件（`FILE/DOCUMENT` 的 `docx` / `pptx` / `xlsx` / `pdf`）会被下载并交给 WeKnora 自带解析器入库，文件名保留原扩展名、按扩展名选择解析器；单个文件上限 64 MiB，超出上限即失败且不重试（重签名只会拿到同样的字节）。下载失败（401/403/404 等）会重新申请签名后重试，最多 3 次。
|
||||
- **开关的默认值与成本**：`include_uploaded_files` 未设置或非法值一律按 `false` 处理，因此**存量数据源升级后的下一次同步（含增量同步）行为不变**，不会突然开始下载和向量化附件。打开后，这些文件的正文会进入知识库，占用存储并产生 embedding 成本；不需要时请保持关闭。关闭时连接校验、资源列表和同步都会跳过上传附件，同步日志给出的原因是 `include_uploaded_files` 未启用，而不是类型不支持。
|
||||
- **开关的作用点**：测试连接（校验）、资源选择列表、全量同步与增量同步共用同一个判据，一处关闭即处处一致。
|
||||
- **同步**：按文档 `modifiedTimestamp`（毫秒）增量读取，缺失时回退 `modifiedTime`；合并重叠选择。全量同步也会对照上次游标对账删除，避免 `sync_mode=full` 漏删。目录遍历不完整时暂缓删除。
|
||||
- **正文**：公开 Blocks API 只返回文档根下的一级块；高亮块等容器若响应里带有 `children` 会继续渲染，否则在元数据中标记 `nested_blocks_unavailable`，避免把残缺正文当成完整成功。
|
||||
- **校验**：测试连接会列出知识库、探测根节点列表，并在根下存在在线文档时试读 Blocks，以便尽早发现缺少 `Wiki.Node.Read` / `Storage.File.Read`。
|
||||
- **校验**：测试连接会列出知识库、探测根节点列表，并在根下存在在线文档时试读 Blocks；对已开启摄取的上传文件只解析存储位置并申请下载 URL，不下载正文，以便尽早发现缺少 `Wiki.Node.Read` / `Storage.File.Read`，同时让校验对多兆字节文件保持廉价。
|
||||
- **失败与恢复**：资源失效不阻断其他范围；失败范围和正文失败文档保留旧版本以便重试。任一范围无法完整扫描时暂缓删除，并保留待核对记录。失效的单独选择需要检查权限或重新选择。
|
||||
- **删除开关**：开启同步删除才移除确认在源端删除的本地知识；不可访问的资源不会直接视为已删除。
|
||||
|
||||
|
||||
Reference in New Issue
Block a user