587 lines
19 KiB
Python
587 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""Read-only inventory and fingerprint utility for LeafWiki snapshots."""
|
|
|
|
from __future__ import annotations
|
|
|
|
# This module is named `inspect.py`, which shadows the standard-library `inspect`
|
|
# module when the script is executed directly (its own directory is prepended to
|
|
# sys.path). Some stdlib modules (e.g. `dataclasses`) import `inspect` lazily and
|
|
# would then re-enter this file, causing a circular import. Drop the script
|
|
# directory from sys.path before importing anything else.
|
|
import os as _os
|
|
import sys as _sys
|
|
|
|
_module_dir = _os.path.dirname(_os.path.abspath(__file__))
|
|
_sys.path[:] = [
|
|
_p for _p in _sys.path
|
|
if _os.path.abspath(_p or _os.getcwd()) != _module_dir
|
|
]
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import mimetypes
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
import sys
|
|
from collections import Counter
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
SQLITE_HEADER = b"SQLite format 3\x00"
|
|
MARKDOWN_SUFFIXES = {".md", ".markdown", ".mdown", ".mkd"}
|
|
ASSET_SUFFIXES = {
|
|
".png", ".jpg", ".jpeg", ".gif", ".webp", ".svg", ".pdf",
|
|
".doc", ".docx", ".xls", ".xlsx", ".ppt", ".pptx",
|
|
".txt", ".csv", ".zip", ".7z", ".tar", ".gz", ".yaml", ".yml",
|
|
}
|
|
WIKI_LINK = re.compile(r"\[\[([^\]]+)\]\]")
|
|
MARKDOWN_LINK = re.compile(r"!?\[[^\]]*\]\(([^)]+)\)")
|
|
FRONTMATTER_KEY = re.compile(r"^([A-Za-z0-9_.-]+)\s*:")
|
|
|
|
|
|
@dataclass
|
|
class Diagnostic:
|
|
level: str
|
|
code: str
|
|
path: str | None
|
|
message: str
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"level": self.level,
|
|
"code": self.code,
|
|
"path": self.path,
|
|
"message": self.message,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class Inspection:
|
|
root: Path
|
|
include_root_path: bool = False
|
|
entries: list[dict[str, Any]] = field(default_factory=list)
|
|
diagnostics: list[Diagnostic] = field(default_factory=list)
|
|
|
|
def add_diagnostic(
|
|
self,
|
|
level: str,
|
|
code: str,
|
|
message: str,
|
|
relative_path: str | None = None,
|
|
) -> None:
|
|
self.diagnostics.append(
|
|
Diagnostic(level=level, code=code, path=relative_path, message=message)
|
|
)
|
|
|
|
def build_report(self) -> dict[str, Any]:
|
|
ordered_entries = sorted(self.entries, key=lambda item: item["path"])
|
|
fingerprint_items = [
|
|
{
|
|
"path": entry["path"],
|
|
"kind": entry["kind"],
|
|
"size": entry.get("size"),
|
|
"sha256": entry.get("sha256"),
|
|
"linkTarget": entry.get("linkTarget"),
|
|
}
|
|
for entry in ordered_entries
|
|
]
|
|
encoded = json.dumps(
|
|
fingerprint_items,
|
|
ensure_ascii=False,
|
|
separators=(",", ":"),
|
|
sort_keys=True,
|
|
).encode("utf-8")
|
|
manifest_hash = hashlib.sha256(encoded).hexdigest()
|
|
counts = Counter(entry["kind"] for entry in ordered_entries)
|
|
|
|
source: dict[str, Any] = {
|
|
"name": self.root.name,
|
|
"manifestSha256": manifest_hash,
|
|
"entryCount": len(ordered_entries),
|
|
}
|
|
if self.include_root_path:
|
|
source["root"] = str(self.root)
|
|
|
|
return {
|
|
"schemaVersion": 1,
|
|
"tool": "ItWorx LeafWiki read-only inspector",
|
|
"generatedAt": datetime.now(timezone.utc).isoformat(),
|
|
"source": source,
|
|
"summary": {
|
|
"countsByKind": dict(sorted(counts.items())),
|
|
"diagnosticCounts": dict(
|
|
sorted(Counter(item.level for item in self.diagnostics).items())
|
|
),
|
|
},
|
|
"entries": ordered_entries,
|
|
"diagnostics": [
|
|
item.to_dict()
|
|
for item in sorted(
|
|
self.diagnostics,
|
|
key=lambda item: (item.level, item.path or "", item.code),
|
|
)
|
|
],
|
|
}
|
|
|
|
|
|
def hash_file(path: Path, chunk_size: int = 1024 * 1024) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
while chunk := handle.read(chunk_size):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def is_inside(candidate: Path, root: Path) -> bool:
|
|
try:
|
|
candidate.resolve(strict=False).relative_to(root.resolve(strict=True))
|
|
return True
|
|
except (ValueError, OSError):
|
|
return False
|
|
|
|
|
|
def relative_posix(path: Path, root: Path) -> str:
|
|
return path.relative_to(root).as_posix()
|
|
|
|
|
|
def read_prefix(path: Path, length: int = 4096) -> bytes:
|
|
with path.open("rb") as handle:
|
|
return handle.read(length)
|
|
|
|
|
|
def inspect_markdown(path: Path, inspection: Inspection, entry: dict[str, Any]) -> None:
|
|
try:
|
|
text = path.read_text(encoding="utf-8")
|
|
encoding = "utf-8"
|
|
except UnicodeDecodeError:
|
|
try:
|
|
text = path.read_text(encoding="utf-8-sig")
|
|
encoding = "utf-8-sig"
|
|
except UnicodeDecodeError as error:
|
|
inspection.add_diagnostic(
|
|
"error",
|
|
"markdown.decode",
|
|
f"Markdown is not valid UTF-8: {error}",
|
|
entry["path"],
|
|
)
|
|
entry["markdown"] = {"encoding": "unknown", "readable": False}
|
|
return
|
|
|
|
lines = text.splitlines()
|
|
frontmatter_keys: list[str] = []
|
|
frontmatter_title: str | None = None
|
|
body_start = 0
|
|
|
|
if lines and lines[0].strip() == "---":
|
|
for index, line in enumerate(lines[1:], start=1):
|
|
if line.strip() == "---":
|
|
body_start = index + 1
|
|
break
|
|
match = FRONTMATTER_KEY.match(line)
|
|
if match:
|
|
key = match.group(1)
|
|
frontmatter_keys.append(key)
|
|
if key.lower() == "title":
|
|
frontmatter_title = line.split(":", 1)[1].strip().strip("\"'")
|
|
else:
|
|
inspection.add_diagnostic(
|
|
"warning",
|
|
"markdown.frontmatter-unclosed",
|
|
"Frontmatter opening marker has no closing marker.",
|
|
entry["path"],
|
|
)
|
|
|
|
first_heading = None
|
|
for line in lines[body_start:]:
|
|
if line.startswith("# "):
|
|
first_heading = line[2:].strip()
|
|
break
|
|
|
|
wiki_links = []
|
|
for raw in WIKI_LINK.findall(text):
|
|
target, separator, label = raw.partition("|")
|
|
wiki_links.append(
|
|
{
|
|
"target": target.strip(),
|
|
"label": label.strip() if separator else None,
|
|
}
|
|
)
|
|
|
|
markdown_links = [target.strip() for target in MARKDOWN_LINK.findall(text)]
|
|
|
|
entry["markdown"] = {
|
|
"readable": True,
|
|
"encoding": encoding,
|
|
"lineCount": len(lines),
|
|
"frontmatterKeys": sorted(set(frontmatter_keys)),
|
|
"title": frontmatter_title or first_heading or path.stem,
|
|
"wikiLinks": wiki_links,
|
|
"markdownLinkTargets": markdown_links,
|
|
}
|
|
|
|
|
|
def quote_identifier(value: str) -> str:
|
|
return '"' + value.replace('"', '""') + '"'
|
|
|
|
|
|
def inspect_sqlite(path: Path, inspection: Inspection, entry: dict[str, Any]) -> None:
|
|
database: dict[str, Any] = {
|
|
"readOnly": True,
|
|
"immutable": True,
|
|
"tables": [],
|
|
"views": [],
|
|
"indexes": [],
|
|
}
|
|
|
|
try:
|
|
uri = path.resolve(strict=True).as_uri() + "?mode=ro&immutable=1"
|
|
connection = sqlite3.connect(uri, uri=True, timeout=5)
|
|
connection.row_factory = sqlite3.Row
|
|
except (sqlite3.Error, OSError) as error:
|
|
inspection.add_diagnostic(
|
|
"error",
|
|
"sqlite.open",
|
|
f"Could not open SQLite database read-only: {error}",
|
|
entry["path"],
|
|
)
|
|
entry["sqlite"] = database
|
|
return
|
|
|
|
try:
|
|
connection.execute("PRAGMA query_only = ON")
|
|
database["applicationId"] = connection.execute(
|
|
"PRAGMA application_id"
|
|
).fetchone()[0]
|
|
database["userVersion"] = connection.execute(
|
|
"PRAGMA user_version"
|
|
).fetchone()[0]
|
|
database["pageSize"] = connection.execute("PRAGMA page_size").fetchone()[0]
|
|
|
|
objects = connection.execute(
|
|
"""
|
|
SELECT type, name, tbl_name, sql
|
|
FROM sqlite_master
|
|
WHERE name NOT LIKE 'sqlite_%'
|
|
ORDER BY type, name
|
|
"""
|
|
).fetchall()
|
|
|
|
schema_fingerprint_data: list[dict[str, Any]] = []
|
|
|
|
for obj in objects:
|
|
object_type = obj["type"]
|
|
name = obj["name"]
|
|
schema_fingerprint_data.append(
|
|
{
|
|
"type": object_type,
|
|
"name": name,
|
|
"table": obj["tbl_name"],
|
|
"sql": obj["sql"],
|
|
}
|
|
)
|
|
|
|
if object_type == "table":
|
|
table: dict[str, Any] = {
|
|
"name": name,
|
|
"columns": [],
|
|
"foreignKeys": [],
|
|
"rowCount": None,
|
|
}
|
|
safe_name = quote_identifier(name)
|
|
|
|
try:
|
|
columns = connection.execute(
|
|
f"PRAGMA table_info({safe_name})"
|
|
).fetchall()
|
|
table["columns"] = [
|
|
{
|
|
"position": column["cid"],
|
|
"name": column["name"],
|
|
"declaredType": column["type"],
|
|
"notNull": bool(column["notnull"]),
|
|
"primaryKeyPosition": column["pk"],
|
|
"hasDefault": column["dflt_value"] is not None,
|
|
}
|
|
for column in columns
|
|
]
|
|
except sqlite3.Error as error:
|
|
inspection.add_diagnostic(
|
|
"warning",
|
|
"sqlite.columns",
|
|
f"Could not inspect columns for {name}: {error}",
|
|
entry["path"],
|
|
)
|
|
|
|
try:
|
|
foreign_keys = connection.execute(
|
|
f"PRAGMA foreign_key_list({safe_name})"
|
|
).fetchall()
|
|
table["foreignKeys"] = [
|
|
{
|
|
"from": key["from"],
|
|
"toTable": key["table"],
|
|
"toColumn": key["to"],
|
|
"onUpdate": key["on_update"],
|
|
"onDelete": key["on_delete"],
|
|
}
|
|
for key in foreign_keys
|
|
]
|
|
except sqlite3.Error as error:
|
|
inspection.add_diagnostic(
|
|
"warning",
|
|
"sqlite.foreign-keys",
|
|
f"Could not inspect foreign keys for {name}: {error}",
|
|
entry["path"],
|
|
)
|
|
|
|
try:
|
|
table["rowCount"] = connection.execute(
|
|
f"SELECT COUNT(*) FROM {safe_name}"
|
|
).fetchone()[0]
|
|
except sqlite3.Error as error:
|
|
inspection.add_diagnostic(
|
|
"warning",
|
|
"sqlite.row-count",
|
|
f"Could not count rows in {name}: {error}",
|
|
entry["path"],
|
|
)
|
|
|
|
database["tables"].append(table)
|
|
elif object_type == "view":
|
|
database["views"].append({"name": name})
|
|
elif object_type == "index":
|
|
database["indexes"].append(
|
|
{"name": name, "table": obj["tbl_name"], "hasSql": obj["sql"] is not None}
|
|
)
|
|
|
|
schema_bytes = json.dumps(
|
|
schema_fingerprint_data,
|
|
ensure_ascii=False,
|
|
separators=(",", ":"),
|
|
sort_keys=True,
|
|
).encode("utf-8")
|
|
database["schemaSha256"] = hashlib.sha256(schema_bytes).hexdigest()
|
|
except sqlite3.Error as error:
|
|
inspection.add_diagnostic(
|
|
"error",
|
|
"sqlite.inspect",
|
|
f"SQLite inspection failed: {error}",
|
|
entry["path"],
|
|
)
|
|
finally:
|
|
connection.close()
|
|
|
|
entry["sqlite"] = database
|
|
|
|
|
|
def inspect_json(path: Path, inspection: Inspection, entry: dict[str, Any]) -> None:
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as error:
|
|
inspection.add_diagnostic(
|
|
"warning",
|
|
"json.parse",
|
|
f"JSON could not be parsed: {error}",
|
|
entry["path"],
|
|
)
|
|
entry["json"] = {"valid": False}
|
|
return
|
|
|
|
summary: dict[str, Any] = {"valid": True, "rootType": type(value).__name__}
|
|
if isinstance(value, dict):
|
|
summary["topLevelKeys"] = sorted(str(key) for key in value.keys())
|
|
elif isinstance(value, list):
|
|
summary["itemCount"] = len(value)
|
|
entry["json"] = summary
|
|
|
|
|
|
def classify_regular_file(path: Path, prefix: bytes) -> str:
|
|
if prefix.startswith(SQLITE_HEADER):
|
|
return "sqlite"
|
|
if path.suffix.lower() in MARKDOWN_SUFFIXES:
|
|
return "markdown"
|
|
if path.suffix.lower() == ".json":
|
|
return "json"
|
|
if path.suffix.lower() in ASSET_SUFFIXES:
|
|
return "asset"
|
|
return "unknown"
|
|
|
|
|
|
def walk_source(inspection: Inspection) -> None:
|
|
root = inspection.root
|
|
|
|
for current_path, directory_names, file_names in os.walk(
|
|
root,
|
|
topdown=True,
|
|
followlinks=False,
|
|
):
|
|
current = Path(current_path)
|
|
|
|
symlink_directories = []
|
|
normal_directories = []
|
|
for name in directory_names:
|
|
candidate = current / name
|
|
if candidate.is_symlink():
|
|
symlink_directories.append(name)
|
|
else:
|
|
normal_directories.append(name)
|
|
directory_names[:] = sorted(normal_directories)
|
|
|
|
for name in sorted(symlink_directories):
|
|
path = current / name
|
|
target = os.readlink(path)
|
|
resolved = (path.parent / target).resolve(strict=False)
|
|
relative = relative_posix(path, root)
|
|
inspection.entries.append(
|
|
{
|
|
"path": relative,
|
|
"kind": "symlink",
|
|
"linkTarget": target,
|
|
"targetInsideRoot": is_inside(resolved, root),
|
|
"isDirectoryLink": True,
|
|
}
|
|
)
|
|
if not is_inside(resolved, root):
|
|
inspection.add_diagnostic(
|
|
"warning",
|
|
"symlink.escape",
|
|
"Symlink target resolves outside the source root and was not followed.",
|
|
relative,
|
|
)
|
|
|
|
for name in sorted(file_names):
|
|
path = current / name
|
|
relative = relative_posix(path, root)
|
|
|
|
if path.is_symlink():
|
|
target = os.readlink(path)
|
|
resolved = (path.parent / target).resolve(strict=False)
|
|
inspection.entries.append(
|
|
{
|
|
"path": relative,
|
|
"kind": "symlink",
|
|
"linkTarget": target,
|
|
"targetInsideRoot": is_inside(resolved, root),
|
|
"isDirectoryLink": False,
|
|
}
|
|
)
|
|
if not is_inside(resolved, root):
|
|
inspection.add_diagnostic(
|
|
"warning",
|
|
"symlink.escape",
|
|
"Symlink target resolves outside the source root and was not followed.",
|
|
relative,
|
|
)
|
|
continue
|
|
|
|
try:
|
|
stat = path.stat()
|
|
prefix = read_prefix(path)
|
|
kind = classify_regular_file(path, prefix)
|
|
entry: dict[str, Any] = {
|
|
"path": relative,
|
|
"kind": kind,
|
|
"size": stat.st_size,
|
|
"sha256": hash_file(path),
|
|
"modifiedTimeNs": stat.st_mtime_ns,
|
|
}
|
|
|
|
if kind == "markdown":
|
|
inspect_markdown(path, inspection, entry)
|
|
elif kind == "sqlite":
|
|
inspect_sqlite(path, inspection, entry)
|
|
elif kind == "json":
|
|
inspect_json(path, inspection, entry)
|
|
elif kind == "asset":
|
|
media_type, _ = mimetypes.guess_type(path.name)
|
|
entry["asset"] = {
|
|
"extension": path.suffix.lower(),
|
|
"guessedMediaType": media_type or "application/octet-stream",
|
|
}
|
|
|
|
inspection.entries.append(entry)
|
|
except (OSError, sqlite3.Error) as error:
|
|
inspection.add_diagnostic(
|
|
"error",
|
|
"file.inspect",
|
|
f"Could not inspect file: {error}",
|
|
relative,
|
|
)
|
|
|
|
|
|
def output_is_inside_source(output: Path, source: Path) -> bool:
|
|
try:
|
|
output.resolve(strict=False).relative_to(source.resolve(strict=True))
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(
|
|
description="Inventory and fingerprint a LeafWiki snapshot without modifying it."
|
|
)
|
|
parser.add_argument("source", type=Path, help="Read-only source directory")
|
|
parser.add_argument(
|
|
"--output",
|
|
type=Path,
|
|
help="Write JSON report to this path. It must be outside the source tree.",
|
|
)
|
|
parser.add_argument(
|
|
"--include-root-path",
|
|
action="store_true",
|
|
help="Include the absolute source path in the report.",
|
|
)
|
|
parser.add_argument(
|
|
"--compact",
|
|
action="store_true",
|
|
help="Emit compact JSON instead of indented JSON.",
|
|
)
|
|
return parser
|
|
|
|
|
|
def main(argv: Iterable[str] | None = None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
source = args.source.resolve(strict=False)
|
|
|
|
if not source.exists() or not source.is_dir():
|
|
print(f"Source directory does not exist: {source}", file=sys.stderr)
|
|
return 2
|
|
|
|
if args.output is not None:
|
|
output = args.output.resolve(strict=False)
|
|
if output_is_inside_source(output, source):
|
|
print("Refusing to write the report inside the source tree.", file=sys.stderr)
|
|
return 2
|
|
else:
|
|
output = None
|
|
|
|
inspection = Inspection(source, include_root_path=args.include_root_path)
|
|
walk_source(inspection)
|
|
report = inspection.build_report()
|
|
|
|
rendered = json.dumps(
|
|
report,
|
|
ensure_ascii=False,
|
|
indent=None if args.compact else 2,
|
|
sort_keys=False,
|
|
) + "\n"
|
|
|
|
if output is None:
|
|
sys.stdout.write(rendered)
|
|
else:
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = output.with_name(output.name + ".tmp")
|
|
temporary.write_text(rendered, encoding="utf-8", newline="\n")
|
|
os.replace(temporary, output)
|
|
|
|
return 1 if any(item.level == "error" for item in inspection.diagnostics) else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|