fix(catalog): 打通共享授权扫描与统一来源清单显示
build / build (push) Successful in 1m18s

This commit is contained in:
姚章浩
2026-10-26 18:36:11 +08:00
parent c8f05ee46f
commit e744114ff4
10 changed files with 276 additions and 50 deletions
+16 -1
View File
@@ -2,11 +2,26 @@ package httpapi
import (
"context"
postgresstore "knowledge-governance/internal/platform/postgres"
"net/http"
"net/http/httptest"
"testing"
"time"
"knowledge-governance/internal/dingtalkdiscovery"
)
func TestDWSNameConflictIsNotASuccessfulRegistration(t *testing.T) {
store := &fakeDingtalkRegistrar{fakeJobStore: &fakeJobStore{}, err: postgresstore.ErrDingtalkSpaceNameTaken}
lister := fakeDingtalkSources{result: dingtalkdiscovery.Result{KnowledgeBases: []dingtalkdiscovery.KnowledgeBase{{Name: "wiki", SpaceID: "testspace"}}}}
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodPost, "/api/v1/dingtalk-spaces", nil)
registerSpaceThroughDWS(rec, req, Config{Store: store, DingtalkSources: lister, Now: time.Now}, "testspace", "test-op")
if rec.Code != http.StatusConflict {
t.Fatalf("name conflict must not claim success: %d", rec.Code)
}
}
// 登记走 dws 不是侧车:解析出来的键、名称、根节点直接决定新来源的确定性 id。
func TestResolveSpaceViaDWSUsesTheSpaceCredentialsAlreadyHeld(t *testing.T) {
lister := fakeDingtalkSources{result: dingtalkdiscovery.Result{
@@ -42,4 +57,4 @@ func TestResolveSpaceViaDWSLeavesUnknownSpacesToTheFallback(t *testing.T) {
if _, _, _, _, ok := resolveSpaceViaDWS(context.Background(), lister, "not-a-space-url"); ok {
t.Fatal("a URL without a space id must not resolve")
}
}
}
+2 -2
View File
@@ -290,8 +290,8 @@ func registerSpaceThroughDWS(w http.ResponseWriter, r *http.Request, config Conf
})
switch {
case errors.Is(err, postgresstore.ErrDingtalkSpaceNameTaken):
writeJSON(w, http.StatusOK, map[string]any{"data": map[string]string{"name": name, "already_present": "true"}})
return true, true
writeJSON(w, http.StatusConflict, map[string]string{"error": "space_name_conflict", "hint": "已有同名来源,但空间标识不一致;请核对已有来源,不能按名称判定添加成功。"})
return true, false
case errors.Is(err, postgresstore.ErrDingtalkSpaceIdentityConflict):
writeJSON(w, http.StatusConflict, map[string]string{
"error": "space_identity_conflict",
+2 -1
View File
@@ -71,8 +71,9 @@ func NewHandler(config Config) (*Handler, error) {
}
func (handler *Handler) Handle(ctx context.Context, job jobs.Job) error {
var request Request
// 进度回写会把 stage/progress/scanned_* 等字段写回同一份 payload;重试时严格解码
// 会把这些合法字段当成未知字段,把真实失败原因盖成 catalog_refresh_invalid_payload。
decoder := json.NewDecoder(bytes.NewReader(job.Payload))
decoder.DisallowUnknownFields()
if err := decoder.Decode(&request); err != nil || request.ConnectionID == "" || request.SourceName == "" || (request.SourceType != "pp" && request.SourceType != "dingtalk" && request.SourceType != "alist") || job.LeaseOwner == "" {
return errors.New("catalog_refresh_invalid_payload")
}
@@ -110,6 +110,18 @@ func (fakeRunner) Run(_ context.Context, _ Request, report func(Progress) error)
}
return report(Progress{Stage: "completed", Progress: 100, Message: "完成"})
}
func TestHandlerAcceptsPersistedScanCountersOnRetry(t *testing.T) {
store := &fakeStore{}
handler, _ := NewHandler(Config{Store: store, Runner: fakeRunner{}})
job := jobs.Job{ID: "retry", LeaseOwner: "worker", Payload: []byte(`{"connection_id":"c","source_type":"dingtalk","source_name":"wiki","current_directory":"目录","scanned_directories":2,"discovered_directories":3,"scanned_files":4,"estimated_total_files":5,"total_is_estimate":true}`)}
if err := handler.Handle(context.Background(), job); err != nil {
t.Fatal(err)
}
if len(store.updates) != 2 {
t.Fatal("runner was not called")
}
}
func TestHandlerRunsLiveRefreshAndPersistsProgress(t *testing.T) {
store := &fakeStore{}
handler, err := NewHandler(Config{Store: store, Runner: fakeRunner{}, Now: func() time.Time { return time.Now() }})
+139
View File
@@ -0,0 +1,139 @@
#!/usr/bin/env python3
"""Read a complete DingTalk space through shared dws credentials.
Only a complete traversal produces an inventory. Source keys are retained so
refreshing an existing source never changes directory annotation identities.
"""
import argparse
from collections import deque
import json
import os
import pathlib
import subprocess
import sys
def dws(args, timeout=180):
env = {k: v for k, v in os.environ.items() if k in (
'PATH', 'HOME', 'DWS_HOME', 'DWS_CONFIG_DIR', 'DWS_DISABLE_KEYCHAIN',
'SSL_CERT_FILE', 'SSL_CERT_DIR', 'LANG')}
if env.get('DWS_HOME'):
env['HOME'] = env['DWS_HOME']
process = subprocess.run([os.environ.get('DWS_BIN', 'dws')] + args,
env=env, capture_output=True, text=True, timeout=timeout)
# Never include upstream stderr/URLs/credentials in the job's public message.
if process.returncode != 0:
raise RuntimeError('dws read rejected; check shared authorization and source access')
payload = json.loads(process.stdout)
if payload.get('ok') is not True or not isinstance(payload.get('data'), dict):
raise RuntimeError('dws returned an invalid response')
return payload['data']
def pages(args, key):
items, cursors, ids = [], set(), set()
cursor = ''
while True:
data = dws(args + (['--cursor', cursor] if cursor else []))
batch = data.get(key)
if not isinstance(batch, list) or not isinstance(data.get('hasMore'), bool):
raise RuntimeError('dws directory response is incomplete')
for item in batch:
if not isinstance(item, dict) or not item.get('nodeId'):
raise RuntimeError('dws directory contains an invalid node')
node_id = item['nodeId']
if node_id in ids:
raise RuntimeError('dws directory contains duplicate nodes')
ids.add(node_id)
items.append(item)
if not data['hasMore']:
return items
cursor = data.get('nextCursor')
if not isinstance(cursor, str) or not cursor or cursor in cursors:
raise RuntimeError('dws pagination did not advance')
cursors.add(cursor)
if len(cursors) >= 10000:
raise RuntimeError('dws pagination safety limit reached')
def list_wiki(workspace, folder):
args = ['wiki', '+node-list', '--workspace', workspace, '--limit', '50']
if folder:
args += ['--folder', folder]
return pages(args, 'nodes')
def list_drive(space, folder):
args = ['drive', '+list', '--space-id', space, '--limit', '50']
if folder:
args += ['--folder', folder]
return pages(args, 'files')
def normalise(item, parent):
node_id = item.get('nodeId')
kind = str(item.get('nodeType') or item.get('type') or '').lower()
name = str(item.get('name') or '')
if not node_id or not name or kind not in ('folder', 'file'):
raise RuntimeError('dws node identity or type is missing')
is_folder = kind == 'folder'
extension = str(item.get('extension') or '').lstrip('.')
if not is_folder and extension and not name.lower().endswith('.' + extension.lower()):
name += '.' + extension
size = item.get('file_size') or item.get('fileSize') or item.get('sizeBytes') or item.get('size')
updated = item.get('content_updated_time') or item.get('updateTime') or item.get('modifiedTime') or item.get('createTime')
return node_id, {
'name': name, 'parent_dentry_uuid': parent,
'dentry_type': 'folder' if is_folder else 'file',
'file_size': size, 'content_updated_time': updated,
}, is_folder
def walk(space_id, space_kind, root_id, emit):
nodes, seen, queue = {}, set(), deque([root_id])
files = 0
while queue:
current = queue.popleft()
if current in seen:
raise RuntimeError('dws directory cycle')
seen.add(current)
folder = current if current != root_id else ''
items = list_wiki(space_id, folder) if space_kind == 'wiki' else list_drive(space_id, folder)
for item in items:
node_id, node, is_folder = normalise(item, current)
if node_id == root_id or node_id in nodes:
raise RuntimeError('dws node appears under multiple parents')
nodes[node_id] = node
if is_folder:
queue.append(node_id)
else:
files += 1
emit(len(seen), len(queue), files)
return nodes
def main():
parser = argparse.ArgumentParser()
parser.add_argument('--root-key', required=True)
parser.add_argument('--name', required=True)
parser.add_argument('--space-id', required=True)
parser.add_argument('--root-id', required=True)
parser.add_argument('--space-kind', choices=('wiki', 'drive'), required=True)
parser.add_argument('--output', required=True)
args = parser.parse_args()
def emit(done, pending, files):
print(f'progress root={args.root_id} folders={done} pending={pending} files={files}', flush=True)
nodes = walk(args.space_id, args.space_kind, args.root_id, emit)
payload = {'roots': {args.root_key: {
'configured_root_id': args.root_id, 'live_space_name': args.name, 'nodes': nodes}}}
pathlib.Path(args.output).write_text(json.dumps(payload, ensure_ascii=False), encoding='utf-8')
if __name__ == '__main__':
try:
main()
except Exception:
print('dws directory scan failed; no snapshot was produced', file=sys.stderr)
sys.exit(1)
+28 -30
View File
@@ -54,38 +54,36 @@ def refresh_pp():
emit('building',78,'本地内容清单暂不可用,将记录为新资料待校验')
return build_pp_live_source(inventory,manifest)
def dingtalk_root_key(connection_id, space_id):
from source_catalog_import import stable_connection_id
key='space-'+space_id
if stable_connection_id('dingtalk-'+key)==connection_id:return key
# Legacy identities remain authoritative for existing directory annotations.
inventory=json.loads(pathlib.Path('/sources/dingtalk/data/dingtalk-internal/inventory-state.json').read_text())
for key in inventory.get('roots',{}):
if stable_connection_id('dingtalk-'+key)==connection_id:return key
raise RuntimeError('DingTalk source identity cannot be resolved safely')
def refresh_dingtalk(name,connection_id=''):
root=pathlib.Path('/sources/dingtalk/data/dingtalk-internal/inventory-state.json');state=json.loads(root.read_text())
# Prefer the root registered on the connection row (spaces/<root>?space=):
# name matching is the legacy fallback for the imported sources.
root_id=''
if connection_id and re.fullmatch(r'[A-Za-z0-9_.:-]{1,36}',connection_id):
try:
rows=subprocess.run(['psql','-X','-At','-d',os.environ.get('DATABASE_URL',''),'-c',"SELECT base_url FROM source_connections WHERE id='"+connection_id+"'"],capture_output=True,text=True,timeout=30)
locator=rows.stdout.strip() if rows.returncode==0 else ''
if locator.startswith('spaces/'):root_id=locator.split('?',1)[0][len('spaces/'):]
except Exception:root_id=''
matches=[(key,value) for key,value in state['roots'].items() if (root_id and value.get('configured_root_id')==root_id) or value.get('live_space_name')==name]
if root_id and len(matches)>1:matches=[m for m in matches if m[1].get('configured_root_id')==root_id]
if len(matches)!=1:raise RuntimeError('DingTalk source root missing')
key,current=matches[0];output=pathlib.Path('/tmp')/('dingtalk-'+key+'-live.json')
if output.exists():output.unlink()
command=['python3','/sources/dingtalk/browser_dingtalk_scan.py','--capture','/run/source-secrets/dingtalk_capture.json','--output',str(output),'--root',f"{key}={current['configured_root_id']}",'--checkpoint-every','10']
emit('connecting',8,'正在验证钉钉专用浏览器登录')
process=subprocess.Popen(command,stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,text=True)
assert process.stdout is not None
last=10
for line in process.stdout:
if line.startswith('progress root='):
values={part.split('=',1)[0]:part.split('=',1)[1] for part in line.strip().split() if '=' in part}
done=int(values.get('folders','0'));pending=int(values.get('pending','0'));total=done+pending;files=int(values.get('files','0'))
current_progress=10+(65*done//total if total else 65);last=max(last,current_progress)
emit('scanning',last,f"正在扫描目录:已完成 {done} 个文件夹,待处理 {pending} 个",done,total,scanned_directories=done,discovered_directories=total,scanned_files=files,estimated_total_files=files,total_is_estimate=True)
if process.wait()!=0 or not output.is_file():raise RefreshBlocked('钉钉登录会话已失效,请在来源链接重新扫码')
payload=json.loads(output.read_text());sources=build_dingtalk_sources(payload);match=[source for source in sources if source.get('root_node_id')==current['configured_root_id'] or source['name']==name]
if len(match)!=1:raise RuntimeError('DingTalk refreshed source missing')
from dws_dingtalk_scan import walk
if not re.fullmatch(r'[A-Za-z0-9_.:-]{1,36}',connection_id):raise RuntimeError('invalid source identity')
rows=subprocess.run(['psql','-X','-At','-v','ON_ERROR_STOP=1','-d',os.environ.get('DATABASE_URL',''),'-c',"SELECT base_url FROM source_connections WHERE id='"+connection_id+"'"],capture_output=True,text=True,timeout=30)
if rows.returncode:raise RuntimeError('DingTalk source lookup failed')
match=re.fullmatch(r'spaces/([A-Za-z0-9_-]+)\?space=([A-Za-z0-9_-]+)',rows.stdout.strip())
if not match:raise RuntimeError('DingTalk source root missing')
root_id,space_id=match.groups()
root_key=dingtalk_root_key(connection_id,space_id)
kind='drive' if space_id.isdigit() else 'wiki'
emit('connecting',8,'正在用 dws 读取来源目录')
def report(done,pending,files):
total=done+pending
emit('scanning',10+(65*done//total if total else 65),f"正在扫描目录:已完成 {done} 个文件夹,待处理 {pending} 个",done,total,scanned_directories=done,discovered_directories=total,scanned_files=files,estimated_total_files=files,total_is_estimate=True)
try:nodes=walk(space_id,kind,root_id,report)
except Exception as error:raise RefreshBlocked('dws 目录读取失败,请检查共享授权及来源读取权限') from error
sources=build_dingtalk_sources({'roots':{root_key:{'configured_root_id':root_id,'live_space_name':name,'nodes':nodes}}})
if len(sources)!=1:raise RuntimeError('DingTalk refreshed source missing')
emit('building',80,'正在校验钉钉父子目录和文件数量')
return match[0]
return sources[0]
def refresh_alist(name):
sys.path.insert(0,'/sources/alist');from alist_readonly_inventory import AListReadOnlyClient,join_path,read_connections
+51
View File
@@ -0,0 +1,51 @@
import unittest
from unittest.mock import patch
import dws_dingtalk_scan as scan
from source_catalog_import import build_dingtalk_sources, validate_source
class ScanTests(unittest.TestCase):
def test_follows_cursor_instead_of_accepting_truncated_inventory(self):
pages = [dict(nodes=[{'nodeId': 'f1', 'name': 'one', 'type': 'file'}], hasMore=True, nextCursor='next'),
dict(nodes=[{'nodeId': 'f2', 'name': 'two', 'type': 'file'}], hasMore=False, nextCursor='')]
with patch.object(scan, 'dws', side_effect=pages) as cli:
self.assertEqual(len(scan.list_wiki('wiki', '')), 2)
self.assertIn('--cursor', cli.call_args_list[1].args[0])
def test_rejects_missing_items_and_repeated_cursor(self):
for data in [{'hasMore': False}, {'nodes': [], 'hasMore': True, 'nextCursor': 'same'}]:
with patch.object(scan, 'dws', return_value=data):
with self.assertRaises(RuntimeError):
scan.list_wiki('wiki', '')
def test_files_with_children_are_not_walked_as_folders(self):
item = {'nodeId': 'doc', 'name': 'document', 'type': 'file', 'hasChildren': True}
self.assertFalse(scan.normalise(item, 'root')[2])
def test_output_retains_file_extension_and_valid_hierarchy(self):
with patch.object(scan, 'list_wiki', side_effect=[
[{'nodeId': 'dir', 'name': '目录', 'type': 'folder'}],
[{'nodeId': 'doc', 'name': '说明', 'type': 'file', 'extension': 'adoc', 'fileSize': 10}]
]):
nodes = scan.walk('space', 'wiki', 'root', lambda *args: None)
self.assertEqual(nodes['doc']['name'], '说明.adoc')
source = build_dingtalk_sources({'roots': {'space-space': {
'configured_root_id': 'root', 'live_space_name': '测试', 'nodes': nodes}}})[0]
validate_source(source)
self.assertEqual(source['file_count'], 1)
self.assertEqual(source['folder_count'], 2)
def test_child_environment_uses_shared_home_without_database_secret(self):
import os
with patch.dict(os.environ, {'DWS_HOME': '/shared/home', 'DATABASE_URL': 'private'}, clear=False):
with patch.object(scan.subprocess, 'run') as run:
run.return_value.returncode = 0
run.return_value.stdout = '{"ok":true,"data":{}}'
scan.dws(['wiki', '+node-list'])
env = run.call_args.kwargs['env']
self.assertEqual(env['HOME'], '/shared/home')
self.assertNotIn('DATABASE_URL', env)
if __name__ == '__main__':
unittest.main()
@@ -3,6 +3,17 @@ import live_catalog_refresh as refresh
class S4ProgressPayloadTests(unittest.TestCase):
def test_source_key_preserves_legacy_directory_identity(self):
from source_catalog_import import stable_connection_id
from unittest.mock import patch
legacy = {'roots': {'company': {'configured_root_id': 'root'}}}
with patch.object(refresh.pathlib.Path, 'read_text', return_value=json.dumps(legacy)):
self.assertEqual(refresh.dingtalk_root_key(stable_connection_id('dingtalk-company'), 'space'), 'company')
self.assertEqual(refresh.dingtalk_root_key(stable_connection_id('dingtalk-space-space'), 'space'), 'space-space')
with patch.object(refresh.pathlib.Path, 'read_text', return_value=json.dumps(legacy)):
with self.assertRaises(RuntimeError):
refresh.dingtalk_root_key('sc-unrecognized', 'space')
def collect(self, fn):
buffer=io.StringIO()
with contextlib.redirect_stdout(buffer):
+9 -10
View File
@@ -197,7 +197,9 @@ describe('SystemSettingsView', () => {
expect(html).not.toContain(removed)
// 需求(2026-10-14 收敛):数据源配置只剩两个入口 —— dws 导入(按当前钉钉账号发现)
// 与 PP 导入;AList/静态钉钉知识库等来源类型不再出现在这里(也没有折叠开关兜着)。
for (const hidden of ['公司级技术支持知识库', '专用浏览器扫码会话', '显示 AList/钉钉知识库来源'])
expect(html).toContain('公司级技术支持知识库')
expect(html).toContain('dws 共享授权')
for (const hidden of ['专用浏览器扫码会话', '显示 AList/钉钉知识库来源'])
expect(html).not.toContain(hidden)
// 统一入口后不再按类型分标题:PP 与钉钉共用同一张清单与同一个添加向导。
expect(html).not.toContain('PP 导入')
@@ -289,11 +291,9 @@ describe('SystemSettingsView', () => {
expect(servicesHtml).toContain('端点探测:暂无真实探测结果')
// V1: left-nav section controls, per-source schedule editing, danger modal & history affordances.
expect(html).toContain('settings-nav')
// 非 PP 来源在这里根本不渲染卡片:连它们卡片上的调度、停用、修订入口一起消失。
for (const gone of ['查看修订历史', '保存并启用调度', '停用来源', '下次自动刷新']) expect(html).not.toContain(gone)
expect(html).toContain('本页只列 PP 来源')
// 手工来源(本地手工验收)也不再列在这里,它的专属提示自然不出现。
expect(html).not.toContain('本地手工导入来源不支持定时刷新')
for (const kept of ['查看修订历史', '保存并启用调度', '停用来源', '下次自动刷新']) expect(html).toContain(kept)
expect(html).not.toContain('本页只列 PP 来源')
expect(html).toContain('本地手工导入来源不支持定时刷新')
// dead buttons removed
expect(html).not.toContain('添加内容源')
expect(html).not.toContain('管理员编辑')
@@ -551,13 +551,12 @@ describe('SystemSettingsView 数据源配置的默认清单', () => {
/>
</AuthContext.Provider>
)
it('只有两个入口:PP 来源列在 PP 导入下,AList/钉钉知识库/手工都不出现', () => {
it('统一来源清单显示已接入的 PP 和钉钉来源,不依赖目录快照', () => {
const html = view()
expect(html).toContain('PP资料下载中心')
expect(html).toContain('PP 资料下载中心')
// 三个非 PP 来源一个都不该渲染,也没有任何折叠开关。
for (const hidden of ['国内工业', 'ATOM知识库', '本地手工验收', '显示 AList', '折叠'])
expect(html).not.toContain(hidden)
for (const name of ['国内工业', 'ATOM知识库']) expect(html).toContain(name)
expect(html).not.toContain('显示 AList')
// 统一入口:一个「+ 添加数据源」向导,不再有按类型分家的 dws 导入面板标题。
expect(html).toContain('+ 添加数据源')
expect(html).not.toContain('PP 导入')
+6 -6
View File
@@ -77,7 +77,7 @@ export const ppSourceConnections = (connections: SourceConnectionListItem[]) =>
connections.filter((item) => item.source_type === 'pp')
const sourceAuthLabels: Record<string, string> = {
dingtalk: '专用浏览器扫码会话',
dingtalk: 'dws 共享授权',
pp: '公开只读访问',
alist: '直连 API Token',
manual: '离线导入,无在线认证',
@@ -358,8 +358,8 @@ export function SystemSettingsView({
)
const [dirty, setDirty] = useState(false)
// 本页只列 PP 来源;dws 导入由下面的发现面板负责(其余类型已归档下架,未删除)。
const ppConnections = ppSourceConnections(connections)
// 统一入口后,清单里列全部来源(钉钉知识库 / 钉盘 / PP),不再只显示 PP。
const listedConnections = connections
useEffect(() => {
if (!settings) return
@@ -444,9 +444,9 @@ export function SystemSettingsView({
/>
) : null}
<div className="settings-source-group" id="settings-sources-pp">
{ppConnections.length ? (
{listedConnections.length ? (
<div className="settings-source-list">
{ppConnections.map((connection) => (
{listedConnections.map((connection) => (
<SourceSettingsCard
key={connection.id}
connection={connection}
@@ -458,7 +458,7 @@ export function SystemSettingsView({
)}
</div>
) : (
<div className="empty-state">本页只列 PP 来源;现在还没有已接入的 PP 来源。</div>
<div className="empty-state">还没有已接入的来源:用上面的「+ 添加数据源」加入。</div>
)}
</div>
</section>