#!/usr/bin/env python3 """Sync zhctprompt and company WeDrive indexes into local WeKnora RAG. This is a safe knowledge mirror. It imports zhctprompt text assets plus every row from the enterprise WeDrive metadata indexes. It does not bulk-copy or parse raw WeDrive binaries, credentials, logs, private configs, databases, or browser profiles. """ from __future__ import annotations import argparse import csv import datetime as dt import hashlib import json import os import re import sys import time import urllib.error import urllib.parse import urllib.request from collections import Counter, defaultdict from dataclasses import dataclass from pathlib import Path from typing import Any from zoneinfo import ZoneInfo REPO_ROOT = Path(__file__).resolve().parents[2] NOTEBOOKLM_SCRIPTS = Path("/Users/jack/code/099-github/notebooklm-mcp/scripts") if str(NOTEBOOKLM_SCRIPTS) not in sys.path: sys.path.insert(0, str(NOTEBOOKLM_SCRIPTS)) from sync_daily_inputs_to_weknora import ( # noqa: E402 DEFAULT_BASE_URL, DEFAULT_EMBEDDING_MODEL_DIMENSION, DEFAULT_EMBEDDING_MODEL_NAME, KnowledgeDoc, WeKnoraClient, ) SHANGHAI = ZoneInfo("Asia/Shanghai") OUTPUT_ROOT = REPO_ROOT / "work" / "2026-06-09-zhctprompt-weknora-rag-sync" DEFAULT_KB_NAME = "ZHCT Prompt OS 企业微盘知识库" DEFAULT_KB_DESCRIPTION = ( "zhctprompt 控制项目、企业微信微盘派生索引、项目规则、skills、模块入口、" "任务队列和可审计项目资料的本地 RAG 知识库。微盘原始二进制仍留在企业微信微盘，" "本知识库只保存可检索 metadata、分类、source path 和项目内安全文本。" ) PROJECT_TAGS = [ ("zhct:index", "#08979c"), ("zhct:rules", "#722ed1"), ("zhct:standards", "#531dab"), ("zhct:skills", "#2f54eb"), ("zhct:modules", "#13c2c2"), ("zhct:control", "#1677ff"), ("zhct:tasks", "#fa8c16"), ("zhct:work", "#52c41a"), ("zhct:company-wedrive", "#d48806"), ("zhct:rag-sync", "#595959"), ] TEXT_EXTENSIONS = { ".cfg", ".conf", ".css", ".csv", ".html", ".ini", ".js", ".json", ".jsonl", ".md", ".py", ".sh", ".sql", ".toml", ".tsv", ".txt", ".xml", ".yaml", ".yml", } EXCLUDED_DIR_NAMES = { ".git", ".pytest_cache", "__pycache__", "node_modules", "private", "runtime", "tmp", "venv", } EXCLUDED_FILE_EXTENSIONS = { ".7z", ".avi", ".bin", ".bmp", ".db", ".dmg", ".doc", ".docx", ".DS_Store", ".gif", ".gz", ".heic", ".jpeg", ".jpg", ".log", ".m4a", ".mov", ".mp3", ".mp4", ".pdf", ".png", ".ppt", ".pptx", ".sqlite", ".tgz", ".wasm", ".webm", ".webp", ".xls", ".xlsx", ".zip", } EXCLUDED_FILE_NAMES = { ".env", ".env.local", ".env.production", ".netrc", ".DS_Store", "cookies.json", "id_rsa", "id_ed25519", "package-lock.json", "pnpm-lock.yaml", "yarn.lock", } EXCLUDED_REL_PREFIXES = { "config/local", "docker/local-config-templates", "docker/zhctapp_", "work/2026-06-09-zhctprompt-weknora-rag-sync", } SECRET_LINE_PATTERNS = [ re.compile( r"(?i)(api[_-]?key|access[_-]?token|refresh[_-]?token|authorization|bearer|cookie|password|secret|private[_-]?key)\s*[:=]\s*['\"]?[^'\"\s]{8,}" ), re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"), ] @dataclass class FileRecord: path: str group: str tag: str size: int sha256: str content: str def now_iso() -> str: return dt.datetime.now(tz=SHANGHAI).isoformat(timespec="seconds") def derive_health_url(base_url: str) -> str: parsed = urllib.parse.urlparse(base_url) if not parsed.scheme or not parsed.netloc: raise ValueError(f"invalid WeKnora base URL: {base_url}") return urllib.parse.urlunparse((parsed.scheme, parsed.netloc, "/health", "", "", "")) def chunked(items: list[KnowledgeDoc], size: int) -> list[list[KnowledgeDoc]]: if size <= 0: return [items] return [items[index : index + size] for index in range(0, len(items), size)] def rel(path: Path) -> str: return path.relative_to(REPO_ROOT).as_posix() def is_under_prefix(path: str, prefixes: set[str]) -> bool: return any(path == prefix or path.startswith(prefix.rstrip("/") + "/") for prefix in prefixes) def classify_project_file(path: str) -> tuple[str, str]: if path in {"AGENTS.md", "README.md", "HEARTBEAT.md", "PROJECT_IPO_USAGE_GUIDE.md"}: return "00-root-rules", "zhct:rules" if path.startswith("tasks/"): return "05-tasks", "zhct:tasks" if path.startswith("control/"): return "10-control-" + path.split("/")[1], "zhct:control" if path.startswith("standards-stack/agent-skills/skills/"): parts = path.split("/") skill = parts[3] if len(parts) > 3 else "unknown" return f"20-skills-{skill}", "zhct:skills" if path.startswith("standards-stack/"): parts = path.split("/") return f"15-standards-{parts[1] if len(parts) > 1 else 'root'}", "zhct:standards" if path.startswith("modules/"): parts = path.split("/") return f"30-modules-{parts[1] if len(parts) > 1 else 'root'}", "zhct:modules" if path.startswith("work_company_knowledge/"): return "35-company-wedrive-derived-index", "zhct:company-wedrive" if path.startswith("work_") or path.startswith("work/") or path.startswith("product_"): return "40-work-" + path.split("/")[0], "zhct:work" if path.startswith("docker/"): return "50-docker", "zhct:control" return "60-other", "zhct:index" def should_skip_project_file(path: Path, max_file_bytes: int) -> str | None: path_str = rel(path) parts = set(path_str.split("/")[:-1]) if parts & EXCLUDED_DIR_NAMES: return "excluded_dir" if is_under_prefix(path_str, EXCLUDED_REL_PREFIXES): return "excluded_private_or_output_prefix" if path.name in EXCLUDED_FILE_NAMES: return "excluded_filename" if path.suffix in EXCLUDED_FILE_EXTENSIONS: return "excluded_extension" if path.suffix not in TEXT_EXTENSIONS and path.name not in {"AGENTS.md", "README.md", "HEARTBEAT.md"}: return "non_text_extension" try: size = path.stat().st_size except OSError: return "stat_failed" if size <= 0: return "empty" if size > max_file_bytes: return "too_large" return None def redact_secret_like_lines(text: str) -> tuple[str, int]: redacted = 0 lines = [] for line in text.splitlines(): if any(pattern.search(line) for pattern in SECRET_LINE_PATTERNS): lines.append("[REDACTED_SECRET_LIKE_LINE]") redacted += 1 else: lines.append(line) return "\n".join(lines), redacted def read_project_record(path: Path) -> tuple[FileRecord | None, int]: path_str = rel(path) raw = path.read_bytes() digest = hashlib.sha256(raw).hexdigest() text = raw.decode("utf-8", errors="ignore") text, redacted = redact_secret_like_lines(text) group, tag = classify_project_file(path_str) content = "\n".join( [ f"## File: `{path_str}`", f"- sha256: `{digest}`", f"- bytes: {len(raw)}", "", "```text", text.strip(), "```", "", ] ) return FileRecord(path=path_str, group=group, tag=tag, size=len(raw), sha256=digest, content=content), redacted def collect_project_files(max_file_bytes: int) -> tuple[list[FileRecord], list[dict[str, Any]], Counter[str], int]: records: list[FileRecord] = [] skipped: list[dict[str, Any]] = [] reasons: Counter[str] = Counter() redacted_total = 0 for dirpath, dirnames, filenames in os.walk(REPO_ROOT): current = Path(dirpath) kept_dirs = [] for dirname in sorted(dirnames): child = current / dirname child_rel = rel(child) if dirname in EXCLUDED_DIR_NAMES: reasons["pruned_dir"] += 1 skipped.append({"path": child_rel + "/", "reason": "pruned_dir"}) continue if is_under_prefix(child_rel, EXCLUDED_REL_PREFIXES): reasons["pruned_private_or_output_dir"] += 1 skipped.append({"path": child_rel + "/", "reason": "pruned_private_or_output_dir"}) continue kept_dirs.append(dirname) dirnames[:] = kept_dirs for filename in sorted(filenames): path = current / filename if not path.is_file(): continue path_str = rel(path) reason = should_skip_project_file(path, max_file_bytes) if reason: reasons[reason] += 1 skipped.append({"path": path_str, "reason": reason}) continue try: record, redacted = read_project_record(path) except Exception as exc: reasons["read_failed"] += 1 skipped.append({"path": path_str, "reason": "read_failed", "error": str(exc)[:240]}) continue if record: records.append(record) redacted_total += redacted return records, skipped, reasons, redacted_total def package_project_records(records: list[FileRecord], max_package_chars: int) -> list[KnowledgeDoc]: by_group: dict[str, list[FileRecord]] = defaultdict(list) for record in records: by_group[record.group].append(record) docs: list[KnowledgeDoc] = [] for group in sorted(by_group): current: list[FileRecord] = [] current_chars = 0 part = 1 for record in by_group[group]: if current and current_chars + len(record.content) > max_package_chars: docs.append(build_project_doc(group, part, current)) current = [] current_chars = 0 part += 1 current.append(record) current_chars += len(record.content) if current: docs.append(build_project_doc(group, part, current)) return docs def build_project_doc(group: str, part: int, records: list[FileRecord]) -> KnowledgeDoc: tag = Counter(record.tag for record in records).most_common(1)[0][0] title = f"[ZHCT项目知识包] {group} part {part:02d}" paths = "\n".join(f"- `{record.path}` ({record.size} bytes, sha256 `{record.sha256[:12]}`)" for record in records) content = "\n".join( [ f"# {title}", "", f"- source_repo: `{REPO_ROOT}`", f"- group: `{group}`", f"- files: {len(records)}", f"- generated_at: {now_iso()}", "- privacy_boundary: excludes credentials, local private config, logs, databases, caches, browser profiles, and large binaries; secret-like lines are redacted.", "", "## Included Files", paths, "", "## Content", "", *[record.content for record in records], ] ) return KnowledgeDoc( title=title, content=content.rstrip() + "\n", tag=tag, channel="zhctprompt-rag-sync:project", source_kind="zhctprompt-project", item_key=f"zhctprompt:project:{group}:{part:02d}", ) def read_wedrive_inventory_rows(limit: int | None = None) -> list[dict[str, str]]: path = REPO_ROOT / "work_company_knowledge" / "indexes" / "file-inventory.tsv" if not path.exists(): return [] rows: list[dict[str, str]] = [] with path.open("r", encoding="utf-8", newline="") as handle: reader = csv.DictReader(handle, delimiter="\t") for row in reader: rows.append({k: str(v or "") for k, v in row.items()}) if limit and len(rows) >= limit: break return rows def package_wedrive_rows(rows: list[dict[str, str]], max_package_chars: int) -> list[KnowledgeDoc]: grouped: dict[tuple[str, str], list[dict[str, str]]] = defaultdict(list) for row in rows: grouped[(row.get("source", "unknown"), row.get("knowledge_area", "unknown"))].append(row) docs: list[KnowledgeDoc] = [] for (source, area) in sorted(grouped): bucket = grouped[(source, area)] part = 1 lines: list[str] = [] current_chars = 0 for row in bucket: line = ( f"- `{row.get('file_id')}` | {row.get('source')} | {row.get('knowledge_area')} | " f"{row.get('role_lane')} | {row.get('stage')} | {row.get('file_kind')} | " f"{row.get('mtime')} | {row.get('name')} | rel=`{row.get('rel_path')}` | " f"abs=`{row.get('abs_path')}`" ) if lines and current_chars + len(line) > max_package_chars: docs.append(build_wedrive_doc(source, area, part, lines)) lines = [] current_chars = 0 part += 1 lines.append(line) current_chars += len(line) if lines: docs.append(build_wedrive_doc(source, area, part, lines)) return docs def build_wedrive_doc(source: str, area: str, part: int, lines: list[str]) -> KnowledgeDoc: title = f"[微盘索引包] {source} {area} part {part:02d}" content = "\n".join( [ f"# {title}", "", f"- generated_at: {now_iso()}", f"- source: `{source}`", f"- knowledge_area: `{area}`", f"- file_rows: {len(lines)}", "- boundary: this package mirrors WeDrive metadata and source paths only; raw binaries and full document bodies remain in enterprise WeDrive.", "", "## File Rows", *lines, ] ) return KnowledgeDoc( title=title, content=content.rstrip() + "\n", tag="zhct:company-wedrive", channel="zhctprompt-rag-sync:wedrive-index", source_kind="company-wedrive-index", item_key=f"zhctprompt:wedrive:{source}:{area}:{part:02d}", ) def build_index_doc( project_records: list[FileRecord], skipped: list[dict[str, Any]], reasons: Counter[str], redacted_total: int, wedrive_rows: list[dict[str, str]], project_docs_count: int, wedrive_docs_count: int, ) -> KnowledgeDoc: by_tag = Counter(record.tag for record in project_records) by_group = Counter(record.group for record in project_records) wedrive_by_source = Counter(row.get("source", "unknown") for row in wedrive_rows) wedrive_by_area = Counter(row.get("knowledge_area", "unknown") for row in wedrive_rows) content = "\n".join( [ "# zhctprompt + 企业微盘 RAG 同步索引", "", f"- generated_at: {now_iso()}", f"- source_repo: `{REPO_ROOT}`", f"- target_kb_name: `{DEFAULT_KB_NAME}`", f"- project_files_included: {len(project_records)}", f"- wedrive_file_rows_included: {len(wedrive_rows)}", f"- project_packages: {project_docs_count}", f"- wedrive_index_packages: {wedrive_docs_count}", f"- redacted_secret_like_lines: {redacted_total}", "- boundary: project text and WeDrive metadata only; no raw WeDrive binaries, local private configs, tokens, cookies, databases, logs, browser profiles, or full private transcripts.", "", "## Project Files By Tag", "\n".join(f"- {tag}: {count}" for tag, count in by_tag.most_common()), "", "## Project Files By Group", "\n".join(f"- {group}: {count}" for group, count in by_group.most_common(80)), "", "## WeDrive Rows By Source", "\n".join(f"- {source}: {count}" for source, count in wedrive_by_source.most_common()), "", "## WeDrive Rows By Knowledge Area", "\n".join(f"- {area}: {count}" for area, count in wedrive_by_area.most_common()), "", "## Skipped Project File Reasons", "\n".join(f"- {reason}: {count}" for reason, count in reasons.most_common()), "", "## Skipped Sample", "\n".join(f"- `{item['path']}` -> {item['reason']}" for item in skipped[:250]) or "- none", ] ) return KnowledgeDoc( title="[ZHCT索引] zhctprompt 企业微盘安全 RAG 镜像", content=content.rstrip() + "\n", tag="zhct:index", channel="zhctprompt-rag-sync:index", source_kind="zhctprompt-rag-index", item_key="zhctprompt:index", ) def ensure_knowledge_base(client: WeKnoraClient, name: str, description: str) -> str: response = client.request("GET", "/knowledge-bases") for item in response.get("data", []) if isinstance(response, dict) else []: if item.get("name") == name: return str(item["id"]) embedding_model_id = client.ensure_embedding_model(DEFAULT_EMBEDDING_MODEL_NAME, DEFAULT_EMBEDDING_MODEL_DIMENSION) payload = { "name": name, "description": description, "type": "document", "is_temporary": False, "embedding_model_id": embedding_model_id, "summary_model_id": "deltaf-lite-summary-disabled", "storage_provider_config": {"provider": "local"}, "chunking_config": { "chunk_size": 1200, "chunk_overlap": 160, "separators": ["\n\n", "\n", "。", "；"], }, "indexing_strategy": { "keyword_enabled": True, "vector_enabled": False, "graph_enabled": False, "wiki_enabled": False, }, "question_generation_config": {"enabled": False, "question_count": 0}, "asr_config": {"enabled": False, "language": "", "model_id": ""}, "extract_config": None, "faq_config": None, } created = client.request("POST", "/knowledge-bases", payload) return str(created["data"]["id"]) def load_failed_doc_identifiers(manifest_path: Path) -> tuple[set[str], set[str]]: if not manifest_path.exists(): raise FileNotFoundError(f"resume manifest not found: {manifest_path}") payload = json.loads(manifest_path.read_text(encoding="utf-8")) sync_result = payload.get("sync_result", {}) if isinstance(payload, dict) else {} failed_titles: set[str] = set() failed_keys: set[str] = set() for item in sync_result.get("errors", []) if isinstance(sync_result, dict) else []: if not isinstance(item, dict): continue title = str(item.get("title") or "").strip() item_key = str(item.get("item_key") or "").strip() if title: failed_titles.add(title) if item_key: failed_keys.add(item_key) if not failed_titles and not failed_keys and sync_result.get("resume_failed") and not sync_result.get("applied"): for item in payload.get("target_documents", []) if isinstance(payload, dict) else []: if not isinstance(item, dict): continue title = str(item.get("title") or "").strip() item_key = str(item.get("item_key") or "").strip() if title: failed_titles.add(title) if item_key: failed_keys.add(item_key) return failed_titles, failed_keys def filter_resume_failed_docs(docs: list[KnowledgeDoc], manifest_path: Path) -> list[KnowledgeDoc]: failed_titles, failed_keys = load_failed_doc_identifiers(manifest_path) if not failed_titles and not failed_keys: return [] return [ doc for doc in docs if doc.title in failed_titles or doc.item_key in failed_keys ] def resolve_resume_manifest(candidate: Path | None) -> Path: if candidate: return candidate search_paths = [OUTPUT_ROOT / "latest" / "manifest.json"] snapshots = OUTPUT_ROOT / "snapshots" if snapshots.exists(): search_paths.extend(sorted(snapshots.glob("*/manifest.json"), reverse=True)) for path in search_paths: if not path.exists(): continue failed_titles, failed_keys = load_failed_doc_identifiers(path) if failed_titles or failed_keys: return path return OUTPUT_ROOT / "latest" / "manifest.json" def health_check(url: str, timeout: float) -> tuple[bool, str]: try: with urllib.request.urlopen(url, timeout=timeout) as response: body = response.read().decode("utf-8", errors="ignore")[:240] return 200 <= response.status < 300, body except urllib.error.HTTPError as exc: body = exc.read().decode("utf-8", errors="ignore")[:240] return False, f"HTTP {exc.code}: {body}" except Exception as exc: return False, str(exc)[:240] def wait_for_health( url: str, *, timeout: float, attempts: int, sleep_seconds: float, backoff_multiplier: float, ) -> dict[str, Any]: events: list[dict[str, Any]] = [] delay = max(sleep_seconds, 0) total_attempts = max(attempts, 1) for attempt in range(1, total_attempts + 1): ok, detail = health_check(url, timeout) event = {"attempt": attempt, "ok": ok, "detail": detail, "checked_at": now_iso()} events.append(event) if ok: return {"ok": True, "events": events} if attempt < total_attempts and delay > 0: time.sleep(delay) delay *= max(backoff_multiplier, 1.0) return {"ok": False, "events": events} def sync_docs( client: WeKnoraClient, kb_id: str, tag_ids: dict[str, str], docs: list[KnowledgeDoc], *, batch_size: int, sleep_between_docs: float, retry_attempts: int, retry_sleep_seconds: float, retry_backoff_multiplier: float, health_url: str, health_timeout: float, health_attempts: int, health_sleep_seconds: float, health_backoff_multiplier: float, ) -> dict[str, Any]: result: dict[str, Any] = { "target_packages": len(docs), "attempted": 0, "upserted": 0, "errors": [], "batches": [], "doc_results": [], "health_url": health_url, } batches = chunked(docs, batch_size) for batch_index, batch in enumerate(batches, start=1): before_health = wait_for_health( health_url, timeout=health_timeout, attempts=health_attempts, sleep_seconds=health_sleep_seconds, backoff_multiplier=health_backoff_multiplier, ) batch_record: dict[str, Any] = { "batch_index": batch_index, "packages": len(batch), "started_at": now_iso(), "before_health": before_health, "upserted": 0, "errors": 0, } result["batches"].append(batch_record) if not before_health["ok"]: for doc in batch: error = { "title": doc.title, "item_key": doc.item_key, "batch_index": batch_index, "error": "health_check_failed_before_batch", "health_events": before_health["events"], } result["errors"].append(error) result["doc_results"].append({**error, "status": "skipped"}) batch_record["errors"] += 1 batch_record["finished_at"] = now_iso() continue for doc_index, doc in enumerate(batch, start=1): result["attempted"] += 1 last_error = "" manual_id = "" retry_events: list[dict[str, Any]] = [] delay = max(retry_sleep_seconds, 0) for attempt in range(1, max(retry_attempts, 1) + 1): try: manual_id = client.upsert_manual(kb_id, doc, tag_ids) last_error = "" break except Exception as exc: last_error = str(exc)[:500] retry_events.append({"attempt": attempt, "error": last_error, "failed_at": now_iso()}) if attempt < max(retry_attempts, 1) and delay > 0: time.sleep(delay) delay *= max(retry_backoff_multiplier, 1.0) if last_error: error = { "title": doc.title, "item_key": doc.item_key, "batch_index": batch_index, "error": last_error, "attempts": max(retry_attempts, 1), "retry_events": retry_events, } result["errors"].append(error) result["doc_results"].append({**error, "status": "failed"}) batch_record["errors"] += 1 after_failure_health = wait_for_health( health_url, timeout=health_timeout, attempts=health_attempts, sleep_seconds=health_sleep_seconds, backoff_multiplier=health_backoff_multiplier, ) result["doc_results"][-1]["after_failure_health"] = after_failure_health else: result["upserted"] += 1 batch_record["upserted"] += 1 result["doc_results"].append( { "title": doc.title, "item_key": doc.item_key, "batch_index": batch_index, "status": "upserted", "manual_id": manual_id, } ) if sleep_between_docs > 0 and (batch_index < len(batches) or doc_index < len(batch)): time.sleep(sleep_between_docs) batch_record["after_health"] = wait_for_health( health_url, timeout=health_timeout, attempts=health_attempts, sleep_seconds=health_sleep_seconds, backoff_multiplier=health_backoff_multiplier, ) batch_record["finished_at"] = now_iso() return result def write_outputs( docs: list[KnowledgeDoc], target_docs: list[KnowledgeDoc], project_records: list[FileRecord], skipped: list[dict[str, Any]], reasons: Counter[str], wedrive_rows: list[dict[str, str]], sync_result: dict[str, Any], ) -> dict[str, str]: latest = OUTPUT_ROOT / "latest" snapshot = OUTPUT_ROOT / "snapshots" / dt.datetime.now(tz=SHANGHAI).strftime("%Y-%m-%d-%H%M%S") latest.mkdir(parents=True, exist_ok=True) snapshot.mkdir(parents=True, exist_ok=True) manifest = { "schema": "zhctprompt-weknora-rag-sync.v1", "generated_at": now_iso(), "source_repo": str(REPO_ROOT), "knowledge_base_name": sync_result.get("knowledge_base_name"), "knowledge_base_id": sync_result.get("knowledge_base_id"), "project_files_included": len(project_records), "project_files_skipped": len(skipped), "project_skip_reasons": dict(reasons), "wedrive_file_rows_included": len(wedrive_rows), "sync_result": sync_result, "documents": [ { "title": doc.title, "tag": doc.tag, "channel": doc.channel, "source_kind": doc.source_kind, "item_key": doc.item_key, } for doc in docs ], "target_documents": [ { "title": doc.title, "tag": doc.tag, "channel": doc.channel, "source_kind": doc.source_kind, "item_key": doc.item_key, } for doc in target_docs ], "included_project_files": [ {"path": r.path, "group": r.group, "tag": r.tag, "size": r.size, "sha256": r.sha256} for r in project_records ], "skipped_project_sample": skipped[:1000], } summary = "\n".join( [ "# WeKnora ZHCT Prompt RAG Sync", "", f"- generated_at: {manifest['generated_at']}", f"- source_repo: `{REPO_ROOT}`", f"- applied: {sync_result.get('applied')}", f"- knowledge_base: {sync_result.get('knowledge_base_name')}", f"- knowledge_base_id: {sync_result.get('knowledge_base_id') or 'n/a'}", f"- project_files_included: {len(project_records)}", f"- wedrive_file_rows_included: {len(wedrive_rows)}", f"- packages: {len(docs)}", f"- target_packages: {len(target_docs)}", f"- upserted: {sync_result.get('upserted')}", f"- errors: {len(sync_result.get('errors', []))}", "", "## Project Skip Reasons", "\n".join(f"- {reason}: {count}" for reason, count in reasons.most_common()), "", "## Package Titles", "\n".join(f"- `{doc.tag}` {doc.title}" for doc in docs), ] ) for root in (latest, snapshot): (root / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") (root / "summary.md").write_text(summary.rstrip() + "\n", encoding="utf-8") return {"latest_manifest": str(latest / "manifest.json"), "latest_summary": str(latest / "summary.md"), "snapshot": str(snapshot)} def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--base-url", default=os.getenv("WEKNORA_BASE_URL", DEFAULT_BASE_URL)) parser.add_argument("--kb-name", default=os.getenv("WEKNORA_ZHCTPROMPT_KB", DEFAULT_KB_NAME)) parser.add_argument("--max-file-bytes", type=int, default=180_000) parser.add_argument("--max-package-chars", type=int, default=70_000) parser.add_argument("--dry-run", action="store_true") parser.add_argument("--resume-failed", action="store_true", help="only retry packages that failed in a previous manifest") parser.add_argument( "--resume-manifest", type=Path, default=None, help="manifest to read when --resume-failed is used", ) parser.add_argument("--batch-size", type=int, default=0, help="packages per health-checked batch; <=0 means one batch") parser.add_argument("--sleep-between-docs", type=float, default=0.0) parser.add_argument("--retry-attempts", type=int, default=2) parser.add_argument("--retry-sleep-seconds", type=float, default=3.0) parser.add_argument("--retry-backoff-multiplier", type=float, default=2.0) parser.add_argument("--health-url", default="", help="defaults to /health") parser.add_argument("--health-timeout", type=float, default=5.0) parser.add_argument("--health-attempts", type=int, default=3) parser.add_argument("--health-sleep-seconds", type=float, default=5.0) parser.add_argument("--health-backoff-multiplier", type=float, default=2.0) args = parser.parse_args() project_records, skipped, reasons, redacted_total = collect_project_files(args.max_file_bytes) wedrive_rows = read_wedrive_inventory_rows() project_docs = package_project_records(project_records, args.max_package_chars) wedrive_docs = package_wedrive_rows(wedrive_rows, args.max_package_chars) index_doc = build_index_doc( project_records, skipped, reasons, redacted_total, wedrive_rows, len(project_docs), len(wedrive_docs), ) docs = [index_doc, *project_docs, *wedrive_docs] resume_manifest = resolve_resume_manifest(args.resume_manifest) if args.resume_failed else None target_docs = filter_resume_failed_docs(docs, resume_manifest) if resume_manifest else docs sync_result: dict[str, Any] = { "applied": False, "knowledge_base_name": args.kb_name, "knowledge_base_id": "", "resume_failed": args.resume_failed, "resume_manifest": str(resume_manifest) if resume_manifest else "", "packages_total": len(docs), "target_packages": len(target_docs), "batch_size": args.batch_size, "sleep_between_docs": args.sleep_between_docs, "upserted": 0, "errors": [], "batches": [], "doc_results": [], } if not args.dry_run: client = WeKnoraClient(args.base_url) client.auto_setup() kb_id = ensure_knowledge_base(client, args.kb_name, DEFAULT_KB_DESCRIPTION) tag_ids = client.ensure_tags(kb_id, PROJECT_TAGS) sync_result["knowledge_base_id"] = kb_id sync_result["applied"] = True sync_result.update( sync_docs( client, kb_id, tag_ids, target_docs, batch_size=args.batch_size, sleep_between_docs=args.sleep_between_docs, retry_attempts=args.retry_attempts, retry_sleep_seconds=args.retry_sleep_seconds, retry_backoff_multiplier=args.retry_backoff_multiplier, health_url=args.health_url or derive_health_url(args.base_url), health_timeout=args.health_timeout, health_attempts=args.health_attempts, health_sleep_seconds=args.health_sleep_seconds, health_backoff_multiplier=args.health_backoff_multiplier, ) ) paths = write_outputs(docs, target_docs, project_records, skipped, reasons, wedrive_rows, sync_result) print( json.dumps( { "project_files_included": len(project_records), "project_files_skipped": len(skipped), "wedrive_file_rows_included": len(wedrive_rows), "packages": len(docs), "target_packages": len(target_docs), "sync_result": sync_result, "outputs": paths, }, ensure_ascii=False, indent=2, ) ) return 2 if sync_result["errors"] else 0 if __name__ == "__main__": raise SystemExit(main())