"""Bot-view access control helpers for an Obsidian wiki vault. The vault is the mother knowledge base. Bot views are derived, filtered exports that can be indexed separately by downstream chatbots. This module keeps the boundary simple and inspectable: explicit frontmatter wins, missing fields are conservatively inferred from path/tags/title, and each export writes a manifest explaining what was included or blocked. """ from __future__ import annotations import json import re import shutil from dataclasses import dataclass from pathlib import Path from typing import Any from obsidian_wiki.file_assets import clean_config_value, registry_path_for_vault, utc_now BOT_PROFILES_RELATIVE_PATH = Path("_meta") / "bot-profiles.json" LIST_ITEM_RE = re.compile(r"^\s*-\s+(.+?)\s*$") INLINE_LIST_RE = re.compile(r"^\[(.*)\]$") WIKILINK_RE = re.compile(r"\[\[([^|\]#]+)(?:[|#][^\]]*)?\]\]") @dataclass(frozen=True) class PageDecision: path: Path rel_path: str include: bool reason: str visibility: str data_classification: str business_domains: list[str] allowed_bots: list[str] sources: list[str] @dataclass(frozen=True) class SourceDecision: archive_path: str include: bool reason: str doc_type: str visibility: str data_classification: str business_categories: list[str] def profiles_path_for_vault(vault_path: str | Path) -> Path: vault = Path(clean_config_value(str(vault_path))).expanduser() return vault / BOT_PROFILES_RELATIVE_PATH def load_profiles(vault_path: str | Path) -> dict[str, Any]: path = profiles_path_for_vault(vault_path) if not path.is_file(): raise FileNotFoundError(f"bot profiles not found: {path}") raw = json.loads(path.read_text(encoding="utf-8")) return raw.get("profiles", raw) def list_profiles(vault_path: str | Path) -> list[str]: return sorted(load_profiles(vault_path)) def iter_markdown_pages(vault_path: str | Path, *, include_meta: bool = False) -> list[Path]: vault = Path(clean_config_value(str(vault_path))).expanduser() out: list[Path] = [] for path in vault.rglob("*.md"): rel_parts = path.relative_to(vault).parts if not rel_parts: continue if rel_parts[0] in {".obsidian", "_archives"}: continue if rel_parts[:2] == ("_sources", "archive"): continue if not include_meta and rel_parts[0] in {"_meta", "_sources"}: continue if rel_parts[0] in {"index.md", "log.md", "hot.md", "AGENTS.md", "_insights.md", "欢迎.md"}: continue out.append(path) return sorted(out) def parse_frontmatter(text: str) -> tuple[dict[str, Any], str]: if not text.startswith("---\n"): return {}, text parts = text.split("---", 2) if len(parts) < 3: return {}, text raw = parts[1].strip("\n") body = parts[2].lstrip("\n") data: dict[str, Any] = {} lines = raw.splitlines() i = 0 while i < len(lines): line = lines[i] if not line.strip() or line.lstrip().startswith("#") or ":" not in line: i += 1 continue key, value = line.split(":", 1) key = key.strip() value = value.strip() if value == "": items: list[str] = [] j = i + 1 while j < len(lines) and (lines[j].startswith(" ") or lines[j].startswith("\t")): match = LIST_ITEM_RE.match(lines[j]) if match: items.append(_strip_scalar(match.group(1))) j += 1 data[key] = items if items else "" i = j continue data[key] = _parse_scalar(value) i += 1 return data, body def _parse_scalar(value: str) -> Any: match = INLINE_LIST_RE.match(value) if match: inner = match.group(1).strip() if not inner: return [] return [_strip_scalar(part.strip()) for part in inner.split(",")] return _strip_scalar(value) def _strip_scalar(value: str) -> str: value = value.strip() if len(value) >= 2 and (value[0], value[-1]) in {('"', '"'), ("'", "'")}: return value[1:-1] return value def as_list(value: Any) -> list[str]: if value is None or value == "": return [] if isinstance(value, list): return [str(item).strip() for item in value if str(item).strip()] return [str(value).strip()] def page_decision(vault_path: str | Path, path: Path, profile_name: str, profile: dict[str, Any]) -> PageDecision: vault = Path(clean_config_value(str(vault_path))).expanduser() text = path.read_text(encoding="utf-8", errors="ignore") frontmatter, body = parse_frontmatter(text) rel_path = str(path.relative_to(vault)) visibility = infer_visibility(frontmatter) data_classification = infer_data_classification(rel_path, frontmatter) business_domains = infer_business_domains(rel_path, frontmatter) allowed_bots = as_list(frontmatter.get("allowed_bots")) excluded_bots = as_list(frontmatter.get("excluded_bots")) sources = as_list(frontmatter.get("sources")) include, reason = _allowed_common( profile_name=profile_name, profile=profile, visibility=visibility, data_classification=data_classification, business_domains=business_domains, allowed_bots=allowed_bots, excluded_bots=excluded_bots, ) if include and profile.get("require_explicit_allow") and profile_name not in allowed_bots and "*" not in allowed_bots: include, reason = False, "profile requires explicit allowed_bots" if include and profile.get("page_path_allow_patterns"): if not _matches_any(rel_path, profile["page_path_allow_patterns"]): include, reason = False, "page path not in allow patterns" if include and profile.get("page_path_block_patterns"): if _matches_any(rel_path, profile["page_path_block_patterns"]): include, reason = False, "page path blocked" # Prevent allowed exports from containing wikilinks to hidden pages as live # links. The text stays useful, but hidden targets are plain text after export. if include and profile.get("deny_unresolved_links", True): pass return PageDecision( path=path, rel_path=rel_path, include=include, reason=reason, visibility=visibility, data_classification=data_classification, business_domains=business_domains, allowed_bots=allowed_bots, sources=sources, ) def infer_visibility(frontmatter: dict[str, Any]) -> str: explicit = str(frontmatter.get("visibility") or "").strip() if explicit: return explicit.removeprefix("visibility/") tags = as_list(frontmatter.get("tags")) for tag in tags: if tag.startswith("visibility/"): return tag.split("/", 1)[1] return "public" def infer_data_classification(rel_path: str, frontmatter: dict[str, Any]) -> str: explicit = str(frontmatter.get("data_classification") or "").strip() if explicit: return explicit blob = f"{rel_path} {frontmatter.get('title','')} {' '.join(as_list(frontmatter.get('tags')))}".lower() visibility = infer_visibility(frontmatter) if visibility == "pii": return "personal-sensitive" if any(key in blob for key in ("order", "订单", "customer-market", "sales-analysis", "客户行业")): return "commercial-confidential" if any(key in blob for key in ("complaint", "客诉", "quality", "索赔", "8d", "不合格")): return "quality-confidential" if any(key in blob for key in ("cross-reference", "product-parameter", "图纸", "drawing", "3d", "工程")): return "technical-confidential" if any(key in blob for key in ("website", "官网", "molex-mini50", "double-85", "standards")): return "public" if visibility == "public" else "internal" return "internal" if visibility == "internal" else "public" def infer_business_domains(rel_path: str, frontmatter: dict[str, Any]) -> list[str]: explicit = as_list(frontmatter.get("business_domains")) if explicit: return explicit blob = f"{rel_path} {frontmatter.get('title','')} {' '.join(as_list(frontmatter.get('tags')))}".lower() domains: set[str] = set() if any(key in blob for key in ("order", "订单", "sales-analysis", "销售")): domains.update({"orders", "sales", "customers"}) if any(key in blob for key in ("customer-market", "客户", "market-segmentation", "市场")): domains.update({"customers", "market"}) if any(key in blob for key in ("complaint", "客诉", "quality", "索赔", "8d", "不合格")): domains.update({"quality", "customers"}) if any(key in blob for key in ("product", "产品", "fpc", "btb", "wtb", "io", "fakra", "connector", "连接器")): domains.update({"products", "sales", "engineering"}) if any(key in blob for key in ("drawing", "图纸", "3d", "cad", "standard", "标准", "materials", "材料", "plating")): domains.update({"engineering"}) if any(key in blob for key in ("company", "公司", "group-company", "集团")): domains.update({"company"}) if any(key in blob for key in ("website", "官网", "公开")): domains.update({"public-product", "company"}) if not domains: domains.add("general") return sorted(domains) def _allowed_common( *, profile_name: str, profile: dict[str, Any], visibility: str, data_classification: str, business_domains: list[str], allowed_bots: list[str], excluded_bots: list[str], ) -> tuple[bool, str]: if profile_name in excluded_bots or "*" in excluded_bots: return False, "explicitly excluded" if allowed_bots and profile_name not in allowed_bots and "*" not in allowed_bots: return False, "not in explicit allowed_bots" if visibility not in set(profile.get("allowed_visibility", ["public"])): return False, f"visibility {visibility} not allowed" if data_classification not in set(profile.get("allowed_data_classification", ["public"])): return False, f"classification {data_classification} not allowed" blocked_domains = set(profile.get("blocked_business_domains", [])) overlap = blocked_domains & set(business_domains) if overlap: return False, f"blocked domain(s): {', '.join(sorted(overlap))}" allowed_domains = set(profile.get("allowed_business_domains", [])) if allowed_domains and not (allowed_domains & set(business_domains)): return False, "no allowed business domain" return True, "allowed" def _matches_any(text: str, patterns: list[str]) -> bool: return any(re.search(pattern, text) for pattern in patterns) def load_source_registry(vault_path: str | Path) -> list[dict[str, Any]]: registry = registry_path_for_vault(vault_path) if not registry.is_file(): return [] rows: list[dict[str, Any]] = [] for raw in registry.read_text(encoding="utf-8", errors="ignore").splitlines(): if not raw.strip(): continue try: record = json.loads(raw) except json.JSONDecodeError: continue if record.get("record_type") == "source_archive": rows.append(record) return rows def source_decision(profile_name: str, profile: dict[str, Any], record: dict[str, Any]) -> SourceDecision: archive_path = str(record.get("archive_path") or "") visibility = str(record.get("visibility") or "internal").removeprefix("visibility/") doc_type = str(record.get("doc_type") or "other") data_classification = str(record.get("data_classification") or infer_source_data_classification(record)) business_categories = as_list(record.get("business_categories")) allowed_bots = as_list(record.get("allowed_bots")) excluded_bots = as_list(record.get("excluded_bots")) domains = infer_source_domains(record) include, reason = _allowed_common( profile_name=profile_name, profile=profile, visibility=visibility, data_classification=data_classification, business_domains=domains, allowed_bots=allowed_bots, excluded_bots=excluded_bots, ) if include and doc_type in set(profile.get("blocked_doc_types", [])): include, reason = False, f"doc_type {doc_type} blocked" allowed_doc_types = set(profile.get("allowed_doc_types", [])) if include and allowed_doc_types and doc_type not in allowed_doc_types: include, reason = False, f"doc_type {doc_type} not allowed" if include and profile.get("include_source_files") is False: include, reason = False, "profile does not include source files" return SourceDecision( archive_path=archive_path, include=include, reason=reason, doc_type=doc_type, visibility=visibility, data_classification=data_classification, business_categories=business_categories, ) def infer_source_data_classification(record: dict[str, Any]) -> str: blob = " ".join( [ str(record.get("doc_type") or ""), str(record.get("title") or ""), str(record.get("summary") or ""), " ".join(as_list(record.get("business_categories"))), ] ).lower() if any(key in blob for key in ("order", "订单", "raw-excel", "price", "价格")): return "commercial-confidential" if any(key in blob for key in ("complaint", "客诉", "claim", "索赔", "8d", "不合格")): return "quality-confidential" if any(key in blob for key in ("drawing", "3d", "cad", "step", "规格书", "图纸")): return "technical-confidential" if any(key in blob for key in ("website", "company_profile", "standard")) and record.get("visibility") == "public": return "public" return "internal" def infer_source_domains(record: dict[str, Any]) -> list[str]: cats = " ".join(as_list(record.get("business_categories"))).lower() doc_type = str(record.get("doc_type") or "").lower() domains: set[str] = set() if "order" in cats or doc_type == "order_data": domains.update({"orders", "sales", "customers"}) if "customer" in cats: domains.add("customers") if "quality" in cats or "complaint" in doc_type or "claim" in doc_type: domains.update({"quality", "customers"}) if "product" in cats or doc_type in {"product_parameter", "cross_reference", "test_report"}: domains.update({"products", "sales", "engineering"}) if "drawing" in cats or doc_type == "drawing": domains.add("engineering") if "company" in cats or doc_type == "company_profile": domains.add("company") if "website" in cats or doc_type == "website_snapshot": domains.update({"public-product", "company"}) return sorted(domains or {"general"}) def audit_profile(vault_path: str | Path, profile_name: str) -> dict[str, Any]: profiles = load_profiles(vault_path) if profile_name not in profiles: raise KeyError(f"unknown bot profile: {profile_name}") profile = profiles[profile_name] page_decisions = [page_decision(vault_path, path, profile_name, profile) for path in iter_markdown_pages(vault_path)] source_decisions = [source_decision(profile_name, profile, record) for record in load_source_registry(vault_path)] return { "profile": profile_name, "description": profile.get("description", ""), "pages_allowed": sum(1 for item in page_decisions if item.include), "pages_blocked": sum(1 for item in page_decisions if not item.include), "sources_allowed": sum(1 for item in source_decisions if item.include), "sources_blocked": sum(1 for item in source_decisions if not item.include), "page_reasons": _reason_counts(page_decisions), "source_reasons": _reason_counts(source_decisions), "allowed_pages": [item.rel_path for item in page_decisions if item.include], } def _reason_counts(decisions: list[Any]) -> dict[str, int]: out: dict[str, int] = {} for item in decisions: out[item.reason] = out.get(item.reason, 0) + 1 return dict(sorted(out.items(), key=lambda pair: (-pair[1], pair[0]))) def export_profile( vault_path: str | Path, profile_name: str, out_dir: str | Path, *, include_files: bool = False, ) -> dict[str, Any]: vault = Path(clean_config_value(str(vault_path))).expanduser() profiles = load_profiles(vault) if profile_name not in profiles: raise KeyError(f"unknown bot profile: {profile_name}") profile = profiles[profile_name] dest = Path(out_dir).expanduser() / profile_name wiki_dest = dest / "wiki" if dest.exists(): shutil.rmtree(dest) wiki_dest.mkdir(parents=True, exist_ok=True) page_decisions = [page_decision(vault, path, profile_name, profile) for path in iter_markdown_pages(vault)] allowed_rel_paths = {item.rel_path for item in page_decisions if item.include} copied_pages: list[str] = [] page_sources: set[str] = set() for decision in page_decisions: if not decision.include: continue target = wiki_dest / decision.rel_path target.parent.mkdir(parents=True, exist_ok=True) text = decision.path.read_text(encoding="utf-8", errors="ignore") text = sanitize_hidden_wikilinks(text, allowed_rel_paths) target.write_text(text, encoding="utf-8") copied_pages.append(decision.rel_path) page_sources.update(src for src in decision.sources if src.startswith("_sources/archive/")) source_records = {str(record.get("archive_path") or ""): record for record in load_source_registry(vault)} copied_sources: list[str] = [] if include_files: for source, record in sorted(source_records.items()): decision = source_decision(profile_name, {**profile, "include_source_files": True}, record) if not decision.include: continue src_path = vault / source if not src_path.is_file(): continue target = dest / "sources" / source target.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(src_path, target) copied_sources.append(source) manifest = { "profile": profile_name, "description": profile.get("description", ""), "generated_at": utc_now(), "source_vault": str(vault), "pages": copied_pages, "page_count": len(copied_pages), "source_files": copied_sources, "source_file_count": len(copied_sources), "blocked_page_reasons": _reason_counts([item for item in page_decisions if not item.include]), } meta_dir = dest / "_bot-view" meta_dir.mkdir(parents=True, exist_ok=True) (meta_dir / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8") (meta_dir / "index.md").write_text(render_export_index(profile_name, copied_pages), encoding="utf-8") return manifest def sanitize_hidden_wikilinks(text: str, allowed_rel_paths: set[str]) -> str: allowed_stems = {Path(path).stem for path in allowed_rel_paths} def repl(match: re.Match[str]) -> str: target = match.group(1).strip() stem = Path(target).stem if target in allowed_rel_paths or stem in allowed_stems: return match.group(0) return stem return WIKILINK_RE.sub(repl, text) def render_export_index(profile_name: str, pages: list[str]) -> str: lines = [ "---", f"title: Bot View Export - {profile_name}", "category: meta", "tags: [bot-view, export]", f"updated: {utc_now()}", "---", "", f"# Bot View Export - {profile_name}", "", "| Page |", "|---|", ] for page in pages: lines.append(f"| `{page}` |") return "\n".join(lines) + "\n"