A Python package for collecting and extracting text from documents across multiple formats and sources.
Project description
CodexCollector
A Python package for collecting and extracting text from documents across multiple formats and sources.
Overview
CodexCollector provides a unified interface for document ingestion from both filesystem paths and web URLs. It automatically handles format detection, text extraction, metadata collection, and error logging, making it easy to build text corpora from diverse document sources.
Key Capabilities:
- Extract text from multiple document formats (Word, PowerPoint, PDF, plain text)
- Collect documents from local filesystems with recursive directory traversal
- Scrape and download documents from websites with configurable crawling depth
- Automatic encoding detection for text files
- Metadata extraction (creation dates) where available
- Comprehensive error logging for failed ingestions
- Rate limiting and timeout controls for web requests
- File size validation and filtering
Installation
Requirements
- Python ≥3.10
- Dependencies installed automatically via pip/uv
Using uv (recommended)
# Clone the repository
git clone https://github.com/NotAndroid37/codexcollector.git
cd codexcollector
# Install dependencies
uv sync
# Or install in development mode
uv pip install -e .
Using pip
# Clone the repository
git clone https://github.com/NotAndroid37/codexcollector.git
cd codexcollector
# Install dependencies
pip install -e .
Quick Start
Basic Filesystem Collection
from codexcollector import CodexCollector
# Initialize collector with default settings
collector = CodexCollector()
# Collect documents from a directory
corpus = collector.collect('/path/to/documents')
# Access collected documents
for doc_id, document in corpus.items():
print(f"Document {doc_id}: {document['filename']}")
print(f"Source: {document['source']}")
print(f"Text length: {len(document['text'])} characters")
print(f"Date: {document['date']}")
Basic Web Collection
from codexcollector import CodexCollector
# Configure for web scraping
collector = CodexCollector(
request_delay=2.0, # 2 seconds between requests
timeout=30,
max_crawl_depth=2 # Crawl starting page + 2 levels deep
)
# Collect documents from a website
corpus = collector.collect('https://example.com/docs')
# Check for errors
if collector.error_log:
print(f"Failed to process {len(collector.error_log)} files")
for source, error in collector.error_log.items():
print(f" {source}: {error}")
Configuration
CodexCollector Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
supported_extensions |
set[str] |
{'.pdf', '.docx', '.doc', '.pptx', '.ppt', '.txt', '.md', '.rtf'} |
File extensions to process (must include leading dot) |
max_file_size_mb |
int |
100 |
Maximum file size in MB (0 = no limit) |
request_delay |
float |
1.0 |
Delay between web requests in seconds |
timeout |
int |
30 |
HTTP request timeout in seconds |
excluded_dirs |
set[str] |
{'.git', '__pycache__', ...} |
Directory names to skip during traversal |
max_crawl_depth |
int |
2 |
Maximum depth for recursive web crawling (0 = starting page only) |
default_encoding |
str |
'utf-8' |
Default encoding when chardet fails or as fallback |
max_collection_time |
int |
0 |
Maximum collection time in seconds (0 = no limit) |
Default excluded directories:
{'.git', '__pycache__', '.pytest_cache', 'node_modules', '.vscode',
'.idea', 'venv', '.env', 'logs', 'tmp', 'temp'}
Advanced Usage
Custom File Filtering
from codexcollector import CodexCollector
# Only collect PDFs and Word documents, limit file size
collector = CodexCollector(
supported_extensions={'.pdf', '.docx'},
max_file_size_mb=25,
excluded_dirs={'archive', 'backup', 'temp'}
)
corpus = collector.collect('/data/documents')
Web Scraping with Depth Control
from codexcollector import CodexCollector
# Configure deep crawling with rate limiting
collector = CodexCollector(
max_crawl_depth=3, # Starting page + 3 levels
request_delay=1.5, # Respectful crawling
timeout=60,
max_collection_time=600 # 10 minute timeout
)
corpus = collector.collect('https://research-papers.example.com')
Progress Tracking
from codexcollector import CodexCollector
def progress_callback(current, total):
"""Called after each document is processed."""
percentage = (current / total) * 100
print(f"Progress: {current}/{total} ({percentage:.1f}%)")
collector = CodexCollector()
corpus = collector.collect(
'/large/document/collection',
progress_callback=progress_callback
)
Custom Text Encoding
from codexcollector import CodexCollector
# Use Latin-1 as fallback for Western European documents
collector = CodexCollector(
default_encoding='latin-1'
)
corpus = collector.collect('/path/to/european/docs')
Collection with Timeout
from codexcollector import CodexCollector
# Limit collection to 5 minutes
collector = CodexCollector(max_collection_time=300)
corpus = collector.collect('https://large-site.example.com')
print(f"Collected {len(corpus)} documents")
print(f"Errors: {len(collector.error_log)}")
Output Format
Corpus Structure
The collect() method returns a dictionary (corpus) mapping integer IDs to document dictionaries:
{
0: {
'filename': 'report.pdf',
'source': '/path/to/report.pdf',
'text': 'Document content...',
'date': '2024-01-15T10:30:00'
},
1: {
'filename': 'presentation.pptx',
'source': 'https://example.com/presentation.pptx',
'text': 'Slide text content...',
'date': None
}
}
Document Dictionary Fields
| Field | Type | Description |
|---|---|---|
filename |
str |
Name of the file extracted from path or URL |
source |
str | None |
Original source path or URL |
text |
str |
Extracted text content (empty string on failure) |
date |
str | None |
ISO format creation date from metadata, or None if unavailable |
Type Safety
CodexCollector provides TypedDict definitions for better type hints:
from codexcollector import Document, CorpusType
# Type-safe document creation
doc: Document = {
'filename': 'example.pdf',
'source': '/path/to/example.pdf',
'text': 'Content here',
'date': '2024-01-01T00:00:00'
}
# Type-safe corpus
corpus: CorpusType = {0: doc}
Error Handling
CodexCollector implements graceful degradation - individual document failures don't halt collection:
from codexcollector import CodexCollector
collector = CodexCollector()
corpus = collector.collect('/path/to/documents')
# Check error log for failures
if collector.error_log:
print(f"\nFailed to process {len(collector.error_log)} files:")
for source, error in collector.error_log.items():
print(f" {source}")
print(f" Error: {error}")
# Successful documents are still in the corpus
print(f"\nSuccessfully collected {len(corpus)} documents")
Custom Exceptions
from codex_exceptions import IngestionError, ConfigurationError
try:
collector = CodexCollector(max_file_size_mb=-1)
except ConfigurationError as e:
print(f"Invalid configuration: {e}")
# IngestionError is caught internally and logged to error_log
Supported File Formats
| Format | Extensions | Library | Metadata |
|---|---|---|---|
| Word (modern) | .docx |
python-docx |
Creation date |
| Word (legacy) | .doc |
docx2txt |
None |
| PowerPoint | .pptx, .ppt |
python-pptx |
Creation date |
.pdf |
pypdf |
Creation date | |
| Plain text | .txt, .md, .rtf |
Built-in | None |
Note: For files without embedded metadata, the file system modification time is used as a fallback for the date field.
Architecture
Design Patterns
Protocol-Based Extension: Text extractors implement the TextExtractor protocol, enabling easy addition of new format support without modifying core code.
class TextExtractor(Protocol):
def extract_text(self, file_path: Path) -> str: ...
def extract_metadata(self, file_path: Path) -> dict[str, str | None]: ...
Composition Over Inheritance: Format-specific extractors are composed into the collector rather than inherited, allowing runtime configuration and testing.
Key Methods
collect(source, progress_callback=None)- Main public API for document collection_is_url(source)- Detects whether source is URL or filesystem path_discover_files_from_path(path)- Recursive filesystem traversal_discover_files_from_url(url)- Web crawling with depth control_extract_from_file(file_path, source)- Format-specific text extraction_download_file(url, temp_path)- Streaming download for web sources
Dependencies
Core Dependencies
beautifulsoup4- HTML parsing for web scrapingchardet- Character encoding detectiondocx2txt- Legacy Word document supportpandas- Data structure supportpypdf- PDF text extractionpython-docx- Modern Word document parsingpython-pptx- PowerPoint parsingrequests- HTTP requests for web scraping
Development Dependencies
ipykernel- Jupyter notebook support
Testing
Run the included test suite:
# Run basic tests
uv run python test_codexcollector.py
# Or using pytest (if installed)
pytest test_codexcollector.py -v
The test suite covers:
- Initialization and configuration validation
- URL vs. file path detection
- Filename extraction from URLs
- Collection timeout checking
- Type safety with TypedDict
Performance Considerations
Large Collections
- Use
max_file_size_mbto skip oversized files - Set
max_collection_timeto prevent indefinite execution - Implement
progress_callbackfor monitoring
Web Scraping
- Adjust
request_delayto respect server rate limits (recommended: ≥1.0 seconds) - Use
max_crawl_depthto limit scope and prevent infinite crawling - Set reasonable
timeoutvalues for slow servers
Text Encoding
- Use
default_encodingto skip chardet for known-encoding collections - For homogeneous sources, setting a specific encoding improves performance
Limitations
- PDF extraction: No OCR support - scanned documents without text layers return empty strings
- Legacy formats:
.docand.pptfiles have limited metadata extraction - Web scraping: Respects same-domain restriction and doesn't handle JavaScript-rendered content
- Encoding detection: May fail on very short files or files with mixed encodings
- File formats: No support for EPUB, ODT, or other formats (extensible via Protocol)
Use Cases
Building Text Corpora
from codexcollector import CodexCollector
# Collect academic papers
collector = CodexCollector(supported_extensions={'.pdf'})
corpus = collector.collect('/research/papers')
# Extract just the text for analysis
texts = [doc['text'] for doc in corpus.values()]
Document Archive Mining
from codexcollector import CodexCollector
# Process large archive with progress tracking
def show_progress(current, total):
if current % 100 == 0:
print(f"Processed {current}/{total} documents...")
collector = CodexCollector(
max_file_size_mb=50,
excluded_dirs={'backup', 'archive_v1', 'duplicates'}
)
corpus = collector.collect(
'/corporate/document/archive',
progress_callback=show_progress
)
Web Documentation Collection
from codexcollector import CodexCollector
# Scrape technical documentation
collector = CodexCollector(
supported_extensions={'.pdf', '.docx'},
max_crawl_depth=3,
request_delay=2.0
)
corpus = collector.collect('https://docs.example.com/api')
Contributing
Contributions are welcome! Areas for improvement:
- Additional format support: EPUB, ODT, etc.
- OCR integration: For scanned PDFs
- Parallel processing: For large collections
- Incremental collection: Track processed files to avoid reprocessing
- Custom stopword filtering: At collection time
- Robots.txt compliance: Enhanced web scraping ethics
Development Process
- Fork the repository
- Create a feature branch (
git checkout -b feature/amazing-feature) - Add tests for new functionality
- Ensure all tests pass (
uv run python test_codexcollector.py) - Submit a pull request
Roadmap
- OCR integration for scanned documents (pytesseract)
- Parallel processing for large corpora (multiprocessing)
- Additional file format support (EPUB, ODT)
- Robots.txt compliance for web scraping
- Incremental collection with state persistence
- Content-type validation before download
- Language-specific encoding presets
- Custom extractor registration API
Known Issues
- Legacy Word (.doc) and PowerPoint (.ppt) files may not extract text correctly in all cases
- Web crawling doesn't handle JavaScript-rendered content
- Temporary files from web downloads are created in current directory (consider using tempfile module)
License
This project is licensed under the MIT License.
Citation
If you use CodexCollector in academic work, please cite:
[Add citation information]
Support
- Issues: GitHub Issues
- Documentation: [Full documentation link]
Built for seamless document collection and text extraction.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file codexcollector-0.1.0.tar.gz.
File metadata
- Download URL: codexcollector-0.1.0.tar.gz
- Upload date:
- Size: 117.4 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.13.5
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
4f43721cefe50d189e0f1ccc85b56491535ef65cd52f1f26c1589ee0c2c40e56
|
|
| MD5 |
0d1996dc8edc96c6e4120a18867d1a84
|
|
| BLAKE2b-256 |
b2bccdb558b7fc05b9fc8c58328de79de7b2f1a6de2917b995350b3520233f96
|
File details
Details for the file codexcollector-0.1.0-py3-none-any.whl.
File metadata
- Download URL: codexcollector-0.1.0-py3-none-any.whl
- Upload date:
- Size: 21.4 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.13.5
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
68d554f69e02a6147dd3e8e05f4b2a51b2c594fd9c1b36da9fe8b4757cc285c7
|
|
| MD5 |
d0857540257f149218fe05c243bf868b
|
|
| BLAKE2b-256 |
f2a8bb53a6e44ad4db51e3cfbf17833c6f10fe1856e792a58227e74df1ee45cf
|