Utils Layers¶

def extract_date(f: File) -> datetime | None:
	"""Extract date from docs file.

	Args:
	    f: The file to extract date from.

	Returns:
	    Optional[datetime]: The date if successful, None otherwise.
	"""
	try:
		return f.note_date
	except AttributeError:
		return None

def extract_tags(meta_dict: dict[str, Any]) -> list[str]:
	"""Normalize frontmatter ``tags`` / ``tag`` into a list of non-empty strings.

	Args:
	    meta_dict: Parsed frontmatter mapping.

	Returns:
	    List of tag strings.
	"""
	raw = meta_dict.get("tags", meta_dict.get("tag", None))
	if raw is None:
		return []
	if isinstance(raw, str):
		parts = [p.strip() for p in re.split(r"[,;]", raw)]
		return [p for p in parts if p]
	if isinstance(raw, (list, tuple)):
		out: list[str] = []
		for item in raw:
			if item is None:
				continue
			s = str(item).strip()
			if s:
				out.append(s)
		return out
	s = str(raw).strip()
	return [s] if s else []

def extract_title(f: File) -> str | None:
	"""Extract title from docs file.

	Args:
	    f: The file to extract title from.

	Returns:
	    Optional[str]: The title if successful, None otherwise.
	"""
	try:
		return f.note_title
	except AttributeError:
		return None

def parse_frontmatter(text: str) -> tuple[dict[str, Any], str]:
	"""Split markdown text into frontmatter dict and body.

	Args:
	    text: Full file contents.

	Returns:
	    Tuple of (metadata dict, body without frontmatter).
	"""
	if not text.startswith("---"):
		return {}, text
	end = text.find("\n---", 3)
	if end == -1:
		return {}, text
	raw = text[3:end].strip()
	body = text[end + 4 :].lstrip("\n")
	if not raw:
		return {}, body
	try:
		loaded = yaml.safe_load(raw)
		meta_dict = loaded if isinstance(loaded, dict) else {}
	except (yaml.YAMLError, AttributeError, TypeError):
		# Fallback: flat key: value (no nested lists).
		meta_dict = {}
		for line in raw.splitlines():
			if ":" not in line:
				continue
			key, value = line.split(":", 1)
			key = key.strip()
			value = value.strip()
			if value.startswith('"') and value.endswith('"'):
				value = value[1:-1]
			meta_dict[key] = value
	return meta_dict, body

def parse_frontmatter_file(path: Path) -> tuple[dict[str, Any], str]:
	"""Read a file and parse its frontmatter.

	Args:
	    path: Path to a markdown (or text) file.

	Returns:
	    Tuple of (metadata dict, body).
	"""
	raw = path.read_text(encoding="utf-8")
	return parse_frontmatter(raw)

def validate_frontmatter(f: File) -> bool:
	"""Validate the frontmatter of the file.

	Args:
	    f: The file to validate.

	Returns:
	    bool: True if the frontmatter is valid, False otherwise.
	"""
	try:
		_, frontmatter = mkdocs_meta.get_data(f.content_string)

		if not frontmatter.get("publish", False):
			logger.debug(f"Skipping {f.src_uri} because it is not published")
			return False

		if "date" not in frontmatter:
			logger.error(f"Invalid frontmatter for {f.src_uri}: 'date' is required")
			return False

		date = frontmatter["date"]
		if not isinstance(date, datetime):
			logger.error(
				f"Invalid frontmatter for {f.src_uri}: 'date' must be a datetime object"
			)
			return False

		f.note_date = date

		if "title" not in frontmatter:
			logger.error(f"Invalid frontmatter for {f.src_uri}: 'title' is required")
			return False

		title = frontmatter["title"]
		if not isinstance(title, str):
			logger.error(
				f"Invalid frontmatter for {f.src_uri}: 'title' must be a string"
			)
			return False

		f.note_title = title
		return True

	except Exception as e:
		logger.error(f"Error validating frontmatter for {f.src_uri}: {e}")
		raise

def scan_notes(files: Files, config) -> tuple[list[File], list[File]]:
	"""Scan notes directory, return all supported note files

	Args:
		files (Files): The list of files to scan
		config: Plugin configuration

	Returns:
		tuple[list[File], list[File]]: (valid notes, invalid files)
	"""
	notes_dir = (
		Path(config.notes_root)
		if isinstance(config.notes_root, str)
		else config.notes_root
	)
	if not notes_dir.exists():
		logger.warning(f"Notes directory does not exist: {notes_dir}")
		return [], []

	notes = []
	invalid_files = []

	try:
		for f in files:
			# Skip non-documentation pages
			if not f.is_documentation_page():
				continue

			# Check if file is within notes_root by comparing absolute paths
			# f.abs_src_path is the absolute path to the source file
			try:
				file_path = Path(f.abs_src_path)
				# Check if the file is within the notes_root directory
				file_path.relative_to(notes_dir)
			except (ValueError, AttributeError):
				# File is not within notes_root
				continue

			# Validate frontmatter
			if validate_frontmatter(f):
				notes.append(f)
			else:
				invalid_files.append(f)
	except Exception as e:
		logger.error(f"Error scanning notes: {e}")
		raise

	return notes, invalid_files

def build_directory_tree(
	root: Path,
	*,
	parent_key: str = "",
	rel_prefix: str = "",
) -> list[TreeNode]:
	"""Build a ``TreeNode`` forest from filesystem hierarchy under ``root``.

	Directories become sections; ``.md`` / ``.ipynb`` leaves become content pages.
	``index.md`` files are omitted from the tree (callers still skip them on sync).

	Args:
	    root: Absolute or relative notes root directory.
	    parent_key: Parent nav key for this level.
	    rel_prefix: Docs-relative path prefix for children.

	Returns:
	    Ordered list of tree nodes for this directory level.
	"""
	root = Path(root)
	if not root.is_dir():
		return []

	items: list[TreeNode] = []
	# Stable ordering: directories first (alpha), then files (alpha).
	entries = sorted(root.iterdir(), key=lambda p: (not p.is_dir(), p.name.lower()))
	for entry in entries:
		if entry.name.startswith("."):
			continue
		if entry.is_dir():
			# Skip co-located asset directories.
			if entry.name == "assets":
				continue
			rel = docs_rel(f"{rel_prefix}/{entry.name}" if rel_prefix else entry.name)
			section_key = rel
			children = build_directory_tree(
				entry, parent_key=section_key, rel_prefix=rel
			)
			if not children:
				continue
			items.append(
				TreeNode(
					key=section_key,
					title=entry.name,
					file_rel=None,
					parent_key=parent_key,
					children=children,
				)
			)
			continue

		suffix = entry.suffix.lower()
		if suffix not in CONTENT_SUFFIXES:
			continue
		rel = docs_rel(f"{rel_prefix}/{entry.name}" if rel_prefix else entry.name)
		if is_index_doc(rel):
			continue
		items.append(
			TreeNode(
				key=rel,
				title=title_from_path(entry),
				file_rel=rel,
				parent_key=parent_key,
			)
		)
	return items

def build_nav_tree(
	nodes: list[Any],
	parent_key: str = "",
	*,
	docs_prefix: str = "docs/",
) -> list[TreeNode]:
	"""Build a ``TreeNode`` forest from awesome-nav YAML nodes."""
	items: list[TreeNode] = []
	for node in nodes:
		if isinstance(node, str):
			rel = resolve_doc_path(node, docs_prefix=docs_prefix)
			items.append(
				TreeNode(
					key=rel,
					title=title_from_path(Path(rel)),
					file_rel=rel,
					parent_key=parent_key,
				)
			)
			continue
		if not isinstance(node, dict):
			continue
		for title, child in node.items():
			title_s = str(title)
			if isinstance(child, str):
				rel = resolve_doc_path(child, docs_prefix=docs_prefix)
				items.append(
					TreeNode(
						key=rel,
						title=title_s,
						file_rel=rel,
						parent_key=parent_key,
					)
				)
			elif isinstance(child, list):
				section_key = f"{parent_key}/{title_s}" if parent_key else title_s
				section = TreeNode(
					key=section_key,
					title=title_s,
					file_rel=None,
					parent_key=parent_key,
					children=build_nav_tree(
						child, section_key, docs_prefix=docs_prefix
					),
				)
				items.append(section)
	return items

def build_page_tree(
	*,
	nav_file: Path | None,
	notes_root: Path,
	docs_dir: Path,
) -> tuple[list[TreeNode], str]:
	"""Resolve page tree: prefer ``.nav.yml``, else ``notes_root`` directory scan.

	Args:
	    nav_file: Path to ``.nav.yml`` (may be None or missing).
	    notes_root: Notes directory for filesystem fallback.
	    docs_dir: Docs root used when stripping prefixes in nav paths.

	Returns:
	    Tuple of (tree, source_label) where source_label is ``"nav.yml"`` or
	    ``"directory"``.
	"""
	docs_prefix = docs_rel(str(docs_dir)).rstrip("/") + "/"
	if nav_file is not None and nav_file.is_file():
		nodes = load_nav_yaml(nav_file)
		return build_nav_tree(nodes, docs_prefix=docs_prefix), "nav.yml"

	logger.warning(
		"Navigation file not found (%s). Falling back to directory scan of "
		"'%s' (directory hierarchy preserved). For custom titles and grouping "
		"that match your site nav, install mkdocs-awesome-nav and add a "
		".nav.yml under your docs directory.",
		nav_file,
		notes_root,
	)
	return build_directory_tree(notes_root), "directory"

def docs_rel(path: str) -> str:
	"""Normalize path separators to forward slashes."""
	return path.replace("\\", "/")

def index_tree(tree: list[TreeNode]) -> dict[str, TreeNode]:
	"""Map node key → node for the entire tree."""
	return {item.key: item for item in walk_tree(tree)}

def is_index_doc(rel: str) -> bool:
	"""True for any docs-relative path whose basename is ``index.md``."""
	return Path(rel).name.lower() == "index.md"

def load_nav_yaml(nav_path: Path) -> list[Any]:
	"""Load the ``nav`` list from an awesome-nav ``.nav.yml`` file."""
	with nav_path.open("r", encoding="utf-8") as f:
		data = yaml.safe_load(f) or {}
	if isinstance(data, dict):
		return data.get("nav", data) if isinstance(data.get("nav", data), list) else []
	if isinstance(data, list):
		return data
	return []

def resolve_doc_path(raw: str, docs_prefix: str = "docs/") -> str:
	"""Strip optional docs prefix and quotes from a nav path entry."""
	raw = raw.strip().strip('"').strip("'")
	prefix = docs_prefix if docs_prefix.endswith("/") else f"{docs_prefix}/"
	# Also accept bare "docs/" when docs_dir is custom.
	if raw.startswith("docs/"):
		raw = raw[5:]
	elif prefix != "docs/" and raw.startswith(prefix):
		raw = raw[len(prefix) :]
	return docs_rel(raw)

def title_from_path(path: Path) -> str:
	"""Default title from a file stem."""
	return path.stem

def walk_tree(items: list[TreeNode]) -> list[TreeNode]:
	"""Depth-first flatten of a tree."""
	ordered: list[TreeNode] = []
	for item in items:
		ordered.append(item)
		ordered.extend(walk_tree(item.children))
	return ordered

def run_sync(options: SyncOptions) -> int:
	"""Main sync entry used by the CLI. Returns a process exit code."""
	options = _apply_env_overrides(options)
	if options.verbose:
		setup_logging(True)

	token = resolve_token(
		options.token,
		options.project_root,
		allow_cursor_mcp_token=options.allow_cursor_mcp_token,
		silence_mcp_token_warning=options.silence_mcp_token_warning,
	)
	if not token and not options.dry_run:
		log.error(
			"Notion token not found. Set NOTION_TOKEN, add a project .env / "
			".notion_token, or enable allow_cursor_mcp_token for developer use."
		)
		return 1

	if not options.database_id or not options.data_source_id:
		log.error(
			"database_id and data_source_id are required "
			"(mkdocs.yml notion_sync or NOTION_WIKI_DATABASE / "
			"NOTION_WIKI_DATA_SOURCE)."
		)
		return 1

	state_path = options.state_path
	sections = options.section
	docs_prefix = _docs_prefix(options.project_root, options.docs_dir)

	# Resolve what to sync.
	path_list: list[str] = []
	if options.paths:
		path_list.extend(options.paths)
	if options.paths_file is not None:
		raw = Path(options.paths_file).read_text(encoding="utf-8")
		path_list.extend(line.strip() for line in raw.splitlines() if line.strip())

	if path_list:
		normalized: set[str] = set()
		for p in path_list:
			normalized.add(_strip_docs_prefix(p, docs_prefix))
		diff = DiffSet(md_changed=normalized)
		base: str | None = "(paths)"
		full = False
	elif options.full:
		diff = DiffSet(nav_changed=True)
		base = None
		full = True
	else:
		base = resolve_git_base(options.project_root, options.base)
		full = base is None
		if full:
			log.info("no git base available → full sync")
			diff = DiffSet(nav_changed=True)
		else:
			log.info("incremental sync since %s", base)
			diff = git_diff(base, options.project_root, options.docs_dir)

	log.info(
		"diff: md=%d deleted=%d assets=%d nav_changed=%s full=%s",
		len(diff.md_changed),
		len(diff.md_deleted),
		len(diff.assets_changed),
		diff.nav_changed,
		full,
	)

	tree, tree_source = build_page_tree(
		nav_file=options.nav_file,
		notes_root=options.notes_root,
		docs_dir=options.docs_dir,
	)
	log.info("page tree source: %s", tree_source)
	nav_index = index_tree(tree)

	# Load / rebuild page map.
	state = load_state(state_path, default_title=options.title_property)
	need_rebuild = options.rebuild_state or not state.pages
	if need_rebuild:
		if options.dry_run and not token:
			log.warning("dry-run without token: empty state")
		else:
			assert token
			log.info("rebuilding page map from Notion wiki…")
			state = rebuild_state_from_wiki(
				token,
				data_source_id=options.data_source_id,
				database_id=options.database_id,
				title_property=options.title_property,
				tree=tree,
				sections=sections,
			)
			if not options.dry_run:
				save_state(state_path, state)
			log.info("mapped %d keys", len(state.pages))
	else:
		state.root_page_id = state.root_page_id or options.database_id
		state.data_source_id = state.data_source_id or options.data_source_id
		state.title_property = state.title_property or options.title_property

	targets, deleted = collect_targets(
		full=full,
		diff=diff,
		nav_index=nav_index,
		docs_dir=options.docs_dir,
		sections=sections,
	)

	if deleted:
		for rel in sorted(deleted):
			log.info("deleted locally (Notion page left intact): %s", rel)

	if not targets:
		log.info("nothing to sync")
		return 0

	tags_cache: TagsSchemaCache | None = None
	if not options.dry_run and token:
		tags_cache = TagsSchemaCache(
			data_source_id=state.data_source_id or options.data_source_id,
			property_name=options.tags_property,
		)

	log.info("syncing %d page(s)", len(targets))
	stats = {"created": 0, "updated": 0, "dry-run": 0, "missing": 0, "failed": 0}
	for item in targets:
		try:
			result = sync_one_page(
				token or "",
				state,
				state_path,
				nav_index,
				item,
				docs_dir=options.docs_dir,
				site_url=options.site_url,
				delay=options.delay,
				dry_run=options.dry_run,
				upload_images=not options.no_images,
				tags_property=options.tags_property,
				tags_cache=tags_cache,
			)
			stats[result] = stats.get(result, 0) + 1
		except urllib.error.HTTPError as exc:
			body = getattr(exc, "reason", "") or ""
			log.error("FAIL %s: %s %s", item.file_rel, exc.code, body)
			stats["failed"] += 1
			if not options.continue_on_error:
				return 1
		except Exception as exc:  # noqa: BLE001 - per-page catch-all for continue_on_error
			log.error("FAIL %s: %s", item.file_rel, exc)
			stats["failed"] += 1
			if not options.continue_on_error:
				return 1

	if not options.dry_run:
		save_state(state_path, state)
	log.info("done %s", json.dumps(stats, ensure_ascii=False))
	return 1 if stats["failed"] else 0