knowledge-governance/scripts/source_catalog_import.py

"""Build immutable, read-only directory snapshots from existing source inventories.

The pure builders never call source APIs. Database activation is added separately;
raw source identifiers, URLs, host paths and credentials are not emitted.
"""
from __future__ import annotations

import csv
import datetime as dt
import hashlib
import json
import argparse
from pathlib import Path
import subprocess
from typing import Iterable, TextIO

PP_ROOTS = ('工业产品','教育产品','具身产品','多足产品','解决方案','公司','软件','生态&配件')
PP_PUBLIC_EXTENSIONS={'.pdf','.docx','.doc','.xlsx','.xls','.pptx','.ppt','.md','.txt','.csv','.json'}


def digest(value:str)->str:
    return hashlib.sha256(value.encode()).hexdigest()


def node_id(source_key:str,raw_key:str)->str:
    return 'sn-'+digest(source_key+'\0'+raw_key)[:32]


def safe_text(value,maximum=1024)->str:
    if not isinstance(value,str): return ''
    symbols={'\n':'␤','\r':'␍','\t':'␉'}
    value=''.join(symbols.get(ch,'�' if ord(ch)<32 else ch) for ch in value).strip()
    if not value or len(value)>maximum: return ''
    return value


def safe_relative_path(value:str)->str:
    value=value.strip('/')
    if len(value)>4096 or any(ord(ch)<32 for ch in value): raise ValueError('unsafe catalog path')
    if any(part in ('.','..') for part in value.split('/')): raise ValueError('unsafe catalog path')
    return value


def normalized_source_path(value:str)->str:
    parts=[]
    for part in value.strip('/').split('/'):
        if part:
            cleaned=safe_text(part)
            if not cleaned:raise ValueError('source path segment is invalid')
            parts.append(cleaned)
    return safe_relative_path('/'.join(parts))


def iso_from_millis(value)->str|None:
    if value in (None,''): return None
    return dt.datetime.fromtimestamp(int(value)/1000,dt.timezone.utc).isoformat().replace('+00:00','Z')


def new_node(source_key:str,raw_key:str,parent_raw_key:str|None,node_type:str,name,relative_path:str,size_bytes=None,modified_at=None,sort_order=0,source_item_hash='',source_locator_hash='',content_sha256=''):
    name=safe_text(name)
    if not name: raise ValueError('catalog node has no safe name')
    if content_sha256:
        content_sha256=str(content_sha256).lower()
        if len(content_sha256)!=64 or any(c not in '0123456789abcdef' for c in content_sha256): raise ValueError('invalid content hash')
    return {
        'key':node_id(source_key,raw_key),
        'parent_key':None if parent_raw_key is None else node_id(source_key,parent_raw_key),
        'node_type':node_type,'name':name,'relative_path':safe_relative_path(relative_path),
        'source_locator_hash':source_locator_hash or digest(source_key+'\0'+raw_key),
        'source_item_hash':source_item_hash,
        'content_sha256':content_sha256 or None,
        'size_bytes':None if size_bytes in (None,'') else int(size_bytes),
        'modified_at':modified_at or None,'sort_order':int(sort_order),
        'direct_child_count':0,'subtree_file_count':0,'subtree_bytes':0,
    }


def finalize_source(source:dict)->dict:
    nodes=source['nodes']; by_key={node['key']:node for node in nodes}
    children={key:[] for key in by_key}
    roots=[]
    for node in nodes:
        if node['parent_key'] is None: roots.append(node)
        elif node['parent_key'] in children: children[node['parent_key']].append(node)
        else: raise ValueError('catalog node has missing parent')
    if len(roots)!=1 or roots[0]['node_type']!='folder': raise ValueError('catalog requires one folder root')
    visiting=set(); memo={}
    def walk(node):
        if node['key'] in memo:return memo[node['key']]
        if node['key'] in visiting:raise ValueError('catalog cycle')
        visiting.add(node['key']); direct=children[node['key']]; node['direct_child_count']=len(direct)
        if node['node_type']=='file': identities={node['source_locator_hash']:node['size_bytes'] or 0}
        else:
            identities={}
            for child in direct:identities.update(walk(child))
        node['subtree_file_count']=len(identities);node['subtree_bytes']=sum(identities.values())
        visiting.remove(node['key']);memo[node['key']]=identities;return identities
    walk(roots[0])
    if len(memo)!=len(nodes):raise ValueError('catalog contains unreachable nodes')
    source['node_count']=len(nodes);source['folder_count']=sum(node['node_type']=='folder' for node in nodes);source['file_count']=roots[0]['subtree_file_count']
    canonical=json.dumps([{k:v for k,v in node.items() if k!='source_item_hash'} for node in sorted(nodes,key=lambda n:n['key'])],ensure_ascii=False,sort_keys=True,separators=(',',':'))
    source['snapshot_sha256']=digest(canonical)
    return source


def build_dingtalk_sources(inventory:dict)->list[dict]:
    result=[]
    for source_key,root in sorted((inventory.get('roots') or {}).items()):
        source_id='dingtalk-'+source_key
        root_raw=safe_text(root.get('configured_root_id'))
        name=safe_text(root.get('live_space_name'))
        if not root_raw or not name:raise ValueError('invalid DingTalk root')
        raw_nodes={key:value for key,value in (root.get('nodes') or {}).items() if not value.get('is_in_trash') and not value.get('illegal')}
        path_cache={root_raw:''};visiting=set()
        def path_for(raw):
            if raw in path_cache:return path_cache[raw]
            if raw in visiting or raw not in raw_nodes:raise ValueError('DingTalk parent chain is incomplete')
            visiting.add(raw);item=raw_nodes[raw];parent=item.get('parent_dentry_uuid')
            parent_path=path_for(parent);name_part=safe_text(item.get('name'))
            if not name_part:raise ValueError('DingTalk node name is invalid')
            path_cache[raw]=safe_relative_path('/'.join(part for part in (parent_path,name_part) if part));visiting.remove(raw);return path_cache[raw]
        nodes=[new_node(source_id,root_raw,None,'folder',name,'',sort_order=-1)]
        for raw,item in raw_nodes.items():
            kind='folder' if item.get('dentry_type')=='folder' else 'file'
            parent=item.get('parent_dentry_uuid')
            if parent!=root_raw and parent not in raw_nodes:raise ValueError('DingTalk parent is missing')
            nodes.append(new_node(source_id,raw,parent,kind,item.get('name'),path_for(raw),item.get('file_size') if kind=='file' else None,iso_from_millis(item.get('content_updated_time') or item.get('updated_time'))))
        result.append(finalize_source({'source_key':source_id,'source_type':'dingtalk','name':name,'nodes':nodes}))
    return result


def build_alist_source(source_key:str,name:str,directories:TextIO,files:TextIO)->dict:
    source_id='alist-'+source_key;name=safe_text(name)
    directory_rows=list(csv.DictReader(directories));file_rows=list(csv.DictReader(files))
    if not directory_rows or directory_rows[0].get('path')!='/':raise ValueError('AList root directory missing')
    nodes=[new_node(source_id,'dir:/',None,'folder',name,'',sort_order=-1)]
    directory_paths={'/'}
    for row in directory_rows[1:]:
        if row.get('status')!='done' or row.get('last_error'):raise ValueError('AList inventory is incomplete')
        path='/' + normalized_source_path(row.get('path',''))
        parent_raw=row.get('parent_path') or '/';parent='/' if parent_raw=='/' else '/'+normalized_source_path(parent_raw);directory_paths.add(path)
        nodes.append(new_node(source_id,'dir:'+path,'dir:'+parent,'folder',Path(path).name,path))
    for row in file_rows:
        path='/' + normalized_source_path(row.get('path',''));parent_raw=row.get('parent_path') or '/';parent='/' if parent_raw=='/' else '/'+normalized_source_path(parent_raw)
        if parent not in directory_paths:raise ValueError('AList file parent missing')
        nodes.append(new_node(source_id,'file:'+path,'dir:'+parent,'file',row.get('name'),path,row.get('size'),row.get('modified')))
    return finalize_source({'source_key':source_id,'source_type':'alist','name':name,'nodes':nodes})


def build_pp_source(manifest:dict,inventory:dict)->dict:
    source_id='pp-download-center';name='PP资料下载中心';root_raw='root';nodes=[new_node(source_id,root_raw,None,'folder',name,'',sort_order=-1)]
    by_id={str(item.get('id')):item for item in inventory.get('items',[])}
    folders=set()
    for order,tag_type in enumerate(PP_ROOTS):
        raw='type:'+tag_type;folders.add(raw);nodes.append(new_node(source_id,raw,root_raw,'folder',tag_type,tag_type,sort_order=order))
    for result in manifest.get('results',[]):
        members=[by_id[str(value)] for value in result.get('source_ids',[]) if str(value) in by_id]
        titles=sorted({safe_text(member.get('fileName')) for member in members}-{''})
        if not titles:
            title=safe_text(Path(result.get('local_path','')).name)
            if not title:
                from urllib.parse import unquote,urlparse
                title=safe_text(Path(unquote(urlparse(str(result.get('url',''))).path)).name)
            titles=[title] if title else []
        if not titles:raise ValueError('PP public file title missing')
        locator=digest(str(result.get('url','')));size=result.get('bytes')
        if size is None:size=max((int(member.get('fileSize') or 0) for member in members),default=0)
        modified=max((safe_text(member.get('updateTime')) for member in members),default='') or None
        tags=set()
        for member in members:
            for tag in member.get('potalTags') or []:
                tag_type=safe_text(tag.get('tagType'));value=safe_text(tag.get('tag'))
                if tag_type in PP_ROOTS and value:tags.add((tag_type,value))
        content_sha=str(result.get('sha256','') or '').lower() if str(result.get('sha256','') or '') else ''
        if not tags:
            nodes.append(new_node(source_id,'file:'+locator,root_raw,'file',titles[0],titles[0],size,modified,source_item_hash=locator,source_locator_hash=locator,content_sha256=content_sha))
        for tag_type,value in sorted(tags,key=lambda pair:(PP_ROOTS.index(pair[0]),pair[1])):
            folder_raw='tag:'+tag_type+':'+value
            if folder_raw not in folders:
                folders.add(folder_raw);nodes.append(new_node(source_id,folder_raw,'type:'+tag_type,'folder',value,tag_type+'/'+value))
            alias_raw='file:'+locator+':'+tag_type+':'+value
            nodes.append(new_node(source_id,alias_raw,folder_raw,'file',titles[0],tag_type+'/'+value+'/'+titles[0],size,modified,source_item_hash=locator,source_locator_hash=locator,content_sha256=content_sha))
    return finalize_source({'source_key':source_id,'source_type':'pp','name':name,'nodes':nodes})


def build_pp_live_source(inventory:dict,manifest:dict|None=None)->dict:
    # P1: the local download manifest is the authoritative content-hash evidence
    # for already-fetched bytes; live portal records absent from it stay null.
    hashes={}
    for entry in (manifest or {}).get('results',[]):
        url=str(entry.get('url') or '').strip()
        value=str(entry.get('sha256') or '').strip().lower()
        if url and len(value)==64 and all(c in '0123456789abcdef' for c in value):
            hashes[url]=value
    results=[]
    for item in inventory.get('items',[]):
        url=str(item.get('fileUrl') or '').strip()
        suffix=str(item.get('fileSuffix') or '').strip().lower()
        if suffix and not suffix.startswith('.'):suffix='.'+suffix
        if not suffix and url:
            from urllib.parse import unquote,urlparse
            suffix=Path(unquote(urlparse(url).path)).suffix.lower()
        if item.get('online')!=1 or (item.get('potalGroup') or {}).get('groupName')!='公开' or suffix not in PP_PUBLIC_EXTENSIONS:
            continue
        if not url:raise ValueError('public PP item has no source URL')
        results.append({'url':url,'source_ids':[str(item.get('id'))],'bytes':int(item.get('fileSize') or 0),'sha256':hashes.get(url,'')})
    merged={}
    for result in results:
        existing=merged.get(result['url'])
        if existing:
            existing['source_ids']=[value for value in dict.fromkeys(existing['source_ids']+result['source_ids'])]
            existing['bytes']=existing['bytes'] or result['bytes']
            existing['sha256']=existing.get('sha256') or result.get('sha256','')
        else:
            merged[result['url']]=dict(result)
    return build_pp_source({'results':list(merged.values())},inventory)


def validate_source(source:dict)->None:
    if source.get('source_type') not in ('dingtalk','alist','pp') or not safe_text(source.get('name')):raise ValueError('invalid source')
    nodes=source.get('nodes') or [];keys=set()
    for node in nodes:
        if node['key'] in keys:raise ValueError('duplicate catalog node')
        keys.add(node['key'])
        if node['node_type'] not in ('folder','file') or not safe_text(node['name']) or node['relative_path'].startswith('/'):
            raise ValueError('invalid catalog node')
        safe_relative_path(node['relative_path'])
        if len(node['source_locator_hash'])!=64:raise ValueError('invalid source locator')
        if node.get('content_sha256') and (len(node['content_sha256'])!=64 or any(c not in '0123456789abcdef' for c in node['content_sha256'])):raise ValueError('invalid content sha256')
    if any(node['parent_key'] is not None and node['parent_key'] not in keys for node in nodes):raise ValueError('orphan catalog node')
    if source.get('node_count')!=len(nodes) or source.get('folder_count')!=sum(node['node_type']=='folder' for node in nodes):raise ValueError('invalid catalog counts')


def load_json(path:Path,allowed_root:Path):
    resolved=path.resolve();root=allowed_root.resolve()
    if path.is_symlink() or not path.is_file() or not resolved.is_relative_to(root):raise ValueError('invalid snapshot path')
    return json.loads(path.read_text())


def stable_connection_id(source_key:str)->str:
    return 'sc-'+digest(source_key)[:32]


def sql_text(value:str)->str:
    return "'"+value.replace("'","''")+"'"


def catalog_statements(source:dict,connection_id:str)->str:
    validate_source(source)
    snapshot_id='ss-'+digest(source['source_key']+'\0'+source['snapshot_sha256'])[:32]
    change_set_id='cs-'+digest(connection_id+'\0'+snapshot_id)[:32]
    payload=json.dumps(source['nodes'],ensure_ascii=False,separators=(',',':')).replace("'","''")
    health=json.dumps({'status':'degraded','reason':'snapshot_only'},separators=(',',':')).replace("'","''")
    return f"""
INSERT INTO source_connections (id,source_type,name,base_url,secret_ref,enabled,schedule,last_health,created_at,updated_at)
VALUES ({sql_text(connection_id)},{sql_text(source['source_type'])},{sql_text(source['name'])},'','',TRUE,'{{}}'::jsonb,'{health}'::jsonb,NOW(),NOW())
ON CONFLICT (id) DO UPDATE SET name=EXCLUDED.name,updated_at=NOW();

INSERT INTO source_catalog_snapshots
    (id,source_connection_id,snapshot_sha256,source_label,node_count,folder_count,file_count,observed_at)
VALUES ({sql_text(snapshot_id)},{sql_text(connection_id)},{sql_text(source['snapshot_sha256'])},{sql_text(source['name'])},{source['node_count']},{source['folder_count']},{source['file_count']},NOW())
ON CONFLICT (source_connection_id,snapshot_sha256) DO NOTHING;

WITH current_snapshot AS (
    SELECT id FROM source_catalog_snapshots
    WHERE source_connection_id={sql_text(connection_id)} AND snapshot_sha256={sql_text(source['snapshot_sha256'])}
), nodes AS (
    SELECT * FROM jsonb_to_recordset('{payload}'::jsonb) AS node(
        key text,parent_key text,node_type text,name text,relative_path text,
        source_locator_hash text,source_item_hash text,content_sha256 text,size_bytes bigint,
        modified_at text,sort_order integer,direct_child_count integer,
        subtree_file_count integer,subtree_bytes bigint
    )
)
INSERT INTO source_catalog_nodes
    (snapshot_id,node_id,parent_node_id,node_type,name,relative_path,source_locator_hash,
     source_item_id,size_bytes,modified_at,sort_order,direct_child_count,subtree_file_count,subtree_bytes,content_sha256)
SELECT snapshot.id,node.key,NULLIF(node.parent_key,''),node.node_type,node.name,node.relative_path,
       node.source_locator_hash,item.id,node.size_bytes,NULLIF(node.modified_at,'')::timestamptz,
       node.sort_order,node.direct_child_count,node.subtree_file_count,node.subtree_bytes,node.content_sha256
FROM current_snapshot snapshot CROSS JOIN nodes node
LEFT JOIN source_items item ON item.source_connection_id={sql_text(connection_id)}
    AND node.source_item_hash<>'' AND item.canonical_uri_hash=node.source_item_hash
ON CONFLICT (snapshot_id,node_id) DO NOTHING;

INSERT INTO source_catalog_change_sets(id,source_connection_id,from_snapshot_id,to_snapshot_id,created_at)
SELECT {sql_text(change_set_id)},{sql_text(connection_id)},
       COALESCE(
         CASE WHEN head.snapshot_id<>target.id THEN head.snapshot_id END,
         (SELECT prior.id FROM source_catalog_snapshots prior
          WHERE prior.source_connection_id={sql_text(connection_id)} AND prior.id<>target.id
            AND prior.observed_at<=target.observed_at
          ORDER BY prior.observed_at DESC,prior.id DESC LIMIT 1)
       ),
       target.id,NOW()
FROM source_catalog_snapshots target
LEFT JOIN source_catalog_heads head ON head.source_connection_id={sql_text(connection_id)}
WHERE target.id={sql_text(snapshot_id)}
ON CONFLICT (to_snapshot_id) DO NOTHING;

SELECT compare_catalog_change_set({sql_text(change_set_id)});

INSERT INTO source_catalog_heads (source_connection_id,snapshot_id,activated_at)
SELECT {sql_text(connection_id)},id,NOW() FROM source_catalog_snapshots
WHERE source_connection_id={sql_text(connection_id)} AND snapshot_sha256={sql_text(source['snapshot_sha256'])}
ON CONFLICT (source_connection_id) DO UPDATE SET snapshot_id=EXCLUDED.snapshot_id,activated_at=EXCLUDED.activated_at;
"""


def catalog_sql(source:dict,connection_id:str,apply=False)->str:
    return "BEGIN;\nSET LOCAL standard_conforming_strings=on;\n"+catalog_statements(source,connection_id)+("COMMIT;\n" if apply else "ROLLBACK;\n")


def checked_csv(path:Path,allowed_root:Path):
    if path.is_symlink() or not path.is_file() or not path.resolve().is_relative_to(allowed_root.resolve()):raise ValueError('invalid CSV snapshot path')
    return path.open(encoding='utf-8-sig',newline='')


def build_local_sources()->list[dict]:
    dingtalk_root=Path('/home/zheromiku/dingtalk_kb_download/data/dingtalk-internal')
    pp_root=Path('/home/zheromiku/dobot-portal-sync/data')
    alist_root=Path('/home/zheromiku/tmp/alist/inventory')
    sources=build_dingtalk_sources(load_json(dingtalk_root/'inventory-state.json',dingtalk_root))
    for slug,label in (('domestic-industrial','国内工业'),('global-industrial','海外工业'),('domestic-education','国内教育'),('global-education','海外教育')):
        with checked_csv(alist_root/slug/'directories.csv',alist_root) as directories,checked_csv(alist_root/slug/'files.csv',alist_root) as files:
            sources.append(build_alist_source(slug,label,directories,files))
    sources.append(build_pp_source(load_json(pp_root/'public-documents/manifest.json',pp_root),load_json(pp_root/'pp-full/inventory-snapshot.json',pp_root)))
    return sources


def psql(sql:str):
    command=['docker','compose','exec','-T','postgres','psql','-X','-v','ON_ERROR_STOP=1','-U','governance','-d','governance','-At']
    return subprocess.run(command,input=sql,text=True,cwd=Path(__file__).resolve().parents[1],capture_output=True)


def main():
    parser=argparse.ArgumentParser(description=__doc__);parser.add_argument('--apply',action='store_true');args=parser.parse_args()
    sources=build_local_sources()
    lookup=psql("SELECT id FROM source_connections WHERE source_type='pp' ORDER BY created_at,id LIMIT 1;\n")
    if lookup.returncode:raise SystemExit('could not resolve PP connection (details withheld)')
    pp_connection=lookup.stdout.strip() or stable_connection_id('pp-download-center')
    statements=[]
    for source in sources:
        connection_id=pp_connection if source['source_type']=='pp' else stable_connection_id(source['source_key'])
        statements.append(catalog_statements(source,connection_id))
    sql="BEGIN;\nSET LOCAL standard_conforming_strings=on;\n"+'\n'.join(statements)+("\nCOMMIT;\n" if args.apply else "\nROLLBACK;\n")
    result=psql(sql)
    if result.returncode:raise SystemExit('catalog import failed; transaction rolled back (details withheld)')
    print('mode='+('apply' if args.apply else 'dry-run'))
    for source in sources:print(f"{source['source_type']} {source['name']}: nodes={source['node_count']} folders={source['folder_count']} files={source['file_count']}")
    print(f"total_sources={len(sources)} total_nodes={sum(s['node_count'] for s in sources)}")


if __name__=='__main__':
    main()