wild-cloud/wild-pc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Add new tools and configurations for Castle platform

Browse Source 
- Introduced `uv.lock` for dependency management with various packages including `pytest`, `colorama`, and `pluggy`.
- Added `pyrightconfig.json` for Python type checking configuration.
- Expanded `recommendations.md` with detailed scaling recommendations and project management strategies.
- Created shared `ruff.toml` for consistent linting across projects.
- Developed `Castle Tools` with various utilities including Android backup, browser automation, document conversion, and search tools.
- Implemented `backup-collect` and `schedule` tools for system administration tasks.
- Enhanced `search` functionality with indexing and querying capabilities using Tantivy.
- Added comprehensive documentation for each tool, including usage examples and installation instructions.
This commit is contained in:
Paul Payne
2026-02-20 16:41:19 -08:00
parent 0d35ac9ffd
commit f39a551aad
152 changed files with 21197 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
tools/search/pyproject.toml Normal file
View File

@@ -0,0 +1,22 @@
[project]
name = "castle-search"
version = "0.1.0"
description = "Castle search and indexing tools"
requires-python = ">=3.11"
dependencies = [
"toml>=0.10.2",
"pymupdf>=1.24.11",
]
[project.scripts]
search = "search.search:main"
pdf-extractor = "search.pdf_extractor:main"
docx-extractor = "search.docx_extractor:main"
text-extractor = "search.text_extractor:main"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["src/search"]

0
tools/search/src/search/__init__.py Normal file
View File

73
tools/search/src/search/docx_extractor.md Normal file
View File

@@ -0,0 +1,73 @@
---
command: docx-extractor
script: search/docx_extractor.py
description: Extract content and metadata from Word .docx files
version: 1.0.0
category: search
system_dependencies:
- pandoc
---
# docx-extractor
Extract text from Microsoft Word (.docx) documents.
## Installation
```bash
cd /data/repos/toolkit
make install
which docx-extractor
```
## Usage
```bash
# Extract from Word doc
docx-extractor document.docx
# Save to file
docx-extractor document.docx > text.txt
# Multiple documents
docx-extractor file1.docx file2.docx
```
### Options
```bash
docx-extractor [files...]
Options:
files DOCX files to extract text from
-h, --help Show help message
```
## Features
- Extract text from .docx files
- Handles headers, footers, and body text
- Preserves basic structure
- Fast extraction
## Examples
```bash
# Single document
docx-extractor report.docx
# Batch extraction
for file in *.docx; do
docx-extractor "$file" > "${file%.docx}.txt"
done
# Search extracted text
docx-extractor document.docx | grep "keyword"
```
## Notes
- Only supports .docx format (not .doc)
- Formatting is stripped
- Tables become plain text
- Images are not extracted

233
tools/search/src/search/docx_extractor.py Normal file
View File

@@ -0,0 +1,233 @@
#!/usr/bin/env python3
"""
docx-extractor: Extract content and metadata from Word .docx files
Reads a Word .docx file and outputs a JSON object containing the file's content
and metadata including filename, size, title, author, creation date, etc.
Usage: docx-extractor [options] [FILE]
Examples:
docx-extractor document.docx # Extract from a Word file
"""
import sys
import os
import json
import argparse
import subprocess
from datetime import datetime
import mimetypes
from pathlib import Path
import re
def get_file_metadata(file_path):
"""Get file metadata including creation time, modification time, size, etc."""
path = Path(file_path)
stat = path.stat()
# Get file modification and creation times
mtime = datetime.fromtimestamp(stat.st_mtime).isoformat()
try:
ctime = datetime.fromtimestamp(stat.st_ctime).isoformat()
except Exception:
ctime = mtime # Fallback if creation time not available
# Get mime type
mime_type, _ = mimetypes.guess_type(file_path)
if not mime_type:
mime_type = "application/vnd.openxmlformats-officedocument.wordprocessingml.document" # Default for .docx
return {
"filename": path.name,
"path": str(path.absolute()),
"size": stat.st_size,
"created_at": ctime,
"modified_at": mtime,
"mime_type": mime_type,
"extension": path.suffix.lstrip(".") if path.suffix else "docx",
}
def extract_docx_metadata(file_path):
"""Extract metadata from a DOCX file using pandoc."""
metadata = {}
try:
# Method 1: Try to extract metadata using pandoc's native metadata extraction
# Convert to JSON to capture metadata fields like title, author, date, etc.
result = subprocess.run(
["pandoc", "--standalone", "--to=json", file_path],
check=True,
capture_output=True,
text=True,
)
try:
# Parse the JSON output
data = json.loads(result.stdout)
# Extract metadata fields from the JSON structure
if "meta" in data:
meta = data["meta"]
# Extract title
if "title" in meta:
title_content = meta["title"].get("c", [])
if title_content:
# Title might be a complex structure, try to extract text
title_parts = []
for part in title_content:
if isinstance(part, dict) and "c" in part:
title_parts.append(part["c"])
elif isinstance(part, str):
title_parts.append(part)
if title_parts:
metadata["title"] = " ".join(title_parts)
# Extract author
if "author" in meta:
author_data = meta["author"]
if isinstance(author_data, list) and author_data:
if "c" in author_data[0]:
metadata["author"] = author_data[0]["c"]
else:
# Try to extract author from complex structures
authors = []
for author in author_data:
if isinstance(author, dict) and "c" in author:
authors.append(author["c"])
if authors:
metadata["author"] = ", ".join(authors)
# Extract date
if "date" in meta:
date_content = meta["date"]
if isinstance(date_content, dict) and "c" in date_content:
metadata["date"] = date_content["c"]
# Extract keywords/tags
if "keywords" in meta:
keywords_data = meta["keywords"]
if isinstance(keywords_data, dict) and "c" in keywords_data:
metadata["keywords"] = keywords_data["c"]
except json.JSONDecodeError:
pass
# Method 2: Extract title from first header if not found in metadata
if "title" not in metadata:
# Convert to markdown and look for the first header
result = subprocess.run(
["pandoc", "-f", "docx", "-t", "markdown", file_path],
check=True,
capture_output=True,
text=True,
)
content = result.stdout.strip()
title_match = re.search(r"^# (.+)$", content, re.MULTILINE)
if title_match:
metadata["title"] = title_match.group(1).strip()
return metadata
except subprocess.CalledProcessError:
# Fall back to basic metadata if pandoc extraction fails
return metadata
except FileNotFoundError:
sys.stderr.write("Warning: pandoc not found, metadata extraction limited\n")
return metadata
def extract_docx_content(file_path):
"""Extract content from a DOCX file using pandoc."""
try:
# Use pandoc to convert DOCX to plain text
result = subprocess.run(
["pandoc", "-f", "docx", "-t", "plain", file_path],
check=True,
capture_output=True,
text=True,
)
return result.stdout.strip()
except subprocess.CalledProcessError as e:
sys.stderr.write(f"Error extracting DOCX content: {e}\n")
if e.stderr:
sys.stderr.write(e.stderr)
return ""
except FileNotFoundError:
sys.stderr.write(
"Error: pandoc is not installed. Please install it with 'apt install pandoc'\n"
)
return ""
def process_file(file_path):
"""Extract content and metadata from the file."""
# Check if input file exists
if not file_path or not os.path.exists(file_path):
sys.stderr.write(f"Error: File '{file_path}' not found.\n")
return {}
# Get file metadata
metadata = get_file_metadata(file_path)
# Extract document metadata
docx_metadata = extract_docx_metadata(file_path)
# Extract content
content = extract_docx_content(file_path)
# Try to extract tags
tags = []
if "keywords" in docx_metadata:
tags = [tag.strip() for tag in docx_metadata["keywords"].split(",")]
# Determine title
title = docx_metadata.get("title", "")
if not title:
# Use filename without extension as fallback title
title = os.path.splitext(os.path.basename(file_path))[0]
# Create result object
result = {
**metadata,
**docx_metadata,
"content": content,
"title": title,
"tags": tags,
}
return result
def main():
parser = argparse.ArgumentParser(
description="Extract content and metadata from Word .docx files",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=__doc__.split("\n\n", 1)[1]
if __doc__
else "", # Use the docstring as extended help
)
parser.add_argument("file", help="Input .docx file")
parser.add_argument(
"-v", "--version", action="version", version="docx-extractor 1.0.0"
)
args = parser.parse_args()
try:
# Process the file
result = process_file(args.file)
# Output the result as JSON
print(json.dumps(result, indent=2))
return 0
except Exception as e:
sys.stderr.write(f"Error: {e}\n")
return 1
if __name__ == "__main__":
sys.exit(main())

70
tools/search/src/search/pdf_extractor.md Normal file
View File

@@ -0,0 +1,70 @@
---
command: pdf-extractor
script: search/pdf_extractor.py
description: Extract content and metadata from PDF files
version: 1.0.0
category: search
---
# pdf-extractor
Extract text from PDF documents.
## Installation
```bash
cd /data/repos/toolkit
make install
which pdf-extractor
```
## Usage
```bash
# Extract from PDF
pdf-extractor document.pdf
# Save to file
pdf-extractor document.pdf > text.txt
# Multiple PDFs
pdf-extractor file1.pdf file2.pdf
```
### Options
```bash
pdf-extractor [files...]
Options:
files PDF files to extract text from
-h, --help Show help message
```
## Features
- Extract text from PDFs
- Handles multi-page documents
- Preserves text layout where possible
- Fast extraction
## Examples
```bash
# Single PDF
pdf-extractor report.pdf
# Batch extraction
for file in *.pdf; do
pdf-extractor "$file" > "${file%.pdf}.txt"
done
# Search extracted text
pdf-extractor document.pdf | grep "keyword"
```
## Notes
- Works with text-based PDFs
- Scanned PDFs (images) require OCR
- Complex layouts may affect text order

168
tools/search/src/search/pdf_extractor.py Executable file
View File

@@ -0,0 +1,168 @@
#!/usr/bin/env python3
"""
pdf-extractor: Extract content and metadata from PDF files
Reads a PDF file and outputs a JSON object containing the file's content
and metadata including filename, size, page count, title, author, etc.
Usage: pdf-extractor [options] [FILE]
Examples:
pdf-extractor document.pdf # Extract from a PDF file
cat document.pdf | pdf-extractor # Read from stdin (if applicable)
"""
import sys
import os
import json
import argparse
import fitz # PyMuPDF
from datetime import datetime
import mimetypes
from pathlib import Path
import tempfile
def get_file_metadata(file_path):
"""Get file metadata including creation time, modification time, size, etc."""
path = Path(file_path)
stat = path.stat()
# Get file modification and creation times
mtime = datetime.fromtimestamp(stat.st_mtime).isoformat()
try:
ctime = datetime.fromtimestamp(stat.st_ctime).isoformat()
except Exception:
ctime = mtime # Fallback if creation time not available
# Get mime type
mime_type, _ = mimetypes.guess_type(file_path)
if not mime_type:
mime_type = "application/pdf" # Default to PDF
return {
"filename": path.name,
"path": str(path.absolute()),
"size": stat.st_size,
"created_at": ctime,
"modified_at": mtime,
"mime_type": mime_type,
"extension": path.suffix.lstrip(".") if path.suffix else "pdf",
}
def extract_pdf_content(file_path):
"""Extract content and metadata from a PDF file."""
try:
doc = fitz.open(file_path)
# Extract PDF metadata
metadata = doc.metadata if doc.metadata else {}
pdf_metadata = {
"page_count": doc.page_count,
"title": metadata.get("title", ""),
"author": metadata.get("author", ""),
"subject": metadata.get("subject", ""),
"keywords": metadata.get("keywords", ""),
"creator": metadata.get("creator", ""),
"producer": metadata.get("producer", ""),
"creation_date": metadata.get("creationDate", ""),
"modification_date": metadata.get("modDate", ""),
}
# Extract text content from all pages
content = ""
for page_num in range(doc.page_count):
page = doc[page_num]
content += page.get_text() # type: ignore
content += "\n\n" # Add spacing between pages
doc.close()
return content, pdf_metadata
except Exception as e:
sys.stderr.write(f"Error extracting PDF content: {e}\n")
return "", {}
def process_file(file_path):
"""Extract content and metadata from the file."""
# Check if we're dealing with stdin or a file
temp_file = None
try:
if not file_path or file_path == "-":
# Reading from stdin, save to temporary file
temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".pdf")
temp_file.write(sys.stdin.buffer.read())
temp_file.close()
file_path = temp_file.name
# Create basic metadata for stdin input
metadata = {
"filename": "stdin",
"path": "stdin",
"size": os.path.getsize(file_path),
"created_at": datetime.now().isoformat(),
"modified_at": datetime.now().isoformat(),
"mime_type": "application/pdf",
"extension": "pdf",
}
else:
# Reading from a file
if not os.path.exists(file_path):
sys.stderr.write(f"Error: File '{file_path}' not found.\n")
return {}
# Get file metadata
metadata = get_file_metadata(file_path)
# Extract PDF content and PDF-specific metadata
content, pdf_metadata = extract_pdf_content(file_path)
# Create result object
result = {
**metadata,
**pdf_metadata,
"content": content,
"title": pdf_metadata.get("title") or metadata["filename"],
"tags": [tag.strip() for tag in pdf_metadata.get("keywords", "").split(",")]
if pdf_metadata.get("keywords")
else [],
}
return result
finally:
# Clean up the temporary file if one was created
if temp_file and os.path.exists(temp_file.name):
os.unlink(temp_file.name)
def main():
parser = argparse.ArgumentParser(
description="Extract content and metadata from PDF files",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=__doc__.split("\n\n", 1)[1]
if __doc__
else "", # Use the docstring as extended help
)
parser.add_argument("file", nargs="?", help="Input file (default: stdin)")
parser.add_argument(
"-v", "--version", action="version", version="pdf-extractor 1.0.0"
)
args = parser.parse_args()
try:
# Process the file
result = process_file(args.file)
# Output the result as JSON
print(json.dumps(result, indent=2))
return 0
except Exception as e:
sys.stderr.write(f"Error: {e}\n")
return 1
if __name__ == "__main__":
sys.exit(main())

123
tools/search/src/search/search.md Normal file
View File

@@ -0,0 +1,123 @@
---
command: search
script: search/search.py
description: Manage self-contained searchable collections of files
version: 1.0.0
category: search
---
# search
Fast full-text search using Tantivy search index. Index and search documents with blazing speed.
## Installation
```bash
cd /data/repos/toolkit
make install
which search
```
## Usage
### Index Documents
```bash
# Index a directory
search index /path/to/documents
# Index specific files
search index file1.txt file2.pdf file3.docx
```
### Search
```bash
# Simple search
search query "search terms"
# Search with field
search query --field title "document title"
# Limit results
search query --limit 10 "search terms"
```
### Options
```bash
search index [paths...] Index files or directories
search query [options] QUERY Search indexed documents
Query options:
QUERY Search query (required)
--field FIELD Search specific field (title, body, path)
--limit N Maximum results to return (default: 20)
-h, --help Show help message
```
## Features
- **Fast searching** - Powered by Tantivy search engine
- **Multiple formats** - Supports PDF, DOCX, TXT, and more
- **Full-text search** - Search entire document contents
- **Field search** - Search specific fields
- **Relevance ranking** - Results sorted by relevance
## Examples
### Index and Search
```bash
# Index your documents
search index ~/Documents
# Search for terms
search query "important meeting notes"
search query "project proposal 2024"
```
### Field-Specific Search
```bash
# Search in title only
search query --field title "Annual Report"
# Search in body
search query --field body "quarterly results"
```
### Batch Operations
```bash
# Index multiple directories
search index ~/Documents ~/Downloads ~/Desktop
# Incremental indexing
search index ~/Documents/new_files
```
## Supported File Types
- **Text files** - .txt, .md, .log
- **PDF documents** - .pdf
- **Word documents** - .docx
- **Other formats** - via text extractors
## Index Location
Index is stored in `~/.local/share/search/index` by default.
## Tips
- Index documents once, search many times
- Re-index when documents change
- Use quotes for exact phrases
- Field searches are faster for specific needs
## Notes
- Initial indexing may take time for large collections
- Index size depends on document collection size
- Supports incremental indexing
- UTF-8 text encoding recommended

832
tools/search/src/search/search.py Executable file
View File

@@ -0,0 +1,832 @@
#!/usr/bin/env python3
"""
search: Manage self-contained searchable collections of files
A modular, low-resource, local search system for files (PDFs, images, videos, etc.)
where each collection is self-contained and can be indexed and queried using Tantivy.
Usage:
search init [options] [directory] Initialize a directory as a search collection
search scan [options] [directory] Find all search collections in a directory tree
search index [options] [directory...] Index all matched files in one or more collections
search query [options] QUERY Search indexed collections using Tantivy
Examples:
search init ~/documents # Initialize a collection
search init --name "Research" ~/papers # Initialize with a specific name
search init --extract "*.pdf=pdf2md {input}" # Add a specific extractor
search scan ~/documents # Find collections
search index ~/documents # Index a single collection
search index ~/documents ~/papers # Index multiple collections
search index --verbose ~/documents # Show detailed indexing progress
search query "neural networks" # Search across all nearby collections
search query "DNA" --in ~/papers # Search in a specific collection
search query "protein" --tag biology # Search with tag filter
search query "machine learning" --limit 10 # Limit number of results
For more detailed help on a specific subcommand:
search init --help
search scan --help
search index --help
search query --help
"""
import argparse
import fnmatch
import json
import os
import re
import subprocess
import sys
from pathlib import Path
from typing import Any, Dict, List, Optional
import toml
# Try to import Tantivy, but don't fail immediately if not available
# This allows --help and other non-search operations to work without Tantivy
tantivy_available = False
try:
import tantivy # type: ignore
tantivy_available = True
except ImportError:
tantivy = None # type: ignore
def check_tantivy():
"""Check if Tantivy is available and exit if not."""
if not tantivy_available:
print(
"Error: Required Tantivy Python bindings not found. Please install them with:"
)
print(" pip install tantivy")
sys.exit(1)
# Constants
SEARCH_DIR = ".search"
CONFIG_FILE = f"{SEARCH_DIR}/config.toml"
CACHE_DIR = f"{SEARCH_DIR}/cache"
INDEX_DIR = f"{SEARCH_DIR}/index"
# Default extractor commands for common file types
DEFAULT_EXTRACTORS = {
"*.md": "text-extractor {input}",
"*.txt": "text-extractor {input}",
"*.pdf": "pdf-extractor {input}",
"*.docx": "docx-extractor {input}",
}
# Default configuration
DEFAULT_CONFIG = {
"name": "Default Collection",
"include": {"patterns": ["*.pdf", "*.md", "*.txt", "*.docx"]},
"exclude": {"patterns": ["*~", "*.bak", "*.tmp", ".git/*", ".search/*"]},
"extractors": {}, # Only define custom extractors here, defaults are used automatically
"output": {"format": "json", "directory": "cache"},
}
def ensure_dir(path: str) -> None:
"""Ensure a directory exists."""
Path(path).mkdir(parents=True, exist_ok=True)
def find_collections(start_dir: str) -> List[str]:
"""Find all search collections starting from a directory."""
collections = []
start_path = Path(start_dir).resolve()
for path in start_path.glob(f"**/{SEARCH_DIR}"):
config_file = path / "config.toml"
if config_file.exists():
collections.append(str(path.parent))
return collections
def load_config(collection_dir: str) -> Dict[str, Any]:
"""Load collection configuration."""
config_path = os.path.join(collection_dir, CONFIG_FILE)
try:
with open(config_path, "r") as f:
return toml.load(f)
except (FileNotFoundError, toml.TomlDecodeError) as e:
sys.stderr.write(f"Error loading config: {e}\n")
return {}
def save_config(collection_dir: str, config: Dict[str, Any]) -> bool:
"""Save collection configuration."""
config_path = os.path.join(collection_dir, CONFIG_FILE)
try:
# Ensure the .search directory exists
ensure_dir(os.path.join(collection_dir, SEARCH_DIR))
# Add a header comment to explain the extractors section
header = """# Search Collection Configuration
#
# Built-in extractors are automatically used for common file types:
# *.txt, *.md: text-extractor {input}
# *.pdf: pdf-extractor {input}
# *.docx: docx-extractor {input}
#
# The [extractors] section below only needs to define custom extractors
# or overrides for specific file types.
#
"""
# Format the config as TOML
config_toml = toml.dumps(config)
# Write the file with header
with open(config_path, "w") as f:
f.write(header)
f.write(config_toml)
return True
except Exception as e:
sys.stderr.write(f"Error saving config: {e}\n")
return False
def filter_files(collection_dir: str, config: Dict[str, Any]) -> List[str]:
"""Filter files based on include/exclude patterns."""
include_patterns = config.get("include", {}).get("patterns", ["*"])
exclude_patterns = config.get("exclude", {}).get("patterns", [])
matched_files = []
# Walk the directory tree
for root, dirs, files in os.walk(collection_dir):
# Skip the .search directory
if os.path.basename(root) == SEARCH_DIR:
continue
# Check if any exclude pattern matches directories
skip_dir = False
for pattern in exclude_patterns:
if any(fnmatch.fnmatch(d, pattern) for d in dirs):
skip_dir = True
break
if skip_dir:
continue
# Match files against include/exclude patterns
for filename in files:
file_path = os.path.join(root, filename)
# Skip if the file is in the .search directory
if SEARCH_DIR in file_path.split(os.sep):
continue
# Check include patterns
if not any(
fnmatch.fnmatch(filename, pattern) for pattern in include_patterns
):
continue
# Check exclude patterns
if any(fnmatch.fnmatch(filename, pattern) for pattern in exclude_patterns):
continue
matched_files.append(file_path)
return matched_files
def get_extractor(filename: str, config: Dict[str, Any]) -> Optional[str]:
"""Find the appropriate extractor command for a file."""
# First check custom extractors defined in config
custom_extractors = config.get("extractors", {})
# Try to match custom extractors first (allowing overrides)
for pattern, command in custom_extractors.items():
if fnmatch.fnmatch(filename, pattern):
return command
# If no custom extractor matched, use the default extractors
for pattern, command in DEFAULT_EXTRACTORS.items():
if fnmatch.fnmatch(filename, pattern):
return command
return None
def extract_metadata(file_path: str, extractor_cmd: str) -> Dict[str, Any]:
"""Execute extractor and parse output as JSON."""
# Replace {input} with the properly quoted file path
# First, escape any single quotes in the path
escaped_path = file_path.replace("'", "'\\''")
# Then wrap in single quotes to handle spaces and special characters
quoted_path = f"'{escaped_path}'"
cmd = extractor_cmd.replace("{input}", quoted_path)
try:
# Run the extractor command
result = subprocess.run(
cmd, shell=True, capture_output=True, text=True, check=True
)
output = result.stdout
# Attempt to parse as JSON
try:
return json.loads(output)
except json.JSONDecodeError:
# If parsing fails, create basic metadata with content
return {"title": os.path.basename(file_path), "content": output}
except subprocess.CalledProcessError as e:
sys.stderr.write(f"Error executing extractor for {file_path}: {e}\n")
return {}
def save_cache(
collection_dir: str,
file_path: str,
metadata: Dict[str, Any],
config: Dict[str, Any],
) -> bool:
"""Save metadata to cache."""
# Ensure the cache directory exists
cache_dir = os.path.join(collection_dir, CACHE_DIR)
ensure_dir(cache_dir)
# Create a cache file name based on the original file path
rel_path = os.path.relpath(file_path, collection_dir)
cache_filename = re.sub(r"[/\\]", "_", rel_path)
# Save as JSON
cache_path = os.path.join(cache_dir, f"{cache_filename}.json")
try:
with open(cache_path, "w") as f:
json.dump(metadata, f, indent=2)
return True
except Exception as e:
sys.stderr.write(f"Error saving cache file {cache_path}: {e}\n")
return False
def create_schema():
"""Create Tantivy schema for document indexing."""
schema_builder = tantivy.SchemaBuilder() # type: ignore
# Define fields for the schema
schema_builder.add_text_field("path", stored=True)
schema_builder.add_text_field("title", stored=True)
schema_builder.add_text_field("content", stored=True)
schema_builder.add_text_field("absolute_path", stored=True)
# Using text fields for tags instead of facets
schema_builder.add_text_field("tag_pdf", stored=True)
schema_builder.add_text_field("tag_test", stored=True)
schema_builder.add_text_field("tag_sample", stored=True)
# Add more common tags as needed
schema_builder.add_text_field(
"metadata", stored=True
) # For storing additional JSON metadata
return schema_builder.build()
def get_or_create_index(collection_dir: str) -> tuple:
"""Get or create a Tantivy index for a collection."""
# Check if Tantivy is available
check_tantivy()
index_dir = os.path.join(collection_dir, INDEX_DIR)
ensure_dir(index_dir)
# For reference only - we don't actually read this
schema_path = os.path.join(index_dir, "schema_info.json")
# Check if index exists
if os.path.exists(os.path.join(index_dir, "meta.json")):
# Try to open existing index
try:
index = tantivy.Index.open(index_dir) # type: ignore
return index, None # We don't need the schema for opened indexes
except Exception as e:
# If opening failed, create a new one
sys.stderr.write(
f"Warning: Could not open existing index, creating new one: {e}\n"
)
# Create new schema and index
schema = create_schema()
# Create basic schema info for reference
schema_info = {
"fields": [
{"name": "path", "type": "text", "stored": True},
{"name": "title", "type": "text", "stored": True},
{"name": "content", "type": "text", "stored": True},
{"name": "absolute_path", "type": "text", "stored": True},
{"name": "tag_pdf", "type": "text", "stored": True},
{"name": "tag_test", "type": "text", "stored": True},
{"name": "tag_sample", "type": "text", "stored": True},
{"name": "metadata", "type": "text", "stored": True},
]
}
# Save schema info for reference
with open(schema_path, "w") as f:
json.dump(schema_info, f, indent=2)
# Create the index
index = tantivy.Index(schema, path=index_dir) # type: ignore
return index, schema
def index_file(collection_dir: str, file_path: str, metadata: Dict[str, Any]) -> bool:
"""Index a file using Tantivy."""
try:
# Get or create index
index, _schema = get_or_create_index(collection_dir)
# Create a writer for adding documents
writer = index.writer()
# Create document
doc = tantivy.Document() # type: ignore
# Get relative path as identifier
rel_path = os.path.relpath(file_path, collection_dir)
# Add fields to document
doc.add_text("path", rel_path)
doc.add_text("absolute_path", os.path.abspath(file_path))
# Add title if available
if "title" in metadata and metadata["title"]:
doc.add_text("title", metadata["title"])
else:
doc.add_text("title", os.path.basename(file_path))
# Add content if available
if "content" in metadata and metadata["content"]:
doc.add_text("content", metadata["content"])
# Add tags as facets if available
if "tags" in metadata and isinstance(metadata["tags"], list):
for tag in metadata["tags"]:
if tag and isinstance(tag, str):
# Add as text since facets are complicated in Tantivy
tag_field = f"tag_{tag.lower().replace(' ', '_').replace(',', '').replace('-', '_')}"
doc.add_text(tag_field, "true")
# Store all metadata as JSON
doc.add_text(
"metadata",
json.dumps(
{
"path": rel_path,
"absolute_path": os.path.abspath(file_path),
**metadata,
}
),
)
# Add document to index
writer.add_document(doc)
# Commit changes
writer.commit()
return True
except Exception as e:
sys.stderr.write(f"Error indexing file {file_path}: {e}\n")
return False
def perform_search(
query_string: str,
collections: List[str],
tag_filter: Optional[str] = None,
limit: int = 20,
) -> List[Dict[str, Any]]:
"""Search across collections using Tantivy."""
# Check if Tantivy is available
check_tantivy()
results = []
# Process each collection
for collection_dir in collections:
index_dir = os.path.join(collection_dir, INDEX_DIR)
if not os.path.exists(index_dir) or not os.path.exists(
os.path.join(index_dir, "meta.json")
):
continue
try:
# Open the index
index = tantivy.Index.open(index_dir) # type: ignore
searcher = index.searcher()
# Parse the query using index's parse_query method
query = index.parse_query(query_string, ["title", "content"])
# Perform the search
search_result = searcher.search(query, limit)
# Extract results - access the hits property which contains doc_address
for hit in search_result.hits:
# Unpack score and doc_address from the hit
score, doc_address = hit
retrieved_doc = searcher.doc(doc_address)
# Create basic metadata
title = ""
path = ""
abs_path = ""
tags = []
score_value = score * 100 # Convert to percentage
# Extract fields using get_first
title_field = retrieved_doc.get_first("title")
if title_field:
title = title_field
path_field = retrieved_doc.get_first("path")
if path_field:
path = path_field
abs_path_field = retrieved_doc.get_first("absolute_path")
if abs_path_field:
abs_path = abs_path_field
metadata_field = retrieved_doc.get_first("metadata")
if metadata_field:
try:
metadata = json.loads(metadata_field)
# If we have tags in metadata, extract them
if "tags" in metadata:
tags = metadata["tags"]
except Exception:
pass
# Skip documents that don't match tag filter
if tag_filter and tag_filter.lower() not in [t.lower() for t in tags]:
continue
# Create result object
data = {
"title": title or os.path.basename(path),
"path": path,
"absolute_path": abs_path,
"tags": tags,
"score": score_value,
"collection": os.path.basename(collection_dir),
"collection_path": collection_dir,
}
results.append(data)
except Exception as e:
sys.stderr.write(f"Error searching in collection {collection_dir}: {e}\n")
# Sort results by score (highest first)
results.sort(key=lambda x: x.get("score", 0), reverse=True)
return results[:limit]
def cmd_init(args: argparse.Namespace) -> int:
"""Initialize a directory as a search collection."""
directory = os.path.abspath(args.directory or os.getcwd())
# Check if the directory exists
if not os.path.isdir(directory):
sys.stderr.write(f"Error: Directory {directory} does not exist.\n")
return 1
# Check if the collection already exists
config_path = os.path.join(directory, CONFIG_FILE)
if os.path.exists(config_path):
if not args.force:
sys.stderr.write(
f"Error: Collection already exists at {directory}. Use --force to overwrite.\n"
)
return 1
# Create the basic configuration
config = DEFAULT_CONFIG.copy()
# Update configuration based on command-line arguments
if args.name:
config["name"] = args.name
if args.include:
config["include"]["patterns"] = args.include
if args.exclude:
config["exclude"]["patterns"] = args.exclude
# Add custom extractors if specified
if args.extract:
extractors = config["extractors"].copy()
for extractor in args.extract:
if "=" in extractor:
pattern, command = extractor.split("=", 1)
extractors[pattern.strip()] = command.strip()
config["extractors"] = extractors
# Create the directory structure
ensure_dir(os.path.join(directory, SEARCH_DIR))
ensure_dir(os.path.join(directory, CACHE_DIR))
ensure_dir(os.path.join(directory, INDEX_DIR))
# Save the configuration
if save_config(directory, config):
print(f"Initialized search collection in {directory}")
print(f"Configuration saved to {config_path}")
return 0
else:
return 1
def cmd_scan(args: argparse.Namespace) -> int:
"""Scan for collections."""
directory = os.path.abspath(args.directory or os.getcwd())
if not os.path.isdir(directory):
sys.stderr.write(f"Error: Directory {directory} does not exist.\n")
return 1
collections = find_collections(directory)
if not collections:
print(f"No search collections found in {directory}")
return 0
print(f"Found {len(collections)} search collection(s):")
for collection in collections:
config = load_config(collection)
name = config.get("name", "Unnamed collection")
print(f" {name}: {collection}")
return 0
def cmd_index(args: argparse.Namespace) -> int:
"""Index files in collections."""
directories = (
[os.path.abspath(d) for d in args.directories]
if args.directories
else [os.getcwd()]
)
collections = []
for directory in directories:
# Check if the directory is a collection
if os.path.exists(os.path.join(directory, CONFIG_FILE)):
collections.append(directory)
else:
# Scan for collections in this directory
found_collections = find_collections(directory)
collections.extend(found_collections)
if not collections:
sys.stderr.write(
"Error: No search collections found in the specified directories.\n"
)
return 1
indexed_count = 0
failed_count = 0
for collection_dir in collections:
config = load_config(collection_dir)
if not config:
sys.stderr.write(
f"Warning: Skipping collection {collection_dir} due to missing or invalid configuration.\n"
)
continue
collection_name = config.get("name", os.path.basename(collection_dir))
print(f"Indexing collection: {collection_name} ({collection_dir})")
# Get matched files
matched_files = filter_files(collection_dir, config)
print(f"Found {len(matched_files)} matching files")
for file_path in matched_files:
rel_path = os.path.relpath(file_path, collection_dir)
try:
# Get the appropriate extractor
extractor = get_extractor(os.path.basename(file_path), config)
if not extractor:
if args.verbose:
print(f" Skipping {rel_path}: No extractor found")
continue
if args.verbose:
print(f" Extracting {rel_path}")
# Extract metadata
metadata = extract_metadata(file_path, extractor)
if not metadata:
if args.verbose:
print(f" Failed to extract metadata from {rel_path}")
failed_count += 1
continue
# Save to cache if caching is enabled
if not args.no_cache:
save_cache(collection_dir, file_path, metadata, config)
# Index the file
if index_file(collection_dir, file_path, metadata):
indexed_count += 1
if args.verbose:
print(f" Successfully indexed {rel_path}")
else:
failed_count += 1
if args.verbose:
print(f" Failed to index {rel_path}")
except Exception as e:
sys.stderr.write(f"Error processing {rel_path}: {e}\n")
failed_count += 1
print(
f"Finished indexing {indexed_count} files across {len(collections)} collections."
)
if failed_count > 0:
print(f"Failed to index {failed_count} files.")
return 0 if failed_count == 0 else 1
def cmd_query(args: argparse.Namespace) -> int:
"""Search collections."""
# Check if Tantivy is available
check_tantivy()
# Determine which collections to search
collections = []
if args.in_dir:
directory = os.path.abspath(args.in_dir)
if os.path.exists(os.path.join(directory, CONFIG_FILE)):
collections.append(directory)
else:
sys.stderr.write(f"Error: {directory} is not a valid search collection.\n")
return 1
else:
# Start from current directory, search up to find collections
dir_path = os.path.abspath(os.getcwd())
while dir_path and dir_path != "/":
if os.path.exists(os.path.join(dir_path, CONFIG_FILE)):
collections.append(dir_path)
break
dir_path = os.path.dirname(dir_path)
# If no collection is found, scan for collections in current directory
if not collections:
collections = find_collections(os.getcwd())
if not collections:
sys.stderr.write(
"Error: No search collections found. Use --in to specify a collection.\n"
)
return 1
# Perform the search
results = perform_search(
query_string=args.query,
collections=collections,
tag_filter=args.tag,
limit=args.limit,
)
if not results:
print(f"No results found for '{args.query}'")
return 0
print(f"Found {len(results)} results for '{args.query}':")
for i, result in enumerate(results, 1):
title = result.get("title", os.path.basename(result.get("path", "Unknown")))
path = result.get("absolute_path", "Unknown path")
score = result.get("score", 0)
collection = result.get("collection", "Unknown collection")
print(f"\n{i}. {title} ({score:.0f}% match) - {collection}")
print(f" Path: {path}")
# Show tags if available
tags = result.get("tags", [])
if tags:
print(f" Tags: {', '.join(tags)}")
# Show a snippet of content if available and detailed output is requested
if args.verbose and "content" in result:
content = result["content"]
if content:
# Truncate and sanitize for display
content = re.sub(r"\s+", " ", content)
snippet = content[:200] + ("..." if len(content) > 200 else "")
print(f" Snippet: {snippet}")
return 0
def main() -> int:
"""Main entry point."""
parser = argparse.ArgumentParser(
description="Manage searchable collections of files",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=__doc__.split("\n\n", 1)[1]
if __doc__
else ""
if __doc__
else "", # Use the docstring as extended help
)
parser.add_argument("--version", action="version", version="search 1.0.0")
# Create subparsers for commands
subparsers = parser.add_subparsers(dest="command", help="Command to run")
# init command
init_parser = subparsers.add_parser(
"init", help="Initialize a directory as a search collection"
)
init_parser.add_argument(
"directory",
nargs="?",
help="Directory to initialize (default: current directory)",
)
init_parser.add_argument("--name", help="Name for the collection")
init_parser.add_argument(
"--include", nargs="+", help='Include patterns (e.g., "*.pdf" "*.md")'
)
init_parser.add_argument(
"--exclude", nargs="+", help='Exclude patterns (e.g., "*.bak" "draft_*")'
)
init_parser.add_argument(
"--extract", nargs="+", help='Extractor patterns (e.g., "*.pdf=pdf2md {input}")'
)
init_parser.add_argument(
"--force",
action="store_true",
help="Force overwrite if collection already exists",
)
# scan command
scan_parser = subparsers.add_parser(
"scan", help="Find all search collections in a directory tree"
)
scan_parser.add_argument(
"directory", nargs="?", help="Directory to scan (default: current directory)"
)
# index command
index_parser = subparsers.add_parser(
"index", help="Index all matched files in one or more collections"
)
index_parser.add_argument(
"directories",
nargs="*",
help="Directories to index (default: current directory)",
)
index_parser.add_argument(
"--verbose", action="store_true", help="Show detailed progress"
)
index_parser.add_argument(
"--no-cache", action="store_true", help="Skip caching extracted metadata"
)
# query command
query_parser = subparsers.add_parser(
"query", help="Search indexed collections using Tantivy"
)
query_parser.add_argument("query", help="Search query")
query_parser.add_argument(
"--in", dest="in_dir", help="Search in specific collection"
)
query_parser.add_argument("--tag", help="Filter by tag")
query_parser.add_argument(
"--limit", type=int, default=20, help="Maximum number of results (default: 20)"
)
query_parser.add_argument(
"--verbose", action="store_true", help="Show more detailed results"
)
# Parse arguments
args = parser.parse_args()
# Execute the appropriate command
if args.command == "init":
return cmd_init(args)
elif args.command == "scan":
return cmd_scan(args)
elif args.command == "index":
return cmd_index(args)
elif args.command == "query":
return cmd_query(args)
else:
parser.print_help()
return 1
if __name__ == "__main__":
sys.exit(main())

71
tools/search/src/search/text_extractor.md Normal file
View File

@@ -0,0 +1,71 @@
---
command: text-extractor
script: search/text_extractor.py
description: Extract content and metadata from text files
version: 1.0.0
category: search
---
# text-extractor
Extract plain text from various document formats.
## Installation
```bash
cd /data/repos/toolkit
make install
which text-extractor
```
## Usage
```bash
# Extract from file
text-extractor document.txt
# Extract to file
text-extractor document.txt > output.txt
# Process multiple files
text-extractor file1.txt file2.txt file3.txt
```
### Options
```bash
text-extractor [files...]
Options:
files Files to extract text from
-h, --help Show help message
```
## Supported Formats
- Plain text (.txt)
- Markdown (.md)
- Log files (.log)
- Other text-based formats
## Features
- Fast text extraction
- Handles multiple files
- Preserves text encoding
- Pipe-friendly output
## Examples
```bash
# Extract single file
text-extractor document.txt
# Batch processing
for file in *.txt; do
text-extractor "$file" > "${file%.txt}_extracted.txt"
done
# Use in pipeline
text-extractor doc.txt | grep "keyword"
```

118
tools/search/src/search/text_extractor.py Executable file
View File

@@ -0,0 +1,118 @@
#!/usr/bin/env python3
"""
text-extractor: Extract content and metadata from text files
Reads a text file and outputs a JSON object containing the file's content
and metadata including filename, size, creation and modification times.
Usage: text-extractor [options] [FILE]
Examples:
text-extractor document.txt # Extract from a text file
cat file.txt | text-extractor # Read from stdin
"""
import sys
import os
import json
import argparse
from datetime import datetime
import mimetypes
from pathlib import Path
def get_file_metadata(file_path):
"""Get file metadata including creation time, modification time, size, etc."""
path = Path(file_path)
stat = path.stat()
# Get file modification and creation times
mtime = datetime.fromtimestamp(stat.st_mtime).isoformat()
try:
ctime = datetime.fromtimestamp(stat.st_ctime).isoformat()
except Exception:
ctime = mtime # Fallback if creation time not available
# Get mime type
mime_type, _ = mimetypes.guess_type(file_path)
if not mime_type:
mime_type = "text/plain" # Default to plain text
return {
"filename": path.name,
"path": str(path.absolute()),
"size": stat.st_size,
"created_at": ctime,
"modified_at": mtime,
"mime_type": mime_type,
"extension": path.suffix.lstrip(".") if path.suffix else "",
}
def process_file(file_path):
"""Extract content and metadata from the file."""
# Get file metadata
if file_path and os.path.exists(file_path):
metadata = get_file_metadata(file_path)
# Read file content
try:
with open(file_path, "r", encoding="utf-8") as f:
content = f.read()
except UnicodeDecodeError:
# Try again with latin-1 encoding
with open(file_path, "r", encoding="latin-1") as f:
content = f.read()
else:
# Reading from stdin
content = sys.stdin.read()
metadata = {
"filename": "stdin",
"path": "stdin",
"size": len(content),
"created_at": datetime.now().isoformat(),
"modified_at": datetime.now().isoformat(),
"mime_type": "text/plain",
"extension": "",
}
# Create result object
result = {
**metadata,
"content": content,
"title": metadata["filename"],
"tags": [], # No tags by default
}
return result
def main():
parser = argparse.ArgumentParser(
description="Extract content and metadata from text files",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=__doc__.split("\n\n", 1)[1]
if __doc__
else "", # Use the docstring as extended help
)
parser.add_argument("file", nargs="?", help="Input file (default: stdin)")
parser.add_argument(
"-v", "--version", action="version", version="text-extractor 1.0.0"
)
args = parser.parse_args()
try:
# Process the file
result = process_file(args.file)
# Output the result as JSON
print(json.dumps(result, indent=2))
return 0
except Exception as e:
sys.stderr.write(f"Error: {e}\n")
return 1
if __name__ == "__main__":
sys.exit(main())