Files
obsidian-wiki/obsidian_wiki/file_assets.py
T
Codex 2b7688bd75
setup / install-leaves-tree-clean (push) Canceled after 0s
fix(file-index): prune removed active assets on rebuild
Treat the source registry as authoritative during file-index rebuilds while preserving staged uploads. This prevents deleted archived sources from remaining searchable.

Tests: python3 -m unittest tests.test_file_assets tests.test_business_facts
2026-07-23 02:18:11 +00:00

769 lines
27 KiB
Python

"""SQLite-backed file asset index for archived wiki sources.
The markdown wiki remains the source of distilled knowledge. This module keeps a
small, portable SQLite database for the operational question "where is the
original file?" so agents can find drawings, 3D files, spreadsheets, reports,
and other source assets without scanning the whole vault.
"""
from __future__ import annotations
import hashlib
import json
import mimetypes
import os
import re
import shutil
import sqlite3
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable
DB_RELATIVE_PATH = Path("_meta") / "file-assets.sqlite"
SOURCE_REGISTRY_RELATIVE_PATH = Path("_meta") / "source-registry.jsonl"
TOKEN_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.+/-]*|[\u4e00-\u9fff]+")
BJD_PART_RE = re.compile(r"\b\d{6}-\d{6}(?:-[A-Za-z0-9]+)?\b")
GENERIC_PART_RE = re.compile(r"\b[A-Z]{1,6}\d{2,}[A-Z0-9-]*\b", re.IGNORECASE)
PITCH_RE = re.compile(
r"(?<!\d)(?:P\s*)?(\d+(?:\.\d+)?)\s*(?:mm)?\s*(?:pitch|间距)"
r"|(?:pitch|间距)\s*(?:P\s*)?(\d+(?:\.\d+)?)\s*(?:mm)?",
re.IGNORECASE,
)
PIN_RE = re.compile(r"(?<!\d)(\d{1,3})\s*(?:pin|p|位|芯)", re.IGNORECASE)
ARCHIVE_HASH_PREFIX_RE = re.compile(r"\b[0-9a-fA-F]{8,64}--")
STRUCTURED_IDENTIFIER_FIELDS = (
"bjd_part_numbers",
"competitor_part_numbers",
"series",
"pitch",
"pin_count",
)
def clean_config_value(value: str) -> str:
"""Normalize config values that may be quoted once or twice."""
value = value.strip()
changed = True
while changed and len(value) >= 2:
changed = False
if (value[0], value[-1]) in {('"', '"'), ("'", "'")}:
value = value[1:-1].strip()
changed = True
return value
def utc_now() -> str:
return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
def db_path_for_vault(vault_path: str | Path) -> Path:
vault = Path(clean_config_value(str(vault_path))).expanduser()
return vault / DB_RELATIVE_PATH
def registry_path_for_vault(vault_path: str | Path) -> Path:
vault = Path(clean_config_value(str(vault_path))).expanduser()
return vault / SOURCE_REGISTRY_RELATIVE_PATH
def connect(db_path: Path, *, read_only: bool = False) -> sqlite3.Connection:
if read_only:
if not db_path.is_file():
raise FileNotFoundError(f"file asset index not found: {db_path}")
immutable = "&immutable=1" if not os.access(db_path.parent, os.W_OK) else ""
uri = f"{db_path.resolve().as_uri()}?mode=ro{immutable}"
conn = sqlite3.connect(uri, uri=True)
conn.row_factory = sqlite3.Row
return conn
db_path.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(db_path))
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL")
return conn
def init_db(db_path: Path) -> None:
with connect(db_path) as conn:
conn.executescript(
"""
CREATE TABLE IF NOT EXISTS file_assets (
file_id TEXT PRIMARY KEY,
status TEXT NOT NULL DEFAULT 'active',
sha256 TEXT NOT NULL,
archive_path TEXT NOT NULL,
original_path TEXT,
original_name TEXT,
file_ext TEXT,
mime_type TEXT,
source_type TEXT,
doc_type TEXT,
project TEXT,
title TEXT,
summary TEXT,
visibility TEXT,
business_categories TEXT,
product_lines TEXT,
customers TEXT,
competitors TEXT,
standards TEXT,
market_tags TEXT,
bjd_part_numbers TEXT,
competitor_part_numbers TEXT,
series TEXT,
pitch TEXT,
pin_count TEXT,
related_pages TEXT,
catalog_paths TEXT,
extracted_text TEXT,
classification_confidence REAL,
classification_evidence TEXT,
needs_review INTEGER NOT NULL DEFAULT 0,
size_bytes INTEGER,
modified_at TEXT,
archived_at TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_file_assets_status
ON file_assets(status);
CREATE INDEX IF NOT EXISTS idx_file_assets_doc_type
ON file_assets(doc_type);
CREATE INDEX IF NOT EXISTS idx_file_assets_project
ON file_assets(project);
CREATE INDEX IF NOT EXISTS idx_file_assets_original_name
ON file_assets(original_name);
CREATE VIRTUAL TABLE IF NOT EXISTS file_assets_fts USING fts5(
file_id UNINDEXED,
original_name,
title,
summary,
archive_path,
doc_type,
product_lines,
business_categories,
customers,
competitors,
standards,
bjd_part_numbers,
competitor_part_numbers,
series,
pitch,
pin_count,
extracted_text
);
"""
)
@dataclass(frozen=True)
class RebuildResult:
db_path: Path
registry_path: Path
indexed: int
skipped: int
@dataclass(frozen=True)
class StageResult:
file_id: str
path: Path
db_path: Path
def rebuild_from_registry(vault_path: str | Path, *, db_path: Path | None = None) -> RebuildResult:
vault = Path(clean_config_value(str(vault_path))).expanduser()
db_path = db_path or db_path_for_vault(vault)
registry_path = registry_path_for_vault(vault)
init_db(db_path)
if not registry_path.is_file():
raise FileNotFoundError(f"source registry not found: {registry_path}")
records: list[dict[str, Any]] = []
skipped = 0
for raw in registry_path.read_text(encoding="utf-8").splitlines():
if not raw.strip():
continue
try:
record = json.loads(raw)
except json.JSONDecodeError:
skipped += 1
continue
if record.get("record_type") == "source_archive":
records.append(record)
with connect(db_path) as conn:
# A rebuild is authoritative for archived sources. Preserve uploaded
# staging rows, but discard active rows that no longer have a registry
# record so deleted sources cannot remain searchable.
conn.execute(
"DELETE FROM file_assets_fts WHERE file_id IN (SELECT file_id FROM file_assets WHERE status = 'active')"
)
conn.execute("DELETE FROM file_assets WHERE status = 'active'")
for record in records:
upsert_asset(conn, asset_from_registry_record(record), status="active")
conn.execute("DELETE FROM file_assets_fts WHERE file_id NOT IN (SELECT file_id FROM file_assets)")
indexed = len(records)
return RebuildResult(db_path=db_path, registry_path=registry_path, indexed=indexed, skipped=skipped)
def asset_from_registry_record(record: dict[str, Any]) -> dict[str, Any]:
source_id = str(record.get("source_id") or "")
sha256 = source_id.removeprefix("sha256:")
archive_path = str(record.get("archive_path") or "")
original_name = str(record.get("original_name") or Path(archive_path).name)
text_blob = " ".join(
str(record.get(key) or "")
for key in ("original_name", "title", "summary", "archive_path", "original_path")
)
text_blob += " " + " ".join(_flatten_to_strings(record.get(key)) for key in _CLASSIFICATION_KEYS)
text_blob += " " + " ".join(_flatten_to_strings(record.get(key)) for key in STRUCTURED_IDENTIFIER_FIELDS)
text_blob = ARCHIVE_HASH_PREFIX_RE.sub("", text_blob)
derived = derive_identifiers(text_blob)
file_ext = Path(original_name).suffix.lower()
confidence = _float_or_none(record.get("classification_confidence"))
bjd_part_numbers = _merge_index_terms(record.get("bjd_part_numbers"), derived["bjd_part_numbers"], uppercase=True)
competitor_part_numbers = _merge_index_terms(
record.get("competitor_part_numbers"),
derived["competitor_part_numbers"],
uppercase=True,
limit=40,
)
series = _merge_index_terms(record.get("series"), derived["series"], uppercase=True, limit=40)
pitch = _merge_index_terms(record.get("pitch"), derived["pitch"], limit=20)
pin_count = _merge_index_terms(record.get("pin_count"), derived["pin_count"], limit=20)
return {
"file_id": source_id or f"sha256:{sha256}",
"sha256": sha256,
"archive_path": archive_path,
"original_path": record.get("original_path"),
"original_name": original_name,
"file_ext": file_ext,
"mime_type": mimetypes.guess_type(original_name)[0],
"source_type": record.get("source_type") or guess_source_type(original_name),
"doc_type": record.get("doc_type") or "other",
"project": record.get("project"),
"title": record.get("title") or Path(original_name).stem,
"summary": record.get("summary"),
"visibility": record.get("visibility") or "internal",
"business_categories": _json_list(record.get("business_categories")),
"product_lines": _json_list(record.get("product_lines")),
"customers": _json_list(record.get("customers")),
"competitors": _json_list(record.get("competitors")),
"standards": _json_list(record.get("standards")),
"market_tags": _json_list(record.get("market_tags")),
"bjd_part_numbers": _json_list(bjd_part_numbers),
"competitor_part_numbers": _json_list(competitor_part_numbers),
"series": _json_list(series),
"pitch": _json_list(pitch),
"pin_count": _json_list(pin_count),
"related_pages": _json_list(record.get("related_pages")),
"catalog_paths": _json_list(record.get("catalog_paths")),
"extracted_text": None,
"classification_confidence": confidence,
"classification_evidence": _json_list(record.get("classification_evidence")),
"needs_review": 1 if confidence is None or confidence < 0.7 or record.get("doc_type") in (None, "other") else 0,
"size_bytes": _int_or_none(record.get("size_bytes")),
"modified_at": record.get("modified_at"),
"archived_at": record.get("archived_at"),
}
_CLASSIFICATION_KEYS = (
"business_categories",
"product_lines",
"customers",
"competitors",
"standards",
"market_tags",
)
def derive_identifiers(text: str) -> dict[str, list[str]]:
bjd_parts = sorted(set(match.group(0).upper() for match in BJD_PART_RE.finditer(text)))
generic_parts = sorted(set(match.group(0).upper() for match in GENERIC_PART_RE.finditer(text)))
competitor_parts = [part for part in generic_parts if part not in bjd_parts]
pitch = sorted(
set(
value
for match in PITCH_RE.finditer(text)
for value in (match.group(1), match.group(2))
if value and _looks_like_pitch(value)
)
)
pin_count = sorted(set(match.group(1) for match in PIN_RE.finditer(text)))
series = sorted(
set(
token.upper()
for token in generic_parts
if token.upper().startswith(("X", "FH", "DF", "BM", "MINI", "HSD", "FAKRA"))
)
)
return {
"bjd_part_numbers": bjd_parts,
"competitor_part_numbers": competitor_parts[:40],
"series": series[:40],
"pitch": pitch[:20],
"pin_count": pin_count[:20],
}
def _looks_like_pitch(value: str) -> bool:
try:
numeric = float(value)
except ValueError:
return False
return 0 < numeric <= 20
def upsert_asset(conn: sqlite3.Connection, asset: dict[str, Any], *, status: str | None = None) -> None:
now = utc_now()
asset = dict(asset)
asset.setdefault("created_at", now)
asset["updated_at"] = now
asset["status"] = status or asset.get("status") or "active"
columns = [
"file_id",
"status",
"sha256",
"archive_path",
"original_path",
"original_name",
"file_ext",
"mime_type",
"source_type",
"doc_type",
"project",
"title",
"summary",
"visibility",
"business_categories",
"product_lines",
"customers",
"competitors",
"standards",
"market_tags",
"bjd_part_numbers",
"competitor_part_numbers",
"series",
"pitch",
"pin_count",
"related_pages",
"catalog_paths",
"extracted_text",
"classification_confidence",
"classification_evidence",
"needs_review",
"size_bytes",
"modified_at",
"archived_at",
"created_at",
"updated_at",
]
placeholders = ", ".join("?" for _ in columns)
update_clause = ", ".join(f"{col}=excluded.{col}" for col in columns if col not in {"file_id", "created_at"})
conn.execute(
f"""
INSERT INTO file_assets ({", ".join(columns)})
VALUES ({placeholders})
ON CONFLICT(file_id) DO UPDATE SET {update_clause}
""",
[asset.get(col) for col in columns],
)
refresh_fts(conn, asset["file_id"])
def refresh_fts(conn: sqlite3.Connection, file_id: str) -> None:
conn.execute("DELETE FROM file_assets_fts WHERE file_id = ?", (file_id,))
row = conn.execute("SELECT * FROM file_assets WHERE file_id = ?", (file_id,)).fetchone()
if row is None:
return
conn.execute(
"""
INSERT INTO file_assets_fts (
file_id, original_name, title, summary, archive_path, doc_type,
product_lines, business_categories, customers, competitors, standards,
bjd_part_numbers, competitor_part_numbers, series, pitch, pin_count, extracted_text
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
[
row["file_id"],
row["original_name"],
row["title"],
row["summary"],
row["archive_path"],
row["doc_type"],
row["product_lines"],
row["business_categories"],
row["customers"],
row["competitors"],
row["standards"],
row["bjd_part_numbers"],
row["competitor_part_numbers"],
row["series"],
row["pitch"],
row["pin_count"],
row["extracted_text"],
],
)
def stage_files(
vault_path: str | Path,
paths: Iterable[str | Path],
*,
project: str = "misc",
doc_type: str = "other",
visibility: str = "internal",
db_path: Path | None = None,
) -> list[StageResult]:
vault = Path(clean_config_value(str(vault_path))).expanduser()
db_path = db_path or db_path_for_vault(vault)
init_db(db_path)
staged_root = vault / "_sources" / "inbox" / project / datetime.now().strftime("%Y-%m")
staged_root.mkdir(parents=True, exist_ok=True)
results: list[StageResult] = []
with connect(db_path) as conn:
for raw_path in paths:
source = Path(raw_path).expanduser().resolve()
if not source.is_file():
raise FileNotFoundError(f"not a file: {source}")
digest = sha256_file(source)
safe = safe_basename(source.name)
staged_path = staged_root / f"{digest[:12]}--{safe}"
if not staged_path.exists():
shutil.copy2(source, staged_path)
stat = source.stat()
derived = derive_identifiers(f"{source.name} {source}")
asset = {
"file_id": f"sha256:{digest}",
"sha256": digest,
"archive_path": str(staged_path.relative_to(vault)),
"original_path": str(source),
"original_name": source.name,
"file_ext": source.suffix.lower(),
"mime_type": mimetypes.guess_type(source.name)[0],
"source_type": guess_source_type(source.name),
"doc_type": doc_type,
"project": project,
"title": source.stem,
"summary": "Staged upload waiting for wiki ingest/classification.",
"visibility": visibility,
"business_categories": _json_list([]),
"product_lines": _json_list([]),
"customers": _json_list([]),
"competitors": _json_list([]),
"standards": _json_list([]),
"market_tags": _json_list([]),
"bjd_part_numbers": _json_list(derived["bjd_part_numbers"]),
"competitor_part_numbers": _json_list(derived["competitor_part_numbers"]),
"series": _json_list(derived["series"]),
"pitch": _json_list(derived["pitch"]),
"pin_count": _json_list(derived["pin_count"]),
"related_pages": _json_list([]),
"catalog_paths": _json_list([]),
"extracted_text": None,
"classification_confidence": 0.0,
"classification_evidence": _json_list(["staged from CLI upload"]),
"needs_review": 1,
"size_bytes": stat.st_size,
"modified_at": _mtime_iso(stat.st_mtime),
"archived_at": utc_now(),
}
upsert_asset(conn, asset, status="staged")
results.append(StageResult(file_id=f"sha256:{digest}", path=staged_path, db_path=db_path))
return results
def search_assets(
vault_path: str | Path,
query: str,
*,
db_path: Path | None = None,
limit: int | None = 10,
doc_type: str | None = None,
source_type: str | None = None,
product_line: str | None = None,
customer: str | None = None,
competitor: str | None = None,
status: str | None = None,
include_staged: bool = True,
read_only: bool = False,
) -> list[dict[str, Any]]:
db_path = db_path or db_path_for_vault(vault_path)
if not db_path.is_file():
raise FileNotFoundError(f"file asset index not found: {db_path}")
filters: list[str] = []
params: list[Any] = []
if not include_staged:
filters.append("status != 'staged'")
if status:
filters.append("status = ?")
params.append(status)
if doc_type:
filters.append("doc_type = ?")
params.append(doc_type)
if source_type:
filters.append("source_type = ?")
params.append(source_type)
where = f"WHERE {' AND '.join(filters)}" if filters else ""
rows: list[dict[str, Any]] = []
with connect(db_path, read_only=read_only) as conn:
for row in conn.execute(f"SELECT * FROM file_assets {where}", params).fetchall():
record = dict(row)
if product_line and product_line.lower() not in _searchable(record.get("product_lines")):
continue
if customer and customer.lower() not in _searchable(record.get("customers")):
continue
if competitor and competitor.lower() not in _searchable(record.get("competitors")):
continue
score = score_asset(record, query)
if score <= 0 and query.strip():
continue
record["score"] = score
record["absolute_path"] = absolute_asset_path(vault_path, record["archive_path"])
rows.append(record)
rows.sort(key=lambda item: (item["score"], item.get("updated_at") or ""), reverse=True)
return rows if limit is None else rows[:limit]
def get_asset_by_file_id(
vault_path: str | Path,
file_id: str,
*,
db_path: Path | None = None,
read_only: bool = False,
) -> dict[str, Any] | None:
db_path = db_path or db_path_for_vault(vault_path)
if not db_path.is_file():
raise FileNotFoundError(f"file asset index not found: {db_path}")
with connect(db_path, read_only=read_only) as conn:
row = conn.execute("SELECT * FROM file_assets WHERE file_id = ?", (file_id,)).fetchone()
if row is None:
return None
record = dict(row)
record["absolute_path"] = absolute_asset_path(vault_path, record["archive_path"])
return record
def score_asset(record: dict[str, Any], query: str) -> int:
query = query.strip()
if not query:
return 1
structured_terms = [
term.lower()
for field in STRUCTURED_IDENTIFIER_FIELDS
for term in _iter_string_values(record.get(field))
]
haystack = _searchable(
[
record.get("original_name"),
record.get("title"),
record.get("summary"),
record.get("archive_path"),
record.get("doc_type"),
record.get("source_type"),
record.get("business_categories"),
record.get("product_lines"),
record.get("customers"),
record.get("competitors"),
record.get("standards"),
record.get("market_tags"),
record.get("bjd_part_numbers"),
record.get("competitor_part_numbers"),
record.get("series"),
record.get("pitch"),
record.get("pin_count"),
record.get("extracted_text"),
]
)
q = query.lower()
tokens = [token.lower() for token in TOKEN_RE.findall(query)]
score = 0
if q in haystack:
score += 25
for token in tokens:
if token in haystack:
score += 7 if len(token) >= 4 else 3
if _structured_term_matches(token, structured_terms):
score += 30 if len(token) >= 4 else 8
if tokens and all(token in haystack for token in tokens):
score += 12
if score > 0 and record.get("doc_type") in {"drawing", "test_report", "product_parameter", "cross_reference"}:
score += 1
return score
def absolute_asset_path(vault_path: str | Path, archive_path: str) -> str:
archive = Path(archive_path)
if archive.is_absolute():
return str(archive)
vault = Path(clean_config_value(str(vault_path))).expanduser()
return str((vault / archive).resolve())
def sha256_file(path: Path) -> str:
hasher = hashlib.sha256()
with path.open("rb") as fh:
for chunk in iter(lambda: fh.read(1024 * 1024), b""):
hasher.update(chunk)
return hasher.hexdigest()
def safe_basename(name: str) -> str:
name = re.sub(r"[/\\:\x00-\x1f]", "_", name.strip())
return name or "unnamed"
def guess_source_type(name: str) -> str:
ext = Path(name).suffix.lower()
if ext == ".pdf":
return "pdf"
if ext in {".xlsx", ".xls", ".csv", ".tsv"}:
return "spreadsheet"
if ext in {".md", ".txt", ".doc", ".docx", ".ppt", ".pptx"}:
return "document"
if ext in {".png", ".jpg", ".jpeg", ".webp", ".gif", ".tif", ".tiff"}:
return "image"
if ext in {".step", ".stp", ".igs", ".iges", ".x_t", ".x_b", ".sldprt", ".sldasm", ".dwg", ".dxf"}:
return "cad"
if ext in {".json", ".jsonl", ".xml", ".yaml", ".yml"}:
return "data"
return "other"
def format_results_table(rows: list[dict[str, Any]]) -> str:
if not rows:
return "No matching file assets found."
lines = ["score type status file"]
lines.append("----- -------------- ------- ----")
for row in rows:
score = str(row.get("score", ""))
doc_type = _clip(str(row.get("doc_type") or row.get("source_type") or ""), 14)
status = _clip(str(row.get("status") or ""), 7)
file_name = row.get("original_name") or Path(str(row.get("archive_path") or "")).name
rel = row.get("archive_path") or ""
lines.append(f"{score:>5} {doc_type:<14} {status:<7} {file_name} ({rel})")
return "\n".join(lines)
def _clip(value: str, width: int) -> str:
return value[: width - 1] + "…" if len(value) > width else value
def _json_list(value: Any) -> str:
if value is None:
data: list[Any] = []
elif isinstance(value, list):
data = value
elif isinstance(value, tuple):
data = list(value)
else:
data = [value]
return json.dumps([item for item in data if item not in (None, "")], ensure_ascii=False)
def _merge_index_terms(*values: Any, uppercase: bool = False, limit: int | None = None) -> list[str]:
terms: list[str] = []
seen: set[str] = set()
for value in values:
for item in _iter_string_values(value):
term = item.strip()
if not term:
continue
if uppercase:
term = term.upper()
key = term.lower()
if key in seen:
continue
seen.add(key)
terms.append(term)
if limit is not None and len(terms) >= limit:
return terms
return terms
def _iter_string_values(value: Any) -> list[str]:
if value is None:
return []
if isinstance(value, str):
stripped = value.strip()
if not stripped:
return []
try:
parsed = json.loads(stripped)
except json.JSONDecodeError:
return [stripped]
if parsed == value:
return [stripped]
return _iter_string_values(parsed)
if isinstance(value, (list, tuple, set)):
items: list[str] = []
for item in value:
items.extend(_iter_string_values(item))
return items
if isinstance(value, dict):
items: list[str] = []
for item in value.values():
items.extend(_iter_string_values(item))
return items
return [str(value)]
def _structured_term_matches(token: str, terms: list[str]) -> bool:
token = token.lower()
return any(token == term or term.startswith(token) or token in term for term in terms)
def _flatten_to_strings(value: Any) -> str:
if value is None:
return ""
if isinstance(value, str):
return value
if isinstance(value, (list, tuple, set)):
return " ".join(_flatten_to_strings(item) for item in value)
if isinstance(value, dict):
return " ".join(_flatten_to_strings(item) for item in value.values())
return str(value)
def _searchable(value: Any) -> str:
if isinstance(value, str):
try:
parsed = json.loads(value)
except json.JSONDecodeError:
return value.lower()
return _flatten_to_strings(parsed).lower()
return _flatten_to_strings(value).lower()
def _float_or_none(value: Any) -> float | None:
try:
return None if value is None or value == "" else float(value)
except (TypeError, ValueError):
return None
def _int_or_none(value: Any) -> int | None:
try:
return None if value is None or value == "" else int(value)
except (TypeError, ValueError):
return None
def _mtime_iso(mtime: float) -> str:
return datetime.fromtimestamp(mtime, timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")