zheromiku
dcdc8d1ddd
全栈知识治理平台首次入库（P1-P10 + 发布链路模拟验证闭环）： - 治理核心：源目录/变更集/人工审核/发布批次/发布执行（fail-closed 闸口） - 测试通道：WeKnora 测试 KB 上传/轮询/验收；poll 完成时写入 parser_generation/index_generation，修复 manual/seed 路径发布闸口永久拦截 - 发布适配器 v071（本地 source=测试角色 / 云端 target=生产角色 模拟验证通过）： * ES 导出/导入 mapping 自适应（.keyword 子字段、可选 tenant_id） * OSS locator 支持嵌套 scheme；chunks.seq_id 由目标序列分配 * bundle 归档同身份重试原子替换；verifier 按 bundle chunk_type 计数 * 回滚 job payload 支持 prior_execution_id（strict 解码） - 三通道发布+回滚演练实测：publish 4/4 verified，检索命中， rollback 4/4 rolled_back，目标端 PG is_enabled=false、ES is_enabled=false - 前端 A·Data-console 风格全站统一，120 用例通过 - 门禁：go vet / go test 全包通过，gofmt 干净
91 lines
3.6 KiB
Python
91 lines
3.6 KiB
Python
"""Enrich blank PP display metadata from existing read-only snapshots.
|
|
|
|
No network/source writes, uploads, visibility changes, or source versions.
|
|
Default runs the scoped update in a transaction and rolls it back.
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
|
|
ROOT = Path('/home/zheromiku/dobot-portal-sync/data')
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def read_snapshot(relative):
|
|
path = ROOT / relative
|
|
for part in [ROOT, *path.parents, path]:
|
|
if part.is_symlink():
|
|
raise ValueError('snapshot symlinks are forbidden')
|
|
if not path.is_file() or not path.resolve().is_relative_to(ROOT.resolve()):
|
|
raise ValueError('invalid snapshot file')
|
|
return json.loads(path.read_text())
|
|
|
|
|
|
def label(value):
|
|
if not isinstance(value, str):
|
|
return ''
|
|
value = value.strip()
|
|
if len(value) > 512 or any(ord(c) < 32 for c in value) or '://' in value:
|
|
return ''
|
|
return value
|
|
|
|
|
|
def build_rows(manifest, inventory):
|
|
by_id = {str(item['id']): item for item in inventory['items']}
|
|
result = {}
|
|
for item in manifest['results']:
|
|
members = [by_id[str(i)] for i in item.get('source_ids', []) if str(i) in by_id]
|
|
titles = sorted({label(m.get('fileName')) for m in members} - {''})
|
|
if not titles:
|
|
filename = label(Path(item.get('local_path', '')).name)
|
|
if filename:
|
|
titles = [filename]
|
|
if not titles:
|
|
continue
|
|
categories = {label((m.get('potalCategory') or {}).get('category')) for m in members}
|
|
directory = next(iter(categories)) if len(categories) == 1 else ''
|
|
# This is a PP category label, NOT a reconstructed ancestor hierarchy.
|
|
if '/' in directory or '\\' in directory or directory in ('.', '..'):
|
|
directory = ''
|
|
source_hash = hashlib.sha256(item['url'].encode()).hexdigest()
|
|
row = {'hash': source_hash, 'title': titles[0], 'directory': directory}
|
|
if source_hash in result and result[source_hash] != row:
|
|
raise ValueError('conflicting display metadata')
|
|
result[source_hash] = row
|
|
return sorted(result.values(), key=lambda x: x['hash'])
|
|
|
|
|
|
def update_sql(rows, apply=False):
|
|
payload = json.dumps(rows, ensure_ascii=False).replace("'", "''")
|
|
return """BEGIN;
|
|
SET LOCAL standard_conforming_strings = on;
|
|
WITH metadata AS (
|
|
SELECT * FROM jsonb_to_recordset('%s'::jsonb) AS m(hash text, title text, directory text)
|
|
), changed AS (
|
|
UPDATE source_items AS i SET title = m.title, normalized_path = m.directory
|
|
FROM metadata m, source_connections c
|
|
WHERE c.id = i.source_connection_id AND c.source_type = 'pp'
|
|
AND i.canonical_uri_hash = m.hash AND i.title = ''
|
|
RETURNING i.id
|
|
) SELECT count(*) AS enriched FROM changed;
|
|
%s;
|
|
""" % (payload, 'COMMIT' if apply else 'ROLLBACK')
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--apply', action='store_true')
|
|
args = parser.parse_args()
|
|
rows = build_rows(read_snapshot('public-documents/manifest.json'), read_snapshot('pp-full/inventory-snapshot.json'))
|
|
print(f'matched={len(rows)} with_category={sum(bool(r["directory"]) for r in rows)} mode={"apply" if args.apply else "dry-run"}', flush=True)
|
|
result = subprocess.run(['docker', 'compose', 'exec', '-T', 'postgres', 'psql', '-X', '-v', 'ON_ERROR_STOP=1', '-U', 'governance', '-d', 'governance', '-At'], input=update_sql(rows, args.apply), text=True, cwd=REPO, capture_output=True)
|
|
if result.returncode:
|
|
raise SystemExit('metadata update failed; transaction rolled back (details withheld)')
|
|
print(result.stdout)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|