nac
777607dc29
Track the source-of-truth code, skills, bootstrap rules, docs, and packaging files before adding new features. Exclude local vault config, generated skill mirrors, temporary exports, virtual environments, and business data files from version control. Validation: inspected staged file list and git ignored status before commit.
501 lines
20 KiB
Python
501 lines
20 KiB
Python
"""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"
|