diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md new file mode 100644 index 00000000..2d96b8ca --- /dev/null +++ b/.github/copilot-instructions.md @@ -0,0 +1,132 @@ +# Copilot Instructions for DocuElevate + +## Project Overview + +DocuElevate is an intelligent document processing system that automates handling, extraction, and processing of documents. It integrates with multiple cloud storage providers (Dropbox, Google Drive, OneDrive, S3, Nextcloud) and uses AI services (OpenAI, Azure Document Intelligence) for metadata extraction and OCR. + +## Tech Stack + +- **Backend**: FastAPI, SQLAlchemy, Celery, Redis +- **Frontend**: Jinja2 templates, Tailwind CSS +- **AI/ML**: OpenAI API, Azure Document Intelligence +- **Auth**: Authentik (OAuth2), Basic Auth +- **Infrastructure**: Docker, Docker Compose, Alembic (migrations) +- **Testing**: Pytest, pytest-asyncio, httpx + +## Core Principles + +### Code Quality +- Always use **Black** for formatting (line length: 120) +- Use **isort** with Black profile for import sorting +- Use **flake8** for linting (ignore E203, W503) +- Use **type hints** for all function parameters and return values +- Write **docstrings** for all public functions, classes, and modules +- Maintain **80% test coverage** for new code + +### Python Conventions +- Use descriptive variable names (e.g., `user_document_path`, not `udp`) +- Follow PEP 8 naming: `snake_case` for functions/variables, `PascalCase` for classes +- Use type hints from `typing` module (Dict, List, Optional, etc.) +- Prefer `pathlib.Path` over string paths for file operations +- Use f-strings for string formatting, not `.format()` or `%` +- Handle exceptions explicitly - avoid bare `except:` clauses + +### Security Best Practices +- **Never commit secrets or credentials** to the repository +- Use environment variables for sensitive configuration (see `.env.demo`) +- Validate and sanitize all user inputs +- Use parameterized queries with SQLAlchemy (never raw SQL with user input) +- Review [SECURITY_AUDIT.md](../SECURITY_AUDIT.md) before making security-related changes +- Run `bandit` to check for security issues in Python code + +### FastAPI Patterns +- Organize endpoints by feature in `app/api/` directory +- Use dependency injection for database sessions and authentication +- Return Pydantic models from endpoints for automatic validation +- Use proper HTTP status codes (200, 201, 400, 401, 403, 404, 500) +- Document endpoints with docstrings for OpenAPI documentation +- Use `async def` for I/O-bound operations + +### Database (SQLAlchemy) +- All models are defined in `app/models.py` +- Use Alembic for schema migrations (create migration for any model change) +- Use declarative base for models +- Define relationships with `relationship()` and proper `back_populates` +- Use database sessions from `app.database.get_db()` dependency +- Always close sessions in `finally` blocks or use context managers + +### Celery Tasks +- Define tasks in `app/tasks/` directory, organized by feature +- Use descriptive task names: `module.action` (e.g., `document.process_ocr`) +- Set appropriate retry policies and error handling +- Log progress and errors using Python's `logging` module +- Use `bind=True` for tasks that need access to task instance +- Keep tasks idempotent when possible + +### Frontend +- Templates are in `frontend/templates/` using Jinja2 +- Static files (CSS, JS, images) in `frontend/static/` +- Use Tailwind CSS utility classes (already configured) +- Keep JavaScript minimal - prefer server-side rendering +- Follow existing template structure and patterns + +### Testing +- Write tests in `tests/` directory, mirroring `app/` structure +- Use pytest markers: `@pytest.mark.unit`, `@pytest.mark.integration`, etc. +- Mock external services (OpenAI, Azure, cloud storage) in tests +- Use `pytest.fixture` for test setup and teardown +- Run tests with: `pytest -v` +- Check coverage with: `pytest --cov=app --cov-report=term-missing` + +### Configuration +- All configuration is in `app/config.py` using Pydantic Settings +- Use environment variables for configuration (12-factor app) +- Provide sensible defaults when possible +- Document all configuration options in `docs/ConfigurationGuide.md` + +### Documentation +- Keep documentation in `docs/` directory in Markdown format +- Update relevant docs when adding features or changing behavior +- User-facing documentation should be clear and include examples +- Reference existing docs: `docs/UserGuide.md`, `docs/API.md`, `docs/DeploymentGuide.md` +- See [AGENTIC_CODING.md](../AGENTIC_CODING.md) for detailed development guide + +### Error Handling +- Use custom exceptions defined in application (follow existing patterns) +- Log errors with context using Python's `logging` module +- Return user-friendly error messages in API responses +- Include error details in development, sanitize in production + +### Dependencies +- Add new dependencies to `requirements.txt` (production) or `requirements-dev.txt` (development) +- Document any new dependencies and their licenses in README.md +- Check for security vulnerabilities with `safety check` +- Pin major versions, allow minor updates (e.g., `fastapi>=0.100.0,<1.0.0`) + +### Git Workflow +- Write clear, descriptive commit messages +- Keep commits focused and atomic +- Run tests and linters before committing +- Pre-commit hooks are configured (`.pre-commit-config.yaml`) +- Follow conventional commits format when appropriate + +### File Organization +- Place API endpoints in `app/api/` organized by feature +- Background tasks go in `app/tasks/` +- Utility functions in `app/utils/` +- UI routes in `app/views/` +- Database models in `app/models.py` +- Configuration in `app/config.py` + +### Common Patterns +- Use `from typing import Optional, Dict, List, Any` for type hints +- Import FastAPI dependencies: `from fastapi import Depends, HTTPException, status` +- Get DB session: `db: Session = Depends(get_db)` +- Current user: `current_user: User = Depends(get_current_user)` +- Logger: `import logging; logger = logging.getLogger(__name__)` + +## Resources +- [AGENTIC_CODING.md](../AGENTIC_CODING.md) - Comprehensive development guide +- [CONTRIBUTING.md](../CONTRIBUTING.md) - Contribution guidelines +- [SECURITY_AUDIT.md](../SECURITY_AUDIT.md) - Security considerations +- [README.md](../README.md) - Project overview and quickstart diff --git a/.github/instructions/documentation.instructions.md b/.github/instructions/documentation.instructions.md new file mode 100644 index 00000000..d815a50d --- /dev/null +++ b/.github/instructions/documentation.instructions.md @@ -0,0 +1,267 @@ +--- +applyTo: "docs/**/*.md" +--- + +# Documentation Instructions + +These instructions apply to all documentation files in the `docs/` directory. + +## Documentation Structure +- User-facing documentation in `docs/` directory +- All documentation in Markdown format +- Follow existing documentation style and structure + +## Existing Documentation +- `docs/UserGuide.md` - How to use DocuElevate +- `docs/API.md` - API reference and examples +- `docs/DeploymentGuide.md` - Deployment instructions +- `docs/ConfigurationGuide.md` - Configuration options +- `docs/Troubleshooting.md` - Common issues and solutions +- `AGENTIC_CODING.md` - Development guide for AI agents +- `CONTRIBUTING.md` - Contribution guidelines +- `README.md` - Project overview and quickstart + +## Markdown Style + +### Headers +```markdown +# H1 - Document Title (only one per file) + +## H2 - Major Sections + +### H3 - Subsections + +#### H4 - Minor subsections (use sparingly) +``` + +### Code Blocks +Always specify the language for syntax highlighting: +````markdown +```python +def example_function(): + """Example Python code.""" + return "Hello, World!" +``` + +```bash +# Shell commands +docker-compose up -d +``` + +```json +{ + "key": "value" +} +``` +```` + +### Lists +```markdown +- Unordered list item 1 +- Unordered list item 2 + - Nested item + - Another nested item + +1. Ordered list item 1 +2. Ordered list item 2 +3. Ordered list item 3 +``` + +### Links +```markdown +[Link text](https://example.com) +[Internal link](./UserGuide.md) +[Link to section](#installation) +``` + +### Images +```markdown +![Alt text](path/to/image.png) + +
+ Descriptive alt text +

Image caption

+
+``` + +### Tables +```markdown +| Column 1 | Column 2 | Column 3 | +|----------|----------|----------| +| Value 1 | Value 2 | Value 3 | +| Value 4 | Value 5 | Value 6 | +``` + +### Admonitions and Notes +```markdown +> **Note:** This is an important note. + +> **Warning:** This is a warning message. + +> **Tip:** This is a helpful tip. +``` + +## Content Guidelines + +### Writing Style +- Use clear, concise language +- Write in second person (you/your) for user-facing docs +- Use present tense +- Avoid jargon; explain technical terms when necessary +- Use active voice +- Keep sentences short and focused + +### Documentation Types + +#### User Documentation +- Focus on **how to use** features, not implementation details +- Include step-by-step instructions +- Provide examples for common use cases +- Add screenshots or diagrams when helpful +- Explain what each feature does and when to use it + +Example: +```markdown +## Uploading Documents + +To upload a document to DocuElevate: + +1. Navigate to the Upload page +2. Click "Choose File" and select your document +3. Select the destination (Dropbox, Google Drive, etc.) +4. Click "Upload" + +The document will be automatically processed and stored in your selected destination. +``` + +#### API Documentation +- Document all endpoints with examples +- Show request and response formats +- Include authentication requirements +- Provide example curl commands +- Document error responses + +Example: +```markdown +### POST /api/documents/upload + +Upload a new document for processing. + +**Authentication:** Required + +**Request:** +```bash +curl -X POST "http://localhost:8000/api/documents/upload" \ + -H "Authorization: Bearer YOUR_TOKEN" \ + -F "file=@document.pdf" +``` + +**Response (201 Created):** +```json +{ + "id": 123, + "filename": "document.pdf", + "status": "processing" +} +``` +``` + +#### Configuration Documentation +- List all configuration options +- Provide default values +- Explain what each option does +- Include example configurations +- Note which options are required vs. optional + +Example: +```markdown +### OPENAI_API_KEY + +**Type:** String +**Required:** Yes +**Default:** None + +Your OpenAI API key for metadata extraction. + +```bash +OPENAI_API_KEY=sk-... +``` +``` + +#### Troubleshooting Documentation +- Start with the symptom/error +- Provide clear diagnosis steps +- Offer solutions +- Include common causes + +Example: +```markdown +### Error: "Connection refused" when starting services + +**Cause:** Docker services are not running or ports are already in use. + +**Solution:** +1. Check if Docker is running: `docker ps` +2. Check port availability: `lsof -i :8000` +3. Restart Docker services: `docker-compose restart` +``` + +## Code Examples +- Always test code examples before including them +- Use realistic examples that users can adapt +- Include comments explaining non-obvious parts +- Show complete examples, not just fragments + +## Version Information +- Update documentation when changing features +- Note version numbers when features are added +- Mark deprecated features clearly + +## Cross-References +- Link to related documentation +- Reference other sections when appropriate +- Keep the documentation interconnected + +Example: +```markdown +For deployment instructions, see the [Deployment Guide](./DeploymentGuide.md). + +For API details, refer to the [API Documentation](./API.md). +``` + +## Updating Documentation +When making code changes: +1. Update relevant documentation in the same PR +2. Check for outdated information +3. Add new sections for new features +4. Update examples if behavior changes +5. Review related documentation for consistency + +## Screenshots and Diagrams +- Use clear, high-quality images +- Annotate screenshots when helpful +- Keep diagrams simple and focused +- Update screenshots when UI changes +- Use consistent styling in diagrams + +## Accessibility +- Use descriptive alt text for images +- Ensure proper heading hierarchy +- Make links descriptive (avoid "click here") +- Use semantic formatting (bold, italic, code) appropriately + +## README.md Specific +- Keep README concise and focused on getting started +- Include badges for build status, version, license +- Show the most important features first +- Link to detailed documentation +- Include quick start instructions +- Add screenshots of the main interface + +## Configuration Guide Updates +When adding new configuration options: +- Add to `docs/ConfigurationGuide.md` +- Include type, default value, and description +- Provide example usage +- Note any dependencies on other config options +- Update `.env.demo` with the new option diff --git a/.github/instructions/frontend.instructions.md b/.github/instructions/frontend.instructions.md new file mode 100644 index 00000000..bc48770f --- /dev/null +++ b/.github/instructions/frontend.instructions.md @@ -0,0 +1,152 @@ +--- +applyTo: "frontend/**/*" +--- + +# Frontend Instructions + +These instructions apply to all files in the `frontend/` directory (templates, CSS, JavaScript, images). + +## Templates (Jinja2) + +### Location and Structure +- All templates in `frontend/templates/` +- Use template inheritance with `base.html` +- Keep templates organized by feature + +### Template Patterns +```jinja2 +{% extends "base.html" %} + +{% block title %}Document Upload - DocuElevate{% endblock %} + +{% block content %} +
+

{{ page_title }}

+ + {% if error_message %} +
+ {{ error_message }} +
+ {% endif %} + +
+ +
+
+{% endblock %} +``` + +### Tailwind CSS Usage +- Use Tailwind utility classes (already configured) +- Follow responsive design: `md:`, `lg:` breakpoints +- Use existing color scheme from the project +- Common patterns: + - Containers: `container mx-auto px-4` + - Buttons: `bg-blue-500 hover:bg-blue-700 text-white font-bold py-2 px-4 rounded` + - Cards: `bg-white shadow-md rounded-lg p-6` + - Forms: `w-full px-3 py-2 border rounded` + +### Static Files +- CSS files in `frontend/static/css/` +- JavaScript in `frontend/static/js/` +- Images in `frontend/static/images/` +- Reference with `{{ url_for('static', path='css/style.css') }}` + +### JavaScript +- Keep JavaScript minimal - prefer server-side rendering +- Use vanilla JavaScript or minimal dependencies +- Place scripts at the end of the body +- Use `defer` or `async` for external scripts +```html + +``` + +### Forms +- Use CSRF protection when needed +- Include proper validation +- Show clear error messages +- Use proper `method` (GET/POST) and `enctype` for file uploads +```html +
+
+ + +
+ +
+``` + +### Accessibility +- Use semantic HTML elements (`nav`, `main`, `article`, `section`) +- Include `alt` text for images +- Use proper heading hierarchy (h1 → h2 → h3) +- Add ARIA labels when needed +- Ensure keyboard navigation works + +### Error Handling +- Display user-friendly error messages +- Use flash messages for feedback +- Show loading states for async operations +```jinja2 +{% with messages = get_flashed_messages(with_categories=true) %} + {% if messages %} + {% for category, message in messages %} +
+ {{ message }} +
+ {% endfor %} + {% endif %} +{% endwith %} +``` + +### URL Generation +- Always use `url_for()` for URLs, never hardcode +- Examples: + - Routes: `{{ url_for('upload_document') }}` + - Static: `{{ url_for('static', path='css/style.css') }}` + - API: `{{ url_for('api_document', document_id=doc.id) }}` + +### Template Variables +- Check if variables exist before using them +- Use filters for formatting +```jinja2 +{% if document %} +

Uploaded: {{ document.created_at|datetime }}

+

Size: {{ document.file_size|filesizeformat }}

+{% else %} +

No document found

+{% endif %} +``` + +### Common Components +- Follow existing patterns for headers, footers, navigation +- Reuse template blocks and includes +- Keep components modular +```jinja2 +{% include 'components/navigation.html' %} +{% include 'components/document_card.html' with document=doc %} +``` + +## UI/UX Guidelines +- Maintain consistent spacing using Tailwind's scale (4, 8, 16, etc.) +- Use the existing color palette from the design +- Ensure mobile responsiveness +- Show loading indicators for long operations +- Provide feedback for user actions (success/error messages) +- Keep the interface clean and minimal + +## Performance +- Optimize images (compress, use appropriate formats) +- Minimize JavaScript bundle size +- Use lazy loading for images when appropriate +- Cache static assets diff --git a/.github/instructions/python-backend.instructions.md b/.github/instructions/python-backend.instructions.md new file mode 100644 index 00000000..3626507b --- /dev/null +++ b/.github/instructions/python-backend.instructions.md @@ -0,0 +1,164 @@ +--- +applyTo: "app/**/*.py" +--- + +# Python Backend Instructions + +These instructions apply to all Python code in the `app/` directory. + +## Code Style +- Use **Black** formatter with 120 character line length +- Use **isort** with Black profile for import organization +- Follow **flake8** rules (ignore E203, W503 as per `.pre-commit-config.yaml`) +- All functions must have type hints for parameters and return values +- Use `from typing import Optional, Dict, List, Any, Union` as needed + +## Import Order (isort with Black profile) +```python +# Standard library imports +import os +from pathlib import Path +from typing import Optional, Dict, List + +# Third-party imports +from fastapi import APIRouter, Depends, HTTPException +from sqlalchemy.orm import Session + +# Local application imports +from app.config import settings +from app.database import get_db +from app.models import Document, User +``` + +## Function Definitions +```python +def process_document( + file_path: Path, + user_id: int, + metadata: Optional[Dict[str, Any]] = None +) -> DocumentMetadata: + """ + Process a document and extract metadata. + + Args: + file_path: Path to the document file + user_id: ID of the user uploading the document + metadata: Optional additional metadata + + Returns: + DocumentMetadata object with extracted information + + Raises: + FileNotFoundError: If file doesn't exist + ProcessingError: If processing fails + """ + pass +``` + +## FastAPI Endpoints +- Use dependency injection for DB sessions and auth +- Return Pydantic models for automatic validation +- Use proper status codes from `fastapi.status` +- Add detailed docstrings for OpenAPI docs +```python +from fastapi import APIRouter, Depends, status +from sqlalchemy.orm import Session + +router = APIRouter(prefix="/api/documents", tags=["documents"]) + +@router.post("/", status_code=status.HTTP_201_CREATED) +async def create_document( + file: UploadFile, + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user) +) -> DocumentResponse: + """Create and process a new document.""" + pass +``` + +## Error Handling +- Use custom exceptions from the application +- Log errors with context using `logging.getLogger(__name__)` +- Return user-friendly error messages +- Never expose internal details in production errors +```python +import logging + +logger = logging.getLogger(__name__) + +try: + result = process_file(file_path) +except FileNotFoundError: + logger.error(f"File not found: {file_path}") + raise HTTPException(status_code=404, detail="File not found") +except Exception as e: + logger.exception(f"Error processing file: {str(e)}") + raise HTTPException(status_code=500, detail="Processing failed") +``` + +## Database Operations +- Use SQLAlchemy ORM, never raw SQL with user input +- Use `get_db()` dependency for sessions +- Always commit in try/except blocks +```python +from sqlalchemy.orm import Session +from app.database import get_db + +def create_document(db: Session, document_data: dict) -> Document: + """Create a new document in the database.""" + db_document = Document(**document_data) + try: + db.add(db_document) + db.commit() + db.refresh(db_document) + return db_document + except Exception as e: + db.rollback() + raise +``` + +## Celery Tasks +- Define in `app/tasks/` directory +- Use descriptive names: `module.action` +- Set retry policies +- Log progress and errors +```python +from celery import shared_task +import logging + +logger = logging.getLogger(__name__) + +@shared_task(bind=True, max_retries=3) +def process_ocr(self, document_id: int) -> Dict[str, Any]: + """Process OCR for a document.""" + try: + # Processing logic + logger.info(f"Processing OCR for document {document_id}") + return {"status": "success"} + except Exception as exc: + logger.exception(f"OCR processing failed for {document_id}") + raise self.retry(exc=exc, countdown=60) +``` + +## Configuration +- All settings in `app/config.py` using Pydantic Settings +- Use environment variables, never hardcode values +- Provide defaults when sensible +```python +from pydantic_settings import BaseSettings + +class Settings(BaseSettings): + openai_api_key: str + max_file_size: int = 10485760 # 10MB default + + class Config: + env_file = ".env" +``` + +## Security +- Never commit secrets +- Validate all user inputs +- Use parameterized queries +- Sanitize file paths +- Check file permissions +- Review SECURITY_AUDIT.md for guidelines diff --git a/.github/instructions/testing.instructions.md b/.github/instructions/testing.instructions.md new file mode 100644 index 00000000..37b3961b --- /dev/null +++ b/.github/instructions/testing.instructions.md @@ -0,0 +1,245 @@ +--- +applyTo: "tests/**/*.py" +--- + +# Testing Instructions + +These instructions apply to all test files in the `tests/` directory. + +## Test Organization +- Mirror the structure of `app/` directory in `tests/` +- Name test files with `test_` prefix (e.g., `test_api.py`) +- Group related tests in classes with `Test` prefix +- Use descriptive test function names: `test___` + +## Pytest Configuration +- Configuration in `pytest.ini` +- Run tests: `pytest -v` +- With coverage: `pytest --cov=app --cov-report=term-missing` +- Run specific markers: `pytest -m unit` or `pytest -m integration` + +## Test Markers +Use pytest markers to categorize tests: +```python +import pytest + +@pytest.mark.unit +def test_document_validation(): + """Test document validation logic.""" + pass + +@pytest.mark.integration +def test_document_upload_api(): + """Test document upload endpoint.""" + pass + +@pytest.mark.slow +def test_large_file_processing(): + """Test processing of large files.""" + pass + +@pytest.mark.requires_external +def test_openai_integration(): + """Test OpenAI API integration.""" + pass +``` + +Available markers: +- `unit` - Unit tests for individual functions/methods +- `integration` - Integration tests for API endpoints and workflows +- `slow` - Tests that take significant time to run +- `security` - Security-related tests +- `requires_external` - Tests requiring external services (OpenAI, Azure, etc.) +- `requires_db` - Tests requiring database +- `requires_redis` - Tests requiring Redis + +## Fixtures +Use pytest fixtures for test setup and teardown: +```python +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker +from app.database import Base + +@pytest.fixture +def db_session(): + """Provide a database session for tests.""" + engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(engine) + Session = sessionmaker(bind=engine) + session = Session() + + yield session + + session.close() + Base.metadata.drop_all(engine) + +@pytest.fixture +def sample_document(): + """Provide a sample document for tests.""" + return { + "filename": "test.pdf", + "content_type": "application/pdf", + "size": 1024 + } +``` + +## API Testing with FastAPI +Use `TestClient` from FastAPI: +```python +from fastapi.testclient import TestClient +from app.main import app + +client = TestClient(app) + +def test_upload_document(): + """Test document upload endpoint.""" + with open("tests/fixtures/sample.pdf", "rb") as f: + response = client.post( + "/api/documents/upload", + files={"file": ("test.pdf", f, "application/pdf")} + ) + + assert response.status_code == 201 + assert "id" in response.json() +``` + +## Async Testing +For async code, use `pytest-asyncio`: +```python +import pytest +import httpx + +@pytest.mark.asyncio +async def test_async_document_processing(): + """Test async document processing.""" + async with httpx.AsyncClient(app=app, base_url="http://test") as client: + response = await client.get("/api/documents/1") + assert response.status_code == 200 +``` + +## Mocking External Services +Always mock external services in tests: +```python +from unittest.mock import Mock, patch + +@pytest.mark.unit +def test_openai_metadata_extraction(mocker): + """Test metadata extraction with mocked OpenAI.""" + mock_response = { + "document_type": "invoice", + "amount": 100.00, + "date": "2024-01-01" + } + + mocker.patch( + "app.utils.openai_client.extract_metadata", + return_value=mock_response + ) + + result = extract_document_metadata("test.pdf") + assert result["document_type"] == "invoice" + +@pytest.mark.unit +def test_azure_ocr_processing(mocker): + """Test OCR with mocked Azure service.""" + mock_text = "Sample extracted text" + + mocker.patch( + "app.utils.azure_client.extract_text", + return_value=mock_text + ) + + result = perform_ocr("test.pdf") + assert result == mock_text +``` + +## Database Testing +```python +@pytest.mark.requires_db +def test_create_document(db_session): + """Test document creation in database.""" + from app.models import Document + + doc = Document( + filename="test.pdf", + user_id=1, + file_path="/tmp/test.pdf" + ) + db_session.add(doc) + db_session.commit() + + assert doc.id is not None + assert doc.filename == "test.pdf" +``` + +## Test Coverage Goals +- Aim for **80% code coverage** for all new code +- Focus on critical paths and error handling +- Test both success and failure scenarios +- Don't test third-party library code + +## Test Structure +Follow the Arrange-Act-Assert pattern: +```python +def test_document_validation(): + """Test that invalid documents are rejected.""" + # Arrange + invalid_document = { + "filename": "", # Empty filename + "size": -1 # Invalid size + } + + # Act + result = validate_document(invalid_document) + + # Assert + assert result.is_valid is False + assert "filename" in result.errors + assert "size" in result.errors +``` + +## Parameterized Tests +Use `pytest.mark.parametrize` for multiple test cases: +```python +@pytest.mark.parametrize("filename,expected", [ + ("document.pdf", True), + ("image.jpg", True), + ("script.exe", False), + ("", False), +]) +def test_allowed_file_types(filename, expected): + """Test file type validation.""" + result = is_allowed_file(filename) + assert result == expected +``` + +## Test Data +- Place test fixtures in `tests/fixtures/` directory +- Use small sample files for testing +- Don't commit large test files +- Clean up test files in teardown + +## Error Testing +Always test error conditions: +```python +def test_missing_file_raises_error(): + """Test that missing files raise appropriate error.""" + with pytest.raises(FileNotFoundError): + process_document("/nonexistent/file.pdf") + +def test_invalid_api_request(): + """Test API error handling.""" + response = client.post("/api/documents/", json={}) + assert response.status_code == 422 # Validation error +``` + +## Best Practices +- Test one thing per test function +- Use descriptive test names +- Keep tests independent (no dependencies between tests) +- Use fixtures for common setup +- Mock external dependencies +- Test edge cases and error conditions +- Keep tests fast (use mocks for slow operations) +- Clean up resources after tests diff --git a/app/api/__init__.py b/app/api/__init__.py index c2efba12..a621adb6 100644 --- a/app/api/__init__.py +++ b/app/api/__init__.py @@ -14,6 +14,7 @@ from app.api.dropbox import router as dropbox_router from app.api.openai import router as openai_router from app.api.azure import router as azure_router from app.api.google_drive import router as google_drive_router +from app.api.logs import router as logs_router # Set up logging logger = logging.getLogger(__name__) @@ -31,3 +32,4 @@ router.include_router(dropbox_router) router.include_router(openai_router) router.include_router(azure_router) router.include_router(google_drive_router) +router.include_router(logs_router) diff --git a/app/api/logs.py b/app/api/logs.py new file mode 100644 index 00000000..36170c88 --- /dev/null +++ b/app/api/logs.py @@ -0,0 +1,164 @@ +""" +Processing logs API endpoints +""" +from fastapi import APIRouter, Request, HTTPException, Depends, Query +from sqlalchemy.orm import Session +from sqlalchemy import desc +from typing import Optional +import logging + +from app.auth import require_login +from app.models import ProcessingLog, FileRecord +from app.api.common import get_db + +# Set up logging +logger = logging.getLogger(__name__) + +router = APIRouter() + +@router.get("/logs") +@require_login +def list_processing_logs( + request: Request, + db: Session = Depends(get_db), + file_id: Optional[int] = Query(None, description="Filter by file ID"), + task_id: Optional[str] = Query(None, description="Filter by task ID"), + limit: int = Query(100, ge=1, le=1000, description="Number of logs to return") +): + """ + Returns a JSON list of ProcessingLog entries. + Protected by `@require_login`, so only logged-in sessions can access. + + Query Parameters: + - file_id: Optional filter by file ID + - task_id: Optional filter by task ID + - limit: Maximum number of logs to return (default 100, max 1000) + + Example response: + [ + { + "id": 1, + "file_id": 123, + "task_id": "abc-123-def", + "step_name": "process_document", + "status": "success", + "message": "Processing completed", + "timestamp": "2025-05-01T12:34:56.789000" + }, + ... + ] + """ + query = db.query(ProcessingLog) + + # Apply filters + if file_id is not None: + query = query.filter(ProcessingLog.file_id == file_id) + if task_id is not None: + query = query.filter(ProcessingLog.task_id == task_id) + + # Order by timestamp descending and limit + logs = query.order_by(desc(ProcessingLog.timestamp)).limit(limit).all() + + # Return a simple list of dicts + result = [] + for log in logs: + result.append({ + "id": log.id, + "file_id": log.file_id, + "task_id": log.task_id, + "step_name": log.step_name, + "status": log.status, + "message": log.message, + "timestamp": log.timestamp.isoformat() if log.timestamp else None + }) + return result + +@router.get("/logs/file/{file_id}") +@require_login +def get_file_processing_logs( + request: Request, + file_id: int, + db: Session = Depends(get_db) +): + """ + Get all processing logs for a specific file. + Returns logs ordered by timestamp (oldest first to show processing flow). + + Also includes file metadata if the file exists. + """ + # Check if file exists + file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first() + if not file_record: + raise HTTPException( + status_code=404, + detail=f"File with ID {file_id} not found" + ) + + # Get all logs for this file + logs = db.query(ProcessingLog).filter( + ProcessingLog.file_id == file_id + ).order_by(ProcessingLog.timestamp).all() + + # Build response + log_list = [] + for log in logs: + log_list.append({ + "id": log.id, + "task_id": log.task_id, + "step_name": log.step_name, + "status": log.status, + "message": log.message, + "timestamp": log.timestamp.isoformat() if log.timestamp else None + }) + + return { + "file": { + "id": file_record.id, + "original_filename": file_record.original_filename, + "file_size": file_record.file_size, + "mime_type": file_record.mime_type, + "created_at": file_record.created_at.isoformat() if file_record.created_at else None + }, + "logs": log_list, + "total_logs": len(log_list) + } + +@router.get("/logs/task/{task_id}") +@require_login +def get_task_processing_logs( + request: Request, + task_id: str, + db: Session = Depends(get_db) +): + """ + Get all processing logs for a specific task. + Returns logs ordered by timestamp (oldest first to show processing flow). + """ + # Get all logs for this task + logs = db.query(ProcessingLog).filter( + ProcessingLog.task_id == task_id + ).order_by(ProcessingLog.timestamp).all() + + if not logs: + raise HTTPException( + status_code=404, + detail=f"No logs found for task {task_id}" + ) + + # Build response + log_list = [] + for log in logs: + log_list.append({ + "id": log.id, + "file_id": log.file_id, + "step_name": log.step_name, + "status": log.status, + "message": log.message, + "timestamp": log.timestamp.isoformat() if log.timestamp else None + }) + + return { + "task_id": task_id, + "logs": log_list, + "total_logs": len(log_list) + } diff --git a/app/tasks/convert_to_pdf.py b/app/tasks/convert_to_pdf.py index 34fec4a5..1631ea1f 100644 --- a/app/tasks/convert_to_pdf.py +++ b/app/tasks/convert_to_pdf.py @@ -7,25 +7,32 @@ import json from celery import shared_task from app.config import settings from app.tasks.process_document import process_document +from app.utils import log_task_progress logger = logging.getLogger(__name__) -@shared_task -def convert_to_pdf(file_path): +@shared_task(bind=True) +def convert_to_pdf(self, file_path): """ Converts a file to PDF using Gotenberg's API. Determines the appropriate Gotenberg endpoint based on the file's MIME type. On success, saves the PDF locally and enqueues it for processing. """ + task_id = self.request.id + logger.info(f"[{task_id}] Starting PDF conversion: {file_path}") + log_task_progress(task_id, "convert_to_pdf", "in_progress", f"Converting file: {os.path.basename(file_path)}") + gotenberg_url = getattr(settings, "gotenberg_url", None) if not gotenberg_url: - logger.error("Gotenberg URL is not configured in settings.") + logger.error(f"[{task_id}] Gotenberg URL is not configured in settings.") + log_task_progress(task_id, "convert_to_pdf", "failure", "Gotenberg URL not configured") return # Try to guess the MIME type based on file content and extension mime_type, encoding = mimetypes.guess_type(file_path) file_ext = os.path.splitext(file_path)[1].lower() - logger.info(f"Guessed MIME type for '{file_path}' is: {mime_type}, extension: {file_ext}") + logger.info(f"[{task_id}] Guessed MIME type for '{file_path}' is: {mime_type}, extension: {file_ext}") + log_task_progress(task_id, "detect_file_type", "success", f"File type: {mime_type or file_ext}") # Determine which Gotenberg endpoint to use endpoint = None @@ -146,11 +153,13 @@ def convert_to_pdf(file_path): logger.warning(f"Using fallback conversion for unknown type: {mime_type} / {file_ext}") if not endpoint: - logger.error(f"Could not determine Gotenberg endpoint for file type: {mime_type}") + logger.error(f"[{task_id}] Could not determine Gotenberg endpoint for file type: {mime_type}") + log_task_progress(task_id, "convert_to_pdf", "failure", f"Unknown file type: {mime_type}") return None try: - logger.info(f"Converting {file_path} using endpoint: {endpoint}") + logger.info(f"[{task_id}] Converting {file_path} using endpoint: {endpoint}") + log_task_progress(task_id, "call_gotenberg", "in_progress", "Calling Gotenberg API") # Send the conversion request to Gotenberg response = requests.post(endpoint, files=files, data=form_data) @@ -161,19 +170,25 @@ def convert_to_pdf(file_path): with open(converted_file_path, "wb") as out_file: out_file.write(response.content) - logger.info(f"Converted file saved as PDF: {converted_file_path}") + logger.info(f"[{task_id}] Converted file saved as PDF: {converted_file_path}") + log_task_progress(task_id, "call_gotenberg", "success", "PDF conversion successful") + log_task_progress(task_id, "convert_to_pdf", "success", f"Converted to PDF: {os.path.basename(converted_file_path)}") # Enqueue the PDF for further processing process_document.delay(converted_file_path) return converted_file_path else: + error_msg = f"Status code: {response.status_code}" logger.error( - f"Conversion failed for {file_path}. " - f"Status code: {response.status_code}, " + f"[{task_id}] Conversion failed for {file_path}. " + f"{error_msg}, " f"Response: {response.text[:500]}..." ) + log_task_progress(task_id, "call_gotenberg", "failure", error_msg) + log_task_progress(task_id, "convert_to_pdf", "failure", f"Conversion failed: {error_msg}") return None except Exception as e: - logger.exception(f"Error converting {file_path} to PDF: {e}") + logger.exception(f"[{task_id}] Error converting {file_path} to PDF: {e}") + log_task_progress(task_id, "convert_to_pdf", "failure", f"Exception: {str(e)}") return None diff --git a/app/tasks/embed_metadata_into_pdf.py b/app/tasks/embed_metadata_into_pdf.py index 25545c70..a33ae8c1 100644 --- a/app/tasks/embed_metadata_into_pdf.py +++ b/app/tasks/embed_metadata_into_pdf.py @@ -3,6 +3,7 @@ import os import shutil import tempfile +import logging import PyPDF2 # Replace fitz with PyPDF2 import json from app.config import settings @@ -11,6 +12,19 @@ from app.tasks.finalize_document_storage import finalize_document_storage # Import the shared Celery instance from app.celery_app import celery +from app.utils import log_task_progress +from app.database import SessionLocal +from app.models import FileRecord + +logger = logging.getLogger(__name__) + +# Directory constants - defined here to avoid hardcoded strings (BAN-B108) +# Note: These are application-specific subdirectories within settings.workdir, +# not system temporary directories. The workdir is a configurable path specific +# to this application. For actual temporary file creation, tempfile module is +# used (see line 70: tempfile.NamedTemporaryFile) +TMP_SUBDIR = "tmp" +PROCESSED_SUBDIR = "processed" def unique_filepath(directory, base_filename, extension=".pdf"): """ @@ -39,8 +53,8 @@ def persist_metadata(metadata, final_pdf_path): json.dump(metadata, f, ensure_ascii=False, indent=2) return json_path -@celery.task(base=BaseTaskWithRetry) -def embed_metadata_into_pdf(local_file_path: str, extracted_text: str, metadata: dict): +@celery.task(base=BaseTaskWithRetry, bind=True) +def embed_metadata_into_pdf(self, local_file_path: str, extracted_text: str, metadata: dict, file_id: int = None): """ Embeds extracted metadata into the PDF's standard metadata fields. The mapping is as follows: @@ -54,13 +68,25 @@ def embed_metadata_into_pdf(local_file_path: str, extracted_text: str, metadata: where is derived from metadata["filename"]. Additionally, the metadata is persisted to a JSON file with the same base name. """ + task_id = self.request.id + logger.info(f"[{task_id}] Starting metadata embedding for: {local_file_path}") + log_task_progress(task_id, "embed_metadata_into_pdf", "in_progress", f"Embedding metadata into {os.path.basename(local_file_path)}", file_id=file_id) + + # Get file_id from database if not provided (fallback only, prefer passing file_id explicitly) + if file_id is None: + with SessionLocal() as db: + file_record = db.query(FileRecord).filter_by(local_filename=local_file_path).first() + if file_record: + file_id = file_record.id + # Check for file existence; if not found, try the known shared tmp directory. if not os.path.exists(local_file_path): - alt_path = os.path.join(settings.workdir, "tmp", os.path.basename(local_file_path)) + alt_path = os.path.join(settings.workdir, TMP_SUBDIR, os.path.basename(local_file_path)) if os.path.exists(alt_path): local_file_path = alt_path else: - print(f"[ERROR] Local file {local_file_path} not found, cannot embed metadata.") + logger.error(f"[{task_id}] Local file {local_file_path} not found, cannot embed metadata.") + log_task_progress(task_id, "embed_metadata_into_pdf", "failure", "File not found", file_id=file_id) return {"error": "File not found"} # Work on a safe copy in a secure temporary directory @@ -75,7 +101,8 @@ def embed_metadata_into_pdf(local_file_path: str, extracted_text: str, metadata: shutil.copy(original_file, processed_file) try: - print(f"[DEBUG] Embedding metadata into {processed_file}...") + logger.info(f"[{task_id}] Embedding metadata into {processed_file}...") + log_task_progress(task_id, "modify_pdf", "in_progress", "Modifying PDF metadata", file_id=file_id) # Open the PDF and modify metadata with open(processed_file, 'rb') as file: @@ -98,49 +125,59 @@ def embed_metadata_into_pdf(local_file_path: str, extracted_text: str, metadata: with open(processed_file, 'wb') as output_file: pdf_writer.write(output_file) - print(f"[INFO] Metadata embedded successfully in {processed_file}") + logger.info(f"[{task_id}] Metadata embedded successfully in {processed_file}") + log_task_progress(task_id, "modify_pdf", "success", "PDF metadata embedded", file_id=file_id) # Use the suggested filename from metadata; if not provided, use the original basename. suggested_filename = metadata.get("filename", os.path.splitext(os.path.basename(local_file_path))[0]) # Remove any extension and then add .pdf suggested_filename = os.path.splitext(suggested_filename)[0] # Define the final directory based on settings.workdir and ensure it exists. - final_dir = os.path.join(settings.workdir, "processed") + final_dir = os.path.join(settings.workdir, PROCESSED_SUBDIR) os.makedirs(final_dir, exist_ok=True) # Get a unique filepath in case of collisions. final_file_path = unique_filepath(final_dir, suggested_filename, extension=".pdf") + logger.info(f"[{task_id}] Moving file to: {final_file_path}") + log_task_progress(task_id, "move_to_processed", "in_progress", f"Moving to processed: {suggested_filename}.pdf", file_id=file_id) # Move the processed file using shutil.move to handle cross-device moves. shutil.move(processed_file, final_file_path) # Ensure the temporary file is deleted if it still exists. if os.path.exists(processed_file): os.remove(processed_file) + log_task_progress(task_id, "move_to_processed", "success", f"Moved to: {os.path.basename(final_file_path)}", file_id=file_id) # Persist the metadata into a JSON file with the same base name. + logger.info(f"[{task_id}] Persisting metadata to JSON") + log_task_progress(task_id, "save_metadata_json", "in_progress", "Saving metadata JSON", file_id=file_id) json_path = persist_metadata(metadata, final_file_path) - print(f"[INFO] Metadata persisted to {json_path}") + logger.info(f"[{task_id}] Metadata persisted to {json_path}") + log_task_progress(task_id, "save_metadata_json", "success", f"Saved: {os.path.basename(json_path)}", file_id=file_id) # Trigger the next step: final storage. - finalize_document_storage.delay(original_file, final_file_path, metadata) + logger.info(f"[{task_id}] Queueing final storage task") + log_task_progress(task_id, "embed_metadata_into_pdf", "success", "Metadata embedded, queuing finalization", file_id=file_id) + finalize_document_storage.delay(original_file, final_file_path, metadata, file_id) # After triggering final storage, delete the original file if it is in workdir/tmp. - workdir_tmp = os.path.join(settings.workdir, "tmp") + workdir_tmp = os.path.join(settings.workdir, TMP_SUBDIR) if original_file.startswith(workdir_tmp) and os.path.exists(original_file): try: os.remove(original_file) - print(f"[INFO] Deleted original file from {original_file}") + logger.info(f"[{task_id}] Deleted original file from {original_file}") except Exception as e: - print(f"[ERROR] Could not delete original file {original_file}: {e}") + logger.error(f"[{task_id}] Could not delete original file {original_file}: {e}") return {"file": final_file_path, "metadata_file": json_path, "status": "Metadata embedded"} except Exception as e: - print(f"[ERROR] Failed to embed metadata into {processed_file}: {e}") + logger.exception(f"[{task_id}] Failed to embed metadata into {processed_file}: {e}") + log_task_progress(task_id, "embed_metadata_into_pdf", "failure", f"Exception: {str(e)}", file_id=file_id) # Clean up temporary file in case of error if os.path.exists(processed_file): try: os.remove(processed_file) - print(f"[INFO] Cleaned up temporary file {processed_file}") + logger.info(f"[{task_id}] Cleaned up temporary file {processed_file}") except Exception as cleanup_error: - print(f"[ERROR] Could not clean up temporary file {processed_file}: {cleanup_error}") + logger.error(f"[{task_id}] Could not clean up temporary file {processed_file}: {cleanup_error}") return {"error": str(e)} diff --git a/app/tasks/extract_metadata_with_gpt.py b/app/tasks/extract_metadata_with_gpt.py index 2aab5283..52d465bb 100644 --- a/app/tasks/extract_metadata_with_gpt.py +++ b/app/tasks/extract_metadata_with_gpt.py @@ -2,6 +2,7 @@ import json import re +import os from app.config import settings from app.tasks.retry_config import BaseTaskWithRetry from app.tasks.embed_metadata_into_pdf import embed_metadata_into_pdf @@ -10,6 +11,9 @@ from app.tasks.embed_metadata_into_pdf import embed_metadata_into_pdf from app.celery_app import celery import openai import logging +from app.utils import log_task_progress +from app.database import SessionLocal +from app.models import FileRecord logger = logging.getLogger(__name__) @@ -41,9 +45,23 @@ def extract_json_from_text(text): return text[start:end+1] return None -@celery.task(base=BaseTaskWithRetry) -def extract_metadata_with_gpt(filename: str, cleaned_text: str): +@celery.task(base=BaseTaskWithRetry, bind=True) +def extract_metadata_with_gpt(self, filename: str, cleaned_text: str, file_id: int = None): """Uses OpenAI to classify document metadata.""" + task_id = self.request.id + logger.info(f"[{task_id}] Starting metadata extraction for: {filename}") + log_task_progress(task_id, "extract_metadata_with_gpt", "in_progress", f"Extracting metadata for {filename}", file_id=file_id) + + # Get file_id from database if not provided + if file_id is None: + tmp_dir = os.path.join(settings.workdir, "tmp") + file_path = os.path.join(tmp_dir, filename) + if os.path.exists(file_path): + with SessionLocal() as db: + file_record = db.query(FileRecord).filter_by(local_filename=file_path).first() + if file_record: + file_id = file_record.id + prompt = f""" You are a specialized document analyzer trained to extract structured metadata from documents. Your task is to analyze the given text and return a well-structured JSON object. @@ -77,7 +95,8 @@ Return only valid JSON with no additional commentary. """ try: - print(f"[DEBUG] Sending classification request for {filename}...") + logger.info(f"[{task_id}] Sending classification request for {filename}...") + log_task_progress(task_id, "call_openai", "in_progress", "Calling OpenAI API", file_id=file_id) completion = client.chat.completions.create( model=settings.openai_model, messages=[ @@ -88,21 +107,27 @@ Return only valid JSON with no additional commentary. ) content = completion.choices[0].message.content - print(f"[DEBUG] Raw classification response for {filename}: {content}") + logger.info(f"[{task_id}] Raw classification response for {filename}: {content[:200]}...") + log_task_progress(task_id, "call_openai", "success", "Received OpenAI response", file_id=file_id) json_text = extract_json_from_text(content) if not json_text: - print(f"[ERROR] Could not find valid JSON in GPT response for {filename}.") + logger.error(f"[{task_id}] Could not find valid JSON in GPT response for {filename}.") + log_task_progress(task_id, "extract_metadata_with_gpt", "failure", "Invalid JSON in response", file_id=file_id) return {} metadata = json.loads(json_text) - print(f"[DEBUG] Extracted metadata: {metadata}") + logger.info(f"[{task_id}] Extracted metadata: {metadata}") + log_task_progress(task_id, "parse_metadata", "success", f"Parsed metadata: {list(metadata.keys())}", file_id=file_id) # Trigger the next step: embedding metadata into the PDF - embed_metadata_into_pdf.delay(filename, cleaned_text, metadata) + logger.info(f"[{task_id}] Queueing metadata embedding task") + log_task_progress(task_id, "extract_metadata_with_gpt", "success", "Metadata extracted, queuing embed task", file_id=file_id) + embed_metadata_into_pdf.delay(filename, cleaned_text, metadata, file_id) return {"s3_file": filename, "metadata": metadata} except Exception as e: - print(f"[ERROR] OpenAI classification failed for {filename}: {e}") + logger.exception(f"[{task_id}] OpenAI classification failed for {filename}: {e}") + log_task_progress(task_id, "extract_metadata_with_gpt", "failure", f"Exception: {str(e)}", file_id=file_id) return {} diff --git a/app/tasks/finalize_document_storage.py b/app/tasks/finalize_document_storage.py index e81a4be3..5f672423 100644 --- a/app/tasks/finalize_document_storage.py +++ b/app/tasks/finalize_document_storage.py @@ -1,28 +1,52 @@ #!/usr/bin/env python3 +import logging import os from app.config import settings from app.tasks.retry_config import BaseTaskWithRetry # Import the shared Celery instance from app.celery_app import celery -# 1) Import the aggregator task +# Import the aggregator task and validator from app.tasks.send_to_all import send_to_all_destinations, get_configured_services_from_validator # Import notification utility from app.utils.notification import notify_file_processed +# Import database and logging utils from main +from app.utils import log_task_progress +from app.database import SessionLocal +from app.models import FileRecord -@celery.task(base=BaseTaskWithRetry) -def finalize_document_storage(original_file: str, processed_file: str, metadata: dict): +logger = logging.getLogger(__name__) + + +@celery.task(base=BaseTaskWithRetry, bind=True) +def finalize_document_storage(self, original_file: str, processed_file: str, metadata: dict, file_id: int = None): """ Final storage step after embedding metadata. We will now call 'send_to_all_destinations' to push the final PDF to Dropbox/Nextcloud/Paperless. After uploading, send a notification about the processed file. """ - print(f"[INFO] Finalizing document storage for {processed_file}") + task_id = self.request.id + logger.info(f"[{task_id}] Finalizing document storage for {processed_file}") + + # 1. Update Database Status (From Main) + log_task_progress(task_id, "finalize_document_storage", "in_progress", f"Finalizing: {os.path.basename(processed_file)}", file_id=file_id) + + # Get file_id from database if not provided (fallback logic from Main) + if file_id is None: + with SessionLocal() as db: + # Only as a last resort, try to find by exact match on local_filename + tmp_path = os.path.join(settings.workdir, "tmp", os.path.basename(original_file)) + file_record = db.query(FileRecord).filter( + FileRecord.local_filename == tmp_path + ).first() + if file_record: + file_id = file_record.id - # Determine which destinations are configured + # 2. Determine Configured Destinations (From Copilot) + # This is needed for the notification message later configured_destinations = [] try: configured_services = get_configured_services_from_validator() @@ -33,16 +57,21 @@ def finalize_document_storage(original_file: str, processed_file: str, metadata: display_name = service_name.replace('_', ' ').title() configured_destinations.append(display_name) except Exception as e: - print(f"[WARNING] Could not determine configured destinations: {e}") + logger.warning(f"[WARNING] Could not determine configured destinations: {e}") configured_destinations = ["configured destinations"] - # 2) Enqueue uploads to all destinations (Dropbox, Nextcloud, Paperless) + # 3. Queue Uploads (Merged) + # Uses Main branch signature to ensure file_id is passed, but keeps logic structure + logger.info(f"[{task_id}] Queueing uploads to all destinations") + log_task_progress(task_id, "finalize_document_storage", "success", "Queuing uploads to destinations", file_id=file_id) + # Note: send_to_all_destinations is asynchronous and queues upload tasks - send_to_all_destinations.delay(processed_file) + # We pass 'True' (delete_after) and 'file_id' as per Main branch requirements + send_to_all_destinations.delay(processed_file, True, file_id) - # 3) Send notification about successful file processing + # 4. Send Notification (From Copilot) # Note: This notification is sent after processing is complete but while uploads - # are being queued. The message reflects that uploads are being initiated. + # are being queued. try: # Get file information file_size = os.path.getsize(processed_file) if os.path.exists(processed_file) else 0 @@ -55,9 +84,9 @@ def finalize_document_storage(original_file: str, processed_file: str, metadata: destinations=configured_destinations ) except Exception as e: - print(f"[WARNING] Failed to send file processed notification: {e}") + logger.warning(f"[WARNING] Failed to send file processed notification: {e}") return { "status": "Completed", "file": processed_file - } + } \ No newline at end of file diff --git a/app/tasks/process_document.py b/app/tasks/process_document.py index fa795600..05cfa74b 100644 --- a/app/tasks/process_document.py +++ b/app/tasks/process_document.py @@ -4,6 +4,7 @@ import os import uuid import shutil import mimetypes +import logging import PyPDF2 # Replace fitz with PyPDF2 from app.config import settings @@ -13,11 +14,13 @@ from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt from app.celery_app import celery from app.database import SessionLocal from app.models import FileRecord -from app.utils import hash_file +from app.utils import hash_file, log_task_progress + +logger = logging.getLogger(__name__) -@celery.task(base=BaseTaskWithRetry) -def process_document(original_local_file: str): +@celery.task(base=BaseTaskWithRetry, bind=True) +def process_document(self, original_local_file: str): """ Process a document file and trigger appropriate text extraction. @@ -28,24 +31,34 @@ def process_document(original_local_file: str): - Check for embedded text. If present, run local GPT extraction - Otherwise, queue Azure Document Intelligence processing """ + task_id = self.request.id + logger.info(f"[{task_id}] Starting document processing: {original_local_file}") + log_task_progress(task_id, "process_document", "in_progress", f"Processing file: {original_local_file}") if not os.path.exists(original_local_file): - print(f"[ERROR] File {original_local_file} not found.") + logger.error(f"[{task_id}] File {original_local_file} not found.") + log_task_progress(task_id, "process_document", "failure", "File not found") return {"error": "File not found"} # 0. Compute the file hash and check for duplicates + logger.info(f"[{task_id}] Computing file hash...") + log_task_progress(task_id, "hash_file", "in_progress", "Computing file hash") filehash = hash_file(original_local_file) original_filename = os.path.basename(original_local_file) file_size = os.path.getsize(original_local_file) mime_type, _ = mimetypes.guess_type(original_local_file) if not mime_type: mime_type = "application/octet-stream" + + logger.info(f"[{task_id}] File hash: {filehash[:10]}..., Size: {file_size} bytes, MIME: {mime_type}") + log_task_progress(task_id, "hash_file", "success", f"Hash: {filehash[:10]}..., Size: {file_size} bytes") # Acquire DB session in the task with SessionLocal() as db: existing = db.query(FileRecord).filter_by(filehash=filehash).one_or_none() if existing: - print(f"[INFO] Duplicate file detected (hash={filehash[:10]}...) Skipping processing.") + logger.info(f"[{task_id}] Duplicate file detected (hash={filehash[:10]}...) Skipping processing.") + log_task_progress(task_id, "process_document", "success", "Duplicate file detected, skipping", file_id=existing.id) return { "status": "duplicate_file", "file_id": existing.id, @@ -53,6 +66,8 @@ def process_document(original_local_file: str): } # Not a duplicate -> insert a new record + logger.info(f"[{task_id}] Creating new file record in database") + log_task_progress(task_id, "create_file_record", "in_progress", "Creating file record") new_record = FileRecord( filehash=filehash, original_filename=original_filename, @@ -63,6 +78,8 @@ def process_document(original_local_file: str): db.add(new_record) db.commit() db.refresh(new_record) + logger.info(f"[{task_id}] File record created with ID: {new_record.id}") + log_task_progress(task_id, "create_file_record", "success", f"File record ID: {new_record.id}", file_id=new_record.id) # 1. Generate a UUID-based filename and place it in /workdir/tmp file_ext = os.path.splitext(original_local_file)[1] @@ -73,14 +90,19 @@ def process_document(original_local_file: str): os.makedirs(tmp_dir, exist_ok=True) new_local_path = os.path.join(tmp_dir, new_filename) + logger.info(f"[{task_id}] Copying file to: {new_local_path}") + log_task_progress(task_id, "copy_file", "in_progress", f"Copying file to {new_filename}", file_id=new_record.id) # Copy the file instead of moving it shutil.copy(original_local_file, new_local_path) + log_task_progress(task_id, "copy_file", "success", f"File copied to {new_filename}", file_id=new_record.id) # Update the DB with final local filename new_record.local_filename = new_local_path db.commit() # 2. Check for embedded text (outside the DB session to avoid long open transactions) + logger.info(f"[{task_id}] Checking for embedded text in PDF") + log_task_progress(task_id, "check_text", "in_progress", "Checking for embedded text", file_id=new_record.id) with open(new_local_path, 'rb') as file: pdf_reader = PyPDF2.PdfReader(file) has_text = False @@ -90,19 +112,30 @@ def process_document(original_local_file: str): break if has_text: - print(f"[INFO] PDF {original_local_file} contains embedded text. Processing locally.") + logger.info(f"[{task_id}] PDF {original_local_file} contains embedded text. Processing locally.") + log_task_progress(task_id, "check_text", "success", "Embedded text found, extracting locally", file_id=new_record.id) # Extract text locally + logger.info(f"[{task_id}] Extracting text from PDF") + log_task_progress(task_id, "extract_text", "in_progress", "Extracting text locally", file_id=new_record.id) extracted_text = "" with open(new_local_path, 'rb') as file: pdf_reader = PyPDF2.PdfReader(file) for page in pdf_reader.pages: extracted_text += page.extract_text() + "\n" + + logger.info(f"[{task_id}] Extracted {len(extracted_text)} characters") + log_task_progress(task_id, "extract_text", "success", f"Extracted {len(extracted_text)} characters", file_id=new_record.id) # Call metadata extraction directly - extract_metadata_with_gpt.delay(new_filename, extracted_text) - return {"file": new_local_path, "status": "Text extracted locally"} + logger.info(f"[{task_id}] Queueing metadata extraction") + log_task_progress(task_id, "process_document", "success", "Queued for metadata extraction", file_id=new_record.id) + extract_metadata_with_gpt.delay(new_filename, extracted_text, new_record.id) + return {"file": new_local_path, "status": "Text extracted locally", "file_id": new_record.id} # 3. If no embedded text, queue Azure Document Intelligence processing - process_with_azure_document_intelligence.delay(new_filename) - return {"file": new_local_path, "status": "Queued for OCR"} + logger.info(f"[{task_id}] No embedded text found. Queueing Azure Document Intelligence processing") + log_task_progress(task_id, "check_text", "success", "No embedded text, queuing OCR", file_id=new_record.id) + log_task_progress(task_id, "process_document", "success", "Queued for OCR processing", file_id=new_record.id) + process_with_azure_document_intelligence.delay(new_filename, new_record.id) + return {"file": new_local_path, "status": "Queued for OCR", "file_id": new_record.id} diff --git a/app/tasks/process_with_azure_document_intelligence.py b/app/tasks/process_with_azure_document_intelligence.py index 1dbed541..dd799ccb 100644 --- a/app/tasks/process_with_azure_document_intelligence.py +++ b/app/tasks/process_with_azure_document_intelligence.py @@ -76,7 +76,7 @@ def check_page_rotation(result, filename): return rotation_data @celery.task(base=BaseTaskWithRetry) -def process_with_azure_document_intelligence(filename: str): +def process_with_azure_document_intelligence(filename: str, file_id: int = None): """ Processes a PDF document using Azure Document Intelligence and overlays OCR text onto the local temporary file (stored under /tmp). @@ -88,6 +88,10 @@ def process_with_azure_document_intelligence(filename: str): 3. Saves the OCR-processed PDF locally in the same location as before. 4. Checks for page rotation and triggers page rotation if needed. 5. Triggers downstream metadata extraction. + + Args: + filename: Name of the file to process + file_id: Optional file ID to pass through to subsequent tasks """ try: tmp_file_path = os.path.join(settings.workdir, "tmp", filename) @@ -139,7 +143,7 @@ def process_with_azure_document_intelligence(filename: str): logger.info(f"Extracted text for {filename}: {len(extracted_text)} characters") # Trigger page rotation task if rotation is detected, otherwise proceed to metadata extraction - rotate_pdf_pages.delay(filename, extracted_text, rotation_data) + rotate_pdf_pages.delay(filename, extracted_text, rotation_data, file_id) return {"file": filename, "searchable_pdf": searchable_pdf_path, "cleaned_text": extracted_text} except Exception as e: diff --git a/app/tasks/rotate_pdf_pages.py b/app/tasks/rotate_pdf_pages.py index 6ac38d4e..f96bb5c0 100644 --- a/app/tasks/rotate_pdf_pages.py +++ b/app/tasks/rotate_pdf_pages.py @@ -47,7 +47,7 @@ def determine_rotation_angle(detected_angle): return rotation_value @celery.task(base=BaseTaskWithRetry) -def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None): +def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None, file_id: int = None): """ Rotates pages in a PDF document based on detected rotation angles. @@ -55,6 +55,7 @@ def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None): filename: The name of the file to rotate extracted_text: The extracted text from the document rotation_data: Optional rotation data dictionary {page_index: angle} + file_id: Optional file ID to pass through to subsequent tasks """ try: pdf_path = os.path.join(settings.workdir, "tmp", filename) @@ -64,7 +65,7 @@ def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None): # Skip rotation if no rotation data provided if not rotation_data: logger.info(f"No rotation data provided for {filename}, proceeding with metadata extraction") - extract_metadata_with_gpt.delay(filename, extracted_text) + extract_metadata_with_gpt.delay(filename, extracted_text, file_id) return {"file": filename, "status": "no_rotation_needed"} # Standardize rotation_data keys to integers @@ -77,7 +78,7 @@ def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None): if not any(abs(angle) > 0 for angle in normalized_rotation_data.values()): logger.info(f"No significant rotations detected in {filename}, proceeding with metadata extraction") - extract_metadata_with_gpt.delay(filename, extracted_text) + extract_metadata_with_gpt.delay(filename, extracted_text, file_id) return {"file": filename, "status": "no_rotation_needed"} logger.info(f"Rotating {len(normalized_rotation_data)} pages in {filename}") @@ -117,7 +118,7 @@ def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None): logger.info(f"Detected rotations in {filename} but no rotations were actually applied (angles too small or not multiples of 90°)") # Continue with metadata extraction - extract_metadata_with_gpt.delay(filename, extracted_text) + extract_metadata_with_gpt.delay(filename, extracted_text, file_id) return { "file": filename, @@ -129,5 +130,5 @@ def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None): except Exception as e: logger.error(f"Error rotating PDF {filename}: {e}") # Continue with metadata extraction despite rotation failure - extract_metadata_with_gpt.delay(filename, extracted_text) + extract_metadata_with_gpt.delay(filename, extracted_text, file_id) return {"file": filename, "status": "rotation_failed", "error": str(e)} diff --git a/app/tasks/send_to_all.py b/app/tasks/send_to_all.py index ba9eab89..33a9774f 100644 --- a/app/tasks/send_to_all.py +++ b/app/tasks/send_to_all.py @@ -16,6 +16,9 @@ from app.tasks.upload_to_onedrive import upload_to_onedrive from app.tasks.upload_to_s3 import upload_to_s3 from app.utils.config_validator import get_provider_status from app.celery_app import celery +from app.utils import log_task_progress +from app.database import SessionLocal +from app.models import FileRecord logger = logging.getLogger(__name__) @@ -104,8 +107,8 @@ def get_configured_services_from_validator(): return result -@celery.task(base=BaseTaskWithRetry) -def send_to_all_destinations(file_path: str, use_validator=True): +@celery.task(base=BaseTaskWithRetry, bind=True) +def send_to_all_destinations(self, file_path: str, use_validator=True, file_id: int = None): """ Distribute a file to all configured storage destinations. @@ -113,11 +116,29 @@ def send_to_all_destinations(file_path: str, use_validator=True): file_path: Path to the file to distribute use_validator: Whether to use the config validator to determine enabled services (if False, falls back to individual checks) + file_id: Optional file ID to associate with logs """ + task_id = self.request.id + if not os.path.exists(file_path): + logger.error(f"[{task_id}] File not found: {file_path}") + log_task_progress(task_id, "send_to_all_destinations", "failure", "File not found", file_id=file_id) raise FileNotFoundError(f"File not found: {file_path}") - logger.info(f"Sending {file_path} to all configured destinations") + logger.info(f"[{task_id}] Sending {file_path} to all configured destinations") + log_task_progress(task_id, "send_to_all_destinations", "in_progress", f"Distributing: {os.path.basename(file_path)}", file_id=file_id) + + # Get file_id from database if not provided (fallback only, prefer passing file_id explicitly) + if file_id is None: + with SessionLocal() as db: + # Only as a last resort, try to find by basename match + # This should not be needed if file_id is passed correctly through the chain + file_record = db.query(FileRecord).filter( + FileRecord.local_filename == os.path.join(settings.workdir, "tmp", os.path.basename(file_path)) + ).first() + if file_record: + file_id = file_record.id + results = {} # Define service configurations @@ -179,12 +200,13 @@ def send_to_all_destinations(file_path: str, use_validator=True): if use_validator: try: configured_services = get_configured_services_from_validator() - logger.info(f"Configured services according to validator: {configured_services}") + logger.info(f"[{task_id}] Configured services according to validator: {configured_services}") except Exception as e: - logger.warning(f"Failed to get configuration from validator: {str(e)}") + logger.warning(f"[{task_id}] Failed to get configuration from validator: {str(e)}") use_validator = False # Process each service + queued_count = 0 for service in services: service_name = service["name"] @@ -192,24 +214,31 @@ def send_to_all_destinations(file_path: str, use_validator=True): is_configured = False if use_validator and service_name in configured_services: is_configured = configured_services[service_name] - logger.debug(f"{service_name} configuration from validator: {is_configured}") + logger.debug(f"[{task_id}] {service_name} configuration from validator: {is_configured}") else: try: is_configured = service["should_upload"]() - logger.debug(f"{service_name} configuration from function: {is_configured}") + logger.debug(f"[{task_id}] {service_name} configuration from function: {is_configured}") except Exception as e: - logger.error(f"Error checking configuration for {service_name}: {str(e)}") + logger.error(f"[{task_id}] Error checking configuration for {service_name}: {str(e)}") is_configured = False # Queue the upload task if service is configured if is_configured: - logger.info(f"Queueing {file_path} for {service_name} upload") + logger.info(f"[{task_id}] Queueing {file_path} for {service_name} upload") + log_task_progress(task_id, f"queue_{service_name}", "in_progress", f"Queueing upload to {service_name}", file_id=file_id) try: - task = service["upload_func"].delay(file_path) + task = service["upload_func"].delay(file_path, file_id) results[f"{service_name}_task_id"] = task.id + queued_count += 1 + log_task_progress(task_id, f"queue_{service_name}", "success", f"Queued for {service_name}", file_id=file_id) except Exception as e: - logger.error(f"Failed to queue {service_name} task: {str(e)}") + logger.error(f"[{task_id}] Failed to queue {service_name} task: {str(e)}") results[f"{service_name}_error"] = str(e) + log_task_progress(task_id, f"queue_{service_name}", "failure", f"Failed: {str(e)}", file_id=file_id) + + logger.info(f"[{task_id}] Queued {queued_count} upload tasks") + log_task_progress(task_id, "send_to_all_destinations", "success", f"Queued {queued_count} uploads", file_id=file_id) return { "status": "Queued", diff --git a/app/tasks/upload_to_dropbox.py b/app/tasks/upload_to_dropbox.py index 03415953..d7930115 100644 --- a/app/tasks/upload_to_dropbox.py +++ b/app/tasks/upload_to_dropbox.py @@ -9,6 +9,9 @@ from app.config import settings from app.tasks.retry_config import BaseTaskWithRetry from app.celery_app import celery from app.utils.filename_utils import get_unique_filename, sanitize_filename, extract_remote_path +from app.utils import log_task_progress +from app.database import SessionLocal +from app.models import FileRecord logger = logging.getLogger(__name__) @@ -99,21 +102,31 @@ def get_dropbox_client(): logger.error(f"Error creating Dropbox client: {str(e)}") raise -@celery.task(base=BaseTaskWithRetry) -def upload_to_dropbox(file_path: str): +@celery.task(base=BaseTaskWithRetry, bind=True) +def upload_to_dropbox(self, file_path: str, file_id: int = None): """ Upload a file to Dropbox. + + Args: + file_path: Path to the file to upload + file_id: Optional file ID to associate with logs """ + task_id = self.request.id + logger.info(f"[{task_id}] Starting Dropbox upload: {file_path}") + log_task_progress(task_id, "upload_to_dropbox", "in_progress", f"Uploading to Dropbox: {os.path.basename(file_path)}", file_id=file_id) + if not os.path.exists(file_path): error_msg = f"File not found: {file_path}" - logger.error(error_msg) + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_dropbox", "failure", error_msg, file_id=file_id) raise FileNotFoundError(error_msg) # Check if Dropbox is properly configured if not (hasattr(settings, 'dropbox_app_key') and settings.dropbox_app_key and hasattr(settings, 'dropbox_app_secret') and settings.dropbox_app_secret and hasattr(settings, 'dropbox_refresh_token') and settings.dropbox_refresh_token): - logger.info("Dropbox upload skipped: Missing configuration") + logger.info(f"[{task_id}] Dropbox upload skipped: Missing configuration") + log_task_progress(task_id, "upload_to_dropbox", "success", "Skipped: Not configured", file_id=file_id) return {"status": "Skipped", "reason": "Dropbox settings not configured"} filename = os.path.basename(file_path) @@ -144,7 +157,8 @@ def upload_to_dropbox(file_path: str): dropbox_path = get_unique_filename(remote_full_path, check_exists_in_dropbox) # Upload the file - logger.info(f"Uploading {filename} to Dropbox at {dropbox_path}") + logger.info(f"[{task_id}] Uploading {filename} to Dropbox at {dropbox_path}") + log_task_progress(task_id, "upload_file", "in_progress", f"Uploading to {dropbox_path}", file_id=file_id) with open(file_path, 'rb') as file_data: # Use files_upload_session for large files to avoid timeouts file_size = os.path.getsize(file_path) @@ -179,7 +193,8 @@ def upload_to_dropbox(file_path: str): mode=dropbox.files.WriteMode.overwrite ) - logger.info(f"Successfully uploaded {filename} to Dropbox at {dropbox_path}") + logger.info(f"[{task_id}] Successfully uploaded {filename} to Dropbox at {dropbox_path}") + log_task_progress(task_id, "upload_to_dropbox", "success", f"Uploaded to Dropbox: {dropbox_path}", file_id=file_id) return { "status": "Completed", "file_path": file_path, @@ -187,14 +202,17 @@ def upload_to_dropbox(file_path: str): } except AuthError: - error_msg = f"[ERROR] Authentication failed while uploading {filename} to Dropbox. Check token." - logger.error(error_msg) + error_msg = f"Authentication failed while uploading {filename} to Dropbox. Check token." + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_dropbox", "failure", error_msg, file_id=file_id) raise Exception(error_msg) except ApiError as e: - error_msg = f"[ERROR] Failed to upload {filename} to Dropbox: {e}" - logger.error(error_msg) + error_msg = f"Failed to upload {filename} to Dropbox: {e}" + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_dropbox", "failure", error_msg, file_id=file_id) raise Exception(error_msg) except Exception as e: - error_msg = f"[ERROR] Unexpected error uploading {filename} to Dropbox: {e}" - logger.error(error_msg) + error_msg = f"Unexpected error uploading {filename} to Dropbox: {e}" + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_dropbox", "failure", error_msg, file_id=file_id) raise Exception(error_msg) diff --git a/app/tasks/upload_to_nextcloud.py b/app/tasks/upload_to_nextcloud.py index 2a3cebc0..3dd98a22 100644 --- a/app/tasks/upload_to_nextcloud.py +++ b/app/tasks/upload_to_nextcloud.py @@ -8,17 +8,29 @@ from app.config import settings from app.celery_app import celery from app.tasks.retry_config import BaseTaskWithRetry from app.utils.filename_utils import get_unique_filename, sanitize_filename, extract_remote_path +from app.utils import log_task_progress +from app.database import SessionLocal +from app.models import FileRecord logger = logging.getLogger(__name__) -@celery.task(base=BaseTaskWithRetry) -def upload_to_nextcloud(file_path: str): +@celery.task(base=BaseTaskWithRetry, bind=True) +def upload_to_nextcloud(self, file_path: str, file_id: int = None): """ Upload a file to Nextcloud WebDAV. + + Args: + file_path: Path to the file to upload + file_id: Optional file ID to associate with logs """ + task_id = self.request.id + logger.info(f"[{task_id}] Starting Nextcloud upload: {file_path}") + log_task_progress(task_id, "upload_to_nextcloud", "in_progress", f"Uploading to Nextcloud: {os.path.basename(file_path)}", file_id=file_id) + if not os.path.exists(file_path): error_msg = f"File not found: {file_path}" - logger.error(error_msg) + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id) raise FileNotFoundError(error_msg) # For Nextcloud, we need to check for 'nextcloud_upload_url' instead of 'nextcloud_url' @@ -26,7 +38,8 @@ def upload_to_nextcloud(file_path: str): if not (getattr(settings, 'nextcloud_upload_url', None) and getattr(settings, 'nextcloud_username', None) and getattr(settings, 'nextcloud_password', None)): - logger.info("Nextcloud upload skipped: Missing configuration") + logger.info(f"[{task_id}] Nextcloud upload skipped: Missing configuration") + log_task_progress(task_id, "upload_to_nextcloud", "success", "Skipped: Not configured", file_id=file_id) return {"status": "Skipped", "reason": "Nextcloud settings not configured"} filename = os.path.basename(file_path) @@ -99,7 +112,8 @@ def upload_to_nextcloud(file_path: str): ) # Upload the file - logger.info(f"Uploading {filename} to Nextcloud at {full_url}") + logger.info(f"[{task_id}] Uploading {filename} to Nextcloud at {full_url}") + log_task_progress(task_id, "upload_file", "in_progress", f"Uploading to {remote_path}", file_id=file_id) with open(file_path, 'rb') as file_data: response = requests.put( full_url, @@ -110,7 +124,8 @@ def upload_to_nextcloud(file_path: str): ) if response.status_code in (201, 204): # Created or No Content - logger.info(f"Successfully uploaded {filename} to Nextcloud at {remote_path}") + logger.info(f"[{task_id}] Successfully uploaded {filename} to Nextcloud at {remote_path}") + log_task_progress(task_id, "upload_to_nextcloud", "success", f"Uploaded to Nextcloud: {remote_path}", file_id=file_id) return { "status": "Completed", "file_path": file_path, @@ -118,11 +133,13 @@ def upload_to_nextcloud(file_path: str): "response_code": response.status_code } else: - error_msg = f"[ERROR] Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}" - logger.error(error_msg) + error_msg = f"Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}" + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id) raise Exception(error_msg) except Exception as e: - error_msg = f"[ERROR] Failed to upload {filename} to Nextcloud: {str(e)}" - logger.error(error_msg) + error_msg = f"Failed to upload {filename} to Nextcloud: {str(e)}" + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id) raise Exception(error_msg) diff --git a/app/tasks/upload_to_paperless.py b/app/tasks/upload_to_paperless.py index bd9b5485..dc8af68b 100644 --- a/app/tasks/upload_to_paperless.py +++ b/app/tasks/upload_to_paperless.py @@ -10,6 +10,9 @@ from typing import Dict, Any from app.config import settings from app.tasks.retry_config import BaseTaskWithRetry from app.celery_app import celery +from app.utils import log_task_progress +from app.database import SessionLocal +from app.models import FileRecord logger = logging.getLogger(__name__) @@ -81,12 +84,24 @@ def poll_task_for_document_id(task_id: str) -> int: f"Task {task_id} didn't reach SUCCESS within {POLL_MAX_ATTEMPTS} attempts." ) -@celery.task(base=BaseTaskWithRetry) -def upload_to_paperless(file_path: str): - """Uploads a file to Paperless-ngx.""" +@celery.task(base=BaseTaskWithRetry, bind=True) +def upload_to_paperless(self, file_path: str, file_id: int = None): + """ + Uploads a file to Paperless-ngx. + + Args: + file_path: Path to the file to upload + file_id: Optional file ID to associate with logs + """ + task_id = self.request.id + logger.info(f"[{task_id}] Starting Paperless upload: {file_path}") + log_task_progress(task_id, "upload_to_paperless", "in_progress", f"Uploading to Paperless: {os.path.basename(file_path)}", file_id=file_id) if not os.path.exists(file_path): - raise FileNotFoundError(f"File not found: {file_path}") + error_msg = f"File not found: {file_path}" + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_paperless", "failure", error_msg, file_id=file_id) + raise FileNotFoundError(error_msg) # Extract filename filename = os.path.basename(file_path) @@ -94,10 +109,13 @@ def upload_to_paperless(file_path: str): # Check if Paperless settings are configured if not settings.paperless_host or not settings.paperless_ngx_api_token: error_msg = "Paperless-ngx credentials are not fully configured" - logger.error(error_msg) + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_paperless", "failure", error_msg, file_id=file_id) raise ValueError(error_msg) # Upload the PDF + logger.info(f"[{task_id}] Posting document to Paperless") + log_task_progress(task_id, "post_document", "in_progress", "Posting to Paperless API", file_id=file_id) post_url = _paperless_api_url("/api/documents/post_document/") with open(file_path, "rb") as f: files = { @@ -110,18 +128,24 @@ def upload_to_paperless(file_path: str): resp = requests.post(post_url, headers=_get_headers(), files=files, data=data) resp.raise_for_status() except requests.exceptions.RequestException as exc: + error_msg = f"Failed to upload to Paperless: {exc}" logger.error( - "Failed to upload document '%s' to Paperless. Error: %s. Response=%s", + f"[{task_id}] Failed to upload document '%s' to Paperless. Error: %s. Response=%s", file_path, exc, getattr(exc.response, "text", "") ) + log_task_progress(task_id, "upload_to_paperless", "failure", error_msg, file_id=file_id) raise raw_task_id = resp.text.strip().strip('"').strip("'") - logger.info(f"Received Paperless task ID: {raw_task_id}") + logger.info(f"[{task_id}] Received Paperless task ID: {raw_task_id}") + log_task_progress(task_id, "post_document", "success", f"Task ID: {raw_task_id}", file_id=file_id) # Poll tasks until success/fail => get doc_id + logger.info(f"[{task_id}] Polling for document ID") + log_task_progress(task_id, "poll_task", "in_progress", "Waiting for Paperless processing", file_id=file_id) doc_id = poll_task_for_document_id(raw_task_id) - logger.info(f"Document {file_path} successfully ingested => ID={doc_id}") + logger.info(f"[{task_id}] Document {file_path} successfully ingested => ID={doc_id}") + log_task_progress(task_id, "upload_to_paperless", "success", f"Uploaded to Paperless: Doc ID {doc_id}", file_id=file_id) return { "status": "Completed", diff --git a/frontend/templates/files.html b/frontend/templates/files.html index afe8f4bb..c421fcc3 100644 --- a/frontend/templates/files.html +++ b/frontend/templates/files.html @@ -43,6 +43,20 @@ .delete-btn:hover { background-color: #fed7d7; } + .view-logs-btn { + color: #3182ce; + cursor: pointer; + padding: 0.25rem 0.5rem; + border-radius: 0.25rem; + background: none; + border: none; + display: flex; + align-items: center; + margin-right: 0.5rem; + } + .view-logs-btn:hover { + background-color: #bee3f8; + } .error-message { background-color: #FEE2E2; border: 1px solid #F87171; @@ -76,8 +90,10 @@ background-color: white; border-radius: 0.5rem; padding: 2rem; - max-width: 500px; + max-width: 800px; width: 90%; + max-height: 80vh; + overflow-y: auto; box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1); } .modal-title { @@ -110,6 +126,48 @@ .modal-btn-delete:hover { background-color: #c53030; } + + /* Logs styles */ + .logs-container { + margin-top: 1rem; + } + .log-entry { + padding: 0.75rem; + border-left: 3px solid #e2e8f0; + margin-bottom: 0.5rem; + background-color: #f7fafc; + border-radius: 0.25rem; + } + .log-entry.success { + border-left-color: #48bb78; + background-color: #f0fff4; + } + .log-entry.failure { + border-left-color: #f56565; + background-color: #fff5f5; + } + .log-entry.in_progress { + border-left-color: #4299e1; + background-color: #ebf8ff; + } + .log-step { + font-weight: 600; + color: #2d3748; + } + .log-message { + color: #4a5568; + margin-top: 0.25rem; + } + .log-timestamp { + font-size: 0.875rem; + color: #718096; + margin-top: 0.25rem; + } + .no-logs { + text-align: center; + padding: 2rem; + color: #718096; + } {% endblock %} @@ -147,9 +205,14 @@ {{ file.mime_type }} {{ file.created_at }} - +
+ + +
{% else %} @@ -173,32 +236,104 @@ + + +