chore: import upstream snapshot with attribution
Ruff / Ruff (push) Has been cancelled
Test / Core Tests (push) Has been cancelled
Test / Offline Coverage Tests (Python 3.10) (push) Has been cancelled
Test / Offline Coverage Tests (Python 3.11) (push) Has been cancelled
Test / Offline Coverage Tests (Python 3.12) (push) Has been cancelled
Test / Offline Coverage Tests (Python 3.13) (push) Has been cancelled
Test / Offline Coverage Tests (Python 3.9) (push) Has been cancelled
Test / Full Coverage (Python 3.11) (push) Has been cancelled
Test / Core Provider Tests (OpenAI) (push) Has been cancelled
Test / Core Provider Tests (Anthropic) (push) Has been cancelled
Test / Core Provider Tests (Google) (push) Has been cancelled
Test / Core Provider Tests (Other) (push) Has been cancelled
Test / Anthropic Tests (push) Has been cancelled
Test / Gemini Tests (push) Has been cancelled
Test / Google GenAI Tests (push) Has been cancelled
Test / Vertex AI Tests (push) Has been cancelled
Test / OpenAI Tests (push) Has been cancelled
Test / Writer Tests (push) Has been cancelled
Test / Auto Client Tests (push) Has been cancelled
ty / type-check (push) Has been cancelled

This commit is contained in:
wehub-resource-sync
2026-07-13 13:36:38 +08:00
commit 97e91a83f3
978 changed files with 159975 additions and 0 deletions
+238
View File
@@ -0,0 +1,238 @@
# Scripts Directory
This directory contains utility scripts for maintaining and improving the Instructor documentation and project structure.
## Available Scripts
### 1. `make_clean.py` - Markdown File Cleaner
**Purpose**: Cleans markdown files by removing special whitespace characters and replacing em dashes with regular dashes.
**What it does**:
- Recursively finds all `.md` files in the `docs/` directory
- Removes special Unicode whitespace characters (non-breaking spaces, zero-width spaces, etc.)
- Replaces em dashes (`—`) and en dashes (``) with regular dashes (`-`)
- Preserves intentional formatting while cleaning problematic characters
**Usage**:
```bash
# Clean all markdown files in docs/
python scripts/make_clean.py
# Dry run to see what would be changed
python scripts/make_clean.py --dry-run
# Clean files in a different directory
python scripts/make_clean.py --docs-dir path/to/docs
```
**Pre-commit Integration**: This script runs automatically on commits that include markdown files in the `docs/` directory.
### 2. `check_blog_excerpts.py` - Blog Post Excerpt Validator
**Purpose**: Ensures all blog posts contain the `<!-- more -->` tag for proper excerpt handling.
**What it does**:
- Scans all markdown files in `docs/blog/posts/`
- Checks for the presence of `<!-- more -->` tags
- Reports files missing the tag
- Exits with error code 1 if any files are missing the tag
**Usage**:
```bash
# Check all blog posts
python scripts/check_blog_excerpts.py
# Check posts in a different directory
python scripts/check_blog_excerpts.py --blog-posts-dir path/to/posts
```
**Pre-commit Integration**: This script runs automatically on commits that include blog post files.
### 3. `make_sitemap.py` - Enhanced Documentation Sitemap Generator
**Purpose**: Generates an enhanced sitemap (`sitemap.yaml`) with AI-powered content analysis and cross-link suggestions.
**What it does**:
- Recursively traverses the `docs/` directory
- Analyzes each markdown file using OpenAI's GPT-4o-mini
- Extracts summaries, keywords, and topics for SEO
- Identifies internal links and references
- Generates cross-link suggestions based on content similarity
- Creates a comprehensive `sitemap.yaml` file
**Features**:
- **Caching**: Reuses analysis for unchanged files (based on content hash)
- **Concurrent Processing**: Processes multiple files simultaneously
- **Cross-linking**: Suggests related documents based on content similarity
- **Retry Logic**: Handles API failures with exponential backoff
**Usage**:
```bash
# Generate sitemap with default settings
python scripts/make_sitemap.py
# Customize settings
python scripts/make_sitemap.py \
--root-dir docs \
--output-file sitemap.yaml \
--max-concurrency 10 \
--min-similarity 0.4
# Use custom API key
python scripts/make_sitemap.py --api-key your-openai-key
```
**Output**: Creates `sitemap.yaml` with structure:
```yaml
file.md:
summary: "Brief description of the content"
keywords: ["keyword1", "keyword2", "keyword3"]
topics: ["topic1", "topic2", "topic3"]
references: ["other-file.md", "another-file.md"]
ai_references: ["ai-detected-reference.md"]
cross_links: ["suggested-related-file.md"]
hash: "content-hash-for-caching"
```
**Requirements**:
- OpenAI API key (set as `OPENAI_API_KEY` environment variable or passed via `--api-key`)
- Dependencies: `openai`, `typer`, `rich`, `tenacity`, `pyyaml`
## Pre-commit Integration
These scripts are integrated into the project's pre-commit hooks to ensure code quality:
- **`make_clean.py`**: Runs on commits with markdown files in `docs/`
- **`check_blog_excerpts.py`**: Runs on commits with blog post files
The hooks are configured in `.pre-commit-config.yaml` and run automatically during the commit process.
## Running Scripts Manually
You can run any script manually for testing or one-time operations:
```bash
# Test markdown cleaning
python scripts/make_clean.py --dry-run
# Check blog excerpts
python scripts/check_blog_excerpts.py
# Generate fresh sitemap
python scripts/make_sitemap.py
```
### 4. `fix_api_calls.py` - API Call Standardization
**Purpose**: Replaces old API call patterns with simplified versions for consistency.
**What it does**:
- Finds and replaces `client.chat.completions.create``client.create`
- Finds and replaces `client.chat.completions.create_partial``client.create_partial`
- Finds and replaces `client.chat.completions.create_iterable``client.create_iterable`
- Finds and replaces `client.chat.completions.create_with_completion``client.create_with_completion`
- Processes all markdown and notebook files in the docs directory
**Usage**:
```bash
# Dry run to see what would be changed
python scripts/fix_api_calls.py --dry-run
# Apply changes to all files
python scripts/fix_api_calls.py
# Process a single file
python scripts/fix_api_calls.py --file docs/index.md
# Custom docs directory
python scripts/fix_api_calls.py --docs-dir path/to/docs
```
### 5. `fix_old_patterns.py` - Client Initialization Pattern Fixer
**Purpose**: Replaces old client initialization patterns with the modern `from_provider` API.
**What it does**:
- Replaces `instructor.from_openai(OpenAI())``instructor.from_provider("openai/model-name")`
- Replaces `instructor.from_anthropic(Anthropic())``instructor.from_provider("anthropic/model-name")`
- Replaces `instructor.patch(OpenAI())``instructor.from_provider("openai/model-name")`
- Handles all supported providers (OpenAI, Anthropic, Google, Cohere, Mistral, Groq, etc.)
- Attempts to extract model names from existing code
**Usage**:
```bash
# Dry run to see what would be changed
python scripts/fix_old_patterns.py --dry-run
# Apply changes to all files
python scripts/fix_old_patterns.py
# Process a single file
python scripts/fix_old_patterns.py --file docs/integrations/openai.md
```
**Note**: Model names are extracted from existing code when possible, but may need manual review for accuracy.
### 6. `audit_patterns.py` - Pattern Auditor
**Purpose**: Audits documentation files to find old patterns that need updating.
**What it does**:
- Finds old API call patterns (`client.chat.completions.*`)
- Finds old initialization patterns (`instructor.from_*`, `instructor.patch`)
- Identifies potentially unused imports
- Reports line numbers for each issue
- Provides summary statistics
**Usage**:
```bash
# Detailed report with line numbers
python scripts/audit_patterns.py
# Summary statistics only
python scripts/audit_patterns.py --summary
# Audit a single file
python scripts/audit_patterns.py --file docs/index.md
# Custom docs directory
python scripts/audit_patterns.py --docs-dir path/to/docs
```
**Output**: Reports issues by file with line numbers, or summary statistics showing total counts per pattern type.
## Adding New Scripts
When adding new scripts to this directory:
1. **Documentation**: Add a section to this README explaining the script's purpose and usage
2. **Pre-commit Integration**: If appropriate, add the script to `.pre-commit-config.yaml`
3. **Error Handling**: Ensure scripts exit with appropriate error codes
4. **Help Text**: Include `--help` functionality for command-line scripts
5. **Testing**: Test scripts manually before committing
## Dependencies
Most scripts use only Python standard library modules. The sitemap generator requires additional dependencies:
```bash
uv add openai typer rich tenacity pyyaml
```
## Troubleshooting
**Pre-commit hooks failing**:
- Check that scripts are executable: `chmod +x scripts/*.py`
- Verify script paths in `.pre-commit-config.yaml`
- Run scripts manually to identify issues
**Sitemap generation issues**:
- Ensure OpenAI API key is set correctly
- Check network connectivity for API calls
- Review error messages for specific file issues
**Markdown cleaning issues**:
- Use `--dry-run` to preview changes
- Check file permissions in the docs directory
- Verify UTF-8 encoding of markdown files
+176
View File
@@ -0,0 +1,176 @@
#!/usr/bin/env python3
"""
Audit documentation files for old patterns that need to be updated.
Reports:
- Old API call patterns (client.chat.completions.*)
- Old initialization patterns (instructor.from_*, instructor.patch)
- Unused imports
"""
import argparse
import re
from collections import defaultdict
from pathlib import Path
from typing import Dict, List
def find_markdown_files(docs_dir: Path) -> List[Path]:
"""Find all markdown files in the docs directory."""
return list(docs_dir.rglob("*.md")) + list(docs_dir.rglob("*.ipynb"))
def audit_api_calls(content: str, file_path: Path) -> Dict[str, List[int]]:
"""Find old API call patterns."""
issues = defaultdict(list)
patterns = {
"client.chat.completions.create": r"client\.chat\.completions\.create\(",
"client.chat.completions.create_partial": r"client\.chat\.completions\.create_partial\(",
"client.chat.completions.create_iterable": r"client\.chat\.completions\.create_iterable\(",
"client.chat.completions.create_with_completion": r"client\.chat\.completions\.create_with_completion\(",
}
for name, pattern in patterns.items():
for match in re.finditer(pattern, content):
line_num = content[: match.start()].count("\n") + 1
issues[name].append(line_num)
return issues
def audit_old_init_patterns(content: str, file_path: Path) -> Dict[str, List[int]]:
"""Find old initialization patterns."""
issues = defaultdict(list)
# Find instructor.from_* patterns
from_pattern = r"instructor\.from_(\w+)\("
for match in re.finditer(from_pattern, content):
provider = match.group(1)
line_num = content[: match.start()].count("\n") + 1
issues[f"instructor.from_{provider}"].append(line_num)
# Find instructor.patch patterns
patch_pattern = r"instructor\.patch\("
for match in re.finditer(patch_pattern, content):
line_num = content[: match.start()].count("\n") + 1
issues["instructor.patch"].append(line_num)
return issues
def audit_unused_imports(content: str, file_path: Path) -> Dict[str, List[int]]:
"""Find potentially unused imports when from_provider is used."""
issues = defaultdict(list)
# Check if from_provider is used
uses_from_provider = "from_provider" in content or "from_provider" in content
if not uses_from_provider:
return issues
# Find provider imports
import_patterns = {
"import openai": r"^import\s+openai\b",
"from openai import": r"^from\s+openai\s+import",
"import anthropic": r"^import\s+anthropic\b",
"from anthropic import": r"^from\s+anthropic\s+import",
}
lines = content.split("\n")
for line_num, line in enumerate(lines, 1):
for name, pattern in import_patterns.items():
if re.search(pattern, line):
# Check if the import is actually used
if name.startswith("import "):
module = name.split()[1]
# Simple check - if module name appears elsewhere, might be used
if content.count(module) <= 2: # Just import and maybe one use
issues[name].append(line_num)
return issues
def process_file(file_path: Path) -> Dict[str, Dict[str, List[int]]]:
"""Process a single file and return all issues."""
try:
content = file_path.read_text(encoding="utf-8")
return {
"api_calls": audit_api_calls(content, file_path),
"old_init": audit_old_init_patterns(content, file_path),
"unused_imports": audit_unused_imports(content, file_path),
}
except Exception as e:
print(f"Error processing {file_path}: {e}")
return {"api_calls": {}, "old_init": {}, "unused_imports": {}}
def main():
parser = argparse.ArgumentParser(
description="Audit documentation files for old patterns"
)
parser.add_argument(
"--docs-dir",
type=Path,
default=Path("docs"),
help="Directory containing documentation files (default: docs)",
)
parser.add_argument(
"--file",
type=Path,
help="Audit a single file instead of all files",
)
parser.add_argument(
"--summary",
action="store_true",
help="Show only summary statistics",
)
args = parser.parse_args()
if args.file:
files = [args.file]
else:
files = find_markdown_files(args.docs_dir)
all_issues = {}
total_counts = defaultdict(int)
for file_path in files:
issues = process_file(file_path)
if any(issues.values()):
all_issues[str(file_path)] = issues
# Count totals
for issue_type, patterns in issues.items():
for pattern, line_nums in patterns.items():
total_counts[f"{issue_type}:{pattern}"] += len(line_nums)
if args.summary:
print("Summary Statistics:")
print("=" * 60)
for key, count in sorted(total_counts.items()):
issue_type, pattern = key.split(":", 1)
print(f" {pattern}: {count} instances")
else:
# Detailed report
for file_path, issues in sorted(all_issues.items()):
print(f"\n{file_path}:")
print("-" * 60)
for issue_type, patterns in issues.items():
if patterns:
print(f" {issue_type.replace('_', ' ').title()}:")
for pattern, line_nums in sorted(patterns.items()):
lines_str = ", ".join(map(str, line_nums[:10]))
if len(line_nums) > 10:
lines_str += f", ... ({len(line_nums)} total)"
print(f" {pattern}: lines {lines_str}")
print(f"\nTotal files with issues: {len(all_issues)}")
print(f"Total issues found: {sum(total_counts.values())}")
if __name__ == "__main__":
main()
+107
View File
@@ -0,0 +1,107 @@
"""Benchmark instructor import costs.
Run with:
python scripts/benchmark_import.py
"""
from __future__ import annotations
import subprocess
import sys
import textwrap
def run_import_benchmark(code: str, label: str) -> dict[str, str]:
"""Run a Python snippet in a subprocess and capture import stats."""
script = textwrap.dedent(
f"""
import sys
import time
import tracemalloc
tracemalloc.start()
start = time.time()
{code}
elapsed = time.time() - start
current, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()
print(f"{{elapsed:.3f}}|{{current}}|{{peak}}|{{len(sys.modules)}}")
"""
)
result = subprocess.run(
[sys.executable, "-c", script],
capture_output=True,
text=True,
timeout=60,
check=False,
)
if result.returncode != 0:
return {
"label": label,
"time": "ERROR",
"memory_mb": "ERROR",
"peak_mb": "ERROR",
"modules": "ERROR",
"error": result.stderr.strip(),
}
elapsed, current, peak, mod_count = result.stdout.strip().split("|")
return {
"label": label,
"time": f"{float(elapsed):.3f}s",
"memory_mb": f"{int(current) / 1024 / 1024:.1f} MB",
"peak_mb": f"{int(peak) / 1024 / 1024:.1f} MB",
"modules": mod_count,
}
def main() -> None:
print("=" * 70)
print(" INSTRUCTOR IMPORT BENCHMARK")
print("=" * 70)
print(f" Python: {sys.version.split()[0]}")
print(f" Executable: {sys.executable}")
print("=" * 70)
print()
benchmarks = [
("import sys", "baseline (sys only)"),
("import instructor", "import instructor"),
("import instructor; _ = instructor.from_provider", "access from_provider"),
("import instructor; _ = instructor.from_openai", "access from_openai"),
("import instructor; _ = instructor.from_anthropic", "access from_anthropic"),
("import instructor; _ = instructor.from_genai", "access from_genai"),
("import instructor; _ = instructor.Mode", "access Mode"),
("import instructor; _ = instructor.Instructor", "access Instructor class"),
("import instructor; _ = instructor.Partial", "access Partial"),
]
results = [run_import_benchmark(code, label) for code, label in benchmarks]
header = f"{'Benchmark':<30} {'Time':>8} {'Memory':>10} {'Peak':>10} {'Modules':>8}"
print(header)
print("-" * len(header))
for result in results:
if result.get("error"):
print(f"{result['label']:<30} {'ERROR':>8} (missing dependency?)")
else:
print(
f"{result['label']:<30} {result['time']:>8} "
f"{result['memory_mb']:>10} {result['peak_mb']:>10} "
f"{result['modules']:>8}"
)
print()
print("Notes:")
print(" - 'import instructor' should stay small because of lazy loading.")
print(" - Provider access loads each SDK on demand.")
if __name__ == "__main__":
main()
+100
View File
@@ -0,0 +1,100 @@
#!/usr/bin/env python3
"""
Check if blog posts contain the <!-- more --> tag for excerpts.
This script:
- Recursively finds all .md files in the docs/blog/posts directory
- Checks if each file contains the <!-- more --> tag
- Reports files that are missing the tag
- Exits with error code 1 if any files are missing the tag
"""
import sys
from pathlib import Path
def check_blog_excerpts(blog_posts_dir: str = "docs/blog/posts") -> bool:
"""
Check if blog posts contain the <!-- more --> tag.
Args:
blog_posts_dir: Path to the blog posts directory (default: "docs/blog/posts")
Returns:
True if all files have the tag, False if any are missing it
"""
blog_path = Path(blog_posts_dir)
if not blog_path.exists():
print(f"Error: Directory '{blog_posts_dir}' does not exist.")
return False
if not blog_path.is_dir():
print(f"Error: '{blog_posts_dir}' is not a directory.")
return False
# Find all markdown files recursively
md_files = list(blog_path.rglob("*.md"))
if not md_files:
print(f"No markdown files found in '{blog_posts_dir}' directory.")
return True
print(f"Checking {len(md_files)} blog post files for <!-- more --> tag...")
missing_tag_files = []
for md_file in md_files:
try:
# Read the file content
with open(md_file, encoding="utf-8") as f:
content = f.read()
# Check if the file contains the <!-- more --> tag
if "<!-- more -->" not in content:
missing_tag_files.append(md_file)
print(f"Missing <!-- more --> tag: {md_file}")
else:
print(f"✓ Has <!-- more --> tag: {md_file}")
except Exception as e:
print(f"Error reading {md_file}: {e}")
missing_tag_files.append(md_file)
# Summary
if missing_tag_files:
print(f"\n❌ Found {len(missing_tag_files)} files missing <!-- more --> tag:")
for file in missing_tag_files:
print(f" - {file}")
print(
f"\nPlease add <!-- more --> tag to these files for proper excerpt handling."
)
return False
else:
print(f"\n✅ All {len(md_files)} blog post files have the <!-- more --> tag!")
return True
def main():
"""Main function to handle command line arguments."""
import argparse
parser = argparse.ArgumentParser(
description="Check if blog posts contain the <!-- more --> tag for excerpts"
)
parser.add_argument(
"--blog-posts-dir",
default="docs/blog/posts",
help="Path to blog posts directory (default: docs/blog/posts)",
)
args = parser.parse_args()
success = check_blog_excerpts(blog_posts_dir=args.blog_posts_dir)
# Exit with appropriate code for pre-commit
sys.exit(0 if success else 1)
if __name__ == "__main__":
main()
+194
View File
@@ -0,0 +1,194 @@
#!/usr/bin/env python3
"""
Check for broken internal links in documentation files.
Finds:
- Broken internal links (missing target files)
- Broken anchor links
- Orphaned pages (no incoming links)
"""
import argparse
import re
from pathlib import Path
def find_markdown_files(docs_dir: Path) -> list[Path]:
"""Find all markdown files in the docs directory."""
return list(docs_dir.rglob("*.md"))
def extract_links(content: str, file_path: Path) -> list[tuple[str, int]]: # noqa: ARG001
"""
Extract internal markdown links from content.
Returns:
List of (link_target, line_number) tuples
"""
links = []
# Match markdown links: [text](url)
for match in re.finditer(r"\[([^\]]+)\]\(([^)]+)\)", content):
link_text = match.group(1)
link_url = match.group(2)
line_num = content[: match.start()].count("\n") + 1
# Skip external links
if link_url.startswith(("http://", "https://", "mailto:", "#")):
continue
links.append((link_url, line_num))
return links
def resolve_link(link_url: str, source_file: Path, docs_dir: Path) -> tuple[bool, str]: # noqa: ARG001
"""
Resolve a relative link and check if target exists.
Returns:
(exists, resolved_path)
"""
# Split anchor if present
if "#" in link_url:
link_path, anchor = link_url.split("#", 1)
else:
link_path = link_url
anchor = None
# Resolve relative path
source_dir = source_file.parent
target_path = (source_dir / link_path).resolve()
# Check if file exists
exists = target_path.exists()
return exists, str(target_path)
def check_file(file_path: Path, docs_dir: Path) -> dict[str, list[tuple[str, int]]]:
"""Check all links in a file."""
issues = {}
try:
content = file_path.read_text(encoding="utf-8")
links = extract_links(content, file_path)
broken_links = []
for link_url, line_num in links:
exists, resolved_path = resolve_link(link_url, file_path, docs_dir)
if not exists:
broken_links.append((link_url, line_num))
if broken_links:
issues["broken_links"] = broken_links
return issues
except Exception as e:
return {"error": [(str(e), 0)]}
def find_orphaned_pages(files: list[Path], docs_dir: Path) -> set[Path]:
"""Find pages with no incoming links."""
all_files = set(files)
referenced_files = set()
for file_path in files:
try:
content = file_path.read_text(encoding="utf-8")
links = extract_links(content, file_path)
for link_url, _ in links:
exists, resolved_path = resolve_link(link_url, file_path, docs_dir)
if exists:
referenced_files.add(Path(resolved_path))
except Exception:
pass
# Files that are not referenced (orphaned)
orphaned = all_files - referenced_files
# Remove index pages and special files from orphaned list
orphaned = {
f
for f in orphaned
if not any(
part in str(f)
for part in ["index.md", "AGENT.md", "repository-overview.md"]
)
}
return orphaned
def main():
parser = argparse.ArgumentParser(
description="Check for broken internal links in documentation"
)
parser.add_argument(
"--docs-dir",
type=Path,
default=Path("docs"),
help="Directory containing documentation files (default: docs)",
)
parser.add_argument(
"--summary",
action="store_true",
help="Show only summary statistics",
)
parser.add_argument(
"--file",
type=Path,
help="Check a single file instead of all files",
)
parser.add_argument(
"--find-orphans",
action="store_true",
help="Find orphaned pages with no incoming links",
)
args = parser.parse_args()
if args.file:
files = [args.file]
else:
files = find_markdown_files(args.docs_dir)
all_issues = {}
total_broken = 0
for file_path in files:
issues = check_file(file_path, args.docs_dir)
if issues:
all_issues[str(file_path)] = issues
if "broken_links" in issues:
total_broken += len(issues["broken_links"])
if args.summary:
print("Summary Statistics:")
print("=" * 60)
print(f" Files with broken links: {len(all_issues)}")
print(f" Total broken links: {total_broken}")
else:
# Detailed report
for file_path, issues in sorted(all_issues.items()):
if "broken_links" in issues:
print(f"\n{file_path}:")
for link_url, line_num in issues["broken_links"]:
print(f" Line {line_num}: {link_url}")
if args.find_orphans:
orphaned = find_orphaned_pages(files, args.docs_dir)
if orphaned:
print("\n" + "=" * 60)
print("Orphaned Pages (no incoming links):")
print("=" * 60)
for file_path in sorted(orphaned):
print(f" {file_path}")
print(f"\nTotal orphaned pages: {len(orphaned)}")
print(f"\nTotal files checked: {len(files)}")
if __name__ == "__main__":
main()
+118
View File
@@ -0,0 +1,118 @@
#!/usr/bin/env python3
"""
Fix API calls in documentation files.
Replaces old API patterns with simplified versions:
- client.chat.completions.create → client.create
- client.chat.completions.create_partial → client.create_partial
- client.chat.completions.create_iterable → client.create_iterable
- client.chat.completions.create_with_completion → client.create_with_completion
"""
import argparse
import re
from pathlib import Path
def find_markdown_files(docs_dir: Path) -> list[Path]:
"""Find all markdown files in the docs directory."""
return list(docs_dir.rglob("*.md")) + list(docs_dir.rglob("*.ipynb"))
def replace_api_calls(content: str, dry_run: bool = False) -> tuple[str, int]: # noqa: ARG001
"""
Replace old API call patterns with simplified versions.
Returns:
Tuple of (new_content, number_of_replacements)
"""
replacements = 0
# Pattern mappings: (old_pattern, new_pattern)
patterns = [
(
r"client\.chat\.completions\.create_with_completion\(",
"client.create_with_completion(",
),
(r"client\.chat\.completions\.create_partial\(", "client.create_partial("),
(r"client\.chat\.completions\.create_iterable\(", "client.create_iterable("),
(r"client\.chat\.completions\.create\(", "client.create("),
]
new_content = content
for old_pattern, new_pattern in patterns:
matches = len(re.findall(old_pattern, new_content))
if matches > 0:
new_content = re.sub(old_pattern, new_pattern, new_content)
replacements += matches
return new_content, replacements
def process_file(file_path: Path, dry_run: bool = False) -> int:
"""Process a single file and return number of replacements."""
try:
content = file_path.read_text(encoding="utf-8")
new_content, replacements = replace_api_calls(content, dry_run)
if replacements > 0:
if dry_run:
print(f"Would fix {replacements} instances in {file_path}")
else:
file_path.write_text(new_content, encoding="utf-8")
print(f"Fixed {replacements} instances in {file_path}")
return replacements
except Exception as e:
print(f"Error processing {file_path}: {e}")
return 0
def main():
parser = argparse.ArgumentParser(
description="Replace old API call patterns with simplified versions"
)
parser.add_argument(
"--docs-dir",
type=Path,
default=Path("docs"),
help="Directory containing documentation files (default: docs)",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Show what would be changed without making changes",
)
parser.add_argument(
"--file",
type=Path,
help="Process a single file instead of all files",
)
args = parser.parse_args()
if args.file:
files = [args.file]
else:
files = find_markdown_files(args.docs_dir)
total_replacements = 0
files_modified = 0
for file_path in files:
replacements = process_file(file_path, args.dry_run)
if replacements > 0:
total_replacements += replacements
files_modified += 1
print(f"\nSummary:")
print(f" Files processed: {len(files)}")
print(f" Files modified: {files_modified}")
print(f" Total replacements: {total_replacements}")
if args.dry_run:
print("\nRun without --dry-run to apply changes")
if __name__ == "__main__":
main()
+54
View File
@@ -0,0 +1,54 @@
#!/usr/bin/env python3
"""Fix doc test formatting issues using --update-examples for each test file."""
import subprocess
import sys
from pathlib import Path
test_files = [
"tests/docs/test_concepts_operations.py",
"tests/docs/test_examples_batch.py",
"tests/docs/test_examples_integrations.py",
"tests/docs/test_examples_multimodal.py",
"tests/docs/test_posts.py",
]
def run_update(test_file: str) -> bool:
"""Run --update-examples on a test file."""
print(f"\n{'=' * 60}")
print(f"Processing: {test_file}")
print(f"{'=' * 60}")
cmd = ["uv", "run", "pytest", test_file, "--update-examples", "-q", "--tb=no"]
try:
result = subprocess.run(
cmd, capture_output=True, text=True, cwd=Path(__file__).parent.parent
)
if result.returncode == 0:
print(f"✓ Successfully updated {test_file}")
return True
else:
# Even with errors, some files might have been updated
print(f"⚠ Completed {test_file} with exit code {result.returncode}")
if result.stdout:
print("STDOUT:", result.stdout[-500:]) # Last 500 chars
return False
except Exception as e:
print(f"✗ Error processing {test_file}: {e}")
return False
if __name__ == "__main__":
success_count = 0
for test_file in test_files:
if run_update(test_file):
success_count += 1
print(f"\n{'=' * 60}")
print(f"Summary: {success_count}/{len(test_files)} files processed")
print(f"{'=' * 60}")
sys.exit(0 if success_count == len(test_files) else 1)
+271
View File
@@ -0,0 +1,271 @@
#!/usr/bin/env python3
"""
Fix old client initialization patterns in documentation files.
Replaces old initialization patterns with from_provider:
- instructor.from_openai(OpenAI()) → instructor.from_provider("openai/model-name")
- instructor.from_anthropic(Anthropic()) → instructor.from_provider("anthropic/model-name")
- instructor.patch(OpenAI()) → instructor.from_provider("openai/model-name")
- Similar patterns for all other providers
"""
import argparse
import re
from pathlib import Path
from typing import List, Tuple
# Mapping of provider names to their from_provider identifiers
PROVIDER_MAPPING = {
"openai": "openai",
"anthropic": "anthropic",
"google": "google",
"cohere": "cohere",
"mistral": "mistral",
"groq": "groq",
"litellm": "litellm",
"ollama": "ollama",
"azure": "azure",
"bedrock": "bedrock",
"vertex": "vertex",
"genai": "google", # Google GenAI
"deepseek": "deepseek",
"fireworks": "fireworks",
"cerebras": "cerebras",
"together": "together",
"anyscale": "anyscale",
"perplexity": "perplexity",
"writer": "writer",
"openrouter": "openrouter",
"sambanova": "sambanova",
"truefoundry": "truefoundry",
"cortex": "cortex",
"databricks": "databricks",
"xai": "xai",
}
def find_markdown_files(docs_dir: Path) -> List[Path]:
"""Find all markdown files in the docs directory."""
return list(docs_dir.rglob("*.md")) + list(docs_dir.rglob("*.ipynb"))
def extract_model_name(content: str, match_start: int, match_end: int) -> str:
"""
Try to extract model name from context around the match.
Looks for common patterns like model="...", model='...', or model_name=...
"""
# Look backwards and forwards for model parameter
context_start = max(0, match_start - 200)
context_end = min(len(content), match_end + 200)
context = content[context_start:context_end]
# Try to find model parameter
model_match = re.search(
r'model\s*[=:]\s*["\']([^"\']+)["\']', context, re.IGNORECASE
)
if model_match:
return model_match.group(1)
# Default model names by provider
return "gpt-4o" # Will need manual review for accuracy
def replace_from_pattern(
content: str, provider: str, dry_run: bool = False
) -> Tuple[str, int]:
"""
Replace instructor.from_PROVIDER(Provider()) patterns.
Pattern: instructor.from_openai(OpenAI(model="..."))
→ instructor.from_provider("openai/model-name")
"""
replacements = 0
# Pattern: instructor.from_PROVIDER(ProviderClass(...))
pattern = rf"instructor\.from_{provider}\((\w+)(\([^)]*\))?\)"
def replacer(match):
nonlocal replacements
provider_class = match.group(1)
args = match.group(2) or ""
# Try to extract model name from args
model_match = re.search(r'model\s*=\s*["\']([^"\']+)["\']', args)
if model_match:
model_name = model_match.group(1)
else:
# Default model - may need manual review
model_name = (
"gpt-4o" if provider == "openai" else "claude-3-5-sonnet-20241022"
)
replacements += 1
return f'instructor.from_provider("{provider}/{model_name}")'
new_content = re.sub(pattern, replacer, content, flags=re.IGNORECASE)
return new_content, replacements
def replace_patch_pattern(content: str, dry_run: bool = False) -> Tuple[str, int]:
"""
Replace instructor.patch(Provider()) patterns.
Pattern: instructor.patch(OpenAI(model="..."))
→ instructor.from_provider("openai/model-name")
"""
replacements = 0
# Pattern: instructor.patch(ProviderClass(...))
# Match common provider classes
provider_classes = "|".join(
[
"OpenAI",
"Anthropic",
"GoogleGenerativeAI",
"Cohere",
"Mistral",
"Groq",
"LiteLLM",
"Ollama",
"Bedrock",
"VertexAI",
]
)
pattern = rf"instructor\.patch\(({provider_classes})(\([^)]*\))?\)"
def replacer(match):
nonlocal replacements
provider_class = match.group(1)
args = match.group(2) or ""
# Map class name to provider identifier
class_to_provider = {
"OpenAI": "openai",
"Anthropic": "anthropic",
"GoogleGenerativeAI": "google",
"Cohere": "cohere",
"Mistral": "mistral",
"Groq": "groq",
"LiteLLM": "litellm",
"Ollama": "ollama",
"Bedrock": "bedrock",
"VertexAI": "vertex",
}
provider = class_to_provider.get(provider_class, "openai")
# Try to extract model name from args
model_match = re.search(r'model\s*=\s*["\']([^"\']+)["\']', args)
if model_match:
model_name = model_match.group(1)
else:
# Default models
defaults = {
"openai": "gpt-4o",
"anthropic": "claude-3-5-sonnet-20241022",
"google": "gemini-1.5-pro",
}
model_name = defaults.get(provider, "gpt-4o")
replacements += 1
return f'instructor.from_provider("{provider}/{model_name}")'
new_content = re.sub(pattern, replacer, content)
return new_content, replacements
def replace_old_patterns(content: str, dry_run: bool = False) -> Tuple[str, int]:
"""
Replace all old initialization patterns.
Returns:
Tuple of (new_content, total_replacements)
"""
total_replacements = 0
new_content = content
# Replace instructor.patch() patterns first
new_content, patch_replacements = replace_patch_pattern(new_content, dry_run)
total_replacements += patch_replacements
# Replace instructor.from_* patterns for each provider
for provider in PROVIDER_MAPPING.keys():
new_content, from_replacements = replace_from_pattern(
new_content, provider, dry_run
)
total_replacements += from_replacements
return new_content, total_replacements
def process_file(file_path: Path, dry_run: bool = False) -> int:
"""Process a single file and return number of replacements."""
try:
content = file_path.read_text(encoding="utf-8")
new_content, replacements = replace_old_patterns(content, dry_run)
if replacements > 0:
if dry_run:
print(f"Would fix {replacements} instances in {file_path}")
else:
file_path.write_text(new_content, encoding="utf-8")
print(f"Fixed {replacements} instances in {file_path}")
return replacements
except Exception as e:
print(f"Error processing {file_path}: {e}")
return 0
def main():
parser = argparse.ArgumentParser(
description="Replace old client initialization patterns with from_provider"
)
parser.add_argument(
"--docs-dir",
type=Path,
default=Path("docs"),
help="Directory containing documentation files (default: docs)",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Show what would be changed without making changes",
)
parser.add_argument(
"--file",
type=Path,
help="Process a single file instead of all files",
)
args = parser.parse_args()
if args.file:
files = [args.file]
else:
files = find_markdown_files(args.docs_dir)
total_replacements = 0
files_modified = 0
for file_path in files:
replacements = process_file(file_path, args.dry_run)
if replacements > 0:
total_replacements += replacements
files_modified += 1
print(f"\nSummary:")
print(f" Files processed: {len(files)}")
print(f" Files modified: {files_modified}")
print(f" Total replacements: {total_replacements}")
if args.dry_run:
print("\nRun without --dry-run to apply changes")
else:
print("\n⚠️ Note: Please review model names - defaults may need adjustment")
if __name__ == "__main__":
main()
+147
View File
@@ -0,0 +1,147 @@
#!/usr/bin/env python3
"""
Clean markdown files in the docs directory.
This script:
- Recursively finds all .md files in the docs directory
- Strips special whitespace characters (non-breaking spaces, zero-width spaces, etc.)
- Replaces em dashes (—) with regular dashes (-)
- Preserves the original file structure
"""
import re
import unicodedata
from pathlib import Path
def clean_markdown_content(content: str) -> str:
"""
Clean markdown content by removing special whitespace and replacing em dashes.
Args:
content: The original markdown content
Returns:
The cleaned markdown content
"""
# Replace em dashes with regular dashes
content = content.replace("", "-")
content = content.replace("", "-") # en dash as well
# Remove special whitespace characters
# This includes non-breaking spaces, zero-width spaces, and other Unicode whitespace
cleaned_lines = []
for line in content.split("\n"):
# Normalize Unicode characters and remove special whitespace
cleaned_line = unicodedata.normalize("NFKC", line)
# Remove zero-width characters and other special whitespace
cleaned_line = re.sub(r"[\u200B\u200C\u200D\uFEFF]", "", cleaned_line)
# Replace non-breaking spaces with regular spaces
cleaned_line = cleaned_line.replace("\u00a0", " ")
# Strip leading/trailing whitespace but preserve intentional indentation
cleaned_line = cleaned_line.rstrip()
cleaned_lines.append(cleaned_line)
return "\n".join(cleaned_lines)
def process_markdown_files(docs_dir: str = "docs", dry_run: bool = False) -> None:
"""
Process all markdown files in the docs directory.
Args:
docs_dir: Path to the docs directory (default: "docs")
dry_run: If True, show what would be changed without modifying files
"""
docs_path = Path(docs_dir)
if not docs_path.exists():
print(f"Error: Directory '{docs_dir}' does not exist.")
return
if not docs_path.is_dir():
print(f"Error: '{docs_dir}' is not a directory.")
return
# Find all markdown files recursively
md_files = list(docs_path.rglob("*.md"))
if not md_files:
print(f"No markdown files found in '{docs_dir}' directory.")
return
mode_text = "DRY RUN - " if dry_run else ""
print(f"{mode_text}Found {len(md_files)} markdown files to process...")
processed_count = 0
modified_count = 0
for md_file in md_files:
try:
# Read the original content
with open(md_file, encoding="utf-8") as f:
original_content = f.read()
# Clean the content
cleaned_content = clean_markdown_content(original_content)
# Check if content was modified
if cleaned_content != original_content:
if dry_run:
print(f"Would modify: {md_file}")
# Show a sample of the changes
original_lines = original_content.split("\n")
cleaned_lines = cleaned_content.split("\n")
for i, (orig, clean) in enumerate(
zip(original_lines, cleaned_lines)
):
if orig != clean:
print(f" Line {i + 1}:")
print(f" Original: {repr(orig)}")
print(f" Cleaned: {repr(clean)}")
# Only show first difference per file
break
else:
# Write the cleaned content back to the file
with open(md_file, "w", encoding="utf-8") as f:
f.write(cleaned_content)
print(f"Modified: {md_file}")
modified_count += 1
else:
if not dry_run:
print(f"No changes needed: {md_file}")
processed_count += 1
except Exception as e:
print(f"Error processing {md_file}: {e}")
action_text = "would be" if dry_run else "were"
print(f"\nProcessing complete!")
print(f"Total files processed: {processed_count}")
print(f"Files {action_text} modified: {modified_count}")
def main():
"""Main function to handle command line arguments."""
import argparse
parser = argparse.ArgumentParser(
description="Clean markdown files by removing special whitespace and replacing em dashes"
)
parser.add_argument(
"--docs-dir", default="docs", help="Path to docs directory (default: docs)"
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Show what would be changed without modifying files",
)
args = parser.parse_args()
process_markdown_files(docs_dir=args.docs_dir, dry_run=args.dry_run)
if __name__ == "__main__":
main()
+207
View File
@@ -0,0 +1,207 @@
import os
from typing import Optional, Literal
import asyncio
from openai import AsyncOpenAI
import typer
from rich.console import Console
from rich.progress import Progress
from rich.table import Table
from pydantic import BaseModel, Field
import instructor
import frontmatter
console = Console()
client = instructor.from_openai(AsyncOpenAI())
async def generate_ai_frontmatter(
client: AsyncOpenAI, title: str, content: str, categories: list[str]
):
"""
Generate a description and categories for the given content using AI.
Args:
client (AsyncOpenAI): The AsyncOpenAI client.
title (str): The title of the markdown file.
content (str): The content of the file.
categories (List[str]): List of all available categories.
Returns:
DescriptionAndCategories: The generated description, categories, tags, and reasoning.
"""
class DescriptionAndCategories(BaseModel):
description: str
reasoning: str = Field(
..., description="The reasoning for the correct categories"
)
tags: list[str]
categories: list[
Literal[
"OpenAI",
"Anthropic",
"LLama",
"LLM Observability",
"Data Processing",
"Python",
"LLM Techniques",
"Pydantic",
"Performance Optimization",
"Data Validation",
"API Development",
"Retrieval Augmented Generation",
]
]
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": "You are an AI assistant that generates SEO-friendly descriptions for markdown files.",
},
{"role": "user", "content": f"Title: {title}\n\nContent: {content}"},
{
"role": "user",
"content": f"Based on the title and content, generate a brief description (max 160 characters) that would be suitable for SEO purposes. Also, select up to 3 relevant categories from the following list: {', '.join(categories)}. Return both the description and the selected categories. The categories should be pretty strict, so only choose one if you're really sure it's the best choice. Also, suggest up to 5 relevant tags.",
},
],
max_tokens=150,
response_model=DescriptionAndCategories,
)
return response
def get_all_categories(root_dir: str) -> set[str]:
"""
Read all markdown files and extract unique categories.
Args:
root_dir (str): The root directory to start processing from.
Returns:
Set[str]: A set of unique categories.
"""
categories = set()
for root, _, files in os.walk(root_dir):
for file in files:
if file.endswith(".md"):
file_path = os.path.join(root, file)
post = frontmatter.load(file_path)
if "categories" in post.metadata:
categories.update(post.metadata["categories"])
return categories
def preview_categories(root_dir: str) -> None:
"""
Preview all categories found in markdown files.
Args:
root_dir (str): The root directory to start processing from.
"""
categories = get_all_categories(root_dir)
table = Table(title="Categories Preview")
table.add_column("Category", style="cyan")
for category in sorted(categories):
table.add_row(category)
console.print(table)
console.print(f"\nTotal categories found: {len(categories)}")
async def process_file(
client: AsyncOpenAI, file_path: str, categories: list[str], enable_comments: bool
) -> None:
"""
Process a single file, adding or updating the description and categories in the front matter.
Args:
client (AsyncOpenAI): The AsyncOpenAI client.
file_path (str): The path to the file to process.
categories (List[str]): List of all available categories.
enable_comments (bool): Whether to enable comments in the front matter.
"""
post = frontmatter.load(file_path)
title = post.metadata.get("title", os.path.basename(file_path))
response = await generate_ai_frontmatter(client, title, post.content, categories)
post.metadata["description"] = response.description
post.metadata["categories"] = response.categories
post.metadata["tags"] = response.tags
if enable_comments:
post.metadata["comments"] = True
with open(file_path, "w", encoding="utf-8") as file:
file.write(frontmatter.dumps(post))
console.print(f"[green]Updated front matter in {file_path}[/green]")
async def process_files(
root_dir: str,
api_key: Optional[str] = None, # noqa: ARG001
use_categories: bool = False,
enable_comments: bool = False,
) -> None:
"""
Process all markdown files in the given directory and its subdirectories.
Args:
root_dir (str): The root directory to start processing from.
api_key (Optional[str]): The OpenAI API key. If not provided, it will be read from the OPENAI_API_KEY environment variable.
use_categories (bool): Whether to first read all files and generate a list of categories.
enable_comments (bool): Whether to enable comments in the front matter.
"""
markdown_files = []
for root, _, files in os.walk(root_dir):
for file in files:
if file.endswith(".md"):
markdown_files.append(os.path.join(root, file))
categories = list(get_all_categories(root_dir)) if use_categories else []
with Progress() as progress:
task = progress.add_task(
"[green]Processing files...", total=len(markdown_files)
)
async def process_and_update(file_path: str) -> None:
await process_file(client, file_path, categories, enable_comments)
progress.update(task, advance=1)
tasks = [process_and_update(file_path) for file_path in markdown_files]
await asyncio.gather(*tasks)
console.print("[bold green]All files processed successfully![/bold green]")
app = typer.Typer()
@app.command()
def main(
root_dir: str = typer.Option("docs", help="Root directory to process"),
api_key: Optional[str] = typer.Option(None, help="OpenAI API key"),
use_categories: bool = typer.Option(False, help="Use categories from all files"),
preview_only: bool = typer.Option(
False, help="Preview categories without processing files"
),
enable_comments: bool = typer.Option(
False, help="Enable comments in the front matter"
),
):
"""
Add or update description in front matter of markdown files in the given directory and its subdirectories.
"""
if preview_only:
preview_categories(root_dir)
else:
asyncio.run(process_files(root_dir, api_key, use_categories, enable_comments))
if __name__ == "__main__":
app()
+314
View File
@@ -0,0 +1,314 @@
import os
import asyncio
import yaml
from typing import Optional, Any
from collections.abc import Generator
from openai import AsyncOpenAI
import typer
from rich.console import Console
from rich.progress import Progress
import hashlib
from asyncio import as_completed
import tenacity
import re
console = Console()
def traverse_docs(
root_dir: str = "docs",
) -> Generator[tuple[str, str, str], None, None]:
"""
Recursively traverse the docs folder and yield the path, content, and content hash of each file.
Args:
root_dir (str): The root directory to start traversing from. Defaults to 'docs'.
Yields:
Tuple[str, str, str]: A tuple containing the relative path from 'docs', the file content, and the content hash.
"""
for root, _, files in os.walk(root_dir):
for file in files:
if file.endswith(".md"): # Assuming we're only interested in Markdown files
file_path = os.path.join(root, file)
relative_path = os.path.relpath(file_path, root_dir)
with open(file_path, encoding="utf-8") as f:
content = f.read()
content_hash = hashlib.md5(content.encode()).hexdigest()
yield relative_path, content, content_hash
def extract_markdown_links(content: str) -> list[str]:
"""
Extract all markdown links from the content.
Args:
content (str): The markdown content to analyze
Returns:
List[str]: List of extracted link paths
"""
# Match markdown links [text](path)
link_pattern = r"\[([^\]]+)\]\(([^)]+)\)"
matches = re.findall(link_pattern, content)
links = []
for _, link_path in matches:
# Filter out external links and anchors
if not link_path.startswith(("http://", "https://", "#", "mailto:")):
# Clean up relative paths
link_path = link_path.strip("/")
if link_path.endswith(".md"):
links.append(link_path)
elif "." not in link_path:
# Assume it's a directory reference, add index.md
links.append(f"{link_path}/index.md")
return links
def normalize_path(path: str, current_path: str) -> str:
"""
Normalize a relative path based on the current file's location.
Args:
path (str): The path to normalize
current_path (str): The current file's path
Returns:
str: The normalized path
"""
if path.startswith("/"):
# Absolute path from docs root
return path.strip("/")
# Relative path
current_dir = os.path.dirname(current_path)
if current_dir:
normalized = os.path.normpath(os.path.join(current_dir, path))
# Remove any leading '../' that go outside docs/
while normalized.startswith("../"):
normalized = normalized[3:]
return normalized
return path
@tenacity.retry(
stop=tenacity.stop_after_attempt(3),
wait=tenacity.wait_exponential(multiplier=1, min=4, max=10),
retry=tenacity.retry_if_exception_type(Exception),
before_sleep=lambda retry_state: console.print(
f"[yellow]Retrying analysis... (Attempt {retry_state.attempt_number})[/yellow]"
),
)
async def analyze_content(
client: AsyncOpenAI, path: str, content: str
) -> dict[str, Any]:
"""
Analyze the content of a file to extract summary, keywords, topics, and references.
Args:
client (AsyncOpenAI): The AsyncOpenAI client.
path (str): The path of the file.
content (str): The content of the file.
Returns:
Dict[str, Any]: Analysis results including summary, keywords, topics, and references.
Raises:
Exception: If all retry attempts fail.
"""
try:
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": """You are a documentation analyzer. Extract and return the following information in a structured format:
1. A concise summary (2-3 sentences) for SEO
2. A list of important keywords (5-10 words/phrases)
3. Main topics/concepts covered (3-5 topics)
4. Any references to other documentation pages mentioned in the text
Return the response in this exact format:
SUMMARY: [Your summary here]
KEYWORDS: [keyword1, keyword2, keyword3, ...]
TOPICS: [topic1, topic2, topic3, ...]
REFERENCES: [referenced_page1.md, referenced_page2.md, ...]
If no references are found, write: REFERENCES: none""",
},
{"role": "user", "content": content},
],
max_tokens=4000,
)
result_text = response.choices[0].message.content
# Parse the structured response
summary = ""
keywords = []
topics = []
references = []
if result_text:
for line in result_text.split("\n"):
line = line.strip()
if line.startswith("SUMMARY:"):
summary = line[8:].strip()
elif line.startswith("KEYWORDS:"):
keywords_text = line[9:].strip()
if keywords_text and keywords_text != "none":
keywords = [k.strip() for k in keywords_text.split(",")]
elif line.startswith("TOPICS:"):
topics_text = line[7:].strip()
if topics_text and topics_text != "none":
topics = [t.strip() for t in topics_text.split(",")]
elif line.startswith("REFERENCES:"):
refs_text = line[11:].strip()
if refs_text and refs_text != "none":
references = [r.strip() for r in refs_text.split(",")]
return {
"summary": summary,
"keywords": keywords,
"topics": topics,
"ai_references": references,
}
except Exception as e:
console.print(f"[bold red]Error analyzing {path}: {str(e)}[/bold red]")
raise
async def generate_sitemap(
root_dir: str,
output_file: str,
api_key: Optional[str] = None,
max_concurrency: int = 5,
) -> None:
"""
Generate a sitemap from the given root directory.
Args:
root_dir (str): The root directory to start traversing from.
output_file (str): The output file to save the sitemap.
api_key (Optional[str]): The OpenAI API key. If not provided, it will be read from the OPENAI_API_KEY environment variable.
max_concurrency (int): The maximum number of concurrent tasks. Defaults to 5.
"""
client = AsyncOpenAI(api_key=api_key)
# Load existing sitemap if it exists
existing_sitemap: dict[str, dict[str, Any]] = {}
if os.path.exists(output_file):
with open(output_file, encoding="utf-8") as sitemap_file:
existing_sitemap = yaml.safe_load(sitemap_file) or {}
sitemap_data: dict[str, dict[str, Any]] = {}
async def process_file(
path: str, content: str, content_hash: str
) -> tuple[str, dict[str, Any]]:
# Check if we can reuse existing data
if (
path in existing_sitemap
and existing_sitemap[path].get("hash") == content_hash
):
# Extract markdown links even for cached content
links = extract_markdown_links(content)
normalized_links = []
for link in links:
normalized = normalize_path(link, path)
if normalized:
normalized_links.append(normalized)
existing_data = existing_sitemap[path].copy()
existing_data["references"] = normalized_links
return path, existing_data
try:
# Extract markdown links
links = extract_markdown_links(content)
normalized_links = []
for link in links:
normalized = normalize_path(link, path)
if normalized:
normalized_links.append(normalized)
# Get AI analysis
analysis = await analyze_content(client, path, content)
return path, {
"summary": analysis["summary"],
"keywords": analysis["keywords"],
"topics": analysis["topics"],
"references": normalized_links,
"ai_references": analysis["ai_references"],
"hash": content_hash,
}
except Exception as e:
console.print(
f"[bold red]Failed to analyze {path} after multiple attempts: {str(e)}[/bold red]"
)
return path, {
"summary": "Failed to generate summary",
"keywords": [],
"topics": [],
"references": normalized_links,
"ai_references": [],
"hash": content_hash,
}
files_to_process: list[tuple[str, str, str]] = list(traverse_docs(root_dir))
total_files = len(files_to_process)
with Progress() as progress:
task = progress.add_task("[green]Processing files...", total=total_files)
semaphore = asyncio.Semaphore(max_concurrency)
async def bounded_process_file(*args):
async with semaphore:
return await process_file(*args)
tasks = [
bounded_process_file(path, content, content_hash)
for path, content, content_hash in files_to_process
]
for completed_task in as_completed(tasks):
path, result = await completed_task
sitemap_data[path] = result
progress.update(task, advance=1)
# Save final results
with open(output_file, "w", encoding="utf-8") as sitemap_file:
yaml.dump(sitemap_data, sitemap_file, default_flow_style=False, sort_keys=True)
console.print(
f"[bold green]Sitemap has been generated and saved to {output_file}[/bold green]"
)
console.print(f"[green]Processed {total_files} files[/green]")
app = typer.Typer()
@app.command()
def main(
root_dir: str = typer.Option("docs", help="Root directory to traverse"),
output_file: str = typer.Option("sitemap.yaml", help="Output file for the sitemap"),
api_key: Optional[str] = typer.Option(None, help="OpenAI API key"),
max_concurrency: int = typer.Option(5, help="Maximum number of concurrent tasks"),
):
"""
Generate a sitemap with keywords, topics, and reference analysis.
"""
asyncio.run(generate_sitemap(root_dir, output_file, api_key, max_concurrency))
if __name__ == "__main__":
app()
+143
View File
@@ -0,0 +1,143 @@
#!/usr/bin/env python3
"""
Validate heading structure in documentation files.
Checks for:
- Multiple H1 tags (should only have one)
- Heading hierarchy violations (e.g., H1 → H3 skipping H2)
- Missing H1 tags
"""
import argparse
import re
from collections import defaultdict
from pathlib import Path
def find_markdown_files(docs_dir: Path) -> list[Path]:
"""Find all markdown files in the docs directory."""
return list(docs_dir.rglob("*.md"))
def extract_headings(content: str) -> list[tuple[int, str, int]]:
"""
Extract all headings from markdown content.
Returns:
List of (level, text, line_number) tuples
"""
headings = []
lines = content.split("\n")
for line_num, line in enumerate(lines, 1):
# Match markdown headings: # Title, ## Title, etc.
match = re.match(r"^(#{1,6})\s+(.+)$", line)
if match:
level = len(match.group(1))
text = match.group(2).strip()
headings.append((level, text, line_num))
return headings
def validate_headings(headings: list[tuple[int, str, int]]) -> dict[str, list[str]]:
"""Validate heading structure."""
issues = {}
if not headings:
issues["no_headings"] = ["No headings found in file"]
return issues
# Check for H1
h1_headings = [h for h in headings if h[0] == 1]
if not h1_headings:
issues["missing_h1"] = ["No H1 heading found"]
elif len(h1_headings) > 1:
issues["multiple_h1"] = [
f"Line {line}: {text}" for level, text, line in h1_headings
]
# Check heading hierarchy
prev_level = 0
hierarchy_violations = []
for level, text, line_num in headings:
if prev_level > 0 and level > prev_level + 1:
hierarchy_violations.append(
f"Line {line_num}: Skipped from H{prev_level} to H{level}: {text[:50]}"
)
prev_level = level
if hierarchy_violations:
issues["hierarchy_violations"] = hierarchy_violations
return issues
def process_file(file_path: Path) -> dict[str, list[str]]:
"""Process a single file and return issues."""
try:
content = file_path.read_text(encoding="utf-8")
headings = extract_headings(content)
return validate_headings(headings)
except Exception as e:
return {"error": [str(e)]}
def main():
parser = argparse.ArgumentParser(
description="Validate heading structure in documentation files"
)
parser.add_argument(
"--docs-dir",
type=Path,
default=Path("docs"),
help="Directory containing documentation files (default: docs)",
)
parser.add_argument(
"--summary",
action="store_true",
help="Show only summary statistics",
)
parser.add_argument(
"--file",
type=Path,
help="Validate a single file instead of all files",
)
args = parser.parse_args()
if args.file:
files = [args.file]
else:
files = find_markdown_files(args.docs_dir)
all_issues = {}
total_counts = defaultdict(int)
for file_path in files:
issues = process_file(file_path)
if issues:
all_issues[str(file_path)] = issues
for issue_type, messages in issues.items():
total_counts[issue_type] += len(messages)
if args.summary:
print("Summary Statistics:")
print("=" * 60)
for issue_type, count in sorted(total_counts.items()):
print(f" {issue_type.replace('_', ' ').title()}: {count}")
else:
# Detailed report
for file_path, issues in sorted(all_issues.items()):
print(f"\n{file_path}:")
for issue_type, messages in issues.items():
print(f" {issue_type.replace('_', ' ').title()}:")
for message in messages:
print(f" {message}")
print(f"\nTotal files checked: {len(files)}")
print(f"Files with issues: {len(all_issues)}")
if __name__ == "__main__":
main()
+195
View File
@@ -0,0 +1,195 @@
#!/usr/bin/env python3
"""
Validate frontmatter meta tags in documentation files.
Checks for:
- Missing title/description
- Title length (50-60 chars recommended)
- Description length (150-160 chars recommended)
- Duplicate titles/descriptions
"""
import argparse
import re
from collections import defaultdict
from pathlib import Path
from typing import Dict, List
def find_markdown_files(docs_dir: Path) -> List[Path]:
"""Find all markdown files in the docs directory."""
return list(docs_dir.rglob("*.md"))
def extract_frontmatter(content: str) -> Dict[str, str]:
"""Extract frontmatter from markdown content."""
frontmatter = {}
# Match YAML frontmatter between --- markers
match = re.match(r"^---\s*\n(.*?)\n---\s*\n", content, re.DOTALL)
if not match:
return frontmatter
yaml_content = match.group(1)
# Extract title
title_match = re.search(r"^title:\s*(.+)$", yaml_content, re.MULTILINE)
if title_match:
frontmatter["title"] = title_match.group(1).strip(" \"'")
# Extract description
desc_match = re.search(r"^description:\s*(.+)$", yaml_content, re.MULTILINE)
if desc_match:
frontmatter["description"] = desc_match.group(1).strip(" \"'")
# Extract keywords
keywords_match = re.search(r"^keywords:\s*(.+)$", yaml_content, re.MULTILINE)
if keywords_match:
frontmatter["keywords"] = keywords_match.group(1).strip(" \"'")
return frontmatter
def validate_file(file_path: Path) -> Dict[str, List[str]]:
"""Validate a single file's frontmatter."""
issues = {}
try:
content = file_path.read_text(encoding="utf-8")
frontmatter = extract_frontmatter(content)
# Check for missing frontmatter
if not frontmatter:
issues["missing_frontmatter"] = ["No frontmatter found"]
return issues
# Check title
if "title" not in frontmatter:
issues["missing_title"] = ["Title missing from frontmatter"]
else:
title = frontmatter["title"]
title_len = len(title)
if title_len < 50:
issues["title_too_short"] = [
f"Title is {title_len} chars (recommend 50-60 for SEO)"
]
elif title_len > 60:
issues["title_too_long"] = [
f"Title is {title_len} chars (recommend 50-60 for SEO)"
]
# Check description
if "description" not in frontmatter:
issues["missing_description"] = ["Description missing from frontmatter"]
else:
desc = frontmatter["description"]
desc_len = len(desc)
if desc_len < 150:
issues["description_too_short"] = [
f"Description is {desc_len} chars (recommend 150-160 for SEO)"
]
elif desc_len > 160:
issues["description_too_long"] = [
f"Description is {desc_len} chars (recommend 150-160 for SEO)"
]
return issues
except Exception as e:
return {"error": [str(e)]}
def main():
parser = argparse.ArgumentParser(
description="Validate frontmatter meta tags in documentation files"
)
parser.add_argument(
"--docs-dir",
type=Path,
default=Path("docs"),
help="Directory containing documentation files (default: docs)",
)
parser.add_argument(
"--summary",
action="store_true",
help="Show only summary statistics",
)
parser.add_argument(
"--file",
type=Path,
help="Validate a single file instead of all files",
)
parser.add_argument(
"--check-duplicates",
action="store_true",
help="Check for duplicate titles and descriptions",
)
args = parser.parse_args()
if args.file:
files = [args.file]
else:
files = find_markdown_files(args.docs_dir)
all_issues = {}
total_counts = defaultdict(int)
# Track titles and descriptions for duplicate checking
titles = defaultdict(list)
descriptions = defaultdict(list)
for file_path in files:
issues = validate_file(file_path)
if issues:
all_issues[str(file_path)] = issues
for issue_type, messages in issues.items():
total_counts[issue_type] += len(messages)
# Collect titles and descriptions for duplicate checking
if args.check_duplicates:
content = file_path.read_text(encoding="utf-8")
frontmatter = extract_frontmatter(content)
if "title" in frontmatter:
titles[frontmatter["title"]].append(str(file_path))
if "description" in frontmatter:
descriptions[frontmatter["description"]].append(str(file_path))
if args.summary:
print("Summary Statistics:")
print("=" * 60)
for issue_type, count in sorted(total_counts.items()):
print(f" {issue_type.replace('_', ' ').title()}: {count} files")
else:
# Detailed report
for file_path, issues in sorted(all_issues.items()):
print(f"\n{file_path}:")
for issue_type, messages in issues.items():
for message in messages:
print(f" - {message}")
# Check for duplicates
if args.check_duplicates:
print("\n" + "=" * 60)
print("Duplicate Titles:")
print("=" * 60)
for title, file_list in sorted(titles.items()):
if len(file_list) > 1:
print(f"\n{title}")
for f in file_list:
print(f" - {f}")
print("\n" + "=" * 60)
print("Duplicate Descriptions:")
print("=" * 60)
for desc, file_list in sorted(descriptions.items()):
if len(file_list) > 1:
print(f"\n{desc}")
for f in file_list:
print(f" - {f}")
print(f"\nTotal files checked: {len(files)}")
print(f"Files with issues: {len(all_issues)}")
if __name__ == "__main__":
main()