diff --git a/internal/datasource/connector/feishu/client.go b/internal/datasource/connector/feishu/client.go index 2884f0f6..3bc9e59c 100644 --- a/internal/datasource/connector/feishu/client.go +++ b/internal/datasource/connector/feishu/client.go @@ -7,6 +7,7 @@ import ( "fmt" "io" "net/http" + "net/url" "strings" "sync" "time" @@ -248,6 +249,25 @@ func (c *Client) ListWikiNodes(ctx context.Context, spaceID string, parentNodeTo return allNodes, nil } +// GetWikiNode returns metadata for a single wiki node. +func (c *Client) GetWikiNode(ctx context.Context, spaceID string, nodeToken string) (wikiNode, error) { + path := fmt.Sprintf("/open-apis/wiki/v2/spaces/get_node?token=%s", url.QueryEscape(nodeToken)) + + var resp wikiNodeInfoResponse + if err := c.doRequest(ctx, http.MethodGet, path, nil, &resp); err != nil { + return wikiNode{}, fmt.Errorf("get wiki node: %w", err) + } + if resp.Code != 0 { + return wikiNode{}, fmt.Errorf("get wiki node error: code=%d msg=%s", resp.Code, resp.Msg) + } + + node := resp.Data.Node + if node.SpaceID == "" { + node.SpaceID = spaceID + } + return node, nil +} + // ListAllWikiNodesRecursive recursively lists all nodes under a wiki space. // It walks the tree depth-first to discover all nested documents. func (c *Client) ListAllWikiNodesRecursive(ctx context.Context, spaceID string) ([]wikiNode, error) { @@ -291,6 +311,75 @@ func (c *Client) ListAllWikiNodesRecursive(ctx context.Context, spaceID string) return allNodes, nil } +// ListWikiNodesRecursiveFrom returns a wiki node and all descendants below it. +func (c *Client) ListWikiNodesRecursiveFrom(ctx context.Context, spaceID string, nodeToken string) ([]wikiNode, error) { + if nodeToken == "" { + return c.ListAllWikiNodesRecursive(ctx, spaceID) + } + + root, err := c.GetWikiNode(ctx, spaceID, nodeToken) + if err != nil { + return nil, err + } + + nodes, err := c.listWikiNodeDescendants(ctx, spaceID, root) + if err != nil { + return append([]wikiNode{root}, nodes...), err + } + return append([]wikiNode{root}, nodes...), nil +} + +func (c *Client) listWikiNodeDescendants(ctx context.Context, spaceID string, root wikiNode) ([]wikiNode, error) { + if !root.HasChild { + return nil, nil + } + + children, err := c.ListWikiNodes(ctx, spaceID, root.NodeToken) + if err != nil { + wrappedErr := fmt.Errorf("list children of %s: %w", root.NodeToken, err) + logger.Warnf(ctx, "[Feishu] partial wiki node listing failure: space=%s node=%s err=%v", + spaceID, root.NodeToken, err) + return nil, &partialWikiNodeListError{ + Failures: []wikiNodeListFailure{{ + Node: root, + Err: wrappedErr, + }}, + } + } + + var allNodes []wikiNode + var failures []wikiNodeListFailure + var walk func(nodes []wikiNode) + + walk = func(nodes []wikiNode) { + for _, node := range nodes { + allNodes = append(allNodes, node) + if !node.HasChild { + continue + } + + grandChildren, err := c.ListWikiNodes(ctx, spaceID, node.NodeToken) + if err != nil { + wrappedErr := fmt.Errorf("list children of %s: %w", node.NodeToken, err) + failures = append(failures, wikiNodeListFailure{ + Node: node, + Err: wrappedErr, + }) + logger.Warnf(ctx, "[Feishu] partial wiki node listing failure: space=%s node=%s err=%v", + spaceID, node.NodeToken, err) + continue + } + walk(grandChildren) + } + } + + walk(children) + if len(failures) > 0 { + return allNodes, &partialWikiNodeListError{Failures: failures} + } + return allNodes, nil +} + // GetDocumentRawContent retrieves the raw text content of a Feishu docx document. // This returns plain text (not rich text / block structure). // Deprecated: prefer ExportAndDownload which preserves formatting. diff --git a/internal/datasource/connector/feishu/connector.go b/internal/datasource/connector/feishu/connector.go index f5625786..0c9ce672 100644 --- a/internal/datasource/connector/feishu/connector.go +++ b/internal/datasource/connector/feishu/connector.go @@ -26,6 +26,8 @@ func (c *Connector) Type() string { return types.ConnectorTypeFeishu } +const feishuWikiNodeResourceSeparator = ":" + // Validate verifies that the Feishu configuration is valid by testing connectivity. func (c *Connector) Validate(ctx context.Context, config *types.DataSourceConfig) error { feishuConfig, err := parseFeishuConfig(config) @@ -41,7 +43,7 @@ func (c *Connector) Validate(ctx context.Context, config *types.DataSourceConfig return nil } -// ListResources lists all accessible Feishu Wiki spaces as selectable resources. +// ListResources lists all accessible Feishu Wiki spaces and nodes as selectable resources. func (c *Connector) ListResources(ctx context.Context, config *types.DataSourceConfig) ([]types.Resource, error) { feishuConfig, err := parseFeishuConfig(config) if err != nil { @@ -62,10 +64,23 @@ func (c *Connector) ListResources(ctx context.Context, config *types.DataSourceC Type: "wiki_space", Description: space.Description, URL: fmt.Sprintf("https://feishu.cn/wiki/%s", space.SpaceID), + HasChildren: true, Metadata: map[string]interface{}{ "visibility": space.Visibility, + "space_id": space.SpaceID, }, }) + + nodes, err := client.ListAllWikiNodesRecursive(ctx, space.SpaceID) + if err != nil { + var partialErr *partialWikiNodeListError + if !errors.As(err, &partialErr) { + return nil, fmt.Errorf("list feishu wiki nodes in space %s: %w", space.SpaceID, err) + } + } + for _, node := range nodes { + resources = append(resources, wikiNodeToResource(space.SpaceID, node)) + } } return resources, nil @@ -82,26 +97,27 @@ func (c *Connector) FetchAll(ctx context.Context, config *types.DataSourceConfig var allItems []types.FetchedItem - for _, spaceID := range resourceIDs { - // List all nodes in this wiki space recursively - nodes, err := client.ListAllWikiNodesRecursive(ctx, spaceID) + for _, resourceID := range resourceIDs { + spaceID, nodeToken := parseWikiResourceID(resourceID) + // List all nodes in this wiki space or selected node subtree recursively + nodes, err := client.ListWikiNodesRecursiveFrom(ctx, spaceID, nodeToken) if err != nil { var partialErr *partialWikiNodeListError if !errors.As(err, &partialErr) { - return nil, fmt.Errorf("list nodes in space %s: %w", spaceID, err) + return nil, fmt.Errorf("list nodes for resource %s: %w", resourceID, err) } - allItems = appendWikiNodeListFailureItems(allItems, spaceID, partialErr.Failures) + allItems = appendWikiNodeListFailureItems(allItems, spaceID, resourceID, partialErr.Failures) } // Fetch content for each document node for _, node := range nodes { - item, err := c.fetchNodeContent(ctx, client, node, spaceID) + item, err := c.fetchNodeContent(ctx, client, node, spaceID, resourceID) if err != nil { // Log error but continue with other nodes allItems = append(allItems, types.FetchedItem{ ExternalID: node.NodeToken, Title: node.Title, - SourceResourceID: spaceID, + SourceResourceID: resourceID, Metadata: map[string]string{ "error": err.Error(), }, @@ -145,25 +161,26 @@ func (c *Connector) FetchIncremental(ctx context.Context, config *types.DataSour // Get resource IDs from config resourceIDs := config.ResourceIDs if len(resourceIDs) == 0 { - return nil, nil, fmt.Errorf("no resource IDs (wiki space IDs) configured") + return nil, nil, fmt.Errorf("no resource IDs (wiki space IDs or wiki node IDs) configured") } - for _, spaceID := range resourceIDs { - // List all nodes in this space - nodes, err := client.ListAllWikiNodesRecursive(ctx, spaceID) + for _, resourceID := range resourceIDs { + spaceID, nodeToken := parseWikiResourceID(resourceID) + // List all nodes in this wiki space or selected node subtree + nodes, err := client.ListWikiNodesRecursiveFrom(ctx, spaceID, nodeToken) var partialErr *partialWikiNodeListError if err != nil { if !errors.As(err, &partialErr) { - return nil, nil, fmt.Errorf("list nodes in space %s: %w", spaceID, err) + return nil, nil, fmt.Errorf("list nodes for resource %s: %w", resourceID, err) } - changedItems = appendWikiNodeListFailureItems(changedItems, spaceID, partialErr.Failures) + changedItems = appendWikiNodeListFailureItems(changedItems, spaceID, resourceID, partialErr.Failures) } - newCursor.SpaceNodeTimes[spaceID] = make(map[string]string) + newCursor.SpaceNodeTimes[resourceID] = make(map[string]string) if partialErr != nil && prevCursor.SpaceNodeTimes != nil { - if prevTimes, ok := prevCursor.SpaceNodeTimes[spaceID]; ok { + if prevTimes, ok := prevCursor.SpaceNodeTimes[resourceID]; ok { for nodeToken, editTime := range prevTimes { - newCursor.SpaceNodeTimes[spaceID][nodeToken] = editTime + newCursor.SpaceNodeTimes[resourceID][nodeToken] = editTime } } } @@ -179,11 +196,11 @@ func (c *Connector) FetchIncremental(ctx context.Context, config *types.DataSour if editTimeStr == "" { editTimeStr = node.NodeEditTime // fallback for nodes that don't have obj_edit_time } - newCursor.SpaceNodeTimes[spaceID][node.NodeToken] = editTimeStr + newCursor.SpaceNodeTimes[resourceID][node.NodeToken] = editTimeStr // Check if node has changed since last sync if prevCursor.SpaceNodeTimes != nil { - if prevTimes, ok := prevCursor.SpaceNodeTimes[spaceID]; ok { + if prevTimes, ok := prevCursor.SpaceNodeTimes[resourceID]; ok { if prevEditTime, exists := prevTimes[node.NodeToken]; exists { if prevEditTime == editTimeStr { // Node unchanged, skip @@ -194,13 +211,13 @@ func (c *Connector) FetchIncremental(ctx context.Context, config *types.DataSour } // Node is new or changed — fetch its content - item, err := c.fetchNodeContent(ctx, client, node, spaceID) + item, err := c.fetchNodeContent(ctx, client, node, spaceID, resourceID) if err != nil { // Record failed items changedItems = append(changedItems, types.FetchedItem{ ExternalID: node.NodeToken, Title: node.Title, - SourceResourceID: spaceID, + SourceResourceID: resourceID, Metadata: map[string]string{ "error": err.Error(), }, @@ -214,14 +231,14 @@ func (c *Connector) FetchIncremental(ctx context.Context, config *types.DataSour // Detect deleted nodes if partialErr == nil && prevCursor.SpaceNodeTimes != nil { - if prevTimes, ok := prevCursor.SpaceNodeTimes[spaceID]; ok { + if prevTimes, ok := prevCursor.SpaceNodeTimes[resourceID]; ok { for nodeToken := range prevTimes { if !currentNodes[nodeToken] { // Node was deleted changedItems = append(changedItems, types.FetchedItem{ ExternalID: nodeToken, IsDeleted: true, - SourceResourceID: spaceID, + SourceResourceID: resourceID, }) } } @@ -242,7 +259,7 @@ func (c *Connector) FetchIncremental(ctx context.Context, config *types.DataSour return changedItems, nextSyncCursor, nil } -func appendWikiNodeListFailureItems(items []types.FetchedItem, spaceID string, failures []wikiNodeListFailure) []types.FetchedItem { +func appendWikiNodeListFailureItems(items []types.FetchedItem, spaceID string, resourceID string, failures []wikiNodeListFailure) []types.FetchedItem { for _, failure := range failures { node := failure.Node title := node.Title @@ -252,7 +269,7 @@ func appendWikiNodeListFailureItems(items []types.FetchedItem, spaceID string, f items = append(items, types.FetchedItem{ ExternalID: node.NodeToken, Title: title, - SourceResourceID: spaceID, + SourceResourceID: resourceID, Metadata: map[string]string{ "error": failure.Err.Error(), "channel": types.ChannelFeishu, @@ -273,7 +290,7 @@ func appendWikiNodeListFailureItems(items []types.FetchedItem, spaceID string, f // - file → drive download → original file (PDF/Word/image/etc.) // - mindnote → skip (no API) // - slides → skip (no API) -func (c *Connector) fetchNodeContent(ctx context.Context, client *Client, node wikiNode, spaceID string) (*types.FetchedItem, error) { +func (c *Connector) fetchNodeContent(ctx context.Context, client *Client, node wikiNode, spaceID string, resourceID string) (*types.FetchedItem, error) { if !isSupportedDocType(node.ObjType) { return nil, nil } @@ -314,7 +331,7 @@ func (c *Connector) fetchNodeContent(ctx context.Context, client *Client, node w FileName: fileName, URL: fmt.Sprintf("https://feishu.cn/wiki/%s", node.NodeToken), UpdatedAt: editTime, - SourceResourceID: spaceID, + SourceResourceID: resourceID, Metadata: baseMeta, }, nil @@ -339,7 +356,7 @@ func (c *Connector) fetchNodeContent(ctx context.Context, client *Client, node w FileName: fileName, URL: fmt.Sprintf("https://feishu.cn/wiki/%s", node.NodeToken), UpdatedAt: editTime, - SourceResourceID: spaceID, + SourceResourceID: resourceID, Metadata: baseMeta, }, nil @@ -350,6 +367,48 @@ func (c *Connector) fetchNodeContent(ctx context.Context, client *Client, node w // --- Helper functions --- +func makeWikiNodeResourceID(spaceID, nodeToken string) string { + return spaceID + feishuWikiNodeResourceSeparator + nodeToken +} + +func parseWikiResourceID(resourceID string) (spaceID string, nodeToken string) { + spaceID, nodeToken, _ = strings.Cut(resourceID, feishuWikiNodeResourceSeparator) + return spaceID, nodeToken +} + +func wikiNodeToResource(spaceID string, node wikiNode) types.Resource { + parentID := spaceID + if node.ParentNodeID != "" { + parentID = makeWikiNodeResourceID(spaceID, node.ParentNodeID) + } + + name := node.Title + if name == "" { + name = node.NodeToken + } + + modifiedAt := parseFeishuTimestamp(node.ObjEditTime) + if modifiedAt.IsZero() { + modifiedAt = parseFeishuTimestamp(node.NodeEditTime) + } + + return types.Resource{ + ExternalID: makeWikiNodeResourceID(spaceID, node.NodeToken), + Name: name, + Type: "wiki_node", + URL: fmt.Sprintf("https://feishu.cn/wiki/%s", node.NodeToken), + ParentID: parentID, + HasChildren: node.HasChild, + ModifiedAt: modifiedAt, + Metadata: map[string]interface{}{ + "space_id": spaceID, + "node_token": node.NodeToken, + "obj_token": node.ObjToken, + "obj_type": node.ObjType, + }, + } +} + // parseFeishuConfig extracts and validates Feishu-specific configuration. func parseFeishuConfig(config *types.DataSourceConfig) (*Config, error) { if config == nil { diff --git a/internal/datasource/connector/feishu/connector_test.go b/internal/datasource/connector/feishu/connector_test.go index 89deefdb..1e70bbf7 100644 --- a/internal/datasource/connector/feishu/connector_test.go +++ b/internal/datasource/connector/feishu/connector_test.go @@ -161,6 +161,20 @@ func fakeFeishuWithChildFailure(topNodes []wikiNode, failingParentToken string) func fakeFeishuHierarchy(topNodes []wikiNode, childNodes map[string][]wikiNode, failingParentToken string) (*httptest.Server, *Config) { mux := http.NewServeMux() + nodeByToken := make(map[string]wikiNode) + for _, node := range topNodes { + node.SpaceID = "space1" + nodeByToken[node.NodeToken] = node + } + for parentToken, nodes := range childNodes { + for _, node := range nodes { + node.SpaceID = "space1" + if node.ParentNodeID == "" { + node.ParentNodeID = parentToken + } + nodeByToken[node.NodeToken] = node + } + } mux.HandleFunc("/open-apis/auth/v3/tenant_access_token/internal", func(w http.ResponseWriter, r *http.Request) { writeJSON(w, tokenResponse{ @@ -170,6 +184,21 @@ func fakeFeishuHierarchy(topNodes []wikiNode, childNodes map[string][]wikiNode, }) }) + mux.HandleFunc("/open-apis/wiki/v2/spaces", func(w http.ResponseWriter, r *http.Request) { + writeJSON(w, wikiSpaceListResponse{ + apiResponse: apiResponse{Code: 0}, + Data: struct { + Items []wikiSpace `json:"items"` + HasMore bool `json:"has_more"` + PageToken string `json:"page_token"` + }{ + Items: []wikiSpace{ + {SpaceID: "space1", Name: "Test Space", Description: "desc", Visibility: "public"}, + }, + }, + }) + }) + mux.HandleFunc("/open-apis/wiki/v2/spaces/space1/nodes", func(w http.ResponseWriter, r *http.Request) { parentToken := r.URL.Query().Get("parent_node_token") if parentToken != "" && parentToken == failingParentToken { @@ -180,6 +209,14 @@ func fakeFeishuHierarchy(topNodes []wikiNode, childNodes map[string][]wikiNode, nodes := topNodes if parentToken != "" { nodes = childNodes[parentToken] + for i := range nodes { + if nodes[i].ParentNodeID == "" { + nodes[i].ParentNodeID = parentToken + } + if nodes[i].SpaceID == "" { + nodes[i].SpaceID = "space1" + } + } } writeJSON(w, wikiNodeListResponse{ apiResponse: apiResponse{Code: 0}, @@ -193,6 +230,23 @@ func fakeFeishuHierarchy(topNodes []wikiNode, childNodes map[string][]wikiNode, }) }) + mux.HandleFunc("/open-apis/wiki/v2/spaces/get_node", func(w http.ResponseWriter, r *http.Request) { + nodeToken := r.URL.Query().Get("token") + node, ok := nodeByToken[nodeToken] + if !ok { + writeJSON(w, wikiNodeInfoResponse{ + apiResponse: apiResponse{Code: 1663, Msg: "node not found"}, + }) + return + } + writeJSON(w, wikiNodeInfoResponse{ + apiResponse: apiResponse{Code: 0}, + Data: struct { + Node wikiNode `json:"node"` + }{Node: node}, + }) + }) + mux.HandleFunc("/open-apis/drive/v1/files/", func(w http.ResponseWriter, r *http.Request) { if strings.HasSuffix(r.URL.Path, "/download") { w.Header().Set("Content-Type", "application/octet-stream") @@ -411,6 +465,44 @@ func TestConnectorListResources(t *testing.T) { } } +func TestConnectorListResources_IncludesWikiNodeTree(t *testing.T) { + topNodes := []wikiNode{ + {NodeToken: "nt-root", ObjToken: "obj-root", ObjType: "docx", Title: "Root", HasChild: true, ObjEditTime: "100"}, + {NodeToken: "nt-peer", ObjToken: "obj-peer", ObjType: "docx", Title: "Peer", ObjEditTime: "200"}, + } + childNodes := map[string][]wikiNode{ + "nt-root": { + {NodeToken: "nt-child", ObjToken: "obj-child", ObjType: "docx", Title: "Child", ObjEditTime: "300"}, + }, + } + ts, cfg := fakeFeishuHierarchy(topNodes, childNodes, "") + defer ts.Close() + + resources, err := NewConnector().ListResources(context.Background(), makeConfig(cfg, nil)) + if err != nil { + t.Fatalf("ListResources() error: %v", err) + } + if len(resources) != 4 { + t.Fatalf("want 4 resources (space + 3 nodes), got %d: %+v", len(resources), resources) + } + + byID := make(map[string]types.Resource) + for _, resource := range resources { + byID[resource.ExternalID] = resource + } + root := byID["space1:nt-root"] + if root.ParentID != "space1" || !root.HasChildren { + t.Fatalf("root resource wrong: %+v", root) + } + child := byID["space1:nt-child"] + if child.ParentID != "space1:nt-root" || child.Name != "Child" { + t.Fatalf("child resource wrong: %+v", child) + } + if child.Metadata["space_id"] != "space1" || child.Metadata["node_token"] != "nt-child" { + t.Fatalf("child metadata wrong: %+v", child.Metadata) + } +} + // ────────────────────────────────────────────────────────────────────── // FetchAll: test all supported doc types // ────────────────────────────────────────────────────────────────────── @@ -635,6 +727,46 @@ func TestFetchAll_ChildNodeListErrorReturnsPartialItems(t *testing.T) { } } +func TestFetchAll_WikiNodeResourceSyncsSelectedSubtree(t *testing.T) { + topNodes := []wikiNode{ + {NodeToken: "nt-root", ObjToken: "obj-root", ObjType: "file", Title: "Root.pdf", NodeEditTime: "100", HasChild: true}, + {NodeToken: "nt-peer", ObjToken: "obj-peer", ObjType: "file", Title: "Peer.pdf", NodeEditTime: "200"}, + } + childNodes := map[string][]wikiNode{ + "nt-root": { + {NodeToken: "nt-child", ObjToken: "obj-child", ObjType: "file", Title: "Child.pdf", NodeEditTime: "300"}, + }, + } + ts, cfg := fakeFeishuHierarchy(topNodes, childNodes, "") + defer ts.Close() + + resourceID := "space1:nt-root" + items, err := NewConnector().FetchAll(context.Background(), makeConfig(cfg, []string{resourceID}), []string{resourceID}) + if err != nil { + t.Fatalf("FetchAll() error: %v", err) + } + if len(items) != 2 { + t.Fatalf("want selected root and child only, got %d: %+v", len(items), items) + } + + got := make(map[string]types.FetchedItem) + for _, item := range items { + got[item.ExternalID] = item + } + if _, ok := got["nt-root"]; !ok { + t.Fatal("selected root node was not fetched") + } + if _, ok := got["nt-child"]; !ok { + t.Fatal("child node was not fetched") + } + if _, ok := got["nt-peer"]; ok { + t.Fatal("peer outside selected subtree must not be fetched") + } + if got["nt-root"].SourceResourceID != resourceID || got["nt-child"].SourceResourceID != resourceID { + t.Fatalf("SourceResourceID should preserve selected resource id: %+v", got) + } +} + // ────────────────────────────────────────────────────────────────────── // FetchIncremental tests // ──────────────────────────────────────────────────────────────────────