diff --git a/.env.demo b/.env.demo index fb0f0bc0..a24f4427 100644 --- a/.env.demo +++ b/.env.demo @@ -1,47 +1,47 @@ -# **Config Variables** -DATABASE_URL=sqlite:///./app/database.db -REDIS_URL=redis://redis:6379/0 -WORKDIR=/workdir -AWS_REGION="eu-central-1" -AZURE_REGION="eastus" -AZURE_ENDPOINT="https://.cognitiveservices.azure.com/" -S3_BUCKET_NAME= -NEXTCLOUD_UPLOAD_URL=https://nextcloud.example.com/remote.php/dav/files/ -NEXTCLOUD_FOLDER="" -PAPERLESS_NGX_URL=https://paperless.example.com/api/documents/post_document/ -PAPERLESS_HOST=https://paperless.example.com - -# **Tokens/API Credentials** -AWS_ACCESS_KEY_ID="" -AWS_SECRET_ACCESS_KEY="" -OPENAI_API_KEY="" -PAPERLESS_NGX_API_TOKEN= -DROPBOX_APP_KEY= -DROPBOX_APP_SECRET= -DROPBOX_REFRESH_TOKEN= -AZURE_AI_KEY= - -# **User Credentials** -ADMIN_USERNAME=admin -ADMIN_PASSWORD=your_secure_password -NEXTCLOUD_USERNAME= -NEXTCLOUD_PASSWORD= -IMAP1_USERNAME= -IMAP1_PASSWORD= -IMAP2_USERNAME= -IMAP2_PASSWORD= - -# **IMAP Settings** -IMAP1_HOST=mail.example.com -IMAP1_PORT=993 -IMAP1_SSL=true -IMAP1_POLL_INTERVAL_MINUTES=5 -IMAP1_DELETE_AFTER_PROCESS=false - -IMAP2_HOST=imap.gmail.com -IMAP2_PORT=993 -IMAP2_SSL=true -IMAP2_POLL_INTERVAL_MINUTES=10 -IMAP2_DELETE_AFTER_PROCESS=false - +# **Config Variables** +DATABASE_URL=sqlite:///./app/database.db +REDIS_URL=redis://redis:6379/0 +WORKDIR=/workdir +AWS_REGION="eu-central-1" +AZURE_REGION="eastus" +AZURE_ENDPOINT="https://.cognitiveservices.azure.com/" +S3_BUCKET_NAME= +NEXTCLOUD_UPLOAD_URL=https://nextcloud.example.com/remote.php/dav/files/ +NEXTCLOUD_FOLDER="" +PAPERLESS_NGX_URL=https://paperless.example.com/api/documents/post_document/ +PAPERLESS_HOST=https://paperless.example.com + +# **Tokens/API Credentials** +AWS_ACCESS_KEY_ID="" +AWS_SECRET_ACCESS_KEY="" +OPENAI_API_KEY="" +PAPERLESS_NGX_API_TOKEN= +DROPBOX_APP_KEY= +DROPBOX_APP_SECRET= +DROPBOX_REFRESH_TOKEN= +AZURE_AI_KEY= + +# **User Credentials** +ADMIN_USERNAME=admin +ADMIN_PASSWORD=your_secure_password +NEXTCLOUD_USERNAME= +NEXTCLOUD_PASSWORD= +IMAP1_USERNAME= +IMAP1_PASSWORD= +IMAP2_USERNAME= +IMAP2_PASSWORD= + +# **IMAP Settings** +IMAP1_HOST=mail.example.com +IMAP1_PORT=993 +IMAP1_SSL=true +IMAP1_POLL_INTERVAL_MINUTES=5 +IMAP1_DELETE_AFTER_PROCESS=false + +IMAP2_HOST=imap.gmail.com +IMAP2_PORT=993 +IMAP2_SSL=true +IMAP2_POLL_INTERVAL_MINUTES=10 +IMAP2_DELETE_AFTER_PROCESS=false + GOTENBERG_URL=http://gotenberg:3000 \ No newline at end of file diff --git a/.github/dependabot.yml b/.github/dependabot.yml index d698d8d0..1813cae9 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -1,12 +1,12 @@ -# To get started with Dependabot version updates, you'll need to specify which -# package ecosystems to update and where the package manifests are located. -# Please see the documentation for all configuration options: -# https://docs.github.com/code-security/dependabot/dependabot-version-updates/configuration-options-for-the-dependabot.yml-file - -version: 2 -updates: - - package-ecosystem: "" # See documentation for possible values - directory: "/" # Location of package manifests - schedule: - interval: "weekly" - +# To get started with Dependabot version updates, you'll need to specify which +# package ecosystems to update and where the package manifests are located. +# Please see the documentation for all configuration options: +# https://docs.github.com/code-security/dependabot/dependabot-version-updates/configuration-options-for-the-dependabot.yml-file + +version: 2 +updates: + - package-ecosystem: "" # See documentation for possible values + directory: "/" # Location of package manifests + schedule: + interval: "weekly" + diff --git a/.github/workflows/deploy.yaml b/.github/workflows/deploy.yaml index 7692e4dc..933198dc 100644 --- a/.github/workflows/deploy.yaml +++ b/.github/workflows/deploy.yaml @@ -1,18 +1,18 @@ -name: Deploy to Production - -on: - workflow_run: - workflows: ["Build and Push Docker Image"] - types: - - completed - -permissions: - contents: read - -jobs: - deploy: - runs-on: ubuntu-latest - steps: - - name: Call Deployment Webhook - run: | - curl -X POST https://docker2.kuechenserver.org/api/stacks/webhooks/960c7d8e-97ec-4175-a8dc-73f037b02349 +name: Deploy to Production + +on: + workflow_run: + workflows: ["Build and Push Docker Image"] + types: + - completed + +permissions: + contents: read + +jobs: + deploy: + runs-on: ubuntu-latest + steps: + - name: Call Deployment Webhook + run: | + curl -X POST https://docker2.kuechenserver.org/api/stacks/webhooks/960c7d8e-97ec-4175-a8dc-73f037b02349 diff --git a/.github/workflows/docker-build.yaml b/.github/workflows/docker-build.yaml index c6cfa9e0..eba37419 100644 --- a/.github/workflows/docker-build.yaml +++ b/.github/workflows/docker-build.yaml @@ -1,54 +1,54 @@ -name: Build and Push Docker Image - -permissions: - contents: read - packages: write - -on: - push: - branches: - - main - - develop - pull_request: - branches: - - main - -jobs: - build: - runs-on: ubuntu-latest - steps: - - name: Checkout Code - uses: actions/checkout@v3 - - - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v2 - - - name: Log in to Docker Hub - uses: docker/login-action@v2 - with: - username: ${{ secrets.DOCKER_USERNAME }} - password: ${{ secrets.DOCKER_PASSWORD }} - - - name: Log in to GitHub Container Registry - uses: docker/login-action@v2 - with: - registry: ghcr.io - username: ${{ github.actor }} - password: ${{ secrets.GITHUB_TOKEN }} - - - name: Build and Push - uses: docker/build-push-action@v4 - with: - # Specify target platforms - platforms: linux/amd64 - context: . - file: Dockerfile - push: true - tags: | - christianlouis/document-processor:latest - christianlouis/document-processor:${{ github.sha }} - ghcr.io/${{ github.repository_owner }}/document-processor:latest - ghcr.io/${{ github.repository_owner }}/document-processor:${{ github.sha }} - # Cache options (optional) - cache-from: type=gha - cache-to: type=gha,mode=max +name: Build and Push Docker Image + +permissions: + contents: read + packages: write + +on: + push: + branches: + - main + - develop + pull_request: + branches: + - main + +jobs: + build: + runs-on: ubuntu-latest + steps: + - name: Checkout Code + uses: actions/checkout@v3 + + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@v2 + + - name: Log in to Docker Hub + uses: docker/login-action@v2 + with: + username: ${{ secrets.DOCKER_USERNAME }} + password: ${{ secrets.DOCKER_PASSWORD }} + + - name: Log in to GitHub Container Registry + uses: docker/login-action@v2 + with: + registry: ghcr.io + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Build and Push + uses: docker/build-push-action@v4 + with: + # Specify target platforms + platforms: linux/amd64 + context: . + file: Dockerfile + push: true + tags: | + christianlouis/document-processor:latest + christianlouis/document-processor:${{ github.sha }} + ghcr.io/${{ github.repository_owner }}/document-processor:latest + ghcr.io/${{ github.repository_owner }}/document-processor:${{ github.sha }} + # Cache options (optional) + cache-from: type=gha + cache-to: type=gha,mode=max diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index da1d96fa..9cb57e5e 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -1,40 +1,40 @@ -name: Run Tests & Linting - -on: [push, pull_request] - -jobs: - test: - runs-on: ubuntu-latest - steps: - - name: Checkout Code - uses: actions/checkout@v3 - - - name: Set up Python - uses: actions/setup-python@v4 - with: - python-version: "3.11" - - - name: Install Dependencies - run: | - python -m pip install --upgrade pip - pip install -r requirements.txt - pip install pytest flake8 black mypy pylint - - # - name: Run Tests - # run: pytest tests/ - - - name: Run Linter (Flake8) - run: flake8 app/ - continue-on-error: true - - - name: Run Code Formatter (Black) - run: black --check app/ - continue-on-error: true - - - name: Run Type Checker (Mypy) - run: mypy app/ - continue-on-error: true - - - name: Run Linter (Pylint) - run: pylint app/ - continue-on-error: true +name: Run Tests & Linting + +on: [push, pull_request] + +jobs: + test: + runs-on: ubuntu-latest + steps: + - name: Checkout Code + uses: actions/checkout@v3 + + - name: Set up Python + uses: actions/setup-python@v4 + with: + python-version: "3.11" + + - name: Install Dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + pip install pytest flake8 black mypy pylint + + # - name: Run Tests + # run: pytest tests/ + + - name: Run Linter (Flake8) + run: flake8 app/ + continue-on-error: true + + - name: Run Code Formatter (Black) + run: black --check app/ + continue-on-error: true + + - name: Run Type Checker (Mypy) + run: mypy app/ + continue-on-error: true + + - name: Run Linter (Pylint) + run: pylint app/ + continue-on-error: true diff --git a/.gitignore b/.gitignore index cf6712b9..5cdea576 100644 --- a/.gitignore +++ b/.gitignore @@ -1,171 +1,171 @@ -# Byte-compiled / optimized / DLL files -__pycache__/ -*.py[cod] -*$py.class - -# C extensions -*.so - -# Distribution / packaging -.Python -build/ -develop-eggs/ -dist/ -downloads/ -eggs/ -.eggs/ -lib/ -lib64/ -parts/ -sdist/ -var/ -wheels/ -share/python-wheels/ -*.egg-info/ -.installed.cfg -*.egg -MANIFEST -.env - -# PyInstaller -# Usually these files are written by a python script from a template -# before PyInstaller builds the exe, so as to inject date/other infos into it. -*.manifest -*.spec - -# Installer logs -pip-log.txt -pip-delete-this-directory.txt - -# Unit test / coverage reports -htmlcov/ -.tox/ -.nox/ -.coverage -.coverage.* -.cache -nosetests.xml -coverage.xml -*.cover -*.py,cover -.hypothesis/ -.pytest_cache/ -cover/ - -# Translations -*.mo -*.pot - -# Django stuff: -*.log -local_settings.py -db.sqlite3 -db.sqlite3-journal - -# Flask stuff: -instance/ -.webassets-cache - -# Scrapy stuff: -.scrapy - -# Sphinx documentation -docs/_build/ - -# PyBuilder -.pybuilder/ -target/ - -# Jupyter Notebook -.ipynb_checkpoints - -# IPython -profile_default/ -ipython_config.py - -# pyenv -# For a library or package, you might want to ignore these files since the code is -# intended to run in multiple environments; otherwise, check them in: -# .python-version - -# pipenv -# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. -# However, in case of collaboration, if having platform-specific dependencies or dependencies -# having no cross-platform support, pipenv may install dependencies that don't work, or not -# install all needed dependencies. -#Pipfile.lock - -# UV -# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control. -# This is especially recommended for binary packages to ensure reproducibility, and is more -# commonly ignored for libraries. -#uv.lock - -# poetry -# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. -# This is especially recommended for binary packages to ensure reproducibility, and is more -# commonly ignored for libraries. -# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control -#poetry.lock - -# pdm -# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. -#pdm.lock -# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it -# in version control. -# https://pdm.fming.dev/latest/usage/project/#working-with-version-control -.pdm.toml -.pdm-python -.pdm-build/ - -# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm -__pypackages__/ - -# Celery stuff -celerybeat-schedule -celerybeat.pid - -# SageMath parsed files -*.sage.py - -# Environments -.venv -env/ -venv/ -ENV/ -env.bak/ -venv.bak/ - -# Spyder project settings -.spyderproject -.spyproject - -# Rope project settings -.ropeproject - -# mkdocs documentation -/site - -# mypy -.mypy_cache/ -.dmypy.json -dmypy.json - -# Pyre type checker -.pyre/ - -# pytype static type analyzer -.pytype/ - -# Cython debug symbols -cython_debug/ - -# PyCharm -# JetBrains specific template is maintained in a separate JetBrains.gitignore that can -# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore -# and can be added to the global gitignore or merged into this file. For a more nuclear -# option (not recommended) you can uncomment the following to ignore the entire idea folder. -#.idea/ - -# PyPI configuration file -.pypirc +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST +.env + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +# For a library or package, you might want to ignore these files since the code is +# intended to run in multiple environments; otherwise, check them in: +# .python-version + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +#Pipfile.lock + +# UV +# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +#uv.lock + +# poetry +# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control +#poetry.lock + +# pdm +# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. +#pdm.lock +# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it +# in version control. +# https://pdm.fming.dev/latest/usage/project/#working-with-version-control +.pdm.toml +.pdm-python +.pdm-build/ + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ + +# PyCharm +# JetBrains specific template is maintained in a separate JetBrains.gitignore that can +# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore +# and can be added to the global gitignore or merged into this file. For a more nuclear +# option (not recommended) you can uncomment the following to ignore the entire idea folder. +#.idea/ + +# PyPI configuration file +.pypirc diff --git a/Dockerfile b/Dockerfile index 059eb7bb..e77ac875 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,30 +1,30 @@ - -# Stage 1: Build dependencies -FROM python:3.11 AS builder - -WORKDIR /app - -COPY requirements.txt /app/ -RUN pip install --no-cache-dir -r requirements.txt - -# Stage 2: Final image -FROM python:3.11-slim - -WORKDIR /app - -# Copy installed dependencies -COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages -COPY --from=builder /usr/local/bin /usr/local/bin - -# Copy application files correctly -COPY ./app /app/app -COPY ./frontend /app/frontend - -# Set Python path explicitly -ENV PYTHONPATH=/app - -# Expose API port -EXPOSE 8000 -WORKDIR /app - -CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"] + +# Stage 1: Build dependencies +FROM python:3.11 AS builder + +WORKDIR /app + +COPY requirements.txt /app/ +RUN pip install --no-cache-dir -r requirements.txt + +# Stage 2: Final image +FROM python:3.11-slim + +WORKDIR /app + +# Copy installed dependencies +COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages +COPY --from=builder /usr/local/bin /usr/local/bin + +# Copy application files correctly +COPY ./app /app/app +COPY ./frontend /app/frontend + +# Set Python path explicitly +ENV PYTHONPATH=/app + +# Expose API port +EXPOSE 8000 +WORKDIR /app + +CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/LICENSE b/LICENSE index 261eeb9e..29f81d81 100644 --- a/LICENSE +++ b/LICENSE @@ -1,201 +1,201 @@ - Apache License - Version 2.0, January 2004 - http://www.apache.org/licenses/ - - TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION - - 1. Definitions. - - "License" shall mean the terms and conditions for use, reproduction, - and distribution as defined by Sections 1 through 9 of this document. - - "Licensor" shall mean the copyright owner or entity authorized by - the copyright owner that is granting the License. - - "Legal Entity" shall mean the union of the acting entity and all - other entities that control, are controlled by, or are under common - control with that entity. For the purposes of this definition, - "control" means (i) the power, direct or indirect, to cause the - direction or management of such entity, whether by contract or - otherwise, or (ii) ownership of fifty percent (50%) or more of the - outstanding shares, or (iii) beneficial ownership of such entity. - - "You" (or "Your") shall mean an individual or Legal Entity - exercising permissions granted by this License. - - "Source" form shall mean the preferred form for making modifications, - including but not limited to software source code, documentation - source, and configuration files. - - "Object" form shall mean any form resulting from mechanical - transformation or translation of a Source form, including but - not limited to compiled object code, generated documentation, - and conversions to other media types. - - "Work" shall mean the work of authorship, whether in Source or - Object form, made available under the License, as indicated by a - copyright notice that is included in or attached to the work - (an example is provided in the Appendix below). - - "Derivative Works" shall mean any work, whether in Source or Object - form, that is based on (or derived from) the Work and for which the - editorial revisions, annotations, elaborations, or other modifications - represent, as a whole, an original work of authorship. For the purposes - of this License, Derivative Works shall not include works that remain - separable from, or merely link (or bind by name) to the interfaces of, - the Work and Derivative Works thereof. - - "Contribution" shall mean any work of authorship, including - the original version of the Work and any modifications or additions - to that Work or Derivative Works thereof, that is intentionally - submitted to Licensor for inclusion in the Work by the copyright owner - or by an individual or Legal Entity authorized to submit on behalf of - the copyright owner. For the purposes of this definition, "submitted" - means any form of electronic, verbal, or written communication sent - to the Licensor or its representatives, including but not limited to - communication on electronic mailing lists, source code control systems, - and issue tracking systems that are managed by, or on behalf of, the - Licensor for the purpose of discussing and improving the Work, but - excluding communication that is conspicuously marked or otherwise - designated in writing by the copyright owner as "Not a Contribution." - - "Contributor" shall mean Licensor and any individual or Legal Entity - on behalf of whom a Contribution has been received by Licensor and - subsequently incorporated within the Work. - - 2. Grant of Copyright License. Subject to the terms and conditions of - this License, each Contributor hereby grants to You a perpetual, - worldwide, non-exclusive, no-charge, royalty-free, irrevocable - copyright license to reproduce, prepare Derivative Works of, - publicly display, publicly perform, sublicense, and distribute the - Work and such Derivative Works in Source or Object form. - - 3. Grant of Patent License. Subject to the terms and conditions of - this License, each Contributor hereby grants to You a perpetual, - worldwide, non-exclusive, no-charge, royalty-free, irrevocable - (except as stated in this section) patent license to make, have made, - use, offer to sell, sell, import, and otherwise transfer the Work, - where such license applies only to those patent claims licensable - by such Contributor that are necessarily infringed by their - Contribution(s) alone or by combination of their Contribution(s) - with the Work to which such Contribution(s) was submitted. If You - institute patent litigation against any entity (including a - cross-claim or counterclaim in a lawsuit) alleging that the Work - or a Contribution incorporated within the Work constitutes direct - or contributory patent infringement, then any patent licenses - granted to You under this License for that Work shall terminate - as of the date such litigation is filed. - - 4. Redistribution. You may reproduce and distribute copies of the - Work or Derivative Works thereof in any medium, with or without - modifications, and in Source or Object form, provided that You - meet the following conditions: - - (a) You must give any other recipients of the Work or - Derivative Works a copy of this License; and - - (b) You must cause any modified files to carry prominent notices - stating that You changed the files; and - - (c) You must retain, in the Source form of any Derivative Works - that You distribute, all copyright, patent, trademark, and - attribution notices from the Source form of the Work, - excluding those notices that do not pertain to any part of - the Derivative Works; and - - (d) If the Work includes a "NOTICE" text file as part of its - distribution, then any Derivative Works that You distribute must - include a readable copy of the attribution notices contained - within such NOTICE file, excluding those notices that do not - pertain to any part of the Derivative Works, in at least one - of the following places: within a NOTICE text file distributed - as part of the Derivative Works; within the Source form or - documentation, if provided along with the Derivative Works; or, - within a display generated by the Derivative Works, if and - wherever such third-party notices normally appear. The contents - of the NOTICE file are for informational purposes only and - do not modify the License. You may add Your own attribution - notices within Derivative Works that You distribute, alongside - or as an addendum to the NOTICE text from the Work, provided - that such additional attribution notices cannot be construed - as modifying the License. - - You may add Your own copyright statement to Your modifications and - may provide additional or different license terms and conditions - for use, reproduction, or distribution of Your modifications, or - for any such Derivative Works as a whole, provided Your use, - reproduction, and distribution of the Work otherwise complies with - the conditions stated in this License. - - 5. Submission of Contributions. Unless You explicitly state otherwise, - any Contribution intentionally submitted for inclusion in the Work - by You to the Licensor shall be under the terms and conditions of - this License, without any additional terms or conditions. - Notwithstanding the above, nothing herein shall supersede or modify - the terms of any separate license agreement you may have executed - with Licensor regarding such Contributions. - - 6. Trademarks. This License does not grant permission to use the trade - names, trademarks, service marks, or product names of the Licensor, - except as required for reasonable and customary use in describing the - origin of the Work and reproducing the content of the NOTICE file. - - 7. Disclaimer of Warranty. Unless required by applicable law or - agreed to in writing, Licensor provides the Work (and each - Contributor provides its Contributions) on an "AS IS" BASIS, - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - implied, including, without limitation, any warranties or conditions - of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A - PARTICULAR PURPOSE. You are solely responsible for determining the - appropriateness of using or redistributing the Work and assume any - risks associated with Your exercise of permissions under this License. - - 8. Limitation of Liability. In no event and under no legal theory, - whether in tort (including negligence), contract, or otherwise, - unless required by applicable law (such as deliberate and grossly - negligent acts) or agreed to in writing, shall any Contributor be - liable to You for damages, including any direct, indirect, special, - incidental, or consequential damages of any character arising as a - result of this License or out of the use or inability to use the - Work (including but not limited to damages for loss of goodwill, - work stoppage, computer failure or malfunction, or any and all - other commercial damages or losses), even if such Contributor - has been advised of the possibility of such damages. - - 9. Accepting Warranty or Additional Liability. While redistributing - the Work or Derivative Works thereof, You may choose to offer, - and charge a fee for, acceptance of support, warranty, indemnity, - or other liability obligations and/or rights consistent with this - License. However, in accepting such obligations, You may act only - on Your own behalf and on Your sole responsibility, not on behalf - of any other Contributor, and only if You agree to indemnify, - defend, and hold each Contributor harmless for any liability - incurred by, or claims asserted against, such Contributor by reason - of your accepting any such warranty or additional liability. - - END OF TERMS AND CONDITIONS - - APPENDIX: How to apply the Apache License to your work. - - To apply the Apache License to your work, attach the following - boilerplate notice, with the fields enclosed by brackets "[]" - replaced with your own identifying information. (Don't include - the brackets!) The text should be enclosed in the appropriate - comment syntax for the file format. We also recommend that a - file or class name and description of purpose be included on the - same "printed page" as the copyright notice for easier - identification within third-party archives. - - Copyright [yyyy] [name of copyright owner] - - Licensed under the Apache License, Version 2.0 (the "License"); - you may not use this file except in compliance with the License. - You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - - Unless required by applicable law or agreed to in writing, software - distributed under the License is distributed on an "AS IS" BASIS, - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - See the License for the specific language governing permissions and - limitations under the License. + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/README.md b/README.md index 1b5e7334..31f12286 100644 --- a/README.md +++ b/README.md @@ -1,144 +1,199 @@ -# Document Processing System - -## Overview - -This project automates the handling, extraction, and processing of documents using a combination of services such as OpenAI, Dropbox, Nextcloud, and Paperless NGX. The system extracts metadata, processes document contents, and stores the results efficiently. It is designed for flexibility and configurability through environment variables, making it easily customizable for different workflows. - -## Features - -- **Document Upload & Storage**: Upload and manage documents via Dropbox and Nextcloud. -- **OCR Processing**: Extract text from scanned documents. -- **Metadata Extraction**: Automatically extract key information using OpenAI's API. -- **Document Management**: Store processed documents and metadata in Paperless NGX for easy retrieval. -- **IMAP Integration**: Fetch documents from multiple IMAP email accounts for processing. - -## Environment Variables - -The project is configured via the `.env` file, where credentials and settings for different services are defined. Below is a breakdown of key configuration variables: - -### General Configuration - -| **Variable** | **Description** | **How to Obtain** | -|-------------|----------------|-------------------| -| `DATABASE_URL` | Path to the SQLite database. | Example: `sqlite:///./app/database.db` | -| `REDIS_URL` | URL for Redis connection. | Example: `redis://redis:6379/0` | -| `WORKDIR` | Working directory for the application. | Example: `/workdir` | -| `NEXTCLOUD_UPLOAD_URL` | Nextcloud WebDAV upload URL. | Example: `https://nextcloud.example.com/remote.php/dav/files/` | -| `NEXTCLOUD_FOLDER` | Folder in Nextcloud for file uploads. | Example: `/Documents/Uploads` | -| `PAPERLESS_NGX_URL` | Paperless NGX API endpoint. | Example: `https://paperless.example.com/api/documents/post_document/` | -| `PAPERLESS_HOST` | Root URL for Paperless NGX. | Example: `https://paperless.example.com` | - -### Tokens/API Credentials - -| **Variable** | **Description** | **How to Obtain** | -|-------------|----------------|-------------------| -| `OPENAI_API_KEY` | API key for OpenAI services. | Get from [OpenAI platform](https://platform.openai.com/account/api-keys). | -| `PAPERLESS_NGX_API_TOKEN` | API token for Paperless NGX. | Obtain from your Paperless NGX instance. | -| `DROPBOX_APP_KEY` | Dropbox API key. | Generate from the [Dropbox Developer Console](https://www.dropbox.com/developers/apps/create). | -| `DROPBOX_APP_SECRET` | Dropbox API secret. | Available in the Dropbox Developer Console. | -| `DROPBOX_REFRESH_TOKEN` | Dropbox OAuth refresh token. | Obtain by following Dropbox's OAuth flow. | - -### User Credentials - -| **Variable** | **Description** | -|-------------|----------------| -| `ADMIN_USERNAME` | Admin username for system access. | -| `ADMIN_PASSWORD` | Admin password for system access. | -| `NEXTCLOUD_USERNAME` | Username for Nextcloud authentication. | -| `NEXTCLOUD_PASSWORD` | Password for Nextcloud authentication. | - -### IMAP Configuration - -| **Variable** | **Description** | -|-------------|----------------| -| `IMAP1_USERNAME` | IMAP username for the first email account. | -| `IMAP1_PASSWORD` | IMAP password for the first email account. | -| `IMAP1_HOST` | Hostname of the first IMAP server. | -| `IMAP1_PORT` | IMAP server port (typically `993`). | -| `IMAP1_SSL` | Enable SSL (`true` or `false`). | -| `IMAP1_POLL_INTERVAL_MINUTES` | Polling interval for IMAP server. | -| `IMAP1_DELETE_AFTER_PROCESS` | Delete emails after processing (`true` or `false`). | - -### Additional Services - -| **Variable** | **Description** | -|-------------|----------------| -| `GOTENBERG_URL` | URL for Gotenberg PDF processing. | - -## Running as a Docker Container - -This project includes a `docker-compose.yml` file that allows for easy deployment using Docker. The following services are defined: - -- **API Service**: Runs the document processing API using `uvicorn`. -- **Worker Service**: Runs the Celery worker for handling document processing tasks. -- **Redis**: Used as a message broker for Celery. -- **Gotenberg**: Provides PDF processing capabilities. - -### Running the Application with Docker Compose - -1. **Ensure Docker and Docker Compose are installed**. -2. **Clone the repository and navigate to the directory**: - ```bash - git clone - cd - ``` -3. **Create and configure the `.env` file**. -4. **Start the services**: - ```bash - docker-compose up -d - ``` -5. The API will be available at `http://localhost:8000`. - -### Services in `docker-compose.yml` - -```yaml -services: - api: - image: christianlouis/document-processor:latest - container_name: document_api - working_dir: /workdir - command: ["sh", "-c", "cd /app && uvicorn app.main:app --host 0.0.0.0 --port 8000"] - environment: - - PYTHONPATH=/app - env_file: - - .env - ports: - - "8000:8000" - depends_on: - - redis - - worker - volumes: - - /var/docparse/workdir:/workdir - - worker: - image: christianlouis/document-processor:latest - container_name: document_worker - working_dir: /workdir - command: ["celery", "-A", "app.celery_worker", "worker", "-B", "--loglevel=info", "-Q", "document_processor,default,celery"] - env_file: - - .env - environment: - - PYTHONPATH=/app - depends_on: - - redis - - gotenberg - volumes: - - /var/docparse/workdir:/workdir - - gotenberg: - image: gotenberg/gotenberg:latest - container_name: gotenberg - - redis: - image: redis:alpine - container_name: document_redis - restart: always -``` - -## To-Do List - -- Refactor AWS-related code to Azure. -- Remove unnecessary environment variables. -- Make upload targets configurable. -- Remove S3 upload functionality. - +# Document Processing System + +## Overview + +This project automates the handling, extraction, and processing of documents using a variety of services, including: + +- **OpenAI** for metadata extraction and text refinement. +- **Dropbox** and **Nextcloud** for file storage and uploads. +- **Paperless NGX** for document indexing and management. +- **Azure Document Intelligence** (optional) for OCR on PDFs (replacing Textract). +- **Gotenberg** for file-to-PDF conversions. +- **AWS S3** (currently implemented but may be removed in the future). + +It is designed for flexibility and configurability through environment variables, making it easily customizable for different workflows. The system can fetch documents from multiple IMAP mailboxes, process them (OCR, metadata extraction, PDF conversion), and store them in the desired destinations. + +## Features + +- **Document Upload & Storage**: + - Manual uploads (via API) to S3, or direct uploads to Dropbox/Nextcloud/Paperless. +- **OCR Processing (Azure)**: + - Extract text from scanned PDFs using Azure Document Intelligence. +- **Metadata Extraction (OpenAI)**: + - Use GPT to classify, label, or otherwise enrich the text with structured metadata. +- **PDF Conversion (Gotenberg)**: + - Convert non-PDF attachments (e.g., Word docs, images) into PDFs. +- **Document Management (Paperless NGX)**: + - Store processed documents and metadata in a Paperless NGX instance. +- **IMAP Integration**: + - Fetch documents from multiple mailboxes (including Gmail) and automatically enqueue them for processing. + +## Environment Variables + +The `.env` file drives all configuration. This table breaks down key variables—some are optional, depending on which services you actually use. + +### Core Settings + +| **Variable** | **Description** | **Example** | +|------------------------|----------------------------------------------------------|--------------------------------| +| `DATABASE_URL` | Path/URL to the SQLite database (or other SQL backend). | `sqlite:///./app/database.db` | +| `REDIS_URL` | URL for Redis, used by Celery for broker & result store. | `redis://redis:6379/0` | +| `WORKDIR` | Working directory for the application. | `/workdir` | +| `GOTENBERG_URL` | Gotenberg PDF processing URL. | `http://gotenberg:3000` | + +### IMAP Configuration (Multiple Mailboxes) + +| **Variable** | **Description** | **Example** | +|-------------------------------|--------------------------------------------------------------|-------------------| +| `IMAP1_HOST` | Hostname for first IMAP server. | `mail.example.com`| +| `IMAP1_PORT` | Port number (usually `993`). | `993` | +| `IMAP1_USERNAME` | IMAP login (first mailbox). | `user@example.com`| +| `IMAP1_PASSWORD` | IMAP password (first mailbox). | `*******` | +| `IMAP1_SSL` | Use SSL (`true`/`false`). | `true` | +| `IMAP1_POLL_INTERVAL_MINUTES` | Frequency in minutes to poll for new mail. | `5` | +| `IMAP1_DELETE_AFTER_PROCESS` | Delete emails after processing (`true`/`false`). | `false` | +| `IMAP2_HOST` | Hostname for second IMAP server (optional). | `imap.gmail.com` | +| `IMAP2_PORT` | Port number for second mailbox. | `993` | +| `IMAP2_USERNAME` | IMAP login for second mailbox. | `you@gmail.com` | +| `IMAP2_PASSWORD` | IMAP password for second mailbox. | `*******` | +| `IMAP2_SSL` | Use SSL for second mailbox (`true`/`false`). | `true` | +| `IMAP2_POLL_INTERVAL_MINUTES` | Frequency in minutes to poll second mailbox. | `10` | +| `IMAP2_DELETE_AFTER_PROCESS` | Delete emails after processing (`true`/`false`) for mailbox.| `false` | + +### OpenAI & Azure Document Intelligence + +| **Variable** | **Description** | **How to Obtain** | +|-----------------------|--------------------------------------------------------------------|-----------------------------------------------------------------------------------------------------------------| +| `OPENAI_API_KEY` | API key for OpenAI services (used for metadata extraction/refinement). | [OpenAI platform](https://platform.openai.com/account/api-keys) | +| `AZURE_AI_KEY` | Azure Document Intelligence key (for OCR). | [Azure Portal](https://portal.azure.com/) | +| `AZURE_REGION` | Azure region of your Document Intelligence instance. | e.g. `eastus`, `westeurope` | +| `AZURE_ENDPOINT` | Endpoint URL for Document Intelligence. | e.g. `https://.cognitiveservices.azure.com/` | + +### Paperless NGX + +| **Variable** | **Description** | +|-------------------------------|-----------------------------------------------------| +| `PAPERLESS_NGX_API_TOKEN` | API token for Paperless NGX. | +| `PAPERLESS_HOST` | Root URL for Paperless NGX (e.g. `https://paperless.example.com`). | + +### Dropbox + +| **Variable** | **Description** | **How to Obtain** | +|-------------------------|--------------------------------------------------|------------------------------------------------------------------------------------| +| `DROPBOX_APP_KEY` | Dropbox API app key. | [Dropbox Developer Console](https://www.dropbox.com/developers/apps/create) | +| `DROPBOX_APP_SECRET` | Dropbox API app secret. | [Dropbox Developer Console](https://www.dropbox.com/developers/apps/create) | +| `DROPBOX_REFRESH_TOKEN` | OAuth2 refresh token for Dropbox. | Follow Dropbox OAuth flow to retrieve | +| `DROPBOX_FOLDER` | Default folder path for Dropbox uploads. | e.g. `"/Documents/Uploads"` | + +### Nextcloud + +| **Variable** | **Description** | +|-------------------------|---------------------------------------------------------------| +| `NEXTCLOUD_UPLOAD_URL` | Nextcloud WebDAV URL (e.g. `https://nc.example.com/remote.php/dav/files/`). | +| `NEXTCLOUD_USERNAME` | Nextcloud login username. | +| `NEXTCLOUD_PASSWORD` | Nextcloud login password. | +| `NEXTCLOUD_FOLDER` | Destination folder in Nextcloud (e.g. `"/Documents/Uploads"`). | + +### AWS S3 + +| **Variable** | **Description** | +|------------------------|---------------------------------------------------------------| +| `AWS_ACCESS_KEY_ID` | AWS Access Key (used for S3 upload). | +| `AWS_SECRET_ACCESS_KEY`| AWS Secret Key (used for S3 upload). | +| `AWS_REGION` | AWS region for S3. | +| `S3_BUCKET_NAME` | Default S3 bucket name if using S3 upload. | + +### General Admin Credentials + +| **Variable** | **Description** | +|--------------------|---------------------------------------------| +| `ADMIN_USERNAME` | Admin username for system access. | +| `ADMIN_PASSWORD` | Admin password for system access. | + +## Running as a Docker Container + +This project uses Celery (with Redis) for asynchronous task management and Gotenberg for PDF conversion. The `docker-compose.yml` file defines these services: + +- **API Service**: Runs the FastAPI application via `uvicorn`. +- **Worker Service**: Runs the Celery worker for processing tasks (PDF conversions, OCR, etc.). +- **Redis**: Provides the message broker & result backend for Celery. +- **Gotenberg**: Offers PDF conversion capabilities. + +### Running the Application with Docker Compose + +1. **Install Docker and Docker Compose** on your system. +2. **Clone the repository** and navigate into it: + ```bash + git clone + cd + ``` +3. **Create and configure the `.env` file**: + - Fill in the variables from the tables above. + - (At minimum, you need `DATABASE_URL`, `REDIS_URL`, `WORKDIR`, plus whichever service creds you plan to use.) +4. **Launch the services**: + ```bash + docker-compose up -d + ``` +5. The API will be available at **`http://localhost:8000`**. + +### Services in `docker-compose.yml` + +Below is the default structure (simplified): + +```yaml +services: + api: + image: christianlouis/document-processor:latest + container_name: document_api + working_dir: /workdir + command: ["sh", "-c", "cd /app && uvicorn app.main:app --host 0.0.0.0 --port 8000"] + environment: + - PYTHONPATH=/app + env_file: + - .env + ports: + - "8000:8000" + depends_on: + - redis + - worker + volumes: + - /var/docparse/workdir:/workdir + + worker: + image: christianlouis/document-processor:latest + container_name: document_worker + working_dir: /workdir + command: ["celery", "-A", "app.celery_worker", "worker", "-B", "--loglevel=info", "-Q", "document_processor,default,celery"] + env_file: + - .env + environment: + - PYTHONPATH=/app + depends_on: + - redis + - gotenberg + volumes: + - /var/docparse/workdir:/workdir + + gotenberg: + image: gotenberg/gotenberg:latest + container_name: gotenberg + + redis: + image: redis:alpine + container_name: document_redis + restart: always +``` + +## To-Do List + +- **Refactor AWS-related code** to rely on Azure or remove if no longer needed. +- **Remove unnecessary environment variables** once final service usage is determined. +- **Make upload targets configurable** (e.g., easily choose only Dropbox, Nextcloud, or Paperless). +- **Potentially remove or consolidate S3 upload code** if Azure is the preferred cloud option. + +--- + +**Questions or Issues?** +- Feel free to open an issue or pull request. +- For local testing or development, use `docker-compose up` and watch the logs via `docker-compose logs -f`. +- Ensure your `.env` aligns with the environment variables listed above. If you see unexpected errors, check for typos or missing values. \ No newline at end of file diff --git a/app/celery_app.py b/app/celery_app.py index 085c16d3..9466bfbf 100644 --- a/app/celery_app.py +++ b/app/celery_app.py @@ -1,20 +1,20 @@ -# app/celery_app.py - -from celery import Celery -from app.config import settings - -celery = Celery( - "document_processor", - broker=settings.redis_url, - backend=settings.redis_url, -) - - -# Optionally add this line to retain connection retry behavior at startup: -celery.conf.broker_connection_retry_on_startup = True - -# Set the default queue and routing so that tasks are enqueued on "document_processor" -celery.conf.task_default_queue = 'document_processor' -celery.conf.task_routes = { - "app.tasks.*": {"queue": "document_processor"}, -} +# app/celery_app.py + +from celery import Celery +from app.config import settings + +celery = Celery( + "document_processor", + broker=settings.redis_url, + backend=settings.redis_url, +) + + +# Optionally add this line to retain connection retry behavior at startup: +celery.conf.broker_connection_retry_on_startup = True + +# Set the default queue and routing so that tasks are enqueued on "document_processor" +celery.conf.task_default_queue = 'document_processor' +celery.conf.task_routes = { + "app.tasks.*": {"queue": "document_processor"}, +} diff --git a/app/celery_worker.py b/app/celery_worker.py index a309798d..926ee7e1 100644 --- a/app/celery_worker.py +++ b/app/celery_worker.py @@ -1,42 +1,42 @@ -#!/usr/bin/env python3 - -from app.config import settings - -# Import the shared Celery instance -from app.celery_app import celery - -# Ensure tasks are loaded -from app import tasks # <— This imports app/tasks.py so Celery can register tasks - -# **Ensure all tasks are imported before Celery starts** -from app.tasks.upload_to_s3 import upload_to_s3 -from app.tasks.process_with_textract import process_with_textract -from app.tasks.refine_text_with_gpt import refine_text_with_gpt -from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt -from app.tasks.embed_metadata_into_pdf import embed_metadata_into_pdf -from app.tasks.convert_to_pdf import convert_to_pdf - -# Import new send tasks -from app.tasks.upload_to_dropbox import upload_to_dropbox -from app.tasks.upload_to_paperless import upload_to_paperless -from app.tasks.upload_to_nextcloud import upload_to_nextcloud -from app.tasks.imap_tasks import pull_all_inboxes -from app.tasks.send_to_all import send_to_all_destinations - -celery.conf.task_routes = { - "app.tasks.*": {"queue": "default"}, -} - -@celery.task -def test_task(): - return "Celery is working!" - -# If you want Celery Beat to run the poll task every minute, add: -from celery.schedules import crontab - -celery.conf.beat_schedule = { - "poll-inboxes-every-minute": { - "task": "app.tasks.imap_tasks.pull_all_inboxes", - "schedule": crontab(minute="*/1"), # every 1 minute - }, +#!/usr/bin/env python3 + +from app.config import settings + +# Import the shared Celery instance +from app.celery_app import celery + +# Ensure tasks are loaded +from app import tasks # <— This imports app/tasks.py so Celery can register tasks + +# **Ensure all tasks are imported before Celery starts** +from app.tasks.upload_to_s3 import upload_to_s3 +from app.tasks.process_with_textract import process_with_textract +from app.tasks.refine_text_with_gpt import refine_text_with_gpt +from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt +from app.tasks.embed_metadata_into_pdf import embed_metadata_into_pdf +from app.tasks.convert_to_pdf import convert_to_pdf + +# Import new send tasks +from app.tasks.upload_to_dropbox import upload_to_dropbox +from app.tasks.upload_to_paperless import upload_to_paperless +from app.tasks.upload_to_nextcloud import upload_to_nextcloud +from app.tasks.imap_tasks import pull_all_inboxes +from app.tasks.send_to_all import send_to_all_destinations + +celery.conf.task_routes = { + "app.tasks.*": {"queue": "default"}, +} + +@celery.task +def test_task(): + return "Celery is working!" + +# If you want Celery Beat to run the poll task every minute, add: +from celery.schedules import crontab + +celery.conf.beat_schedule = { + "poll-inboxes-every-minute": { + "task": "app.tasks.imap_tasks.pull_all_inboxes", + "schedule": crontab(minute="*/1"), # every 1 minute + }, } \ No newline at end of file diff --git a/app/config.py b/app/config.py index 8816b7d6..64774626 100644 --- a/app/config.py +++ b/app/config.py @@ -1,53 +1,53 @@ -#!/usr/bin/env python3 - -from pydantic_settings import BaseSettings -from typing import Optional - -class Settings(BaseSettings): - admin_username: str - admin_password: str - aws_access_key_id: str - aws_secret_access_key: str - aws_region: str - database_url: str - redis_url: str - s3_bucket_name: str - openai_api_key: str - workdir: str - dropbox_app_key: str - dropbox_app_secret: str - dropbox_folder: str - dropbox_refresh_token: str - nextcloud_upload_url: str - nextcloud_username: str - nextcloud_password: str - nextcloud_folder: str - paperless_ngx_api_token: str - paperless_host: str - azure_ai_key: str - azure_region: str - azure_endpoint: str - gotenberg_url: str - - # IMAP 1 - imap1_host: Optional[str] = None - imap1_port: Optional[int] = 993 - imap1_username: Optional[str] = None - imap1_password: Optional[str] = None - imap1_ssl: bool = True - imap1_poll_interval_minutes: int = 5 - imap1_delete_after_process: bool = False - - # IMAP 2 - imap2_host: Optional[str] = None - imap2_port: Optional[int] = 993 - imap2_username: Optional[str] = None - imap2_password: Optional[str] = None - imap2_ssl: bool = True - imap2_poll_interval_minutes: int = 10 - imap2_delete_after_process: bool = False - - class Config: - env_file = ".env" - -settings = Settings() +#!/usr/bin/env python3 + +from pydantic_settings import BaseSettings +from typing import Optional + +class Settings(BaseSettings): + admin_username: str + admin_password: str + aws_access_key_id: str + aws_secret_access_key: str + aws_region: str + database_url: str + redis_url: str + s3_bucket_name: str + openai_api_key: str + workdir: str + dropbox_app_key: str + dropbox_app_secret: str + dropbox_folder: str + dropbox_refresh_token: str + nextcloud_upload_url: str + nextcloud_username: str + nextcloud_password: str + nextcloud_folder: str + paperless_ngx_api_token: str + paperless_host: str + azure_ai_key: str + azure_region: str + azure_endpoint: str + gotenberg_url: str + + # IMAP 1 + imap1_host: Optional[str] = None + imap1_port: Optional[int] = 993 + imap1_username: Optional[str] = None + imap1_password: Optional[str] = None + imap1_ssl: bool = True + imap1_poll_interval_minutes: int = 5 + imap1_delete_after_process: bool = False + + # IMAP 2 + imap2_host: Optional[str] = None + imap2_port: Optional[int] = 993 + imap2_username: Optional[str] = None + imap2_password: Optional[str] = None + imap2_ssl: bool = True + imap2_poll_interval_minutes: int = 10 + imap2_delete_after_process: bool = False + + class Config: + env_file = ".env" + +settings = Settings() diff --git a/app/database.py b/app/database.py index bfff1a6b..a513c72d 100644 --- a/app/database.py +++ b/app/database.py @@ -1,17 +1,17 @@ -#!/usr/bin/env python3 - -from sqlalchemy import create_engine, Column, String, Integer -from sqlalchemy.ext.declarative import declarative_base -from sqlalchemy.orm import sessionmaker -from .config import settings - -Base = declarative_base() -engine = create_engine(settings.database_url, connect_args={"check_same_thread": False}) -SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) - -def get_db(): - db = SessionLocal() - try: - yield db - finally: - db.close() +#!/usr/bin/env python3 + +from sqlalchemy import create_engine, Column, String, Integer +from sqlalchemy.ext.declarative import declarative_base +from sqlalchemy.orm import sessionmaker +from .config import settings + +Base = declarative_base() +engine = create_engine(settings.database_url, connect_args={"check_same_thread": False}) +SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) + +def get_db(): + db = SessionLocal() + try: + yield db + finally: + db.close() diff --git a/app/frontend.py b/app/frontend.py index 8fef9bd7..9cec1c42 100644 --- a/app/frontend.py +++ b/app/frontend.py @@ -1,19 +1,24 @@ -# app/frontend.py (new file or inline in main.py) -from fastapi import APIRouter -from fastapi.responses import FileResponse -from fastapi.staticfiles import StaticFiles -import os - -router = APIRouter() - -# 1) Serve the folder that contains index.html, etc. -# e.g. "frontend" is relative to your project root -frontend_folder = os.path.join(os.path.dirname(__file__), "..", "frontend") - -# If you just want to serve the entire folder as static: -router.mount("/static", StaticFiles(directory=frontend_folder), name="static") - -# 2) For the root route ("/"), return the index.html -@router.get("/ui", response_class=FileResponse) -def serve_ui(): - return os.path.join(frontend_folder, "index.html") +# app/frontend.py (new file or inline in main.py) +from fastapi import APIRouter +from fastapi.responses import FileResponse +from fastapi.staticfiles import StaticFiles +import os + +router = APIRouter() + +# 1) Serve the folder that contains index.html, etc. +# e.g. "frontend" is relative to your project root +frontend_folder = os.path.join(os.path.dirname(__file__), "..", "frontend") + +# If you just want to serve the entire folder as static: +router.mount("/static", StaticFiles(directory=frontend_folder), name="static") + +# 2) For the root route ("/"), return the index.html +@router.get("/ui", response_class=FileResponse) +def serve_ui(): + return os.path.join(frontend_folder, "index.html") + +# 3) Serve favicon.ico from the frontend folder +@router.get("/favicon.ico", response_class=FileResponse) +def favicon(): + return os.path.join(frontend_folder, "favicon.ico") \ No newline at end of file diff --git a/app/main.py b/app/main.py index 59912724..9d03538f 100644 --- a/app/main.py +++ b/app/main.py @@ -1,133 +1,133 @@ -#!/usr/bin/env python3 - -import os -from fastapi import FastAPI, HTTPException, UploadFile, File -from app.config import settings -from app.tasks.upload_to_s3 import upload_to_s3 -from app.tasks.upload_to_dropbox import upload_to_dropbox -from app.tasks.upload_to_paperless import upload_to_paperless -from app.tasks.upload_to_nextcloud import upload_to_nextcloud -from app.tasks.send_to_all import send_to_all_destinations -from app.frontend import router as frontend_router - -app = FastAPI(title="Document Processing API") - -@app.get("/") -def root(): - return {"message": "Document Processing API"} - -@app.post("/process/") -def process(file_path: str): - """ - API Endpoint to start document processing. - This enqueues the first task (upload_to_s3), which handles the full pipeline. - """ - - # If file_path is not absolute, treat it as relative to settings.workdir. - if not os.path.isabs(file_path): - file_path = os.path.join(settings.workdir, file_path) - - if not os.path.exists(file_path): - raise HTTPException(status_code=400, detail=f"File {file_path} not found.") - - task = upload_to_s3.delay(file_path) - return {"task_id": task.id, "status": "queued"} - -@app.post("/send_to_dropbox/") -def send_to_dropbox(file_path: str): - if not os.path.isabs(file_path): - file_path = os.path.join(settings.workdir, 'processed', file_path) - if not os.path.exists(file_path): - raise HTTPException(status_code=400, detail=f"File {file_path} not found.") - task = upload_to_dropbox.delay(file_path) - return {"task_id": task.id, "status": "queued"} - -@app.post("/send_to_paperless/") -def send_to_paperless(file_path: str): - if not os.path.isabs(file_path): - file_path = os.path.join(settings.workdir, 'processed', file_path) - if not os.path.exists(file_path): - raise HTTPException(status_code=400, detail=f"File {file_path} not found.") - task = upload_to_paperless.delay(file_path) - return {"task_id": task.id, "status": "queued"} - -@app.post("/send_to_nextcloud/") -def send_to_nextcloud(file_path: str): - if not os.path.isabs(file_path): - file_path = os.path.join(settings.workdir, 'processed', file_path) - if not os.path.exists(file_path): - raise HTTPException(status_code=400, detail=f"File {file_path} not found.") - task = upload_to_nextcloud.delay(file_path) - return {"task_id": task.id, "status": "queued"} - - -@app.post("/send_to_all_destinations/") -def send_to_all_destinations_endpoint(file_path: str): - """ - Call the aggregator task that sends this file to dropbox, nextcloud, and paperless. - """ - if not os.path.isabs(file_path): - # If not absolute, assume it's in processed subdir - file_path = os.path.join(settings.workdir, 'processed', file_path) - - if not os.path.exists(file_path): - raise HTTPException( - status_code=400, - detail=f"File {file_path} not found." - ) - - task = send_to_all_destinations.delay(file_path) - return {"task_id": task.id, "status": "queued", "file_path": file_path} - - - -@app.post("/processall") -def process_all_pdfs_in_workdir(): - """ - Finds all .pdf files in /processed - and enqueues them for upload_to_s3. - """ - target_dir = settings.workdir - if not os.path.exists(target_dir): - raise HTTPException(status_code=400, detail=f"Directory {target_dir} does not exist.") - - pdf_files = [] - for filename in os.listdir(target_dir): - if filename.lower().endswith(".pdf"): - pdf_files.append(filename) - - if not pdf_files: - return {"message": "No PDF files found in processed directory."} - - task_ids = [] - for pdf in pdf_files: - file_path = os.path.join(target_dir, pdf) - # Enqueue upload_to_s3 - from app.tasks.upload_to_s3 import upload_to_s3 - task = upload_to_s3.delay(file_path) - task_ids.append(task.id) - - return { - "message": f"Enqueued {len(pdf_files)} PDFs to upload_to_s3", - "pdf_files": pdf_files, - "task_ids": task_ids - } - -app.include_router(frontend_router) - -@app.post("/ui-upload") -async def ui_upload(file: UploadFile = File(...)): - # You can store this file in your 'workdir' (like how /process does) or a tmp dir - workdir = "/workdir" - target_path = os.path.join(workdir, file.filename) - - try: - with open(target_path, "wb") as f: - content = await file.read() - f.write(content) - except Exception as e: - raise HTTPException(status_code=500, detail=f"Failed to save file: {e}") - - # Now you can call your existing Celery flow: - task = upload_to_s3.delay(target_path) +#!/usr/bin/env python3 + +import os +from fastapi import FastAPI, HTTPException, UploadFile, File +from app.config import settings +from app.tasks.upload_to_s3 import upload_to_s3 +from app.tasks.upload_to_dropbox import upload_to_dropbox +from app.tasks.upload_to_paperless import upload_to_paperless +from app.tasks.upload_to_nextcloud import upload_to_nextcloud +from app.tasks.send_to_all import send_to_all_destinations +from app.frontend import router as frontend_router + +app = FastAPI(title="Document Processing API") + +@app.get("/") +def root(): + return {"message": "Document Processing API"} + +@app.post("/process/") +def process(file_path: str): + """ + API Endpoint to start document processing. + This enqueues the first task (upload_to_s3), which handles the full pipeline. + """ + + # If file_path is not absolute, treat it as relative to settings.workdir. + if not os.path.isabs(file_path): + file_path = os.path.join(settings.workdir, file_path) + + if not os.path.exists(file_path): + raise HTTPException(status_code=400, detail=f"File {file_path} not found.") + + task = upload_to_s3.delay(file_path) + return {"task_id": task.id, "status": "queued"} + +@app.post("/send_to_dropbox/") +def send_to_dropbox(file_path: str): + if not os.path.isabs(file_path): + file_path = os.path.join(settings.workdir, 'processed', file_path) + if not os.path.exists(file_path): + raise HTTPException(status_code=400, detail=f"File {file_path} not found.") + task = upload_to_dropbox.delay(file_path) + return {"task_id": task.id, "status": "queued"} + +@app.post("/send_to_paperless/") +def send_to_paperless(file_path: str): + if not os.path.isabs(file_path): + file_path = os.path.join(settings.workdir, 'processed', file_path) + if not os.path.exists(file_path): + raise HTTPException(status_code=400, detail=f"File {file_path} not found.") + task = upload_to_paperless.delay(file_path) + return {"task_id": task.id, "status": "queued"} + +@app.post("/send_to_nextcloud/") +def send_to_nextcloud(file_path: str): + if not os.path.isabs(file_path): + file_path = os.path.join(settings.workdir, 'processed', file_path) + if not os.path.exists(file_path): + raise HTTPException(status_code=400, detail=f"File {file_path} not found.") + task = upload_to_nextcloud.delay(file_path) + return {"task_id": task.id, "status": "queued"} + + +@app.post("/send_to_all_destinations/") +def send_to_all_destinations_endpoint(file_path: str): + """ + Call the aggregator task that sends this file to dropbox, nextcloud, and paperless. + """ + if not os.path.isabs(file_path): + # If not absolute, assume it's in processed subdir + file_path = os.path.join(settings.workdir, 'processed', file_path) + + if not os.path.exists(file_path): + raise HTTPException( + status_code=400, + detail=f"File {file_path} not found." + ) + + task = send_to_all_destinations.delay(file_path) + return {"task_id": task.id, "status": "queued", "file_path": file_path} + + + +@app.post("/processall") +def process_all_pdfs_in_workdir(): + """ + Finds all .pdf files in /processed + and enqueues them for upload_to_s3. + """ + target_dir = settings.workdir + if not os.path.exists(target_dir): + raise HTTPException(status_code=400, detail=f"Directory {target_dir} does not exist.") + + pdf_files = [] + for filename in os.listdir(target_dir): + if filename.lower().endswith(".pdf"): + pdf_files.append(filename) + + if not pdf_files: + return {"message": "No PDF files found in processed directory."} + + task_ids = [] + for pdf in pdf_files: + file_path = os.path.join(target_dir, pdf) + # Enqueue upload_to_s3 + from app.tasks.upload_to_s3 import upload_to_s3 + task = upload_to_s3.delay(file_path) + task_ids.append(task.id) + + return { + "message": f"Enqueued {len(pdf_files)} PDFs to upload_to_s3", + "pdf_files": pdf_files, + "task_ids": task_ids + } + +app.include_router(frontend_router) + +@app.post("/ui-upload") +async def ui_upload(file: UploadFile = File(...)): + # You can store this file in your 'workdir' (like how /process does) or a tmp dir + workdir = "/workdir" + target_path = os.path.join(workdir, file.filename) + + try: + with open(target_path, "wb") as f: + content = await file.read() + f.write(content) + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to save file: {e}") + + # Now you can call your existing Celery flow: + task = upload_to_s3.delay(target_path) return {"task_id": task.id, "status": "queued"} \ No newline at end of file diff --git a/app/models.py b/app/models.py index 251c4e2c..0ad894e4 100644 --- a/app/models.py +++ b/app/models.py @@ -1,14 +1,14 @@ -#!/usr/bin/env python3 - -from .database import Base -from sqlalchemy import Column, String, Integer - -class DocumentMetadata(Base): - __tablename__ = "documents" - - id = Column(Integer, primary_key=True, index=True) - filename = Column(String, unique=True, index=True) - sender = Column(String) - recipient = Column(String) - tags = Column(String) - summary = Column(String) +#!/usr/bin/env python3 + +from .database import Base +from sqlalchemy import Column, String, Integer + +class DocumentMetadata(Base): + __tablename__ = "documents" + + id = Column(Integer, primary_key=True, index=True) + filename = Column(String, unique=True, index=True) + sender = Column(String) + recipient = Column(String) + tags = Column(String) + summary = Column(String) diff --git a/app/tasks/convert_to_pdf.py b/app/tasks/convert_to_pdf.py index 3bebc951..f03d039e 100644 --- a/app/tasks/convert_to_pdf.py +++ b/app/tasks/convert_to_pdf.py @@ -1,72 +1,72 @@ -#!/usr/bin/env python3 -import os -import requests -import logging -import mimetypes -from celery import shared_task -from app.config import settings -from app.tasks.upload_to_s3 import upload_to_s3 - -logger = logging.getLogger(__name__) - -@shared_task -def convert_to_pdf(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 S3 upload. - """ - gotenberg_url = getattr(settings, "gotenberg_url", None) - if not gotenberg_url: - logger.error("Gotenberg URL is not configured in settings.") - return - - # Try to guess the MIME type based on file content (using extension-based fallback) - mime_type, encoding = mimetypes.guess_type(file_path) - logger.info(f"Guessed MIME type for '{file_path}' is: {mime_type}") - - endpoint = None - form_key = "files" # Default form key for most endpoints - - if mime_type: - if mime_type == "text/html": - endpoint = f"{gotenberg_url}/forms/chromium/convert/html" - # The Chromium HTML endpoint expects the HTML file to be provided under the key "index.html" - form_key = "index.html" - elif mime_type.startswith("image/"): - # For images, we use the LibreOffice endpoint (which supports image conversion) - endpoint = f"{gotenberg_url}/forms/libreoffice/convert" - elif mime_type.startswith("text/plain"): - endpoint = f"{gotenberg_url}/forms/libreoffice/convert" - elif mime_type in ["text/markdown", "text/x-markdown"]: - # Optionally, you could use the Chromium markdown endpoint if you have an HTML wrapper. - # For now, we'll fallback to LibreOffice. - endpoint = f"{gotenberg_url}/forms/libreoffice/convert" - else: - # For all other MIME types (e.g. Office documents), use the LibreOffice endpoint. - endpoint = f"{gotenberg_url}/forms/libreoffice/convert" - else: - # If MIME detection fails, fallback to extension-based detection. - ext = os.path.splitext(file_path)[1].lower() - if ext in [".html", ".htm"]: - endpoint = f"{gotenberg_url}/forms/chromium/convert/html" - form_key = "index.html" - else: - endpoint = f"{gotenberg_url}/forms/libreoffice/convert" - - try: - with open(file_path, "rb") as f: - files = {form_key: f} - response = requests.post(endpoint, files=files) - - if response.status_code == 200: - converted_file_path = os.path.splitext(file_path)[0] + ".pdf" - 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}") - upload_to_s3.delay(converted_file_path) - return converted_file_path - else: - logger.error(f"Conversion failed for {file_path}. Status code: {response.status_code}") - except Exception as e: - logger.exception(f"Error converting {file_path} to PDF: {e}") +#!/usr/bin/env python3 +import os +import requests +import logging +import mimetypes +from celery import shared_task +from app.config import settings +from app.tasks.upload_to_s3 import upload_to_s3 + +logger = logging.getLogger(__name__) + +@shared_task +def convert_to_pdf(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 S3 upload. + """ + gotenberg_url = getattr(settings, "gotenberg_url", None) + if not gotenberg_url: + logger.error("Gotenberg URL is not configured in settings.") + return + + # Try to guess the MIME type based on file content (using extension-based fallback) + mime_type, encoding = mimetypes.guess_type(file_path) + logger.info(f"Guessed MIME type for '{file_path}' is: {mime_type}") + + endpoint = None + form_key = "files" # Default form key for most endpoints + + if mime_type: + if mime_type == "text/html": + endpoint = f"{gotenberg_url}/forms/chromium/convert/html" + # The Chromium HTML endpoint expects the HTML file to be provided under the key "index.html" + form_key = "index.html" + elif mime_type.startswith("image/"): + # For images, we use the LibreOffice endpoint (which supports image conversion) + endpoint = f"{gotenberg_url}/forms/libreoffice/convert" + elif mime_type.startswith("text/plain"): + endpoint = f"{gotenberg_url}/forms/libreoffice/convert" + elif mime_type in ["text/markdown", "text/x-markdown"]: + # Optionally, you could use the Chromium markdown endpoint if you have an HTML wrapper. + # For now, we'll fallback to LibreOffice. + endpoint = f"{gotenberg_url}/forms/libreoffice/convert" + else: + # For all other MIME types (e.g. Office documents), use the LibreOffice endpoint. + endpoint = f"{gotenberg_url}/forms/libreoffice/convert" + else: + # If MIME detection fails, fallback to extension-based detection. + ext = os.path.splitext(file_path)[1].lower() + if ext in [".html", ".htm"]: + endpoint = f"{gotenberg_url}/forms/chromium/convert/html" + form_key = "index.html" + else: + endpoint = f"{gotenberg_url}/forms/libreoffice/convert" + + try: + with open(file_path, "rb") as f: + files = {form_key: f} + response = requests.post(endpoint, files=files) + + if response.status_code == 200: + converted_file_path = os.path.splitext(file_path)[0] + ".pdf" + 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}") + upload_to_s3.delay(converted_file_path) + return converted_file_path + else: + logger.error(f"Conversion failed for {file_path}. Status code: {response.status_code}") + except Exception as e: + logger.exception(f"Error converting {file_path} to PDF: {e}") diff --git a/app/tasks/embed_metadata_into_pdf.py b/app/tasks/embed_metadata_into_pdf.py index b20a40da..d5d59254 100644 --- a/app/tasks/embed_metadata_into_pdf.py +++ b/app/tasks/embed_metadata_into_pdf.py @@ -1,128 +1,128 @@ -#!/usr/bin/env python3 - -import os -import shutil -import fitz # PyMuPDF for PDF metadata editing -import json -from app.config import settings -from app.tasks.retry_config import BaseTaskWithRetry -from app.tasks.finalize_document_storage import finalize_document_storage - -# Import the shared Celery instance -from app.celery_app import celery - -def unique_filepath(directory, base_filename, extension=".pdf"): - """ - Returns a unique filepath in the specified directory. - If 'base_filename.pdf' exists, it will append an underscore and counter. - """ - candidate = os.path.join(directory, base_filename + extension) - if not os.path.exists(candidate): - return candidate - counter = 1 - while True: - candidate = os.path.join(directory, f"{base_filename}_{counter}{extension}") - if not os.path.exists(candidate): - return candidate - counter += 1 - -def persist_metadata(metadata, final_pdf_path): - """ - Saves the metadata dictionary to a JSON file with the same base name as the final PDF. - For example, if final_pdf_path is "/processed/MyFile.pdf", - the metadata will be saved as "/processed/MyFile.json". - """ - base, _ = os.path.splitext(final_pdf_path) - json_path = base + ".json" - with open(json_path, "w", encoding="utf-8") as f: - 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): - """ - Embeds extracted metadata into the PDF's standard metadata fields. - The mapping is as follows: - - title: uses the extracted metadata "filename" - - author: uses "absender" (or "Unknown" if missing) - - subject: uses "document_type" (or "Unknown") - - keywords: a comma‐separated list from the "tags" field - - After processing, the file is moved to - /processed/ - where is derived from metadata["filename"]. - The output PDF is saved incrementally while preserving its original encryption. - Additionally, the metadata is persisted to a JSON file with the same base name. - """ - # 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)) - 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.") - return {"error": "File not found"} - - # Work on a safe copy in /tmp - tmp_dir = "/tmp" - original_file = local_file_path - processed_file = os.path.join(tmp_dir, f"processed_{os.path.basename(local_file_path)}") - - # Create a safe copy to work on - shutil.copy(original_file, processed_file) - - try: - print(f"[DEBUG] Embedding metadata into {processed_file}...") - - # Open the PDF - doc = fitz.open(processed_file) - # Set PDF metadata using only the standard keys. - doc.set_metadata({ - "title": metadata.get("filename", "Unknown Document"), - "author": metadata.get("absender", "Unknown"), - "subject": metadata.get("document_type", "Unknown"), - "keywords": ", ".join(metadata.get("tags", [])) - }) - # Save incrementally and preserve encryption - doc.save(processed_file, incremental=True, encryption=fitz.PDF_ENCRYPT_KEEP) - doc.close() - - print(f"[INFO] Metadata embedded successfully in {processed_file}") - - # 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") - 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") - - # 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) - - # Persist the metadata into a JSON file with the same base name. - json_path = persist_metadata(metadata, final_file_path) - print(f"[INFO] Metadata persisted to {json_path}") - - # Trigger the next step: final storage. - finalize_document_storage.delay(original_file, final_file_path, metadata) - - # After triggering final storage, delete the original file if it is in workdir/tmp. - workdir_tmp = os.path.join(settings.workdir, "tmp") - 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}") - except Exception as e: - print(f"[ERROR] 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}") - return {"error": str(e)} +#!/usr/bin/env python3 + +import os +import shutil +import fitz # PyMuPDF for PDF metadata editing +import json +from app.config import settings +from app.tasks.retry_config import BaseTaskWithRetry +from app.tasks.finalize_document_storage import finalize_document_storage + +# Import the shared Celery instance +from app.celery_app import celery + +def unique_filepath(directory, base_filename, extension=".pdf"): + """ + Returns a unique filepath in the specified directory. + If 'base_filename.pdf' exists, it will append an underscore and counter. + """ + candidate = os.path.join(directory, base_filename + extension) + if not os.path.exists(candidate): + return candidate + counter = 1 + while True: + candidate = os.path.join(directory, f"{base_filename}_{counter}{extension}") + if not os.path.exists(candidate): + return candidate + counter += 1 + +def persist_metadata(metadata, final_pdf_path): + """ + Saves the metadata dictionary to a JSON file with the same base name as the final PDF. + For example, if final_pdf_path is "/processed/MyFile.pdf", + the metadata will be saved as "/processed/MyFile.json". + """ + base, _ = os.path.splitext(final_pdf_path) + json_path = base + ".json" + with open(json_path, "w", encoding="utf-8") as f: + 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): + """ + Embeds extracted metadata into the PDF's standard metadata fields. + The mapping is as follows: + - title: uses the extracted metadata "filename" + - author: uses "absender" (or "Unknown" if missing) + - subject: uses "document_type" (or "Unknown") + - keywords: a comma‐separated list from the "tags" field + + After processing, the file is moved to + /processed/ + where is derived from metadata["filename"]. + The output PDF is saved incrementally while preserving its original encryption. + Additionally, the metadata is persisted to a JSON file with the same base name. + """ + # 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)) + 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.") + return {"error": "File not found"} + + # Work on a safe copy in /tmp + tmp_dir = "/tmp" + original_file = local_file_path + processed_file = os.path.join(tmp_dir, f"processed_{os.path.basename(local_file_path)}") + + # Create a safe copy to work on + shutil.copy(original_file, processed_file) + + try: + print(f"[DEBUG] Embedding metadata into {processed_file}...") + + # Open the PDF + doc = fitz.open(processed_file) + # Set PDF metadata using only the standard keys. + doc.set_metadata({ + "title": metadata.get("filename", "Unknown Document"), + "author": metadata.get("absender", "Unknown"), + "subject": metadata.get("document_type", "Unknown"), + "keywords": ", ".join(metadata.get("tags", [])) + }) + # Save incrementally and preserve encryption + doc.save(processed_file, incremental=True, encryption=fitz.PDF_ENCRYPT_KEEP) + doc.close() + + print(f"[INFO] Metadata embedded successfully in {processed_file}") + + # 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") + 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") + + # 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) + + # Persist the metadata into a JSON file with the same base name. + json_path = persist_metadata(metadata, final_file_path) + print(f"[INFO] Metadata persisted to {json_path}") + + # Trigger the next step: final storage. + finalize_document_storage.delay(original_file, final_file_path, metadata) + + # After triggering final storage, delete the original file if it is in workdir/tmp. + workdir_tmp = os.path.join(settings.workdir, "tmp") + 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}") + except Exception as e: + print(f"[ERROR] 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}") + return {"error": str(e)} diff --git a/app/tasks/extract_metadata_with_gpt.py b/app/tasks/extract_metadata_with_gpt.py index 22c83b0d..5adf01c0 100644 --- a/app/tasks/extract_metadata_with_gpt.py +++ b/app/tasks/extract_metadata_with_gpt.py @@ -1,97 +1,97 @@ -#!/usr/bin/env python3 - -import json -import re -from openai import OpenAI -from app.config import settings -from app.tasks.retry_config import BaseTaskWithRetry -from app.tasks.embed_metadata_into_pdf import embed_metadata_into_pdf - -# Import the shared Celery instance -from app.celery_app import celery - -client = OpenAI(api_key=settings.openai_api_key) - -def extract_json_from_text(text): - """ - Try to extract a JSON object from the text. - - First, check for a JSON block inside triple backticks. - - If not found, try to extract text from the first '{' to the last '}'. - """ - pattern = r"```(?:json)?\s*(\{.*?\})\s*```" - match = re.search(pattern, text, re.DOTALL) - if match: - return match.group(1) - else: - start = text.find("{") - end = text.rfind("}") - if start != -1 and end != -1 and end > start: - return text[start:end+1] - return None - -@celery.task(base=BaseTaskWithRetry) -def extract_metadata_with_gpt(s3_filename: str, cleaned_text: str): - """Uses OpenAI GPT-4o-mini to classify document metadata.""" - - 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. - -Extract and return the following fields: -1. **filename**: Machine-readable filename (YYYY-MM-DD_DescriptiveTitle, use only letters, numbers, periods, and underscores). -2. **empfaenger**: The recipient, or "Unknown" if not found. -3. **absender**: The sender, or "Unknown" if not found. -4. **correspondent**: The entity or company that issued the document (shortest possible name, e.g., "Amazon" instead of "Amazon EU SARL, German branch"). -5. **kommunikationsart**: One of [Behoerdlicher_Brief, Rechnung, Kontoauszug, Vertrag, Quittung, Privater_Brief, Einladung, Gewerbliche_Korrespondenz, Newsletter, Werbung, Sonstiges]. -6. **kommunikationskategorie**: One of [Amtliche_Postbehoerdliche_Dokumente, Finanz_und_Vertragsdokumente, Geschaeftliche_Kommunikation, Private_Korrespondenz, Sonstige_Informationen]. -7. **document_type**: Precise classification (e.g., Invoice, Contract, Information, Unknown). -8. **tags**: A list of up to 4 relevant thematic keywords. -9. **language**: Detected document language (ISO 639-1 code, e.g., "de" or "en"). -10. **title**: A human-readable title summarizing the document content. -11. **confidence_score**: A numeric value (0-100) indicating the confidence level of the extracted metadata. -12. **reference_number**: Extracted invoice/order/reference number if available. -13. **monetary_amounts**: A list of key monetary values detected in the document. - -### Important Rules: -- **OCR Correction**: Assume the text has been corrected for OCR errors. -- **Tagging**: Max 4 tags, avoiding generic or overly specific terms. -- **Title**: Concise, no addresses, and contains key identifying features. -- **Date Selection**: Use the most relevant date if multiple are found. -- **Output Language**: Maintain the document's original language. - -Extracted text: -{cleaned_text} - -Return only valid JSON with no additional commentary. -""" - - try: - print(f"[DEBUG] Sending classification request for {s3_filename}...") - completion = client.chat.completions.create( - model="gpt-4o-mini", - messages=[ - {"role": "system", "content": "You are an intelligent document classifier."}, - {"role": "user", "content": prompt} - ], - temperature=0 - ) - - content = completion.choices[0].message.content - print(f"[DEBUG] Raw classification response for {s3_filename}: {content}") - - json_text = extract_json_from_text(content) - if not json_text: - print(f"[ERROR] Could not find valid JSON in GPT response for {s3_filename}.") - return {} - - metadata = json.loads(json_text) - print(f"[DEBUG] Extracted metadata: {metadata}") - - # Trigger the next step: embedding metadata into the PDF - embed_metadata_into_pdf.delay(s3_filename, cleaned_text, metadata) - - return {"s3_file": s3_filename, "metadata": metadata} - - except Exception as e: - print(f"[ERROR] OpenAI classification failed for {s3_filename}: {e}") - return {} +#!/usr/bin/env python3 + +import json +import re +from openai import OpenAI +from app.config import settings +from app.tasks.retry_config import BaseTaskWithRetry +from app.tasks.embed_metadata_into_pdf import embed_metadata_into_pdf + +# Import the shared Celery instance +from app.celery_app import celery + +client = OpenAI(api_key=settings.openai_api_key) + +def extract_json_from_text(text): + """ + Try to extract a JSON object from the text. + - First, check for a JSON block inside triple backticks. + - If not found, try to extract text from the first '{' to the last '}'. + """ + pattern = r"```(?:json)?\s*(\{.*?\})\s*```" + match = re.search(pattern, text, re.DOTALL) + if match: + return match.group(1) + else: + start = text.find("{") + end = text.rfind("}") + if start != -1 and end != -1 and end > start: + return text[start:end+1] + return None + +@celery.task(base=BaseTaskWithRetry) +def extract_metadata_with_gpt(s3_filename: str, cleaned_text: str): + """Uses OpenAI GPT-4o-mini to classify document metadata.""" + + 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. + +Extract and return the following fields: +1. **filename**: Machine-readable filename (YYYY-MM-DD_DescriptiveTitle, use only letters, numbers, periods, and underscores). +2. **empfaenger**: The recipient, or "Unknown" if not found. +3. **absender**: The sender, or "Unknown" if not found. +4. **correspondent**: The entity or company that issued the document (shortest possible name, e.g., "Amazon" instead of "Amazon EU SARL, German branch"). +5. **kommunikationsart**: One of [Behoerdlicher_Brief, Rechnung, Kontoauszug, Vertrag, Quittung, Privater_Brief, Einladung, Gewerbliche_Korrespondenz, Newsletter, Werbung, Sonstiges]. +6. **kommunikationskategorie**: One of [Amtliche_Postbehoerdliche_Dokumente, Finanz_und_Vertragsdokumente, Geschaeftliche_Kommunikation, Private_Korrespondenz, Sonstige_Informationen]. +7. **document_type**: Precise classification (e.g., Invoice, Contract, Information, Unknown). +8. **tags**: A list of up to 4 relevant thematic keywords. +9. **language**: Detected document language (ISO 639-1 code, e.g., "de" or "en"). +10. **title**: A human-readable title summarizing the document content. +11. **confidence_score**: A numeric value (0-100) indicating the confidence level of the extracted metadata. +12. **reference_number**: Extracted invoice/order/reference number if available. +13. **monetary_amounts**: A list of key monetary values detected in the document. + +### Important Rules: +- **OCR Correction**: Assume the text has been corrected for OCR errors. +- **Tagging**: Max 4 tags, avoiding generic or overly specific terms. +- **Title**: Concise, no addresses, and contains key identifying features. +- **Date Selection**: Use the most relevant date if multiple are found. +- **Output Language**: Maintain the document's original language. + +Extracted text: +{cleaned_text} + +Return only valid JSON with no additional commentary. +""" + + try: + print(f"[DEBUG] Sending classification request for {s3_filename}...") + completion = client.chat.completions.create( + model="gpt-4o-mini", + messages=[ + {"role": "system", "content": "You are an intelligent document classifier."}, + {"role": "user", "content": prompt} + ], + temperature=0 + ) + + content = completion.choices[0].message.content + print(f"[DEBUG] Raw classification response for {s3_filename}: {content}") + + json_text = extract_json_from_text(content) + if not json_text: + print(f"[ERROR] Could not find valid JSON in GPT response for {s3_filename}.") + return {} + + metadata = json.loads(json_text) + print(f"[DEBUG] Extracted metadata: {metadata}") + + # Trigger the next step: embedding metadata into the PDF + embed_metadata_into_pdf.delay(s3_filename, cleaned_text, metadata) + + return {"s3_file": s3_filename, "metadata": metadata} + + except Exception as e: + print(f"[ERROR] OpenAI classification failed for {s3_filename}: {e}") + return {} diff --git a/app/tasks/finalize_document_storage.py b/app/tasks/finalize_document_storage.py index cf9f6944..6d5216d0 100644 --- a/app/tasks/finalize_document_storage.py +++ b/app/tasks/finalize_document_storage.py @@ -1,26 +1,26 @@ -#!/usr/bin/env python3 - -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 -from app.tasks.send_to_all import send_to_all_destinations - - -@celery.task(base=BaseTaskWithRetry) -def finalize_document_storage(original_file: str, processed_file: str, metadata: dict): - """ - Final storage step after embedding metadata. - We will now call 'send_to_all_destinations' to push the final PDF to Dropbox/Nextcloud/Paperless. - """ - print(f"[INFO] Finalizing document storage for {processed_file}") - - # 2) Enqueue uploads to all destinations (Dropbox, Nextcloud, Paperless) - send_to_all_destinations.delay(processed_file) - - return { - "status": "Completed", - "file": processed_file - } +#!/usr/bin/env python3 + +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 +from app.tasks.send_to_all import send_to_all_destinations + + +@celery.task(base=BaseTaskWithRetry) +def finalize_document_storage(original_file: str, processed_file: str, metadata: dict): + """ + Final storage step after embedding metadata. + We will now call 'send_to_all_destinations' to push the final PDF to Dropbox/Nextcloud/Paperless. + """ + print(f"[INFO] Finalizing document storage for {processed_file}") + + # 2) Enqueue uploads to all destinations (Dropbox, Nextcloud, Paperless) + send_to_all_destinations.delay(processed_file) + + return { + "status": "Completed", + "file": processed_file + } diff --git a/app/tasks/imap_tasks.py b/app/tasks/imap_tasks.py index b00522c7..b41572c2 100644 --- a/app/tasks/imap_tasks.py +++ b/app/tasks/imap_tasks.py @@ -1,412 +1,412 @@ -#!/usr/bin/env python3 -import os -import json -import email -import imaplib -import logging -import redis -import re -from datetime import datetime, timedelta, timezone -from celery import shared_task -from app.config import settings -from app.tasks.upload_to_s3 import upload_to_s3 -from app.tasks.convert_to_pdf import convert_to_pdf # new conversion task - -logger = logging.getLogger(__name__) - -# Initialize Redis connection using Celery's Redis settings -redis_client = redis.StrictRedis.from_url(settings.redis_url, decode_responses=True) - -LOCK_KEY = "imap_lock" # Unique key for locking -LOCK_EXPIRE = 300 # Lock expires in 5 minutes - -# Local cache file for tracking processed emails -CACHE_FILE = os.path.join(settings.workdir, "processed_mails.json") - - -def acquire_lock(): - """Attempt to acquire a Redis-based lock. If acquired, set an expiration.""" - lock_acquired = redis_client.setnx(LOCK_KEY, "locked") - if lock_acquired: - redis_client.expire(LOCK_KEY, LOCK_EXPIRE) - logger.info("Lock acquired for IMAP processing.") - return True - logger.warning("Lock already held. Skipping this cycle.") - return False - - -def release_lock(): - """Release the lock by deleting the Redis key.""" - redis_client.delete(LOCK_KEY) - logger.info("Lock released.") - - -def load_processed_emails(): - """Load the list of already processed emails from a local JSON file.""" - if os.path.exists(CACHE_FILE): - try: - with open(CACHE_FILE, "r") as f: - processed_emails = json.load(f) - processed_emails = cleanup_old_entries(processed_emails) - return processed_emails - except json.JSONDecodeError: - logger.warning("Failed to decode JSON, resetting processed emails cache.") - return {} - return {} - - -def save_processed_emails(processed_emails): - """Save the processed email IDs to a local JSON file.""" - with open(CACHE_FILE, "w") as f: - json.dump(processed_emails, f, indent=4) - - -def cleanup_old_entries(processed_emails): - """Remove entries older than 7 days from the cache to avoid infinite growth.""" - seven_days_ago = datetime.now(timezone.utc) - timedelta(days=7) - valid_emails = {} - for msg_id, date_str in processed_emails.items(): - naive_dt = datetime.strptime(date_str, "%Y-%m-%dT%H:%M:%S") - aware_dt = naive_dt.replace(tzinfo=timezone.utc) - if aware_dt > seven_days_ago: - valid_emails[msg_id] = date_str - return valid_emails - - -@shared_task -def pull_all_inboxes(): - """ - Periodic Celery task that checks all configured IMAP mailboxes - and fetches attachments from new emails. - Ensures only one instance runs at a time using Redis-based locking. - """ - if not acquire_lock(): - logger.info("Skipping execution: Another instance is running.") - return - - try: - logger.info("Starting pull_all_inboxes") - - # Mailbox #1 (non-Gmail) - check_and_pull_mailbox( - mailbox_key="imap1", - host=settings.imap1_host, - port=settings.imap1_port, - username=settings.imap1_username, - password=settings.imap1_password, - use_ssl=settings.imap1_ssl, - delete_after_process=settings.imap1_delete_after_process, - ) - - # Mailbox #2 (Gmail) - check_and_pull_mailbox( - mailbox_key="imap2", - host=settings.imap2_host, - port=settings.imap2_port, - username=settings.imap2_username, - password=settings.imap2_password, - use_ssl=settings.imap2_ssl, - delete_after_process=settings.imap2_delete_after_process, - ) - - logger.info("Finished pull_all_inboxes") - - finally: - release_lock() - - -def check_and_pull_mailbox( - mailbox_key: str, - host: str | None, - port: int | None, - username: str | None, - password: str | None, - use_ssl: bool, - delete_after_process: bool, -): - """Validates config and invokes pulling from the mailbox if valid.""" - if not (host and port and username and password): - logger.warning(f"Mailbox {mailbox_key} is missing config, skipping.") - return - - logger.info(f"Checking mailbox: {mailbox_key}") - pull_inbox( - mailbox_key=mailbox_key, - host=host, - port=port, - username=username, - password=password, - use_ssl=use_ssl, - delete_after_process=delete_after_process, - ) - - -def pull_inbox(mailbox_key, host, port, username, password, use_ssl, - delete_after_process): - """ - Connects to the IMAP inbox, fetches new unread emails from the last 3 days, - and processes attachments while preserving the original unread status. - - For Gmail: - - Attempts to select the localized All Mail folder. - - Runs an X-GM-RAW query: "in:anywhere in:unread newer_than:3d has:attachment". - - For non-Gmail mailboxes, it falls back to selecting the INBOX with a SINCE/UNSEEN filter. - """ - logger.info("Connecting to %s at %s:%s (SSL=%s)", - mailbox_key, host, port, use_ssl) - processed_emails = load_processed_emails() - - try: - mail = imaplib.IMAP4_SSL(host, port) if use_ssl else imaplib.IMAP4(host, port) - mail.login(username, password) - - is_gmail_host = "gmail" in host.lower() - if is_gmail_host: - # For Gmail, try to select the localized All Mail folder. - all_mail_folder = find_all_mail_folder(mail) - if all_mail_folder: - logger.info("Using Gmail All Mail folder: %s", all_mail_folder) - mail.select(f'"{all_mail_folder}"') - else: - logger.warning("Gmail All Mail folder not found, falling back to INBOX.") - mail.select("INBOX") - # Use the X-GM-RAW query for Gmail. - raw_query = "in:anywhere in:unread newer_than:3d has:attachment" - status, search_data = mail.search(None, "X-GM-RAW", f'"{raw_query}"') - else: - # For non-Gmail, select INBOX and use SINCE/UNSEEN query. - mail.select("INBOX") - since_date = (datetime.now(timezone.utc) - timedelta(days=3) - ).strftime("%d-%b-%Y") - status, search_data = mail.search(None, f'(SINCE {since_date} UNSEEN)') - - if status != "OK": - logger.warning("Search failed on mailbox %s. Status=%s", - mailbox_key, status) - mail.close() - mail.logout() - return - - msg_numbers = search_data[0].split() - logger.info("Found %d unread emails in %s.", len(msg_numbers), mailbox_key) - - for num in msg_numbers: - status, msg_data = mail.fetch(num, "(RFC822)") - if status != "OK": - logger.warning("Failed to fetch message %s in %s. Status=%s", - num, mailbox_key, status) - continue - - raw_email = msg_data[0][1] - email_message = email.message_from_bytes(raw_email) - msg_id = email_message.get("Message-ID") - - if not msg_id: - logger.warning("Skipping email without Message-ID in %s", mailbox_key) - continue - - if msg_id in processed_emails: - logger.info("Skipping already processed email %s in %s", msg_id, mailbox_key) - continue - - # For Gmail, check if the email already has the "Ingested" label. - if is_gmail_host: - if email_already_has_label(mail, num, "Ingested"): - logger.info("Skipping email %s in %s, already labeled 'Ingested'.", - msg_id, mailbox_key) - continue - - # Process attachments (and convert non-PDF files). - # We call the function without assigning its return value since it is not used. - fetch_attachments_and_enqueue(email_message) - - if is_gmail_host: - mark_as_processed_with_star(mail, num) - mark_as_processed_with_label(mail, num, label="Ingested") - - processed_emails[msg_id] = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S") - save_processed_emails(processed_emails) - - if delete_after_process: - logger.info("Deleting message %s from %s", num.decode(), mailbox_key) - mail.store(num, "+FLAGS", "\\Deleted") - else: - mail.store(num, "-FLAGS", "\\Seen") - - if delete_after_process: - mail.expunge() - - mail.close() - mail.logout() - logger.info("Finished processing mailbox %s", mailbox_key) - - except Exception as e: - logger.exception("Error pulling mailbox %s: %s", mailbox_key, e) - - -def fetch_attachments_and_enqueue(email_message): - """ - Extracts attachments from the email and processes only allowed file types. - - Allowed file types include: - - PDF: application/pdf - - Microsoft Office files: - - Word: application/msword, - application/vnd.openxmlformats-officedocument.wordprocessingml.document - - Excel: application/vnd.ms-excel, - application/vnd.openxmlformats-officedocument.spreadsheetml.sheet - - PowerPoint: application/vnd.ms-powerpoint, - application/vnd.openxmlformats-officedocument.presentationml.presentation - - Other meaningful attachments: - - Plain text: text/plain - - CSV: text/csv - - Rich Text Format: application/rtf, text/rtf - - Attachments not in this list are skipped. Common image MIME types such as - image/jpeg, image/png, image/gif, image/bmp, image/tiff, and image/webp are - intentionally excluded. - - If the attachment is a PDF, it is enqueued for upload; any other allowed file - is enqueued for conversion to PDF. - - Returns True if at least one allowed attachment was processed. - """ - ALLOWED_MIME_TYPES = { - "application/pdf", - "application/msword", - "application/vnd.openxmlformats-officedocument.wordprocessingml.document", - "application/vnd.ms-excel", - "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", - "application/vnd.ms-powerpoint", - "application/vnd.openxmlformats-officedocument.presentationml.presentation", - "text/plain", - "text/csv", - "application/rtf", - "text/rtf", - } - - has_attachment = False - for part in email_message.walk(): - if part.get_content_maintype() == "multipart": - continue - - filename = part.get_filename() - if not filename: - continue - - mime_type = part.get_content_type() - if mime_type not in ALLOWED_MIME_TYPES: - logger.info("Skipping attachment %s with MIME type %s", - filename, mime_type) - continue - - file_path = os.path.join(settings.workdir, filename) - with open(file_path, "wb") as f: - f.write(part.get_payload(decode=True)) - - if mime_type == "application/pdf": - upload_to_s3.delay(file_path) - logger.info("Enqueued PDF for upload: %s", filename) - elif mime_type in ALLOWED_MIME_TYPES: - # Enqueue conversion to PDF using the Gotenberg service. - convert_to_pdf.delay(file_path) - logger.info("Enqueued file for conversion to PDF: %s", filename) - - has_attachment = True - return has_attachment - - -def email_already_has_label(mail, msg_id, label="Ingested"): - """ - Checks if the given message (msg_id) has the specified Gmail label. - Returns True if the label is found, False otherwise. - """ - try: - label_status, label_data = mail.fetch(msg_id, "(X-GM-LABELS)") - if label_status == "OK" and label_data and len(label_data) > 0: - raw_labels = label_data[0][1].decode("utf-8", errors="ignore") - if label in raw_labels: - return True - except Exception as e: - logger.error("Failed to fetch labels for msg_id=%s: %s", msg_id, e) - return False - - -def mark_as_processed_with_star(mail, msg_id): - """Stars the email in Gmail.""" - try: - mail.store(msg_id, "+FLAGS", "\\Flagged") - logger.info("Email %s starred in Gmail.", msg_id) - except Exception as e: - logger.error("Failed to star email %s: %s", msg_id, e) - - -def mark_as_processed_with_label(mail, msg_id, label="Ingested"): - """Adds a custom label to the email in Gmail.""" - try: - mail.store(msg_id, "+X-GM-LABELS", label) - logger.info("Email %s labeled '%s' in Gmail.", msg_id, label) - except Exception as e: - logger.error("Failed to label email %s with %s: %s", msg_id, label, e) - - -def find_all_mail_folder(mail): - """ - Attempts to select the Gmail All Mail folder using known localized names. - Falls back to using XLIST if needed. - Returns the folder name if found, otherwise None. - """ - COMMON_ALL_MAIL_NAMES = [ - "[Gmail]/Alle Nachrichten", - "[Gmail]/All Mail", - "[Gmail]/Todos", - "[Gmail]/Tutte le mail", - "[Gmail]/Tous les messages", - ] - for candidate in COMMON_ALL_MAIL_NAMES: - status, _ = mail.select(f'"{candidate}"', readonly=True) - if status == "OK": - return candidate - - capabilities = get_capabilities(mail) - if "XLIST" in capabilities: - candidate = find_all_mail_xlist(mail) - if candidate: - return candidate - return None - - -def get_capabilities(mail): - """Returns a list of capabilities supported by the IMAP server.""" - typ, data = mail.capability() - if typ == "OK" and data: - caps = data[0].decode("utf-8", errors="ignore").upper().split() - return caps - return [] - - -def find_all_mail_xlist(mail): - """ - Uses XLIST to discover the mailbox flagged as All Mail. - Returns the folder name if found, otherwise None. - """ - tag = mail._new_tag().decode("ascii") - command_str = f"{tag} XLIST \"\" \"*\"" - mail.send((command_str + "\r\n").encode("utf-8")) - - all_mail_folder = None - while True: - line = mail.readline() - if not line: - break - line_str = line.decode("utf-8", errors="ignore").strip() - if line_str.upper().startswith("* XLIST ") and "\\ALLMAIL" in line_str.upper(): - match = re.search(r'"([^"]+)"$', line_str) - if match: - candidate = match.group(1) - logger.info("Found All Mail folder via XLIST: %s", candidate) - all_mail_folder = candidate - if line_str.startswith(tag): - break - return all_mail_folder +#!/usr/bin/env python3 +import os +import json +import email +import imaplib +import logging +import redis +import re +from datetime import datetime, timedelta, timezone +from celery import shared_task +from app.config import settings +from app.tasks.upload_to_s3 import upload_to_s3 +from app.tasks.convert_to_pdf import convert_to_pdf # new conversion task + +logger = logging.getLogger(__name__) + +# Initialize Redis connection using Celery's Redis settings +redis_client = redis.StrictRedis.from_url(settings.redis_url, decode_responses=True) + +LOCK_KEY = "imap_lock" # Unique key for locking +LOCK_EXPIRE = 300 # Lock expires in 5 minutes + +# Local cache file for tracking processed emails +CACHE_FILE = os.path.join(settings.workdir, "processed_mails.json") + + +def acquire_lock(): + """Attempt to acquire a Redis-based lock. If acquired, set an expiration.""" + lock_acquired = redis_client.setnx(LOCK_KEY, "locked") + if lock_acquired: + redis_client.expire(LOCK_KEY, LOCK_EXPIRE) + logger.info("Lock acquired for IMAP processing.") + return True + logger.warning("Lock already held. Skipping this cycle.") + return False + + +def release_lock(): + """Release the lock by deleting the Redis key.""" + redis_client.delete(LOCK_KEY) + logger.info("Lock released.") + + +def load_processed_emails(): + """Load the list of already processed emails from a local JSON file.""" + if os.path.exists(CACHE_FILE): + try: + with open(CACHE_FILE, "r") as f: + processed_emails = json.load(f) + processed_emails = cleanup_old_entries(processed_emails) + return processed_emails + except json.JSONDecodeError: + logger.warning("Failed to decode JSON, resetting processed emails cache.") + return {} + return {} + + +def save_processed_emails(processed_emails): + """Save the processed email IDs to a local JSON file.""" + with open(CACHE_FILE, "w") as f: + json.dump(processed_emails, f, indent=4) + + +def cleanup_old_entries(processed_emails): + """Remove entries older than 7 days from the cache to avoid infinite growth.""" + seven_days_ago = datetime.now(timezone.utc) - timedelta(days=7) + valid_emails = {} + for msg_id, date_str in processed_emails.items(): + naive_dt = datetime.strptime(date_str, "%Y-%m-%dT%H:%M:%S") + aware_dt = naive_dt.replace(tzinfo=timezone.utc) + if aware_dt > seven_days_ago: + valid_emails[msg_id] = date_str + return valid_emails + + +@shared_task +def pull_all_inboxes(): + """ + Periodic Celery task that checks all configured IMAP mailboxes + and fetches attachments from new emails. + Ensures only one instance runs at a time using Redis-based locking. + """ + if not acquire_lock(): + logger.info("Skipping execution: Another instance is running.") + return + + try: + logger.info("Starting pull_all_inboxes") + + # Mailbox #1 (non-Gmail) + check_and_pull_mailbox( + mailbox_key="imap1", + host=settings.imap1_host, + port=settings.imap1_port, + username=settings.imap1_username, + password=settings.imap1_password, + use_ssl=settings.imap1_ssl, + delete_after_process=settings.imap1_delete_after_process, + ) + + # Mailbox #2 (Gmail) + check_and_pull_mailbox( + mailbox_key="imap2", + host=settings.imap2_host, + port=settings.imap2_port, + username=settings.imap2_username, + password=settings.imap2_password, + use_ssl=settings.imap2_ssl, + delete_after_process=settings.imap2_delete_after_process, + ) + + logger.info("Finished pull_all_inboxes") + + finally: + release_lock() + + +def check_and_pull_mailbox( + mailbox_key: str, + host: str | None, + port: int | None, + username: str | None, + password: str | None, + use_ssl: bool, + delete_after_process: bool, +): + """Validates config and invokes pulling from the mailbox if valid.""" + if not (host and port and username and password): + logger.warning(f"Mailbox {mailbox_key} is missing config, skipping.") + return + + logger.info(f"Checking mailbox: {mailbox_key}") + pull_inbox( + mailbox_key=mailbox_key, + host=host, + port=port, + username=username, + password=password, + use_ssl=use_ssl, + delete_after_process=delete_after_process, + ) + + +def pull_inbox(mailbox_key, host, port, username, password, use_ssl, + delete_after_process): + """ + Connects to the IMAP inbox, fetches new unread emails from the last 3 days, + and processes attachments while preserving the original unread status. + + For Gmail: + - Attempts to select the localized All Mail folder. + - Runs an X-GM-RAW query: "in:anywhere in:unread newer_than:3d has:attachment". + + For non-Gmail mailboxes, it falls back to selecting the INBOX with a SINCE/UNSEEN filter. + """ + logger.info("Connecting to %s at %s:%s (SSL=%s)", + mailbox_key, host, port, use_ssl) + processed_emails = load_processed_emails() + + try: + mail = imaplib.IMAP4_SSL(host, port) if use_ssl else imaplib.IMAP4(host, port) + mail.login(username, password) + + is_gmail_host = "gmail" in host.lower() + if is_gmail_host: + # For Gmail, try to select the localized All Mail folder. + all_mail_folder = find_all_mail_folder(mail) + if all_mail_folder: + logger.info("Using Gmail All Mail folder: %s", all_mail_folder) + mail.select(f'"{all_mail_folder}"') + else: + logger.warning("Gmail All Mail folder not found, falling back to INBOX.") + mail.select("INBOX") + # Use the X-GM-RAW query for Gmail. + raw_query = "in:anywhere in:unread newer_than:3d has:attachment" + status, search_data = mail.search(None, "X-GM-RAW", f'"{raw_query}"') + else: + # For non-Gmail, select INBOX and use SINCE/UNSEEN query. + mail.select("INBOX") + since_date = (datetime.now(timezone.utc) - timedelta(days=3) + ).strftime("%d-%b-%Y") + status, search_data = mail.search(None, f'(SINCE {since_date} UNSEEN)') + + if status != "OK": + logger.warning("Search failed on mailbox %s. Status=%s", + mailbox_key, status) + mail.close() + mail.logout() + return + + msg_numbers = search_data[0].split() + logger.info("Found %d unread emails in %s.", len(msg_numbers), mailbox_key) + + for num in msg_numbers: + status, msg_data = mail.fetch(num, "(RFC822)") + if status != "OK": + logger.warning("Failed to fetch message %s in %s. Status=%s", + num, mailbox_key, status) + continue + + raw_email = msg_data[0][1] + email_message = email.message_from_bytes(raw_email) + msg_id = email_message.get("Message-ID") + + if not msg_id: + logger.warning("Skipping email without Message-ID in %s", mailbox_key) + continue + + if msg_id in processed_emails: + logger.info("Skipping already processed email %s in %s", msg_id, mailbox_key) + continue + + # For Gmail, check if the email already has the "Ingested" label. + if is_gmail_host: + if email_already_has_label(mail, num, "Ingested"): + logger.info("Skipping email %s in %s, already labeled 'Ingested'.", + msg_id, mailbox_key) + continue + + # Process attachments (and convert non-PDF files). + # We call the function without assigning its return value since it is not used. + fetch_attachments_and_enqueue(email_message) + + if is_gmail_host: + mark_as_processed_with_star(mail, num) + mark_as_processed_with_label(mail, num, label="Ingested") + + processed_emails[msg_id] = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S") + save_processed_emails(processed_emails) + + if delete_after_process: + logger.info("Deleting message %s from %s", num.decode(), mailbox_key) + mail.store(num, "+FLAGS", "\\Deleted") + else: + mail.store(num, "-FLAGS", "\\Seen") + + if delete_after_process: + mail.expunge() + + mail.close() + mail.logout() + logger.info("Finished processing mailbox %s", mailbox_key) + + except Exception as e: + logger.exception("Error pulling mailbox %s: %s", mailbox_key, e) + + +def fetch_attachments_and_enqueue(email_message): + """ + Extracts attachments from the email and processes only allowed file types. + + Allowed file types include: + - PDF: application/pdf + - Microsoft Office files: + - Word: application/msword, + application/vnd.openxmlformats-officedocument.wordprocessingml.document + - Excel: application/vnd.ms-excel, + application/vnd.openxmlformats-officedocument.spreadsheetml.sheet + - PowerPoint: application/vnd.ms-powerpoint, + application/vnd.openxmlformats-officedocument.presentationml.presentation + - Other meaningful attachments: + - Plain text: text/plain + - CSV: text/csv + - Rich Text Format: application/rtf, text/rtf + + Attachments not in this list are skipped. Common image MIME types such as + image/jpeg, image/png, image/gif, image/bmp, image/tiff, and image/webp are + intentionally excluded. + + If the attachment is a PDF, it is enqueued for upload; any other allowed file + is enqueued for conversion to PDF. + + Returns True if at least one allowed attachment was processed. + """ + ALLOWED_MIME_TYPES = { + "application/pdf", + "application/msword", + "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + "application/vnd.ms-excel", + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + "application/vnd.ms-powerpoint", + "application/vnd.openxmlformats-officedocument.presentationml.presentation", + "text/plain", + "text/csv", + "application/rtf", + "text/rtf", + } + + has_attachment = False + for part in email_message.walk(): + if part.get_content_maintype() == "multipart": + continue + + filename = part.get_filename() + if not filename: + continue + + mime_type = part.get_content_type() + if mime_type not in ALLOWED_MIME_TYPES: + logger.info("Skipping attachment %s with MIME type %s", + filename, mime_type) + continue + + file_path = os.path.join(settings.workdir, filename) + with open(file_path, "wb") as f: + f.write(part.get_payload(decode=True)) + + if mime_type == "application/pdf": + upload_to_s3.delay(file_path) + logger.info("Enqueued PDF for upload: %s", filename) + elif mime_type in ALLOWED_MIME_TYPES: + # Enqueue conversion to PDF using the Gotenberg service. + convert_to_pdf.delay(file_path) + logger.info("Enqueued file for conversion to PDF: %s", filename) + + has_attachment = True + return has_attachment + + +def email_already_has_label(mail, msg_id, label="Ingested"): + """ + Checks if the given message (msg_id) has the specified Gmail label. + Returns True if the label is found, False otherwise. + """ + try: + label_status, label_data = mail.fetch(msg_id, "(X-GM-LABELS)") + if label_status == "OK" and label_data and len(label_data) > 0: + raw_labels = label_data[0][1].decode("utf-8", errors="ignore") + if label in raw_labels: + return True + except Exception as e: + logger.error("Failed to fetch labels for msg_id=%s: %s", msg_id, e) + return False + + +def mark_as_processed_with_star(mail, msg_id): + """Stars the email in Gmail.""" + try: + mail.store(msg_id, "+FLAGS", "\\Flagged") + logger.info("Email %s starred in Gmail.", msg_id) + except Exception as e: + logger.error("Failed to star email %s: %s", msg_id, e) + + +def mark_as_processed_with_label(mail, msg_id, label="Ingested"): + """Adds a custom label to the email in Gmail.""" + try: + mail.store(msg_id, "+X-GM-LABELS", label) + logger.info("Email %s labeled '%s' in Gmail.", msg_id, label) + except Exception as e: + logger.error("Failed to label email %s with %s: %s", msg_id, label, e) + + +def find_all_mail_folder(mail): + """ + Attempts to select the Gmail All Mail folder using known localized names. + Falls back to using XLIST if needed. + Returns the folder name if found, otherwise None. + """ + COMMON_ALL_MAIL_NAMES = [ + "[Gmail]/Alle Nachrichten", + "[Gmail]/All Mail", + "[Gmail]/Todos", + "[Gmail]/Tutte le mail", + "[Gmail]/Tous les messages", + ] + for candidate in COMMON_ALL_MAIL_NAMES: + status, _ = mail.select(f'"{candidate}"', readonly=True) + if status == "OK": + return candidate + + capabilities = get_capabilities(mail) + if "XLIST" in capabilities: + candidate = find_all_mail_xlist(mail) + if candidate: + return candidate + return None + + +def get_capabilities(mail): + """Returns a list of capabilities supported by the IMAP server.""" + typ, data = mail.capability() + if typ == "OK" and data: + caps = data[0].decode("utf-8", errors="ignore").upper().split() + return caps + return [] + + +def find_all_mail_xlist(mail): + """ + Uses XLIST to discover the mailbox flagged as All Mail. + Returns the folder name if found, otherwise None. + """ + tag = mail._new_tag().decode("ascii") + command_str = f"{tag} XLIST \"\" \"*\"" + mail.send((command_str + "\r\n").encode("utf-8")) + + all_mail_folder = None + while True: + line = mail.readline() + if not line: + break + line_str = line.decode("utf-8", errors="ignore").strip() + if line_str.upper().startswith("* XLIST ") and "\\ALLMAIL" in line_str.upper(): + match = re.search(r'"([^"]+)"$', line_str) + if match: + candidate = match.group(1) + logger.info("Found All Mail folder via XLIST: %s", candidate) + all_mail_folder = candidate + if line_str.startswith(tag): + break + return all_mail_folder diff --git a/app/tasks/process_with_textract.py b/app/tasks/process_with_textract.py index ee100ee4..d3319545 100644 --- a/app/tasks/process_with_textract.py +++ b/app/tasks/process_with_textract.py @@ -1,67 +1,67 @@ -import os -import logging -from azure.core.credentials import AzureKeyCredential -from azure.ai.documentintelligence import DocumentIntelligenceClient -from azure.ai.documentintelligence.models import AnalyzeOutputOption, AnalyzeResult - -from app.config import settings -from app.tasks.retry_config import BaseTaskWithRetry -from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt -from app.celery_app import celery - -logger = logging.getLogger(__name__) - -# Initialize Azure Document Intelligence client -document_intelligence_client = DocumentIntelligenceClient( - endpoint=settings.azure_endpoint, - credential=AzureKeyCredential(settings.azure_ai_key) -) - -@celery.task(base=BaseTaskWithRetry) -def process_with_textract(s3_filename: str): - """ - Processes a PDF document using Azure Document Intelligence and overlays OCR text onto - the local temporary file (stored under /tmp). - - Steps: - 1. Uploads the document for OCR using Azure Document Intelligence. - 2. Retrieves the processed PDF with embedded text. - 3. Saves the OCR-processed PDF locally in the same location as before. - 4. Extracts the text content for metadata processing. - 5. Triggers downstream metadata extraction by calling extract_metadata_with_gpt. - """ - try: - tmp_file_path = os.path.join(settings.workdir, "tmp", s3_filename) - if not os.path.exists(tmp_file_path): - raise FileNotFoundError(f"Local file not found: {tmp_file_path}") - - logger.info(f"Processing {s3_filename} with Azure Document Intelligence OCR.") - - # Open and send the document for processing - with open(tmp_file_path, "rb") as f: - poller = document_intelligence_client.begin_analyze_document( - "prebuilt-read", body=f, output=[AnalyzeOutputOption.PDF] - ) - result: AnalyzeResult = poller.result() - operation_id = poller.details["operation_id"] - - # Retrieve the processed searchable PDF - response = document_intelligence_client.get_analyze_result_pdf( - model_id=result.model_id, result_id=operation_id - ) - searchable_pdf_path = tmp_file_path # Overwrite the original PDF location - with open(searchable_pdf_path, "wb") as writer: - writer.writelines(response) - logger.info(f"Searchable PDF saved at: {searchable_pdf_path}") - - # Extract raw text content from the result - extracted_text = result.content if result.content else "" - logger.info(f"Extracted text for {s3_filename}: {len(extracted_text)} characters") - - # Trigger downstream metadata extraction - extract_metadata_with_gpt.delay(s3_filename, extracted_text) - - return {"s3_file": s3_filename, "searchable_pdf": searchable_pdf_path, "cleaned_text": extracted_text} - except Exception as e: - logger.error(f"Error processing {s3_filename} with Azure Document Intelligence: {e}") - raise +import os +import logging +from azure.core.credentials import AzureKeyCredential +from azure.ai.documentintelligence import DocumentIntelligenceClient +from azure.ai.documentintelligence.models import AnalyzeOutputOption, AnalyzeResult + +from app.config import settings +from app.tasks.retry_config import BaseTaskWithRetry +from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt +from app.celery_app import celery + +logger = logging.getLogger(__name__) + +# Initialize Azure Document Intelligence client +document_intelligence_client = DocumentIntelligenceClient( + endpoint=settings.azure_endpoint, + credential=AzureKeyCredential(settings.azure_ai_key) +) + +@celery.task(base=BaseTaskWithRetry) +def process_with_textract(s3_filename: str): + """ + Processes a PDF document using Azure Document Intelligence and overlays OCR text onto + the local temporary file (stored under /tmp). + + Steps: + 1. Uploads the document for OCR using Azure Document Intelligence. + 2. Retrieves the processed PDF with embedded text. + 3. Saves the OCR-processed PDF locally in the same location as before. + 4. Extracts the text content for metadata processing. + 5. Triggers downstream metadata extraction by calling extract_metadata_with_gpt. + """ + try: + tmp_file_path = os.path.join(settings.workdir, "tmp", s3_filename) + if not os.path.exists(tmp_file_path): + raise FileNotFoundError(f"Local file not found: {tmp_file_path}") + + logger.info(f"Processing {s3_filename} with Azure Document Intelligence OCR.") + + # Open and send the document for processing + with open(tmp_file_path, "rb") as f: + poller = document_intelligence_client.begin_analyze_document( + "prebuilt-read", body=f, output=[AnalyzeOutputOption.PDF] + ) + result: AnalyzeResult = poller.result() + operation_id = poller.details["operation_id"] + + # Retrieve the processed searchable PDF + response = document_intelligence_client.get_analyze_result_pdf( + model_id=result.model_id, result_id=operation_id + ) + searchable_pdf_path = tmp_file_path # Overwrite the original PDF location + with open(searchable_pdf_path, "wb") as writer: + writer.writelines(response) + logger.info(f"Searchable PDF saved at: {searchable_pdf_path}") + + # Extract raw text content from the result + extracted_text = result.content if result.content else "" + logger.info(f"Extracted text for {s3_filename}: {len(extracted_text)} characters") + + # Trigger downstream metadata extraction + extract_metadata_with_gpt.delay(s3_filename, extracted_text) + + return {"s3_file": s3_filename, "searchable_pdf": searchable_pdf_path, "cleaned_text": extracted_text} + except Exception as e: + logger.error(f"Error processing {s3_filename} with Azure Document Intelligence: {e}") + raise diff --git a/app/tasks/refine_text_with_gpt.py b/app/tasks/refine_text_with_gpt.py index 87e3e8c5..fe687b6b 100644 --- a/app/tasks/refine_text_with_gpt.py +++ b/app/tasks/refine_text_with_gpt.py @@ -1,34 +1,34 @@ -#!/usr/bin/env python3 - -from app.config import settings -from openai import OpenAI -from app.tasks.retry_config import BaseTaskWithRetry - -# Import the shared Celery instance -from app.celery_app import celery - - - -client = OpenAI(api_key=settings.openai_api_key) - -@celery.task(base=BaseTaskWithRetry) -def refine_text_with_gpt(s3_filename: str, raw_text: str): - """Uses GPT to clean and refine OCR text.""" - - # Use the Chat Completions endpoint with 'messages' - response = client.chat.completions.create( - model="gpt-4", - messages=[ - {"role": "system", "content": "Clean and format the following text. The idea is that the text you see comes from an OCR system and your task is to eliminate OCR errors. Keep the original language when doing so."}, - {"role": "user", "content": raw_text} - ] - ) - - cleaned_text = response.choices[0].message.content - - # Trigger next task (import locally if needed to avoid circular imports) - from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt - extract_metadata_with_gpt.delay(s3_filename, cleaned_text) - - return {"s3_file": s3_filename, "cleaned_text": cleaned_text} - +#!/usr/bin/env python3 + +from app.config import settings +from openai import OpenAI +from app.tasks.retry_config import BaseTaskWithRetry + +# Import the shared Celery instance +from app.celery_app import celery + + + +client = OpenAI(api_key=settings.openai_api_key) + +@celery.task(base=BaseTaskWithRetry) +def refine_text_with_gpt(s3_filename: str, raw_text: str): + """Uses GPT to clean and refine OCR text.""" + + # Use the Chat Completions endpoint with 'messages' + response = client.chat.completions.create( + model="gpt-4", + messages=[ + {"role": "system", "content": "Clean and format the following text. The idea is that the text you see comes from an OCR system and your task is to eliminate OCR errors. Keep the original language when doing so."}, + {"role": "user", "content": raw_text} + ] + ) + + cleaned_text = response.choices[0].message.content + + # Trigger next task (import locally if needed to avoid circular imports) + from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt + extract_metadata_with_gpt.delay(s3_filename, cleaned_text) + + return {"s3_file": s3_filename, "cleaned_text": cleaned_text} + diff --git a/app/tasks/retry_config.py b/app/tasks/retry_config.py index 254a0d02..6f94c992 100644 --- a/app/tasks/retry_config.py +++ b/app/tasks/retry_config.py @@ -1,9 +1,9 @@ -#!/usr/bin/env python3 - -from celery import Task - -class BaseTaskWithRetry(Task): - autoretry_for = (Exception,) - retry_kwargs = {"max_retries": 3, "countdown": 10} # 3 retries, 10s delay - retry_backoff = True # Exponential backoff - +#!/usr/bin/env python3 + +from celery import Task + +class BaseTaskWithRetry(Task): + autoretry_for = (Exception,) + retry_kwargs = {"max_retries": 3, "countdown": 10} # 3 retries, 10s delay + retry_backoff = True # Exponential backoff + diff --git a/app/tasks/send_to_all.py b/app/tasks/send_to_all.py index fe3cee7f..15f78d42 100644 --- a/app/tasks/send_to_all.py +++ b/app/tasks/send_to_all.py @@ -1,21 +1,21 @@ -# app/tasks/send_to_all.py - -from app.celery_app import celery -from app.tasks.upload_to_dropbox import upload_to_dropbox -from app.tasks.upload_to_nextcloud import upload_to_nextcloud -from app.tasks.upload_to_paperless import upload_to_paperless - -@celery.task -def send_to_all_destinations(file_path: str): - """ - Fires off tasks to upload a single file to Dropbox, Nextcloud, and Paperless. - These tasks run in parallel (Celery returns immediately from each .delay()). - """ - upload_to_dropbox.delay(file_path) - upload_to_nextcloud.delay(file_path) - upload_to_paperless.delay(file_path) - - return { - "status": "All upload tasks enqueued", - "file_path": file_path - } +# app/tasks/send_to_all.py + +from app.celery_app import celery +from app.tasks.upload_to_dropbox import upload_to_dropbox +from app.tasks.upload_to_nextcloud import upload_to_nextcloud +from app.tasks.upload_to_paperless import upload_to_paperless + +@celery.task +def send_to_all_destinations(file_path: str): + """ + Fires off tasks to upload a single file to Dropbox, Nextcloud, and Paperless. + These tasks run in parallel (Celery returns immediately from each .delay()). + """ + upload_to_dropbox.delay(file_path) + upload_to_nextcloud.delay(file_path) + upload_to_paperless.delay(file_path) + + return { + "status": "All upload tasks enqueued", + "file_path": file_path + } diff --git a/app/tasks/upload_to_dropbox.py b/app/tasks/upload_to_dropbox.py index 4875fa99..448360a6 100644 --- a/app/tasks/upload_to_dropbox.py +++ b/app/tasks/upload_to_dropbox.py @@ -1,74 +1,74 @@ -#!/usr/bin/env python3 - -import os -import requests -import dropbox -from app.config import settings -from app.tasks.retry_config import BaseTaskWithRetry -from app.celery_app import celery - -def get_dropbox_access_token(): - """Refresh the Dropbox access token using the stored refresh token from ENV.""" - - token_url = "https://api.dropbox.com/oauth2/token" - headers = {"Content-Type": "application/x-www-form-urlencoded"} - data = { - "grant_type": "refresh_token", - "refresh_token": settings.dropbox_refresh_token, # Now using ENV - "client_id": settings.dropbox_app_key, - "client_secret": settings.dropbox_app_secret, - } - - response = requests.post(token_url, headers=headers, data=data) - - if response.status_code == 200: - return response.json()["access_token"] - else: - error_msg = f"Failed to refresh Dropbox token: {response.status_code} - {response.text}" - print(f"[ERROR] {error_msg}") - raise Exception(error_msg) - -@celery.task(base=BaseTaskWithRetry) -def upload_to_dropbox(file_path: str): - """Uploads a file to Dropbox using the API.""" - - if not os.path.exists(file_path): - raise FileNotFoundError(f"File not found: {file_path}") - - # Extract filename and set target path - filename = os.path.basename(file_path) - dropbox_path = f"{settings.dropbox_folder}/{filename}" - - try: - # Get fresh access token - access_token = get_dropbox_access_token() - dbx = dropbox.Dropbox(access_token) - - file_size = os.path.getsize(file_path) - chunk_size = 4 * 1024 * 1024 # 4MB chunk size - - with open(file_path, "rb") as file_data: - if file_size <= chunk_size: - dbx.files_upload(file_data.read(), dropbox_path) - else: - upload_session_start_result = dbx.files_upload_session_start(file_data.read(chunk_size)) - cursor = dropbox.files.UploadSessionCursor( - session_id=upload_session_start_result.session_id, - offset=file_data.tell(), - ) - commit = dropbox.files.CommitInfo(path=dropbox_path) - - while file_data.tell() < file_size: - if (file_size - file_data.tell()) <= chunk_size: - dbx.files_upload_session_finish(file_data.read(chunk_size), cursor, commit) - else: - dbx.files_upload_session_append_v2(file_data.read(chunk_size), cursor) - cursor.offset = file_data.tell() - - print(f"[INFO] Successfully uploaded {filename} to Dropbox at {dropbox_path}.") - return {"status": "Completed", "file": file_path} - - except Exception as e: - error_msg = f"[ERROR] Failed to upload {filename} to Dropbox: {str(e)}" - print(error_msg) - raise Exception(error_msg) +#!/usr/bin/env python3 + +import os +import requests +import dropbox +from app.config import settings +from app.tasks.retry_config import BaseTaskWithRetry +from app.celery_app import celery + +def get_dropbox_access_token(): + """Refresh the Dropbox access token using the stored refresh token from ENV.""" + + token_url = "https://api.dropbox.com/oauth2/token" + headers = {"Content-Type": "application/x-www-form-urlencoded"} + data = { + "grant_type": "refresh_token", + "refresh_token": settings.dropbox_refresh_token, # Now using ENV + "client_id": settings.dropbox_app_key, + "client_secret": settings.dropbox_app_secret, + } + + response = requests.post(token_url, headers=headers, data=data) + + if response.status_code == 200: + return response.json()["access_token"] + else: + error_msg = f"Failed to refresh Dropbox token: {response.status_code} - {response.text}" + print(f"[ERROR] {error_msg}") + raise Exception(error_msg) + +@celery.task(base=BaseTaskWithRetry) +def upload_to_dropbox(file_path: str): + """Uploads a file to Dropbox using the API.""" + + if not os.path.exists(file_path): + raise FileNotFoundError(f"File not found: {file_path}") + + # Extract filename and set target path + filename = os.path.basename(file_path) + dropbox_path = f"{settings.dropbox_folder}/{filename}" + + try: + # Get fresh access token + access_token = get_dropbox_access_token() + dbx = dropbox.Dropbox(access_token) + + file_size = os.path.getsize(file_path) + chunk_size = 4 * 1024 * 1024 # 4MB chunk size + + with open(file_path, "rb") as file_data: + if file_size <= chunk_size: + dbx.files_upload(file_data.read(), dropbox_path) + else: + upload_session_start_result = dbx.files_upload_session_start(file_data.read(chunk_size)) + cursor = dropbox.files.UploadSessionCursor( + session_id=upload_session_start_result.session_id, + offset=file_data.tell(), + ) + commit = dropbox.files.CommitInfo(path=dropbox_path) + + while file_data.tell() < file_size: + if (file_size - file_data.tell()) <= chunk_size: + dbx.files_upload_session_finish(file_data.read(chunk_size), cursor, commit) + else: + dbx.files_upload_session_append_v2(file_data.read(chunk_size), cursor) + cursor.offset = file_data.tell() + + print(f"[INFO] Successfully uploaded {filename} to Dropbox at {dropbox_path}.") + return {"status": "Completed", "file": file_path} + + except Exception as e: + error_msg = f"[ERROR] Failed to upload {filename} to Dropbox: {str(e)}" + print(error_msg) + raise Exception(error_msg) diff --git a/app/tasks/upload_to_nextcloud.py b/app/tasks/upload_to_nextcloud.py index 6f8fee64..b64c2972 100644 --- a/app/tasks/upload_to_nextcloud.py +++ b/app/tasks/upload_to_nextcloud.py @@ -1,37 +1,37 @@ -#!/usr/bin/env python3 - -import os -import requests -from app.config import settings -from app.tasks.retry_config import BaseTaskWithRetry -from app.celery_app import celery - -@celery.task(base=BaseTaskWithRetry) -def upload_to_nextcloud(file_path: str): - """Uploads a file to Nextcloud in the configured folder.""" - - if not os.path.exists(file_path): - raise FileNotFoundError(f"File not found: {file_path}") - - # Extract filename - filename = os.path.basename(file_path) - - # Construct the full upload URL - nextcloud_url = f"{settings.nextcloud_upload_url}/{settings.nextcloud_folder}/{filename}" - - # Read file content - with open(file_path, "rb") as file_data: - response = requests.put( - nextcloud_url, - auth=(settings.nextcloud_username, settings.nextcloud_password), - data=file_data - ) - - # Check if upload was successful - if response.status_code in (200, 201): - print(f"[INFO] Successfully uploaded {filename} to Nextcloud at {nextcloud_url}.") - return {"status": "Completed", "file": file_path} - else: - error_msg = f"[ERROR] Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}" - print(error_msg) - raise Exception(error_msg) +#!/usr/bin/env python3 + +import os +import requests +from app.config import settings +from app.tasks.retry_config import BaseTaskWithRetry +from app.celery_app import celery + +@celery.task(base=BaseTaskWithRetry) +def upload_to_nextcloud(file_path: str): + """Uploads a file to Nextcloud in the configured folder.""" + + if not os.path.exists(file_path): + raise FileNotFoundError(f"File not found: {file_path}") + + # Extract filename + filename = os.path.basename(file_path) + + # Construct the full upload URL + nextcloud_url = f"{settings.nextcloud_upload_url}/{settings.nextcloud_folder}/{filename}" + + # Read file content + with open(file_path, "rb") as file_data: + response = requests.put( + nextcloud_url, + auth=(settings.nextcloud_username, settings.nextcloud_password), + data=file_data + ) + + # Check if upload was successful + if response.status_code in (200, 201): + print(f"[INFO] Successfully uploaded {filename} to Nextcloud at {nextcloud_url}.") + return {"status": "Completed", "file": file_path} + else: + error_msg = f"[ERROR] Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}" + print(error_msg) + raise Exception(error_msg) diff --git a/app/tasks/upload_to_paperless.py b/app/tasks/upload_to_paperless.py index 8f14c726..135c886b 100644 --- a/app/tasks/upload_to_paperless.py +++ b/app/tasks/upload_to_paperless.py @@ -1,131 +1,131 @@ -#!/usr/bin/env python3 - -import os -import json -import time -import requests -import logging -from typing import Dict, Any - -from app.config import settings -from app.tasks.retry_config import BaseTaskWithRetry -from app.celery_app import celery - -logger = logging.getLogger(__name__) - -POLL_MAX_ATTEMPTS = 10 -POLL_INTERVAL_SEC = 3 - -def _get_headers(): - """Returns HTTP headers for Paperless-ngx API calls.""" - return { - "Authorization": f"Token {settings.paperless_ngx_api_token}" - } - -def _paperless_api_url(path: str) -> str: - """ - Constructs a full Paperless-ngx API URL using `settings.paperless_host`. - Ensures the path is appended with a leading slash if missing. - """ - host = settings.paperless_host.rstrip("/") - if not path.startswith("/"): - path = "/" + path - return f"{host}{path}" - -def poll_task_for_document_id(task_id: str) -> int: - """ - Polls /api/tasks/?task_id= until we get status=SUCCESS or FAILURE, - or until we run out of attempts. - - On SUCCESS: returns the int document_id from 'related_document'. - On FAILURE: raises RuntimeError with the task's 'result' message. - If times out, raises TimeoutError. - """ - url = _paperless_api_url("/api/tasks/") - attempts = 0 - - while attempts < POLL_MAX_ATTEMPTS: - try: - resp = requests.get(url, headers=_get_headers(), params={"task_id": task_id}) - resp.raise_for_status() - tasks_data = resp.json() - except requests.exceptions.RequestException as exc: - logger.warning( - "Failed to poll for task_id='%s'. Attempt=%d Error=%s", - task_id, attempts + 1, exc - ) - time.sleep(POLL_INTERVAL_SEC) - attempts += 1 - continue - - if isinstance(tasks_data, dict) and "results" in tasks_data: - tasks_data = tasks_data["results"] - - if tasks_data: - task_info = tasks_data[0] - status = task_info.get("status") - if status == "SUCCESS": - doc_str = task_info.get("related_document") - if doc_str: - return int(doc_str) - raise RuntimeError( - f"Task {task_id} completed but no doc ID found. Task info: {task_info}" - ) - elif status == "FAILURE": - raise RuntimeError(f"Task {task_id} failed: {task_info.get('result')}") - - attempts += 1 - time.sleep(POLL_INTERVAL_SEC) - - raise TimeoutError( - f"Task {task_id} didn't reach SUCCESS within {POLL_MAX_ATTEMPTS} attempts." - ) - -@celery.task(base=BaseTaskWithRetry) -def upload_to_paperless(file_path: str) -> Dict[str, Any]: - """ - Uploads a PDF to Paperless with minimal metadata (filename and date only). - - 1. Extracts the filename and date from the file. - 2. POSTs the PDF to Paperless => returns a quoted UUID string (task_id). - 3. Polls /api/tasks/?task_id= until SUCCESS or FAILURE => doc_id. - - Returns a dict with status, the paperless_task_id, paperless_document_id, and file_path. - """ - if not os.path.exists(file_path): - raise FileNotFoundError(f"File not found: {file_path}") - - base_name = os.path.basename(file_path) - - # Upload the PDF - post_url = _paperless_api_url("/api/documents/post_document/") - with open(file_path, "rb") as f: - files = { - "document": (base_name, f, "application/pdf"), - } - data = {"title": base_name} # Title = Filename (no additional metadata) - - try: - logger.debug("Posting document to Paperless: file=%s", base_name) - resp = requests.post(post_url, headers=_get_headers(), files=files, data=data) - resp.raise_for_status() - except requests.exceptions.RequestException as exc: - logger.error( - "Failed to upload document '%s' to Paperless. Error: %s. Response=%s", - file_path, exc, getattr(exc.response, "text", "") - ) - raise - - raw_task_id = resp.text.strip().strip('"').strip("'") - logger.info(f"Received Paperless task ID: {raw_task_id}") - - # Poll tasks until success/fail => get doc_id - doc_id = poll_task_for_document_id(raw_task_id) - logger.info(f"Document {file_path} successfully ingested => ID={doc_id}") - - return { - "status": "Completed", - "paperless_task_id": raw_task_id, - "paperless_document_id": doc_id, - "file_path": file_path - } +#!/usr/bin/env python3 + +import os +import json +import time +import requests +import logging +from typing import Dict, Any + +from app.config import settings +from app.tasks.retry_config import BaseTaskWithRetry +from app.celery_app import celery + +logger = logging.getLogger(__name__) + +POLL_MAX_ATTEMPTS = 10 +POLL_INTERVAL_SEC = 3 + +def _get_headers(): + """Returns HTTP headers for Paperless-ngx API calls.""" + return { + "Authorization": f"Token {settings.paperless_ngx_api_token}" + } + +def _paperless_api_url(path: str) -> str: + """ + Constructs a full Paperless-ngx API URL using `settings.paperless_host`. + Ensures the path is appended with a leading slash if missing. + """ + host = settings.paperless_host.rstrip("/") + if not path.startswith("/"): + path = "/" + path + return f"{host}{path}" + +def poll_task_for_document_id(task_id: str) -> int: + """ + Polls /api/tasks/?task_id= until we get status=SUCCESS or FAILURE, + or until we run out of attempts. + + On SUCCESS: returns the int document_id from 'related_document'. + On FAILURE: raises RuntimeError with the task's 'result' message. + If times out, raises TimeoutError. + """ + url = _paperless_api_url("/api/tasks/") + attempts = 0 + + while attempts < POLL_MAX_ATTEMPTS: + try: + resp = requests.get(url, headers=_get_headers(), params={"task_id": task_id}) + resp.raise_for_status() + tasks_data = resp.json() + except requests.exceptions.RequestException as exc: + logger.warning( + "Failed to poll for task_id='%s'. Attempt=%d Error=%s", + task_id, attempts + 1, exc + ) + time.sleep(POLL_INTERVAL_SEC) + attempts += 1 + continue + + if isinstance(tasks_data, dict) and "results" in tasks_data: + tasks_data = tasks_data["results"] + + if tasks_data: + task_info = tasks_data[0] + status = task_info.get("status") + if status == "SUCCESS": + doc_str = task_info.get("related_document") + if doc_str: + return int(doc_str) + raise RuntimeError( + f"Task {task_id} completed but no doc ID found. Task info: {task_info}" + ) + elif status == "FAILURE": + raise RuntimeError(f"Task {task_id} failed: {task_info.get('result')}") + + attempts += 1 + time.sleep(POLL_INTERVAL_SEC) + + raise TimeoutError( + f"Task {task_id} didn't reach SUCCESS within {POLL_MAX_ATTEMPTS} attempts." + ) + +@celery.task(base=BaseTaskWithRetry) +def upload_to_paperless(file_path: str) -> Dict[str, Any]: + """ + Uploads a PDF to Paperless with minimal metadata (filename and date only). + + 1. Extracts the filename and date from the file. + 2. POSTs the PDF to Paperless => returns a quoted UUID string (task_id). + 3. Polls /api/tasks/?task_id= until SUCCESS or FAILURE => doc_id. + + Returns a dict with status, the paperless_task_id, paperless_document_id, and file_path. + """ + if not os.path.exists(file_path): + raise FileNotFoundError(f"File not found: {file_path}") + + base_name = os.path.basename(file_path) + + # Upload the PDF + post_url = _paperless_api_url("/api/documents/post_document/") + with open(file_path, "rb") as f: + files = { + "document": (base_name, f, "application/pdf"), + } + data = {"title": base_name} # Title = Filename (no additional metadata) + + try: + logger.debug("Posting document to Paperless: file=%s", base_name) + resp = requests.post(post_url, headers=_get_headers(), files=files, data=data) + resp.raise_for_status() + except requests.exceptions.RequestException as exc: + logger.error( + "Failed to upload document '%s' to Paperless. Error: %s. Response=%s", + file_path, exc, getattr(exc.response, "text", "") + ) + raise + + raw_task_id = resp.text.strip().strip('"').strip("'") + logger.info(f"Received Paperless task ID: {raw_task_id}") + + # Poll tasks until success/fail => get doc_id + doc_id = poll_task_for_document_id(raw_task_id) + logger.info(f"Document {file_path} successfully ingested => ID={doc_id}") + + return { + "status": "Completed", + "paperless_task_id": raw_task_id, + "paperless_document_id": doc_id, + "file_path": file_path + } diff --git a/app/tasks/upload_to_s3.py b/app/tasks/upload_to_s3.py index 435331bf..10919bed 100644 --- a/app/tasks/upload_to_s3.py +++ b/app/tasks/upload_to_s3.py @@ -1,87 +1,87 @@ -#!/usr/bin/env python3 - -import os -import uuid -import boto3 -import shutil -import fitz # PyMuPDF for checking embedded text -from app.config import settings -from app.tasks.retry_config import BaseTaskWithRetry -from app.tasks.process_with_textract import process_with_textract -from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt - -# Import the shared Celery instance -from app.celery_app import celery - -# Initialize S3 client -s3_client = boto3.client( - "s3", - aws_access_key_id=settings.aws_access_key_id, - aws_secret_access_key=settings.aws_secret_access_key, - region_name=settings.aws_region, -) - -@celery.task(base=BaseTaskWithRetry) -def upload_to_s3(original_local_file: str): - """ - Uploads a file to S3 with a UUID-based filename and triggers processing. - - If the PDF already contains embedded text, skip Textract and extract text locally. - - Otherwise, upload to S3 and process with Textract. - """ - bucket_name = settings.s3_bucket_name - if not bucket_name: - print("[ERROR] S3 bucket name not set.") - return {"error": "Missing S3 bucket name"} - - if not os.path.exists(original_local_file): - print(f"[ERROR] File {original_local_file} not found.") - return {"error": "File not found"} - - # Generate UUID and create a new filename - file_ext = os.path.splitext(original_local_file)[1] # Preserve original file extension - file_uuid = str(uuid.uuid4()) - new_filename = f"{file_uuid}{file_ext}" - - # Construct the new local path using settings.workdir and a 'tmp' subdirectory - tmp_dir = os.path.join(settings.workdir, "tmp") - new_local_path = os.path.join(tmp_dir, new_filename) - - # Ensure the target tmp directory exists - os.makedirs(tmp_dir, exist_ok=True) - - # Copy the file instead of moving it - shutil.copy(original_local_file, new_local_path) - - # Check for embedded text - pdf_doc = fitz.open(new_local_path) - has_text = any(page.get_text() for page in pdf_doc) - pdf_doc.close() - - if has_text: - print(f"[INFO] PDF {original_local_file} contains embedded text. Skipping Textract.") - - # Extract text locally - extracted_text = "" - pdf_doc = fitz.open(new_local_path) - for page in pdf_doc: - extracted_text += page.get_text("text") + "\n" - pdf_doc.close() - - # Call metadata extraction directly - extract_metadata_with_gpt.delay(new_filename, extracted_text) - - return {"file": new_local_path, "status": "Text extracted locally"} - - try: - print(f"[INFO] Uploading {new_local_path} to s3://{bucket_name}/{new_filename}...") - s3_client.upload_file(new_local_path, bucket_name, new_filename) - print(f"[INFO] File uploaded successfully: {new_filename}") - - # Trigger Textract processing if no embedded text was found - process_with_textract.delay(new_filename) - - return {"file": new_local_path, "s3_key": new_filename, "status": "Uploaded to S3 for OCR"} - - except Exception as e: - print(f"[ERROR] Failed to upload {new_local_path} to S3: {e}") - return {"error": str(e)} +#!/usr/bin/env python3 + +import os +import uuid +import boto3 +import shutil +import fitz # PyMuPDF for checking embedded text +from app.config import settings +from app.tasks.retry_config import BaseTaskWithRetry +from app.tasks.process_with_textract import process_with_textract +from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt + +# Import the shared Celery instance +from app.celery_app import celery + +# Initialize S3 client +s3_client = boto3.client( + "s3", + aws_access_key_id=settings.aws_access_key_id, + aws_secret_access_key=settings.aws_secret_access_key, + region_name=settings.aws_region, +) + +@celery.task(base=BaseTaskWithRetry) +def upload_to_s3(original_local_file: str): + """ + Uploads a file to S3 with a UUID-based filename and triggers processing. + - If the PDF already contains embedded text, skip Textract and extract text locally. + - Otherwise, upload to S3 and process with Textract. + """ + bucket_name = settings.s3_bucket_name + if not bucket_name: + print("[ERROR] S3 bucket name not set.") + return {"error": "Missing S3 bucket name"} + + if not os.path.exists(original_local_file): + print(f"[ERROR] File {original_local_file} not found.") + return {"error": "File not found"} + + # Generate UUID and create a new filename + file_ext = os.path.splitext(original_local_file)[1] # Preserve original file extension + file_uuid = str(uuid.uuid4()) + new_filename = f"{file_uuid}{file_ext}" + + # Construct the new local path using settings.workdir and a 'tmp' subdirectory + tmp_dir = os.path.join(settings.workdir, "tmp") + new_local_path = os.path.join(tmp_dir, new_filename) + + # Ensure the target tmp directory exists + os.makedirs(tmp_dir, exist_ok=True) + + # Copy the file instead of moving it + shutil.copy(original_local_file, new_local_path) + + # Check for embedded text + pdf_doc = fitz.open(new_local_path) + has_text = any(page.get_text() for page in pdf_doc) + pdf_doc.close() + + if has_text: + print(f"[INFO] PDF {original_local_file} contains embedded text. Skipping Textract.") + + # Extract text locally + extracted_text = "" + pdf_doc = fitz.open(new_local_path) + for page in pdf_doc: + extracted_text += page.get_text("text") + "\n" + pdf_doc.close() + + # Call metadata extraction directly + extract_metadata_with_gpt.delay(new_filename, extracted_text) + + return {"file": new_local_path, "status": "Text extracted locally"} + + try: + print(f"[INFO] Uploading {new_local_path} to s3://{bucket_name}/{new_filename}...") + s3_client.upload_file(new_local_path, bucket_name, new_filename) + print(f"[INFO] File uploaded successfully: {new_filename}") + + # Trigger Textract processing if no embedded text was found + process_with_textract.delay(new_filename) + + return {"file": new_local_path, "s3_key": new_filename, "status": "Uploaded to S3 for OCR"} + + except Exception as e: + print(f"[ERROR] Failed to upload {new_local_path} to S3: {e}") + return {"error": str(e)} diff --git a/docker-compose.yaml b/docker-compose.yaml index 583566e6..4e67945d 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -1,64 +1,64 @@ - -services: - api: - image: christianlouis/document-processor:latest - container_name: document_api - - # We'll keep the code in /app, but set working_dir to the shared data directory - working_dir: /workdir - - # We'll run uvicorn from the container's /app code - command: ["sh", "-c", "cd /app && uvicorn app.main:app --host 0.0.0.0 --port 8000"] - - # Environment variables - environment: - - PYTHONPATH=/app - env_file: - - .env - - # Expose container's 8000 -> Host's 8000 - ports: - - "8000:8000" - - depends_on: - - redis - - worker - - # Mount the shared working directory for data - volumes: - # optional: mount your code if you want local dev changes to reflect - # - ./app:/app - - /var/docparse/workdir:/workdir - - worker: - image: christianlouis/document-processor:latest - container_name: document_worker - - # same shared working directory - working_dir: /workdir - - command: ["celery", "-A", "app.celery_worker", "worker", "-B", "--loglevel=info", "-Q", "document_processor,default,celery"] - env_file: - - .env - environment: - - PYTHONPATH=/app - - depends_on: - - redis - - gotenberg - - # Mount the shared directory (and optionally your code if you want dev mode) - volumes: - # optional: mount your code if you want local dev changes - # - ./app:/app - - /var/docparse/workdir:/workdir - - gotenberg: - image: gotenberg/gotenberg:latest - container_name: gotenberg - - - redis: - image: redis:alpine - container_name: document_redis - restart: always + +services: + api: + image: christianlouis/document-processor:latest + container_name: document_api + + # We'll keep the code in /app, but set working_dir to the shared data directory + working_dir: /workdir + + # We'll run uvicorn from the container's /app code + command: ["sh", "-c", "cd /app && uvicorn app.main:app --host 0.0.0.0 --port 8000"] + + # Environment variables + environment: + - PYTHONPATH=/app + env_file: + - .env + + # Expose container's 8000 -> Host's 8000 + ports: + - "8000:8000" + + depends_on: + - redis + - worker + + # Mount the shared working directory for data + volumes: + # optional: mount your code if you want local dev changes to reflect + # - ./app:/app + - /var/docparse/workdir:/workdir + + worker: + image: christianlouis/document-processor:latest + container_name: document_worker + + # same shared working directory + working_dir: /workdir + + command: ["celery", "-A", "app.celery_worker", "worker", "-B", "--loglevel=info", "-Q", "document_processor,default,celery"] + env_file: + - .env + environment: + - PYTHONPATH=/app + + depends_on: + - redis + - gotenberg + + # Mount the shared directory (and optionally your code if you want dev mode) + volumes: + # optional: mount your code if you want local dev changes + # - ./app:/app + - /var/docparse/workdir:/workdir + + gotenberg: + image: gotenberg/gotenberg:latest + container_name: gotenberg + + + redis: + image: redis:alpine + container_name: document_redis + restart: always diff --git a/frontend/favicon.ico b/frontend/favicon.ico new file mode 100644 index 00000000..f64bb451 Binary files /dev/null and b/frontend/favicon.ico differ diff --git a/frontend/index.html b/frontend/index.html index 3140069e..b22c3d7d 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -1,89 +1,89 @@ - - - - - - Document Processor - Upload - - - - - -

Upload a File

- -
-

- Drag & drop a file here, or click to select a file. -

- -
- -
- - - - + + + + + + Document Processor - Upload + + + + + +

Upload a File

+ +
+

+ Drag & drop a file here, or click to select a file. +

+ +
+ +
+ + + + diff --git a/requirements.txt b/requirements.txt index ae4fac1f..e61f348f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,15 +1,15 @@ -fastapi[all] -uvicorn -celery -redis -sqlalchemy -pydantic -boto3 -pikepdf -openai -# Add this line explicitly -#pymupdf==1.23.5 -pymupdf -requests -dropbox +fastapi[all] +uvicorn +celery +redis +sqlalchemy +pydantic +boto3 +pikepdf +openai +# Add this line explicitly +#pymupdf==1.23.5 +pymupdf +requests +dropbox azure-ai-documentintelligence \ No newline at end of file