682 lines
22 KiB
Python
682 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
"""Shared helpers for the knowledge supply chain MVP."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import fnmatch
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Final, cast
|
|
|
|
try:
|
|
import jsonschema
|
|
except ImportError as exc:
|
|
raise SystemExit(
|
|
f"缺少依赖 jsonschema，请使用 uvx --with jsonschema --with pyyaml python <script>\n{exc}"
|
|
) from exc
|
|
|
|
try:
|
|
import yaml
|
|
except ImportError as exc:
|
|
raise SystemExit(
|
|
f"缺少依赖 PyYAML，请使用 uvx --with pyyaml --with jsonschema python <script>\n{exc}"
|
|
) from exc
|
|
|
|
|
|
JSONMapping = dict[str, object]
|
|
|
|
SOURCE_CONFIG_SCHEMA_RELATIVE: Final[Path] = Path(
|
|
"platform/contracts/knowledge-source-config.schema.yaml"
|
|
)
|
|
ARTIFACT_SCHEMA_RELATIVE: Final[Path] = Path(
|
|
"platform/contracts/knowledge-artifact.schema.yaml"
|
|
)
|
|
PUBLISH_RECORD_SCHEMA_RELATIVE: Final[Path] = Path(
|
|
"platform/contracts/knowledge-publish-record.schema.yaml"
|
|
)
|
|
DOCUMENT_SCHEMA_RELATIVE: Final[Path] = Path(
|
|
"platform/contracts/knowledge-document.schema.yaml"
|
|
)
|
|
CHUNK_SCHEMA_RELATIVE: Final[Path] = Path(
|
|
"platform/contracts/knowledge-chunk.schema.yaml"
|
|
)
|
|
SOURCE_REGISTRY_RELATIVE: Final[Path] = Path("platform/knowledge/source-registry.yaml")
|
|
ACL_POLICIES_RELATIVE: Final[Path] = Path("platform/knowledge/acl-policies.yaml")
|
|
PUBLISH_TARGETS_RELATIVE: Final[Path] = Path("platform/knowledge/publish-targets.yaml")
|
|
DEFAULT_MAX_CHUNK_CHARACTERS: Final[int] = 1200
|
|
|
|
|
|
class KnowledgeContractError(Exception):
|
|
"""Raised when a knowledge contract is invalid."""
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ChangeEntry:
|
|
"""Represents one git diff path change."""
|
|
|
|
change_type: str
|
|
path: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SourceRule:
|
|
"""Represents one source rule declared by a business repository."""
|
|
|
|
source_id: str
|
|
include: tuple[str, ...]
|
|
exclude: tuple[str, ...]
|
|
visibility: str
|
|
source_type: str
|
|
doc_type: str
|
|
product_area: str
|
|
owner: str
|
|
review_status: str
|
|
tags: tuple[str, ...]
|
|
title_strategy: str
|
|
title_prefix: str
|
|
approved_by: str | None
|
|
effective_from: str | None
|
|
expires_at: str | None
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class KnowledgePaths:
|
|
"""Filesystem layout inside ljwx-deploy for knowledge state."""
|
|
|
|
artifacts_root: Path
|
|
events_path: Path
|
|
publish_records_root: Path
|
|
|
|
|
|
def utc_now() -> str:
|
|
"""Return an ISO-8601 timestamp in UTC."""
|
|
|
|
return datetime.now(tz=UTC).isoformat()
|
|
|
|
|
|
def ensure_directory(path: Path) -> None:
|
|
"""Create a directory tree when it does not exist."""
|
|
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def ensure_parent(path: Path) -> None:
|
|
"""Create the parent directory for a file path."""
|
|
|
|
ensure_directory(path.parent)
|
|
|
|
|
|
def write_json(path: Path, payload: JSONMapping | list[JSONMapping]) -> None:
|
|
"""Write a JSON file with deterministic formatting."""
|
|
|
|
ensure_parent(path)
|
|
path.write_text(
|
|
json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def load_json_mapping(path: Path) -> JSONMapping:
|
|
"""Load a JSON object from disk."""
|
|
|
|
raw = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(raw, dict):
|
|
raise KnowledgeContractError(f"{path} 根节点必须是 object")
|
|
return {str(key): value for key, value in raw.items()}
|
|
|
|
|
|
def load_yaml_mapping(path: Path) -> JSONMapping:
|
|
"""Load a YAML mapping from disk."""
|
|
|
|
raw = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
if raw is None:
|
|
return {}
|
|
if not isinstance(raw, dict):
|
|
raise KnowledgeContractError(f"{path} 根节点必须是 mapping")
|
|
return {str(key): value for key, value in raw.items()}
|
|
|
|
|
|
def write_jsonl(path: Path, rows: list[JSONMapping]) -> None:
|
|
"""Write JSONL rows to disk."""
|
|
|
|
ensure_parent(path)
|
|
with path.open("w", encoding="utf-8") as handle:
|
|
for row in rows:
|
|
handle.write(json.dumps(row, ensure_ascii=False, sort_keys=True))
|
|
handle.write("\n")
|
|
|
|
|
|
def append_jsonl(path: Path, row: JSONMapping) -> None:
|
|
"""Append one JSONL row to disk."""
|
|
|
|
ensure_parent(path)
|
|
with path.open("a", encoding="utf-8") as handle:
|
|
handle.write(json.dumps(row, ensure_ascii=False, sort_keys=True))
|
|
handle.write("\n")
|
|
|
|
|
|
def load_jsonl(path: Path) -> list[JSONMapping]:
|
|
"""Load JSONL records from disk when present."""
|
|
|
|
if not path.exists():
|
|
return []
|
|
rows: list[JSONMapping] = []
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
stripped = line.strip()
|
|
if not stripped:
|
|
continue
|
|
raw = json.loads(stripped)
|
|
if not isinstance(raw, dict):
|
|
raise KnowledgeContractError(f"{path} 存在非 mapping JSONL 记录")
|
|
rows.append({str(key): value for key, value in raw.items()})
|
|
return rows
|
|
|
|
|
|
def load_validator(path: Path) -> jsonschema.Draft202012Validator:
|
|
"""Load a JSON schema validator from YAML."""
|
|
|
|
schema = load_yaml_mapping(path)
|
|
jsonschema.Draft202012Validator.check_schema(schema)
|
|
return jsonschema.Draft202012Validator(schema)
|
|
|
|
|
|
def validate_payload(
|
|
validator: jsonschema.Draft202012Validator,
|
|
payload: JSONMapping,
|
|
payload_name: str,
|
|
) -> None:
|
|
"""Validate one payload against a schema."""
|
|
|
|
try:
|
|
validator.validate(payload)
|
|
except jsonschema.ValidationError as exc:
|
|
raise KnowledgeContractError(
|
|
f"{payload_name} 不符合 schema: {exc.message}"
|
|
) from exc
|
|
|
|
|
|
def require_mapping(value: object, field_path: str) -> JSONMapping:
|
|
"""Require a JSON mapping."""
|
|
|
|
if not isinstance(value, dict):
|
|
raise KnowledgeContractError(f"{field_path} 必须是 mapping")
|
|
return {str(key): item for key, item in value.items()}
|
|
|
|
|
|
def require_list(value: object, field_path: str) -> list[object]:
|
|
"""Require a list."""
|
|
|
|
if not isinstance(value, list):
|
|
raise KnowledgeContractError(f"{field_path} 必须是列表")
|
|
return value
|
|
|
|
|
|
def require_non_empty_str(value: object, field_path: str) -> str:
|
|
"""Require a non-empty string."""
|
|
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise KnowledgeContractError(f"{field_path} 必须是非空字符串")
|
|
return value.strip()
|
|
|
|
|
|
def optional_non_empty_str(value: object, field_path: str) -> str | None:
|
|
"""Read an optional non-empty string."""
|
|
|
|
if value in ("", None):
|
|
return None
|
|
return require_non_empty_str(value, field_path)
|
|
|
|
|
|
def require_str_list(value: object, field_path: str) -> list[str]:
|
|
"""Require a list of strings."""
|
|
|
|
items = require_list(value, field_path)
|
|
results: list[str] = []
|
|
for index, item in enumerate(items):
|
|
results.append(require_non_empty_str(item, f"{field_path}[{index}]"))
|
|
return results
|
|
|
|
|
|
def sha256_text(text: str) -> str:
|
|
"""Return the SHA-256 digest of a text payload."""
|
|
|
|
return hashlib.sha256(text.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def slugify_fragment(value: str) -> str:
|
|
"""Convert a string into a filesystem-safe slug."""
|
|
|
|
normalized = re.sub(r"[^a-zA-Z0-9]+", "-", value).strip("-").lower()
|
|
return normalized or "unknown"
|
|
|
|
|
|
def repo_slug(repo: str) -> str:
|
|
"""Return a stable directory slug for an owner/repo name."""
|
|
|
|
return repo.replace("/", "__")
|
|
|
|
|
|
def build_document_id(source_repo: str, source_path: str) -> str:
|
|
"""Build a deterministic document identifier from repo and path."""
|
|
|
|
repo_name = source_repo.split("/")[-1]
|
|
base_slug = slugify_fragment(f"{repo_name}-{source_path}")
|
|
if len(base_slug) <= 96:
|
|
return base_slug
|
|
digest = hashlib.sha1(f"{source_repo}:{source_path}".encode("utf-8")).hexdigest()[
|
|
:12
|
|
]
|
|
return f"{base_slug[:80]}-{digest}"
|
|
|
|
|
|
def match_globs(
|
|
path: str, includes: tuple[str, ...], excludes: tuple[str, ...]
|
|
) -> bool:
|
|
"""Return True when a path matches include/exclude rules."""
|
|
|
|
included = any(fnmatch.fnmatch(path, pattern) for pattern in includes)
|
|
if not included:
|
|
return False
|
|
return not any(fnmatch.fnmatch(path, pattern) for pattern in excludes)
|
|
|
|
|
|
def parse_diff_entries(path: Path) -> list[ChangeEntry]:
|
|
"""Parse git diff --name-status output into flattened path changes."""
|
|
|
|
if not path.exists():
|
|
return []
|
|
|
|
results: list[ChangeEntry] = []
|
|
for raw_line in path.read_text(encoding="utf-8").splitlines():
|
|
line = raw_line.strip()
|
|
if not line:
|
|
continue
|
|
parts = line.split("\t")
|
|
status = require_non_empty_str(parts[0], "diff.status")
|
|
base_status = status[0]
|
|
if base_status in {"A", "C", "M", "T"}:
|
|
if len(parts) < 2:
|
|
raise KnowledgeContractError(f"无效 diff 行: {line}")
|
|
results.append(ChangeEntry(change_type="upsert", path=parts[-1]))
|
|
continue
|
|
if base_status == "D":
|
|
if len(parts) != 2:
|
|
raise KnowledgeContractError(f"无效删除 diff 行: {line}")
|
|
results.append(ChangeEntry(change_type="delete", path=parts[1]))
|
|
continue
|
|
if base_status == "R":
|
|
if len(parts) != 3:
|
|
raise KnowledgeContractError(f"无效重命名 diff 行: {line}")
|
|
results.append(ChangeEntry(change_type="delete", path=parts[1]))
|
|
results.append(ChangeEntry(change_type="upsert", path=parts[2]))
|
|
continue
|
|
raise KnowledgeContractError(f"暂不支持的 diff 状态: {status}")
|
|
return results
|
|
|
|
|
|
def extract_markdown_title(content: str, source_path: str, strategy: str) -> str:
|
|
"""Resolve a document title from markdown content."""
|
|
|
|
if strategy == "filename":
|
|
return Path(source_path).stem.replace("-", " ").replace("_", " ").strip()
|
|
|
|
for line in content.splitlines():
|
|
stripped = line.strip()
|
|
if stripped.startswith("# "):
|
|
return stripped.removeprefix("# ").strip()
|
|
|
|
return Path(source_path).stem.replace("-", " ").replace("_", " ").strip()
|
|
|
|
|
|
def chunk_markdown(
|
|
content: str, max_characters: int = DEFAULT_MAX_CHUNK_CHARACTERS
|
|
) -> list[str]:
|
|
"""Chunk markdown into paragraph groups."""
|
|
|
|
paragraphs = [item.strip() for item in content.split("\n\n") if item.strip()]
|
|
if not paragraphs:
|
|
return []
|
|
|
|
chunks: list[str] = []
|
|
current_parts: list[str] = []
|
|
current_length = 0
|
|
for paragraph in paragraphs:
|
|
paragraph_length = len(paragraph)
|
|
if current_parts and current_length + paragraph_length > max_characters:
|
|
chunks.append("\n\n".join(current_parts))
|
|
current_parts = [paragraph]
|
|
current_length = paragraph_length
|
|
continue
|
|
current_parts.append(paragraph)
|
|
current_length += paragraph_length
|
|
|
|
if current_parts:
|
|
chunks.append("\n\n".join(current_parts))
|
|
|
|
return chunks
|
|
|
|
|
|
def knowledge_paths(deploy_root: Path, environment: str) -> KnowledgePaths:
|
|
"""Resolve the filesystem layout used by central knowledge state."""
|
|
|
|
base_root = deploy_root / "knowledge"
|
|
return KnowledgePaths(
|
|
artifacts_root=base_root / "artifacts" / environment,
|
|
events_path=base_root / "events" / environment / "events.jsonl",
|
|
publish_records_root=base_root / "publish-records" / environment,
|
|
)
|
|
|
|
|
|
def load_source_rules(
|
|
config_path: Path, deploy_root: Path
|
|
) -> tuple[str, list[SourceRule]]:
|
|
"""Load and validate source config rules."""
|
|
|
|
validator = load_validator(deploy_root / SOURCE_CONFIG_SCHEMA_RELATIVE)
|
|
payload = load_yaml_mapping(config_path)
|
|
validate_payload(validator, payload, config_path.as_posix())
|
|
|
|
service = require_non_empty_str(
|
|
payload.get("service"), "knowledge.sources.yaml.service"
|
|
)
|
|
raw_sources = require_list(payload.get("sources"), "knowledge.sources.yaml.sources")
|
|
|
|
rules: list[SourceRule] = []
|
|
for index, raw_source in enumerate(raw_sources):
|
|
source = require_mapping(raw_source, f"knowledge.sources.yaml.sources[{index}]")
|
|
rules.append(
|
|
SourceRule(
|
|
source_id=require_non_empty_str(
|
|
source.get("id"), f"knowledge.sources.yaml.sources[{index}].id"
|
|
),
|
|
include=tuple(
|
|
require_str_list(
|
|
source.get("include"),
|
|
f"knowledge.sources.yaml.sources[{index}].include",
|
|
)
|
|
),
|
|
exclude=tuple(
|
|
require_str_list(
|
|
source.get("exclude", []),
|
|
f"knowledge.sources.yaml.sources[{index}].exclude",
|
|
)
|
|
),
|
|
visibility=require_non_empty_str(
|
|
source.get("visibility"),
|
|
f"knowledge.sources.yaml.sources[{index}].visibility",
|
|
),
|
|
source_type=require_non_empty_str(
|
|
source.get("source_type"),
|
|
f"knowledge.sources.yaml.sources[{index}].source_type",
|
|
),
|
|
doc_type=require_non_empty_str(
|
|
source.get("doc_type"),
|
|
f"knowledge.sources.yaml.sources[{index}].doc_type",
|
|
),
|
|
product_area=require_non_empty_str(
|
|
source.get("product_area"),
|
|
f"knowledge.sources.yaml.sources[{index}].product_area",
|
|
),
|
|
owner=require_non_empty_str(
|
|
source.get("owner"),
|
|
f"knowledge.sources.yaml.sources[{index}].owner",
|
|
),
|
|
review_status=require_non_empty_str(
|
|
source.get("review_status"),
|
|
f"knowledge.sources.yaml.sources[{index}].review_status",
|
|
),
|
|
tags=tuple(
|
|
require_str_list(
|
|
source.get("tags", []),
|
|
f"knowledge.sources.yaml.sources[{index}].tags",
|
|
)
|
|
),
|
|
title_strategy=optional_non_empty_str(
|
|
source.get("title_strategy"),
|
|
f"knowledge.sources.yaml.sources[{index}].title_strategy",
|
|
)
|
|
or "first_heading",
|
|
title_prefix=optional_non_empty_str(
|
|
source.get("title_prefix"),
|
|
f"knowledge.sources.yaml.sources[{index}].title_prefix",
|
|
)
|
|
or "",
|
|
approved_by=optional_non_empty_str(
|
|
source.get("approved_by"),
|
|
f"knowledge.sources.yaml.sources[{index}].approved_by",
|
|
),
|
|
effective_from=optional_non_empty_str(
|
|
source.get("effective_from"),
|
|
f"knowledge.sources.yaml.sources[{index}].effective_from",
|
|
),
|
|
expires_at=optional_non_empty_str(
|
|
source.get("expires_at"),
|
|
f"knowledge.sources.yaml.sources[{index}].expires_at",
|
|
),
|
|
)
|
|
)
|
|
return service, rules
|
|
|
|
|
|
def load_source_registry(deploy_root: Path) -> dict[str, dict[str, object]]:
|
|
"""Load artifact source registry entries keyed by repository."""
|
|
|
|
payload = load_yaml_mapping(deploy_root / SOURCE_REGISTRY_RELATIVE)
|
|
raw_registry = require_list(
|
|
payload.get("artifact_sources", []),
|
|
"platform/knowledge/source-registry.yaml.artifact_sources",
|
|
)
|
|
registry: dict[str, dict[str, object]] = {}
|
|
for index, raw_entry in enumerate(raw_registry):
|
|
entry = require_mapping(
|
|
raw_entry,
|
|
f"platform/knowledge/source-registry.yaml.artifact_sources[{index}]",
|
|
)
|
|
repo_name = require_non_empty_str(
|
|
entry.get("repo"),
|
|
f"platform/knowledge/source-registry.yaml.artifact_sources[{index}].repo",
|
|
)
|
|
registry[repo_name] = {
|
|
"service": require_non_empty_str(
|
|
entry.get("service"),
|
|
"platform/knowledge/source-registry.yaml.artifact_sources."
|
|
f"[{index}].service",
|
|
),
|
|
"source_ids": require_str_list(
|
|
entry.get("source_ids", []),
|
|
"platform/knowledge/source-registry.yaml.artifact_sources."
|
|
f"[{index}].source_ids",
|
|
),
|
|
"allowed_visibilities": require_str_list(
|
|
entry.get("allowed_visibilities", []),
|
|
"platform/knowledge/source-registry.yaml.artifact_sources."
|
|
f"[{index}].allowed_visibilities",
|
|
),
|
|
}
|
|
return registry
|
|
|
|
|
|
def load_acl_policies(deploy_root: Path) -> dict[str, dict[str, object]]:
|
|
"""Load ACL policies keyed by visibility."""
|
|
|
|
payload = load_yaml_mapping(deploy_root / ACL_POLICIES_RELATIVE)
|
|
raw_policies = require_mapping(
|
|
payload.get("policies"),
|
|
"platform/knowledge/acl-policies.yaml.policies",
|
|
)
|
|
policies: dict[str, dict[str, object]] = {}
|
|
for visibility, raw_policy in raw_policies.items():
|
|
policy = require_mapping(
|
|
raw_policy,
|
|
f"platform/knowledge/acl-policies.yaml.policies.{visibility}",
|
|
)
|
|
policies[visibility] = {
|
|
"acl_expression_template": require_non_empty_str(
|
|
policy.get("acl_expression_template"),
|
|
"platform/knowledge/acl-policies.yaml.policies."
|
|
f"{visibility}.acl_expression_template",
|
|
),
|
|
"allowed_targets": require_str_list(
|
|
policy.get("allowed_targets", []),
|
|
"platform/knowledge/acl-policies.yaml.policies."
|
|
f"{visibility}.allowed_targets",
|
|
),
|
|
"review_required": bool(policy.get("review_required", False)),
|
|
}
|
|
return policies
|
|
|
|
|
|
def load_publish_targets(deploy_root: Path) -> dict[str, dict[str, object]]:
|
|
"""Load publish targets keyed by target id."""
|
|
|
|
payload = load_yaml_mapping(deploy_root / PUBLISH_TARGETS_RELATIVE)
|
|
raw_targets = require_mapping(
|
|
payload.get("targets"),
|
|
"platform/knowledge/publish-targets.yaml.targets",
|
|
)
|
|
targets: dict[str, dict[str, object]] = {}
|
|
for target_id, raw_target in raw_targets.items():
|
|
target = require_mapping(
|
|
raw_target, f"platform/knowledge/publish-targets.yaml.targets.{target_id}"
|
|
)
|
|
targets[target_id] = {
|
|
"type": require_non_empty_str(
|
|
target.get("type"),
|
|
f"platform/knowledge/publish-targets.yaml.targets.{target_id}.type",
|
|
),
|
|
"visibility": require_non_empty_str(
|
|
target.get("visibility"),
|
|
"platform/knowledge/publish-targets.yaml.targets."
|
|
f"{target_id}.visibility",
|
|
),
|
|
"dataset_name": optional_non_empty_str(
|
|
target.get("dataset_name"),
|
|
"platform/knowledge/publish-targets.yaml.targets."
|
|
f"{target_id}.dataset_name",
|
|
),
|
|
}
|
|
return targets
|
|
|
|
|
|
def load_publish_record(path: Path) -> JSONMapping:
|
|
"""Load one publish record file."""
|
|
|
|
return load_json_mapping(path)
|
|
|
|
|
|
def load_publish_records(root: Path) -> dict[str, JSONMapping]:
|
|
"""Load all publish records under a directory."""
|
|
|
|
if not root.exists():
|
|
return {}
|
|
records: dict[str, JSONMapping] = {}
|
|
for path in sorted(root.rglob("*.json")):
|
|
record = load_publish_record(path)
|
|
document_id = require_non_empty_str(
|
|
record.get("document_id"), f"{path}.document_id"
|
|
)
|
|
records[document_id] = record
|
|
return records
|
|
|
|
|
|
def build_acl_expression(
|
|
visibility: str, policies: dict[str, dict[str, object]]
|
|
) -> str:
|
|
"""Resolve the ACL expression for a visibility."""
|
|
|
|
policy = policies.get(visibility)
|
|
if policy is None:
|
|
raise KnowledgeContractError(f"未注册 visibility ACL policy: {visibility}")
|
|
acl_template = policy.get("acl_expression_template")
|
|
return require_non_empty_str(acl_template, f"acl policy for {visibility}")
|
|
|
|
|
|
def build_event(
|
|
*,
|
|
action: str,
|
|
artifact_id: str,
|
|
document_id: str,
|
|
source_repo: str,
|
|
status: str,
|
|
reason: str,
|
|
target_ids: list[str],
|
|
) -> JSONMapping:
|
|
"""Build one append-only event record."""
|
|
|
|
compact_time = utc_now().replace(":", "").replace("-", "")
|
|
return {
|
|
"event_id": f"knowledge-{action}-{document_id}-{compact_time}",
|
|
"timestamp": utc_now(),
|
|
"artifact_id": artifact_id,
|
|
"document_id": document_id,
|
|
"source_repo": source_repo,
|
|
"action": action,
|
|
"status": status,
|
|
"reason": reason,
|
|
"target_ids": target_ids,
|
|
}
|
|
|
|
|
|
def artifact_document_name(document_id: str) -> str:
|
|
"""Return the stable remote document name used in Dify."""
|
|
|
|
return document_id
|
|
|
|
|
|
def artifact_snapshot_root(
|
|
deploy_root: Path,
|
|
environment: str,
|
|
source_repo: str,
|
|
artifact_id: str,
|
|
) -> Path:
|
|
"""Return the snapshot root for one artifact."""
|
|
|
|
return (
|
|
knowledge_paths(deploy_root, environment).artifacts_root
|
|
/ repo_slug(source_repo)
|
|
/ artifact_id
|
|
)
|
|
|
|
|
|
def record_path_for_document(
|
|
deploy_root: Path,
|
|
environment: str,
|
|
source_repo: str,
|
|
document_id: str,
|
|
) -> Path:
|
|
"""Return the publish-record path for a document."""
|
|
|
|
return (
|
|
knowledge_paths(deploy_root, environment).publish_records_root
|
|
/ repo_slug(source_repo)
|
|
/ f"{document_id}.json"
|
|
)
|
|
|
|
|
|
def cast_mapping_list(value: object, field_path: str) -> list[JSONMapping]:
|
|
"""Cast a list payload into a list of mappings."""
|
|
|
|
items = require_list(value, field_path)
|
|
mappings: list[JSONMapping] = []
|
|
for index, item in enumerate(items):
|
|
mappings.append(require_mapping(item, f"{field_path}[{index}]"))
|
|
return mappings
|
|
|
|
|
|
def optional_mapping(value: object, field_path: str) -> JSONMapping | None:
|
|
"""Return an optional mapping."""
|
|
|
|
if value is None:
|
|
return None
|
|
return require_mapping(value, field_path)
|
|
|
|
|
|
def cast_str(value: object) -> str:
|
|
"""Cast a value known to be a string."""
|
|
|
|
return cast(str, value)
|